前向注解
正在阅读支持前向注解(用引号包裹)或使用 from __future__ import annotations future statement
(由 PEP563 引入):
from __future__ import annotations
from pydantic import BaseModel
MyInt = int
class Model(BaseModel):
a: MyInt
# 不使用 future import 时,等价于:
# a: 'MyInt'
print(Model(a='1'))
#> a=1
如下节所示,当你想引用代码中尚未定义的类型时,前向注解非常有用。
内部解析前向注解的逻辑在此章节中有详细描述。
自引用(或“递归”)模型¶
也支持包含自引用字段的模型。这些注解将在模型创建期间被解析。
在模型内部,你可以添加 from __future__ import annotations 导入或将注解包裹在字符串中:
from typing import Optional
from pydantic import BaseModel
class Foo(BaseModel):
a: int = 123
sibling: 'Optional[Foo]' = None
print(Foo())
#> a=123 sibling=None
print(Foo(sibling={'a': '321'}))
#> a=123 sibling=Foo(a=321, sibling=None)
循环引用¶
当处理自引用递归模型时,你可能会在验证输入中遇到循环引用。例如,在验证具有来自属性的反向引用的 ORM 实例时可能发生这种情况。
Pydantic 能够检测到循环引用并抛出适当的 ValidationError,而不是在尝试验证包含循环引用的数据时抛出 RecursionError:
from typing import Optional
from pydantic import BaseModel, ValidationError
class ModelA(BaseModel):
b: 'Optional[ModelB]' = None
class ModelB(BaseModel):
a: Optional[ModelA] = None
cyclic_data = {}
cyclic_data['a'] = {'b': cyclic_data}
print(cyclic_data)
#> {'a': {'b': {...}}}
try:
ModelB.model_validate(cyclic_data)
except ValidationError as exc:
print(exc)
"""
1 validation error for ModelB
a.b
Recursion error - cyclic reference detected [type=recursion_loop, input_value={'a': {'b': {...}}}, input_type=dict]
"""
from typing import Optional
from pydantic import BaseModel, ValidationError
class ModelA(BaseModel):
b: 'Optional[ModelB]' = None
class ModelB(BaseModel):
a: ModelA | None = None
cyclic_data = {}
cyclic_data['a'] = {'b': cyclic_data}
print(cyclic_data)
#> {'a': {'b': {...}}}
try:
ModelB.model_validate(cyclic_data)
except ValidationError as exc:
print(exc)
"""
1 validation error for ModelB
a.b
Recursion error - cyclic reference detected [type=recursion_loop, input_value={'a': {'b': {...}}}, input_type=dict]
"""
由于此错误是在未实际超过最大递归深度的情况下抛出的,你可以捕获并处理抛出的 ValidationError,而无需担心剩余的有限递归深度:
from __future__ import annotations
from collections.abc import Generator
from contextlib import contextmanager
from dataclasses import field
from pydantic import BaseModel, ValidationError, field_validator
def is_recursion_validation_error(exc: ValidationError) -> bool:
errors = exc.errors()
return len(errors) == 1 and errors[0]['type'] == 'recursion_loop'
@contextmanager
def suppress_recursion_validation_error() -> Generator[None]:
try:
yield
except ValidationError as exc:
if not is_recursion_validation_error(exc):
raise exc
class Node(BaseModel):
id: int
children: list[Node] = field(default_factory=list)
@field_validator('children', mode='wrap')
@classmethod
def drop_cyclic_references(cls, children, h):
try:
return h(children)
except ValidationError as exc:
if not (
is_recursion_validation_error(exc)
and isinstance(children, list)
):
raise exc
value_without_cyclic_refs = []
for child in children:
with suppress_recursion_validation_error():
value_without_cyclic_refs.extend(h([child]))
return h(value_without_cyclic_refs)
# 创建包含循环引用的数据,表示图 1 -> 2 -> 3 -> 1
node_data = {'id': 1, 'children': [{'id': 2, 'children': [{'id': 3}]}]}
node_data['children'][0]['children'][0]['children'] = [node_data]
print(Node.model_validate(node_data))
#> id=1 children=[Node(id=2, children=[Node(id=3, children=[])])]
类似地,如果 Pydantic 在序列化过程中遇到递归引用,会立即抛出 ValueError,而不是等待超过最大递归深度:
from pydantic import TypeAdapter
# 创建包含循环引用的数据,表示图 1 -> 2 -> 3 -> 1
node_data = {'id': 1, 'children': [{'id': 2, 'children': [{'id': 3}]}]}
node_data['children'][0]['children'][0]['children'] = [node_data]
try:
# 尝试将循环引用序列化为 JSON
TypeAdapter(dict).dump_json(node_data)
except ValueError as exc:
print(exc)
"""
Error serializing to JSON: ValueError: Circular reference detected (id repeated)
"""
如果需要,也可以处理这种情况:
from dataclasses import field
from typing import Any
from pydantic import (
SerializerFunctionWrapHandler,
TypeAdapter,
field_serializer,
)
from pydantic.dataclasses import dataclass
@dataclass
class NodeReference:
id: int
@dataclass
class Node(NodeReference):
children: list['Node'] = field(default_factory=list)
@field_serializer('children', mode='wrap')
def serialize(
self, children: list['Node'], handler: SerializerFunctionWrapHandler
) -> Any:
"""
Serialize a list of nodes, handling circular references by excluding the children.
"""
try:
return handler(children)
except ValueError as exc:
if not str(exc).startswith('Circular reference'):
raise exc
result = []
for node in children:
try:
serialized = handler([node])
except ValueError as exc:
if not str(exc).startswith('Circular reference'):
raise exc
result.append({'id': node.id})
else:
result.append(serialized)
return result
# 创建一个循环图:
nodes = [Node(id=1), Node(id=2), Node(id=3)]
nodes[0].children.append(nodes[1])
nodes[1].children.append(nodes[2])
nodes[2].children.append(nodes[0])
print(nodes[0])
#> Node(id=1, children=[Node(id=2, children=[Node(id=3, children=[...])])])
# 序列化循环图:
print(TypeAdapter(Node).dump_python(nodes[0]))
"""
{
'id': 1,
'children': [{'id': 2, 'children': [{'id': 3, 'children': [{'id': 1}]}]}],
}