Сделайте поля Pydantic BaseModel необязательными, включая подмодели для PATCH

Когда я впервые задал этот вопрос, я думал, что единственная проблема заключается в том, как превратить все поля Optional во вложенные BaseModel, но на самом деле это было несложно исправить.

Реальная проблема с частичными обновлениями при реализации вызова PATCH заключается в том, что метод Pydantic BaseModel.copy не пытается поддерживать вложенные модели при применении своего параметра update. Это довольно сложная задача для общего случая, учитывая, что у вас могут быть поля, которые являются dicts, lists или sets другого BaseModel, например. Вместо этого он просто распаковывает dict с помощью **: https://github.com/pydantic/pydantic/blob/main/pydantic/main.py#L353

У меня нет надлежащей реализации этого для Pydantic, но, поскольку у меня есть рабочий пример PATCH путем обмана, я собираюсь опубликовать это как ответ и посмотреть, может ли кто-нибудь обвинить его или предоставить лучше, возможно, даже с реализация BaseModel.copy, которая поддерживает обновления для вложенных моделей.

Вместо того, чтобы публиковать реализации отдельно, я собираюсь обновить пример, приведенный в вопросе, чтобы он работал PATCH и был полной демонстрацией PATCH, надеюсь, это поможет другим больше.

Два дополнения: partial и merge. partial — это то, что обозначается как optional в коде вопроса.

partial: Это функция, которая принимает любой BaseModel и возвращает новый BaseModel со всеми полями Optional, включая поля подобъектов. Этого достаточно, чтобы Pydantic разрешил любой подмножество полей, не выдавая ошибку для «отсутствующих полей». Это рекурсивно - не очень популярно - но, учитывая, что это вложенные модели данных, ожидается, что глубина не превысит однозначных цифр.

merge: Метод обновления при копировании BaseModel работает с экземпляром BaseModel, но поддержка всех возможных вариаций типов при спуске по вложенной модели является сложной задачей, а данные базы данных и входящие обновления легко доступны в виде простых Python dict; так что это чит: merge вместо этого является реализацией вложенного dict обновления, и, поскольку данные dict уже были проверены в тот или иной момент, все должно быть в порядке.

Вот полный пример решения:

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