Make Pydantic BaseModel fields optional including sub-models for PATCH

Question

As already asked in similar questions, I want to support PATCH operations for a FastApi application where the caller can specify as many or as few fields as they like, of a Pydantic BaseModel with sub-models, so that efficient PATCH operations can be performed, without the caller having to supply an entire valid model just in order to update two or three of the fields.

I've discovered there are 2 steps in Pydantic PATCH from the tutorial that don't support sub-models. However, Pydantic is far too good for me to criticise it for something that it seems can be built using the tools that Pydantic provides. This question is to request implementation of those 2 things while also supporting sub-models:

1. generate a new DRY BaseModel with all fields optional
2. implement deep copy with update of BaseModel

These problems are already recognised by Pydantic.

• There is discussion of a class based solution to the optional model
• And there two issues open on the deep copy with update

A similar question has been asked one or two times here on SO and there are some great answers with different approaches to generating an all-fields optional version of the nested BaseModel. After considering them all this particular answer by Ziur Olpa seemed to me to be the best, providing a function that takes the existing model with optional and mandatory fields, and returning a new model with all fields optional: https://stackoverflow.com/a/72365032

The beauty of this approach is that you can hide the (actually quite compact) little function in a library and just use it as a dependency so that it appears in-line in the path operation function and there's no other code or boilerplate.

But the implementation provided in the previous answer did not take the step of dealing with sub-objects in the BaseModel being patched.

This question therefore requests an improved implementation of the all-fields-optional function that also deals with sub-objects, as well as a deep copy with update.

I have a simple example as a demonstration of this use-case, which although aiming to be simple for demonstration purposes, also includes a number of fields to more closely reflect the real world examples we see. Hopefully this example provides a test scenario for implementations, saving work:

import logging
from datetime import datetime, date

from collections import defaultdict
from pydantic import BaseModel
from fastapi import FastAPI, HTTPException, status, Depends
from fastapi.encoders import jsonable_encoder

app = FastAPI(title="PATCH demo")
logging.basicConfig(level=logging.DEBUG)


class Collection:
    collection = defaultdict(dict)

    def __init__(self, this, that):
        logging.debug("-".join((this, that)))
        self.this = this
        self.that = that

    def get_document(self):
        document = self.collection[self.this].get(self.that)
        if not document:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail="Not Found",
            )
        logging.debug(document)
        return document

    def save_document(self, document):
        logging.debug(document)
        self.collection[self.this][self.that] = document
        return document


class SubOne(BaseModel):
    original: date
    verified: str = ""
    source: str = ""
    incurred: str = ""
    reason: str = ""
    attachments: list[str] = []


class SubTwo(BaseModel):
    this: str
    that: str
    amount: float
    plan_code: str = ""
    plan_name: str = ""
    plan_type: str = ""
    meta_a: str = ""
    meta_b: str = ""
    meta_c: str = ""


class Document(BaseModel):
    this: str
    that: str
    created: datetime
    updated: datetime

    sub_one: SubOne
    sub_two: SubTwo

    the_code: str = ""
    the_status: str = ""
    the_type: str = ""
    phase: str = ""
    process: str = ""
    option: str = ""


@app.get("/endpoint/{this}/{that}", response_model=Document)
async def get_submission(this: str, that: str) -> Document:

    collection = Collection(this=this, that=that)
    return collection.get_document()


@app.put("/endpoint/{this}/{that}", response_model=Document)
async def put_submission(this: str, that: str, document: Document) -> Document:

    collection = Collection(this=this, that=that)
    return collection.save_document(jsonable_encoder(document))


@app.patch("/endpoint/{this}/{that}", response_model=Document)
async def patch_submission(
    document: Document,
    # document: optional(Document),  # <<< IMPLEMENT optional <<<
    this: str,
    that: str,
) -> Document:

    collection = Collection(this=this, that=that)
    existing = collection.get_document()
    existing = Document(**existing)
    update = document.dict(exclude_unset=True)
    updated = existing.copy(update=update, deep=True)  # <<< FIX THIS <<<
    updated = jsonable_encoder(updated)
    collection.save_document(updated)
    return updated

This example is a working FastAPI application, following the tutorial, and can be run with uvicorn example:app --reload. Except it doesn't work, because there's no all-optional fields model, and Pydantic's deep copy with update actually overwrites sub-models rather than updating them.

In order to test it the following Bash script can be used to run curl requests. Again I'm supplying this just to hopefully make it easier to get started with this question. Just comment out the other commands each time you run it so that the command you want is used. To demonstrate this initial state of the example app working you would run GET (expect 404), PUT (document stored), GET (expect 200 and same document returned), PATCH (expect 200), GET (expect 200 and updated document returned).

host='http://127.0.0.1:8000'
path="/endpoint/A123/B456"

method='PUT'
data='
{
"this":"A123",
"that":"B456",
"created":"2022-12-01T01:02:03.456",
"updated":"2023-01-01T01:02:03.456",
"sub_one":{"original":"2022-12-12","verified":"Y"},
"sub_two":{"this":"A123","that":"B456","amount":0.88,"plan_code":"HELLO"},
"the_code":"BYE"}
'

# method='PATCH'
# data='{"this":"A123","that":"B456","created":"2022-12-01T01:02:03.456","updated":"2023-01-02T03:04:05.678","sub_one":{"original":"2022-12-12","verified":"N"},"sub_two":{"this":"A123","that":"B456","amount":123.456}}' 

method='GET'
data=''

if [[ -n data ]]; then data=" --data '$data'"; fi
curl="curl -K curlrc -X $method '$host$path' $data"
echo $curl >&2
eval $curl

This curlrc will need to be co-located to ensure the content type headers are correct:

--cookie "_cookies"
--cookie-jar "_cookies"
--header "Content-Type: application/json"
--header "Accept: application/json"
--header "Accept-Encoding: compress, gzip"
--header "Cache-Control: no-cache"

So what I'm looking for is the implementation of optional that is commented out in the code, and a fix for existing.copy with the update parameter, that will enable this example to be used with PATCH calls that omit otherwise mandatory fields. The implementation does not have to conform precisely to the commented out line, I just provided that based on Ziur Olpa's previous answer.

Answer 1

When I first posed this question I thought that the only problem was how to turn all fields Optional in a nested BaseModel, but actually that was not difficult to fix.

The real problem with partial updates when implementing a PATCH call is that the Pydantic BaseModel.copy method doesn't attempt to support nested models when applying it's update parameter. That's quite an involved task for the generic case, considering you may have fields that are dicts, lists, or sets of another BaseModel, just for instance. Instead it just unpacks the dict using **: https://github.com/pydantic/pydantic/blob/main/pydantic/main.py#L353

I haven't got a proper implementation of that for Pydantic, but since I've got a working example PATCH by cheating, I'm going to post this as an answer and see if anyone can fault it or provide better, possibly even with an implementation of BaseModel.copy that supports updates for nested models.

Rather than post the implementations separately I am going to update the example given in the question so that it has a working PATCH and being a full demonstration of PATCH hopefully this will help others more.

The two additions are partial and merge. partial is what's referred to as optional in the question code.

partial: This is a function that takes any BaseModel and returns a new BaseModel with all fields Optional, including sub-object fields. That's enough for Pydantic to allow through any sub-set of fields without throwing an error for "missing fields". It's recursive - not really popular - but given these are nested data models the depth is not expected to exceed single digits.

merge: The BaseModel update on copy method operates on an instance of BaseModel - but supporting all the possible type variations when descending through a nested model is the hard part - and the database data, and the incoming update, are easily available as plain Python dicts; so this is the cheat: merge is an implementation of a nested dict update instead, and since the dict data has already been validated at one point or other, it should be fine.

Here's the full example solution:

import logging
from typing import Optional, Type
from datetime import datetime, date
from functools import lru_cache

from pydantic import BaseModel, create_model

from collections import defaultdict
from pydantic import BaseModel
from fastapi import FastAPI, HTTPException, status, Depends, Body
from fastapi.encoders import jsonable_encoder

app = FastAPI(title="Nested model PATCH demo")
logging.basicConfig(level=logging.DEBUG)


class Collection:
    collection = defaultdict(dict)

    def __init__(self, this, that):
        logging.debug("-".join((this, that)))
        self.this = this
        self.that = that

    def get_document(self):
        document = self.collection[self.this].get(self.that)
        if not document:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail="Not Found",
            )
        logging.debug(document)
        return document

    def save_document(self, document):
        logging.debug(document)
        self.collection[self.this][self.that] = document
        return document


class SubOne(BaseModel):
    original: date
    verified: str = ""
    source: str = ""
    incurred: str = ""
    reason: str = ""
    attachments: list[str] = []


class SubTwo(BaseModel):
    this: str
    that: str
    amount: float
    plan_code: str = ""
    plan_name: str = ""
    plan_type: str = ""
    meta_a: str = ""
    meta_b: str = ""
    meta_c: str = ""

class SubThree(BaseModel):
    one: str = ""
    two: str = ""


class Document(BaseModel):
    this: str
    that: str
    created: datetime
    updated: datetime

    sub_one: SubOne
    sub_two: SubTwo
    # sub_three: dict[str, SubThree] = {}  # Hah hah not really

    the_code: str = ""
    the_status: str = ""
    the_type: str = ""
    phase: str = ""
    process: str = ""
    option: str = ""


@lru_cache
def partial(baseclass: Type[BaseModel]) -> Type[BaseModel]:
    """Make all fields in supplied Pydantic BaseModel Optional, for use in PATCH calls.

    Iterate over fields of baseclass, descend into sub-classes, convert fields to Optional and return new model.
    Cache newly created model with lru_cache to ensure it's only created once.
    Use with Body to generate the partial model on the fly, in the PATCH path operation function.

    - https://stackoverflow.com/questions/75167317/make-pydantic-basemodel-fields-optional-including-sub-models-for-patch
    - https://stackoverflow.com/questions/67699451/make-every-fields-as-optional-with-pydantic
    - https://github.com/pydantic/pydantic/discussions/3089
    - https://fastapi.tiangolo.com/tutorial/body-updates/#partial-updates-with-patch
    """
    fields = {}
    for name, field in baseclass.__fields__.items():
        type_ = field.type_
        if type_.__base__ is BaseModel:
            fields[name] = (Optional[partial(type_)], {})
        else:
            fields[name] = (Optional[type_], None) if field.required else (type_, field.default)
    # https://docs.pydantic.dev/usage/models/#dynamic-model-creation
    validators = {"__validators__": baseclass.__validators__}
    return create_model(baseclass.__name__ + "Partial", **fields, __validators__=validators)


def merge(original, update):
    """Update original nested dict with values from update retaining original values that are missing in update.

    - https://github.com/pydantic/pydantic/issues/3785
    - https://github.com/pydantic/pydantic/issues/4177
    - https://docs.pydantic.dev/usage/exporting_models/#modelcopy
    - https://github.com/pydantic/pydantic/blob/main/pydantic/main.py#L353
    """
    for key in update:
        if key in original:
            if isinstance(original[key], dict) and isinstance(update[key], dict):
                merge(original[key], update[key])
            elif isinstance(original[key], list) and isinstance(update[key], list):
                original[key].extend(update[key])
            else:
                original[key] = update[key]
        else:
            original[key] = update[key]
    return original


@app.get("/endpoint/{this}/{that}", response_model=Document)
async def get_submission(this: str, that: str) -> Document:

    collection = Collection(this=this, that=that)
    return collection.get_document()


@app.put("/endpoint/{this}/{that}", response_model=Document)
async def put_submission(this: str, that: str, document: Document) -> Document:

    collection = Collection(this=this, that=that)
    return collection.save_document(jsonable_encoder(document))


@app.patch("/endpoint/{this}/{that}", response_model=Document)
async def patch_submission(
    this: str,
    that: str,
    document: partial(Document),  # <<< IMPLEMENTED partial TO MAKE ALL FIELDS Optional <<<
) -> Document:

    collection = Collection(this=this, that=that)
    existing_document = collection.get_document()
    incoming_document = document.dict(exclude_unset=True)
    # VVV IMPLEMENTED merge INSTEAD OF USING BROKEN PYDANTIC copy WITH update VVV
    updated_document = jsonable_encoder(merge(existing_document, incoming_document))
    collection.save_document(updated_document)
    return updated_document