使用 Pydantic 将每个字段设为可选

Question

我正在使用 FastAPI 和 Pydantic 制作 API。

我想要一些 PATCH 端点，可以一次编辑记录的 1 个或 N 个字段。此外，我希望客户端只传递有效负载中的必要字段。

例子：

class Item(BaseModel):
    name: str
    description: str
    price: float
    tax: float


@app.post("/items", response_model=Item)
async def post_item(item: Item):
    ...

@app.patch("/items/{item_id}", response_model=Item)
async def update_item(item_id: str, item: Item):
    ...

在这个例子中，对于 POST 请求，我希望每个字段都是必需的。但是，在 PATCH 端点中，我不介意有效负载是否仅包含例如描述字段。这就是为什么我希望所有字段都是可选的。

幼稚的方法：

class UpdateItem(BaseModel):
    name: Optional[str] = None
    description: Optional[str] = None
    price: Optional[float] = None
    tax: Optional[float]

但这在代码重复方面会很糟糕。

还有更好的选择吗？

Answer 1

此方法会阻止数据验证

阅读 @Anime Bk 的这篇文章： https ://stackoverflow.com/a/75011200

使用元类的解决方案

我刚刚想出了以下几点：

class AllOptional(pydantic.main.ModelMetaclass):
    def __new__(cls, name, bases, namespaces, **kwargs):
        annotations = namespaces.get('__annotations__', {})
        for base in bases:
            annotations.update(base.__annotations__)
        for field in annotations:
            if not field.startswith('__'):
                annotations[field] = Optional[annotations[field]]
        namespaces['__annotations__'] = annotations
        return super().__new__(cls, name, bases, namespaces, **kwargs)

将其用作：

class UpdatedItem(Item, metaclass=AllOptional):
    pass

所以基本上它将所有非可选字段替换为Optional

欢迎任何编辑！

以你的例子：

from typing import Optional

from fastapi import FastAPI
from pydantic import BaseModel
import pydantic

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str
    price: float
    tax: float


class AllOptional(pydantic.main.ModelMetaclass):
    def __new__(self, name, bases, namespaces, **kwargs):
        annotations = namespaces.get('__annotations__', {})
        for base in bases:
            annotations.update(base.__annotations__)
        for field in annotations:
            if not field.startswith('__'):
                annotations[field] = Optional[annotations[field]]
        namespaces['__annotations__'] = annotations
        return super().__new__(self, name, bases, namespaces, **kwargs)

class UpdatedItem(Item, metaclass=AllOptional):
    pass

# This continues to work correctly
@app.get("/items/{item_id}", response_model=Item)
async def get_item(item_id: int):
    return {
        'name': 'Uzbek Palov',
        'description': 'Palov is my traditional meal',
        'price': 15.0,
        'tax': 0.5,
    }

@app.patch("/items/{item_id}") # does using response_model=UpdatedItem makes mypy sad? idk, i did not check
async def update_item(item_id: str, item: UpdatedItem):
    return item

Answer 2

修改@Drdilyor的解决方案

添加了对模型嵌套的检查。

from pydantic.main import ModelMetaclass, BaseModel
from typing import Any, Dict, Optional, Tuple

class _AllOptionalMeta(ModelMetaclass):
    def __new__(self, name: str, bases: Tuple[type], namespaces: Dict[str, Any], **kwargs):
        annotations: dict = namespaces.get('__annotations__', {})

        for base in bases:
            for base_ in base.__mro__:
                if base_ is BaseModel:
                    break

                annotations.update(base_.__annotations__)

        for field in annotations:
            if not field.startswith('__'):
                annotations[field] = Optional[annotations[field]]

        namespaces['__annotations__'] = annotations

        return super().__new__(mcs, name, bases, namespaces, **kwargs)

Answer 3

问题是，一旦 FastAPIitem: Item在您的路由定义中看到，它将尝试从请求正文初始化类型，并且您有时Item无法根据某些条件（例如取决于所使用的路由）将模型的字段声明为可选。

我有3个解决方案：

解决方案#1：单独的模型

我想说的是，为 POST 和 PATCH 有效负载建立单独的模型似乎是更符合逻辑和可读的方法。是的，这可能会导致重复的代码，但我认为明确定义哪条路线具有全部必需或全部可选模型可以平衡可维护性成本。

FastAPI 文档中有一个部分用于使用使用字段的 PUT 或 PATCH 部分更新模型Optional，最后有一条注释说了类似的内容：

请注意，输入模型仍然经过验证。

因此，如果您希望接收可以省略所有属性的部分更新，则需要一个模型，其中所有属性都标记为可选（使用默认值 或None）。

所以...

class NewItem(BaseModel):
    name: str
    description: str
    price: float
    tax: float

class UpdateItem(BaseModel):
    name: Optional[str] = None
    description: Optional[str] = None
    price: Optional[float] = None
    tax: Optional[float] = None

@app.post('/items', response_model=NewItem)
async def post_item(item: NewItem):
    return item

@app.patch('/items/{item_id}',
           response_model=UpdateItem,
           response_model_exclude_none=True)
async def update_item(item_id: str, item: UpdateItem):
    return item

解决方案 #2：声明为 All-Required，但手动验证 PATCH

您可以将模型定义为具有所有必需的字段，然后将有效负载定义Body为 PATCH 路由上的常规参数，然后Item根据有效负载中的可用内容“手动”初始化实际对象。

from fastapi import Body
from typing import Dict

class Item(BaseModel):
    name: str
    description: str
    price: float
    tax: float

@app.post('/items', response_model=Item)
async def post_item(item: Item):
    return item

@app.patch('/items/{item_id}', response_model=Item)
async def update_item(item_id: str, payload: Dict = Body(...)):
    item = Item(
        name=payload.get('name', ''),
        description=payload.get('description', ''),
        price=payload.get('price', 0.0),
        tax=payload.get('tax', 0.0),
    )
    return item

在这里，Item对象使用有效负载中的任何内容进行初始化，如果没有，则使用一些默认值进行初始化。您必须手动验证是否未传递任何预期字段，例如：

from fastapi import HTTPException

@app.patch('/items/{item_id}', response_model=Item)
async def update_item(item_id: str, payload: Dict = Body(...)):
    # Get intersection of keys/fields
    # Must have at least 1 common
    if not (set(payload.keys()) & set(Item.__fields__)):
        raise HTTPException(status_code=400, detail='No common fields')
    ...

$ cat test2.json
{
    "asda": "1923"
}
$ curl -i -H'Content-Type: application/json' --data @test2.json --request PATCH localhost:8000/items/1
HTTP/1.1 400 Bad Request
content-type: application/json

{"detail":"No common fields"}

POST 路由的行为符合预期：必须传递所有字段。

解决方案#3：声明为全可选但手动验证 POST

Pydantic 的BaseModel方法dict具有exclude_defaults以下exclude_none选项：

• exclude_defaults：是否应从返回的字典中排除等于默认值（无论是否设置）的字段；默认False

• exclude_noneNone：是否应从返回的字典中排除等于的字段；默认False

这意味着，对于 POST 和 PATCH 路由，您可以使用相同的Item模型，但现在具有所有Optional[T] = None字段。item: Item也可以使用相同的参数。

class Item(BaseModel):
    name: Optional[str] = None
    description: Optional[str] = None
    price: Optional[float] = None
    tax: Optional[float] = None

在 POST 路由上，如果未设置所有字段，则exclude_defaults和exclude_none将返回不完整的字典，因此您可能会引发错误。否则，您可以将其用作item新的Item.

@app.post('/items', response_model=Item)
async def post_item(item: Item):
    new_item_values = item.dict(exclude_defaults=True, exclude_none=True)

    # Check if exactly same set of keys/fields
    if set(new_item_values.keys()) != set(Item.__fields__):
        raise HTTPException(status_code=400, detail='Missing some fields..')

    # Use `item` or `new_item_values`
    return item

$ cat test_empty.json
{
}
$ curl -i -H'Content-Type: application/json' --data @test_empty.json --request POST localhost:8000/items
HTTP/1.1 400 Bad Request
content-type: application/json

{"detail":"Missing some fields.."}

$ cat test_incomplete.json 
{
    "name": "test-name",
    "tax": 0.44
}
$ curl -i -H'Content-Type: application/json' --data @test_incomplete.json --request POST localhost:8000/items
HTTP/1.1 400 Bad Request
content-type: application/json

{"detail":"Missing some fields.."}

$ cat test_ok.json
{
    "name": "test-name",
    "description": "test-description",
    "price": 123.456,
    "tax": 0.44
}
$ curl -i -H'Content-Type: application/json' --data @test_ok.json --request POST localhost:8000/items
HTTP/1.1 200 OK
content-type: application/json

{"name":"test-name","description":"test-description","price":123.456,"tax":0.44}

在 PATCH 路由上，如果至少有 1 个值不是默认/无，那么这将是您的更新数据。如果未传入任何预期字段，则使用解决方案 2中的相同验证会失败。

@app.patch('/items/{item_id}', response_model=Item)
async def update_item(item_id: str, item: Item):
    update_item_values = item.dict(exclude_defaults=True, exclude_none=True)

    # Get intersection of keys/fields
    # Must have at least 1 common
    if not (set(update_item_values.keys()) & set(Item.__fields__)):
        raise HTTPException(status_code=400, detail='No common fields')

    update_item = Item(**update_item_values)

    return update_item

$ cat test2.json
{
    "asda": "1923"
}
$ curl -i -s -H'Content-Type: application/json' --data @test2.json --request PATCH localhost:8000/items/1
HTTP/1.1 400 Bad Request
content-type: application/json

{"detail":"No common fields"}

$ cat test2.json
{
    "description": "test-description"
}
$ curl -i -s -H'Content-Type: application/json' --data @test2.json --request PATCH localhost:8000/items/1
HTTP/1.1 200 OK
content-type: application/json

{"name":null,"description":"test-description","price":null,"tax":null}

Answer 4

好消息和坏消息：

坏：它是一个wontfix，即使在pydantic v2：https ://github.com/pydantic/pydantic/issues/3120

好：@adriangb - 的核心开发人员之一pydantic- 提出了一个解决方案，我将其翻译成一个简洁的装饰器。它适用于嵌套模型。

事情是这样的：

from typing import Optional, Type, Any, Tuple
from copy import deepcopy

from pydantic import BaseModel, create_model
from pydantic.fields import FieldInfo


def partial_model(model: Type[BaseModel]):
    def make_field_optional(field: FieldInfo, default: Any = None) -> Tuple[Any, FieldInfo]:
        new = deepcopy(field)
        new.default = default
        new.annotation = Optional[field.annotation]  # type: ignore
        return new.annotation, new
    return create_model(
        f'Partial{model.__name__}',
        __base__=model,
        __module__=model.__module__,
        **{
            field_name: make_field_optional(field_info)
            for field_name, field_info in model.__fields__.items()
        }
    )

原始代码在这里。

用法：

@partial_model
class Model(BaseModel):
    i: int
    f: float
    s: str


Model(i=1)

派丹提克 2

根据 @martintrapp 的输入，该解决方案也适用于pydantic 2. 您唯一需要更新的是

model.__fields__.items()需要更改为model.model_fields.items()使其工作

Answer 5

显然这不适用于 Pydantic 2

使用装饰器

对于大型项目来说，使用单独的模型似乎是一个坏主意。许多有效重复的代码使得维护变得更加困难。这样做的目标是可重用性和灵活性

from typing import Optional, get_type_hints, Type

from pydantic import BaseModel


def make_optional(
    include: Optional[list[str]] = None,
    exclude: Optional[list[str]] = None,
):
    """Return a decorator to make model fields optional"""

    if exclude is None:
        exclude = []

    # Create the decorator
    def decorator(cls: Type[BaseModel]):
        type_hints = get_type_hints(cls)
        fields = cls.__fields__
        if include is None:
            fields = fields.items()
        else:
            # Create iterator for specified fields
            fields = ((name, fields[name]) for name in include if name in fields)
            # Fields in 'include' that are not in the model are simply ignored, as in BaseModel.dict
        for name, field in fields:
            if name in exclude:
                continue
            if not field.required:
                continue
            # Update pydantic ModelField to not required
            field.required = False
            # Update/append annotation
            cls.__annotations__[name] = Optional[type_hints[name]]
        return cls

    return decorator

用法

在 fast-api 模型的背景下

class ModelBase(pydantic.BaseModel):
  a: int
  b: str


class ModelCreate(ModelBase):
  pass

# Make all fields optional
@make_optional()
class ModelUpdate(ModelBase):
  pass

• 默认情况下，所有字段都是可选的。
• include指定哪些字段可选；所有其他字段保持不变。
• exclude指定不影响哪些字段。
• exclude优先于include.

class ModelBase(pydantic.BaseModel):
  a: int
  b: str


class ModelCreate(ModelBase):
  pass

# Make all fields optional
@make_optional()
class ModelUpdate(ModelBase):
  pass

注意：当您从基类继承时，pydantic 似乎会复制字段，这就是为什么可以就地更改它们

Answer 6

对于我的情况，创建一个新类是唯一有效的解决方案，但打包到一个函数中它非常方便：

from pydantic import BaseModel, create_model
from typing import Optional
from functools import lru_cache

@lru_cache(maxsize=None) # avoids creating many classes with same name
def make_optional(baseclass: Type[BaseModel]) -> Type[BaseModel]:
    # Extracts the fields and validators from the baseclass and make fields optional
    fields = baseclass.__fields__
    validators = {'__validators__': baseclass.__validators__}
    optional_fields = {key: (Optional[item.type_], None)
                       for key, item in fields.items()}
    return create_model(f'{baseclass.__name__}Optional', **optional_fields,
                        __validators__=validators)

class Item(BaseModel):
    name: str
    description: str
    price: float
    tax: float

ItemOptional = make_optional(Item)

前后对比：

> Item.__fields__

{'name': ModelField(name='name', type=str, required=True),
 'description': ModelField(name='description', type=str, required=True),
 'price': ModelField(name='price', type=float, required=True),
 'tax': ModelField(name='tax', type=float, required=True)}

> ItemOptional.__fields__

{'name': ModelField(name='name', type=Optional[str], required=False, default=None),
 'description': ModelField(name='description', type=Optional[str], required=False, default=None),
 'price': ModelField(name='price', type=Optional[float], required=False, default=None),
 'tax': ModelField(name='tax', type=Optional[float], required=False, default=None)}

它确实有效，而且如果需要的话，它还允许您过滤掉 dict_compressive 中的某些字段。

此外，在 fastapi 中，这种方法允许您执行以下操作：

@app.post("/items", response_model=Item)
async def post_item(item: Item = Depends()):
    ...

@app.patch("/items/{item_id}", response_model=Item)
async def update_item(item_id: str, item: make_optional(Item) = Depends()):
    ...

这减少了很多样板文件，使用相同的方法，您还可以创建一个函数，使字段可选，并且还可以排除字段，以防您的 Item 有 ID 字段，该 id 将在您的 PATCH 调用中重复。可以这样解决：

def make_optional_no_id(baseclass):
    ... # same as make optional
    optional_fields = {key: (Optional[item.type_], None) 
                       for key, item in fields.items() if key != 'ID'} # take out here ID
    ... # you can also take out also validators of ID

@app.patch("/items/{item_id}", response_model=Item)
async def update_item(item: make_optional_no_id(Item) = Depends()):

Answer 7

\xe2\x9b\x94\xef\xb8\x8f 注意，@Drdilyor\ 的解决方案会阻止字段验证。

\n

看来@Drdilyor的解决方案取消了所有字段验证

\n

假设您有：

\n

from typing import Optional\nimport pydantic\nfrom pydantic import BaseModel, Field\n\nclass AllOptional(pydantic.main.ModelMetaclass):\n    def __new__(self, name, bases, namespaces, **kwargs):\n        annotations = namespaces.get(\'__annotations__\', {})\n        for base in bases:\n            annotations.update(base.__annotations__)\n        for field in annotations:\n            if not field.startswith(\'__\'):\n                annotations[field] = Optional[annotations[field]]\n        namespaces[\'__annotations__\'] = annotations\n        return super().__new__(self, name, bases, namespaces, **kwargs)\n\nclass A(BaseModel):\n    a:int = Field(gt=1)\n\nclass AO(A, metaclass=AllOptional):\n    pass\n\nAO(a=-1) # This will pass through the validation even that it\'s wrong \xe2\x9b\x94\xef\xb8\x8f\n

\n

一个简单的替代方案

\n

class AllOptional(pydantic.main.ModelMetaclass):\n    def __new__(mcls, name, bases, namespaces, **kwargs):\n        cls = super().__new__(mcls, name, bases, namespaces, **kwargs)\n        for field in cls.__fields__.values():\n            field.required=False\n        return cls\n

\n