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

Posted 2023-02-15

【中文标题】使用 Pydantic 将每个字段设为可选

【英文标题】：Make every fields as optional with Pydantic

【发布时间】：2021-08-14 09:00:08

【问题描述】：

我正在使用 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]

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

还有更好的选择吗？

【参考方案1】：

问题是，一旦 FastAPI 在您的路由定义中看到 item: Item，它将尝试从请求正文中初始化 Item 类型，并且您不能将模型的字段声明为可选有时 em> 取决于某些条件，例如取决于使用哪个路由。

我有 3 个解决方案：

解决方案 #1：分离模型

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

FastAPI 文档有一个使用Optional 字段的section for partially updating models with PUT or PATCH，末尾有一条说明类似的内容：

请注意，输入模型仍在验证中。

因此，如果您想接收可以省略所有属性的部分更新，您需要有一个模型，其中所有属性都标记为可选（使用默认值或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

您可以将模型定义为具有所有必需的字段，然后将有效负载定义为 PATCH 路由上的常规 Body 参数，然后根据有效负载中可用的内容“手动”初始化实际的 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 and exclude_none options 用于：

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

exclude_none：是否应该从返回的字典中排除等于None的字段；默认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 将返回不完整的 dict，因此您可以引发错误。否则，您可以使用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 个值不是 default/None，那么这将是您的更新数据。如果没有传入任何预期字段，则使用 解决方案 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

【讨论】：

谢谢！很好的解释。所以，看起来解决方案 2 比 3 好，因为 PATCH 的手动验证必须在两者中完成，而 POST 验证仅在 3 中完成。但我同意解决方案 1 更容易阅读，当你不是一个人在一个项目中.. .

【参考方案2】：

元类解决方案

我刚刚想出了以下内容：

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

所以基本上它将所有非可选字段替换为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") # not using response_model=Item
async def update_item(item_id: str, item: UpdatedItem):
    return item

【讨论】：

把它作为公认的答案，因为它是真正解决问题的唯一解决方案。老实说，这可能是对 Pydantic 的改进！

【参考方案3】：

修改了@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)

【讨论】：

有没有办法让它通用，以便它适用于任何 pydantic 模型，而不是从 PydanticModel 继承？

没关系