使用 Pydantic 使每个字段都是可选的
Make every fields as optional with Pydantic
我正在用 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]
但这在代码重复方面会很糟糕。
还有更好的选择吗？
问题是一旦 FastAPI 在你的路由定义中看到 item: Item，它会尝试从请求体中初始化一个 Item 类型，你不能将你的模型的字段声明为可选的有时取决于一些条件，比如取决于使用的路由
我有 3 个解决方案：
解决方案 #1：分离模型
我想说的是，为 POST 和 PATCH 有效负载使用单独的模型似乎是更合乎逻辑且 可读性 的方法。是的，这可能会导致重复代码，但我认为 清楚地 定义哪个路由具有全必需或全可选模型可以平衡可维护性成本。
FastAPI 文档有一个使用 Optional 字段的 section for partially updating models with PUT or PATCH，最后有一个注释说了类似的话：
Notice that the input model is still validated.
So, if you want to receive partial updates that can omit all the attributes, you need to have a model with all the attributes marked as optional (with default values or 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：声明为全部必需，但手动验证 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 对象使用有效载荷中的任何内容进行初始化，如果没有，则使用一些默认值。您必须手动验证是否传递了 none 个预期字段，例如：
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: whether fields which are equal to their default values (whether set or otherwise) should be excluded from the returned dictionary; default False
exclude_none: whether fields which are equal to None should be excluded from the returned dictionary; default 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 将 return 一个不完整的字典，所以你可以报错.否则，您可以使用 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，那么这将是您的更新数据。如果传入了 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}
元类解决方案
我刚刚想出了以下内容：
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
修改@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)
感谢@Drdilyor 提供了一个很好的解决方案。我制作了一个版本，允许您在子 class 中定义所需的参数（例如，您要更新的项目的 ID 的 ID）：
class AllOptional(ModelMetaclass):
def __new__(self, name, bases, namespaces, **kwargs):
annotations = namespaces.get('__annotations__', {})
for base in bases:
optionals = {
key: Optional[value] if not key.startswith('__') else value for key, value in base.__annotations__.items()
}
annotations.update(optionals)
namespaces['__annotations__'] = annotations
return super().__new__(self, name, bases, namespaces, **kwargs)
对于我的案例来说，创建一个新的 class 是唯一有效的解决方案，但打包到一个函数中非常方便：
from pydantic import BaseModel, create_model
from typing import Optional
def make_optional(baseclass):
# 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_comprehension 中的某些字段。
此外，在 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()):
...
这大大减少了样板文件，使用相同的方法，您还可以制作一个使字段可选的函数，并排除一个字段，以防您的项目有 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()):
我正在用 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]
但这在代码重复方面会很糟糕。
还有更好的选择吗？
问题是一旦 FastAPI 在你的路由定义中看到 item: Item，它会尝试从请求体中初始化一个 Item 类型，你不能将你的模型的字段声明为可选的有时取决于一些条件，比如取决于使用的路由
我有 3 个解决方案：
解决方案 #1：分离模型
我想说的是，为 POST 和 PATCH 有效负载使用单独的模型似乎是更合乎逻辑且 可读性 的方法。是的，这可能会导致重复代码，但我认为 清楚地 定义哪个路由具有全必需或全可选模型可以平衡可维护性成本。
FastAPI 文档有一个使用 Optional 字段的 section for partially updating models with PUT or PATCH，最后有一个注释说了类似的话：
Notice that the input model is still validated.
So, if you want to receive partial updates that can omit all the attributes, you need to have a model with all the attributes marked as optional (with default values or
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：声明为全部必需，但手动验证 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 对象使用有效载荷中的任何内容进行初始化，如果没有，则使用一些默认值。您必须手动验证是否传递了 none 个预期字段，例如：
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: whether fields which are equal to their default values (whether set or otherwise) should be excluded from the returned dictionary; defaultFalse
exclude_none: whether fields which are equal toNoneshould be excluded from the returned dictionary; defaultFalse
这意味着，对于 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 将 return 一个不完整的字典，所以你可以报错.否则，您可以使用 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，那么这将是您的更新数据。如果传入了 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}
元类解决方案
我刚刚想出了以下内容：
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
修改@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)
感谢@Drdilyor 提供了一个很好的解决方案。我制作了一个版本，允许您在子 class 中定义所需的参数（例如，您要更新的项目的 ID 的 ID）：
class AllOptional(ModelMetaclass):
def __new__(self, name, bases, namespaces, **kwargs):
annotations = namespaces.get('__annotations__', {})
for base in bases:
optionals = {
key: Optional[value] if not key.startswith('__') else value for key, value in base.__annotations__.items()
}
annotations.update(optionals)
namespaces['__annotations__'] = annotations
return super().__new__(self, name, bases, namespaces, **kwargs)
对于我的案例来说，创建一个新的 class 是唯一有效的解决方案，但打包到一个函数中非常方便：
from pydantic import BaseModel, create_model
from typing import Optional
def make_optional(baseclass):
# 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_comprehension 中的某些字段。
此外，在 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()):
...
这大大减少了样板文件，使用相同的方法，您还可以制作一个使字段可选的函数，并排除一个字段，以防您的项目有 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()):