使 Pydantic BaseModel 字段可选,包括 PATCH 的子模型

2024-02-04

正如已经问过的similar有疑问,我要支持PATCHFastApi 应用程序的操作,调用者可以根据需要指定 Pydantic 的任意数量的字段BaseModel 有子模型,这样高效PATCH可以执行操作,而调用者不必仅仅为了更新两个或三个字段而提供整个有效模型。

我发现有2 steps在派丹提克PATCH来自tutorial https://fastapi.tiangolo.com/tutorial/body-updates/#partial-updates-with-patch that 不支持子型号。然而,Pydantic 对我来说太好了,无法批评它似乎可以使用 Pydantic 提供的工具构建的东西。这个问题是要求实施这两件事同时还支持子型号:

  1. 生成一个新的DRYBaseModel所有字段都是可选的
  2. 通过更新实现深复制BaseModel

Pydantic 已经认识到这些问题。

  • 讨论 https://github.com/pydantic/pydantic/discussions/3089可选模型的基于类的解决方案
  • 还有那里two https://github.com/pydantic/pydantic/issues/4177 issues https://github.com/pydantic/pydantic/issues/3785在深拷贝上打开并更新

A similar question https://stackoverflow.com/q/67699451已经在这里被问过一两次了,并且有一些很好的答案,采用不同的方法来生成嵌套的全字段可选版本BaseModel。考虑完所有这些之后这个特定的答案 https://stackoverflow.com/a/72365032 by 齐尔·奥尔帕 https://stackoverflow.com/users/10416012/ziur-olpa在我看来,这是最好的,提供了一个函数,该函数采用带有可选和强制字段的现有模型,并返回一个新模型所有字段可选: https://stackoverflow.com/a/72365032 https://stackoverflow.com/a/72365032

这种方法的优点在于,您可以隐藏库中的(实际上非​​常紧凑)小函数,并将其用作依赖项,以便它内联显示在路径操作函数中,并且没有其他代码或样板。

但是前面的答案中提供的实现并没有采取处理子对象的步骤BaseModel正在修补。

因此,这个问题要求改进所有字段可选函数的实现,该函数还处理子对象,以及带有更新的深层复制。

我有一个简单的示例来演示此用例,虽然其目的是为了演示目的而简单,但也包含许多字段以更接近地反映我们看到的现实世界示例。希望这个示例为实现提供一个测试场景,从而节省工作量:

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

此示例是一个正在运行的 FastAPI 应用程序,遵循教程,可以使用以下命令运行uvicorn example:app --reload。但它不起作用,因为没有全可选字段模型,而 Pydantic 的深度复制实际上带有更新覆盖子模型而不是updating them.

为了测试它,可以使用以下 Bash 脚本来运行curl要求。我再次提供这个只是为了希望能够更容易地开始解决这个问题。 只需在每次运行时注释掉其他命令,以便使用您想要的命令。 为了演示示例应用程序的初始状态,您将运行GET(预计 404),PUT(存储的文档),GET(预计返回 200 个和相同的文档),PATCH(预计 200),GET(预计返回 200 和更新的文档)。

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需要位于同一位置以确保内容类型标头正确:

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

所以我正在寻找的是实施optional代码中已注释掉,并修复了existing.copyupdate参数,这将使该示例能够与PATCH省略其他强制字段的调用。 实现不必完全符合注释掉的行,我只是根据齐尔·奥尔帕的 https://stackoverflow.com/users/10416012/ziur-olpa以前的answer https://stackoverflow.com/a/72365032.


当我第一次提出这个问题时,我认为唯一的问题是如何转动所有字段Optional在嵌套的BaseModel,但实际上这并不难解决。

实施时部分更新的真正问题PATCH呼叫是 PydanticBaseModel.copy方法在应用它时不会尝试支持嵌套模型update范围。对于一般情况来说,这是一项相当复杂的任务,考虑到您可能有以下字段dicts, lists, or set另一个的BaseModel,仅举个例子。相反,它只是解压dict using **: https://github.com/pydantic/pydantic/blob/main/pydantic/main.py#L353 https://github.com/pydantic/pydantic/blob/main/pydantic/main.py#L353

我还没有为 Pydantic 正确实现它,但因为我有一个工作示例PATCH通过作弊,我将把这个作为答案发布,看看是否有人可以提出错误或提供更好的,甚至可能实施BaseModel.copy支持嵌套模型的更新。

我不会单独发布实现,而是更新问题中给出的示例,以便它具有有效的功能PATCH并充分展示了PATCH希望这能对其他人有更多帮助。

两个补充是partial and merge. partial就是所谓的optional在问题代码中。

partial: 这是一个函数,可以接受任何BaseModel并返回一个新的BaseModel与所有字段Optional,包括子对象字段。这足以让 Pydantic 允许通过任何字段子集,而不会抛出“缺失字段”的错误。它是递归的 - 并不是很流行 - 但鉴于这些是嵌套数据模型,深度预计不会超过个位数。

merge: The BaseModelupdate on copy 方法对以下实例进行操作BaseModel- 但是在嵌套模型中下降时支持所有可能的类型变化是困难的部分 - 并且数据库数据和传入的更新可以轻松地以纯 Python 形式获得dicts;所以这就是作弊: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
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)

使 Pydantic BaseModel 字段可选,包括 PATCH 的子模型 的相关文章

随机推荐