腾讯云

文章/答案/技术大牛

发布

社区首页 >问答首页 >用Pydantic使每个字段都是可选的

问用Pydantic使每个字段都是可选的
EN

Stack Overflow用户

提问于 2021-05-25 22:16:58

回答 6查看 28.9K关注 0票数 30

我正在用FastAPI和Pydantic制作API。

我希望有一些补丁端点，其中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请求，我希望每个字段都是必需的。但是，在修补程序端点中，我不介意负载是否仅包含描述字段。这就是为什么我希望所有字段都是可选的。

天真的方法：

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

但是，从代码重复的角度来说，这将是很糟糕的。

有更好的选择吗？

python

python-3.x

fastapi

pydantic

回答 6

Stack Overflow用户

回答已采纳

发布于 2021-05-27 22:30:16

元类溶液

我刚刚想出了以下几点：

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}") # does using response_model=UpdatedItem makes mypy sad? idk, i did not check
async def update_item(item_id: str, item: UpdatedItem):
    return item

票数 37

Stack Overflow用户

发布于 2021-05-26 07:32:12

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

我有三个解决方案：

解决方案1:单独的模型

我要说的是，为POST和修补程序有效载荷建立单独的模型似乎是更符合逻辑和更易理解的方法。这可能会导致代码重复，是的，但我认为清楚地定义哪一条路由具有全必需的或全可选的模型可以平衡可维护性成本。

FastAPI文档有一个使用Optional字段的使用PUT或修补程序部分更新模型的部分，最后有一个注释，上面写着类似的内容：

注意，输入模型仍然是有效的。因此，如果希望接收可以省略所有属性的部分更新，则需要有一个模型，该模型将所有属性标记为可选(带有默认值或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:声明为所有必需的，但手动验证修补程序

您可以定义模型具有所有必需的字段，然后将有效负载定义为补丁路由上的常规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:声明为All-可选但手动验证POST

Pydantic的BaseModel的dict方法有选项，用于：

exclude_defaults：是否应从返回的字典中排除等于其默认值(无论设置或其他)的字段；默认False
exclude_none：是否应从返回的字典中排除等于None的字段；默认False

这意味着，对于POST和修补程序路由，您可以使用相同的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}

在修补程序路由上，如果至少有一个值不是默认/无，那么这将是您的更新数据。如果没有传入任何预期字段，则使用来自解决方案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}

票数 6

Stack Overflow用户

发布于 2021-09-12 06:30:10

改进型@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)