跳转到内容

架构

注意

本节是 内部原理 文档的一部分,主要面向贡献者。

从 Pydantic V2 开始,部分代码库是用 Rust 编写的,并独立封装在名为 pydantic-core 的包中。这样做部分是为了提高验证和序列化性能(代价是内部逻辑的定制和扩展性受到限制)。

本架构文档将首先介绍 pydanticpydantic-core 这两个包如何协同工作,然后将详细阐述各种模式(模型定义、验证、序列化、JSON Schema)的架构细节。

Pydantic 库的使用可以分为两部分

  • 模型定义,在 pydantic 包中完成。
  • 模型验证和序列化,在 pydantic-core 包中完成。

模型定义

每当定义 Pydantic BaseModel 时,元类将分析模型的主体以收集一些元素

  • 用于构建模型字段的已定义注解(收集在 model_fields 属性中)。
  • 模型配置,通过 model_config 设置。
  • 额外的验证器/序列化器。
  • 私有属性、类变量、泛型参数化的识别等。

pydanticpydantic-core 之间的通信:核心 schema

然后我们需要一种方法将模型定义中收集到的信息传递给 pydantic-core,以便相应地执行验证和序列化。为此,Pydantic 使用了核心 schema 的概念:一个结构化的(且可序列化的)Python 字典(使用 TypedDict 定义表示),描述了特定的验证和序列化逻辑。它是用于在 pydanticpydantic-core 包之间通信的核心数据结构。每个核心 schema 都有一个必需的 type 键,以及取决于此 type 的额外属性。

核心 schema 的生成由 GenerateSchema 类在一个地方处理(无论是 Pydantic 模型还是其他任何东西)。

注意

无法定义自定义核心 schema。核心 schema 需要被 pydantic-core 包理解,因此我们只支持固定数量的核心 schema 类型。这也是 GenerateSchema 没有真正公开和适当文档化的部分原因。

核心 schema 定义可以在 pydantic_core.core_schema 模块中找到。

对于 Pydantic 模型,将构建一个核心 schema 并将其设置为 __pydantic_core_schema__ 属性。

为了说明核心 schema 的样子,我们将以 bool 核心 schema 为例

class BoolSchema(TypedDict, total=False):
    type: Required[Literal['bool']]
    strict: bool
    ref: str
    metadata: Any
    serialization: SerSchema

当定义一个带有布尔字段的 Pydantic 模型时

from pydantic import BaseModel, Field


class Model(BaseModel):
    foo: bool = Field(strict=True)

foo 字段的核心 schema 将如下所示

{
    'type': 'bool',
    'strict': True,
}

BoolSchema 定义所示,序列化逻辑也定义在核心 schema 中。如果我们要为 foo 定义一个自定义序列化函数 (1),serialization 键将如下所示

  1. 例如使用 field_serializer 装饰器

    class Model(BaseModel):
        foo: bool = Field(strict=True)
    
        @field_serializer('foo', mode='plain')
        def serialize_foo(self, value: bool) -> Any:
            ...
    
{
    'type': 'function-plain',
    'function': <function Model.serialize_foo at 0x111>,
    'is_field_serializer': True,
    'info_arg': False,
    'return_schema': {'type': 'int'},
}

请注意,这也是一个核心 schema 定义,只是它仅在序列化期间与 pydantic-core 相关。

核心 schema 覆盖了广泛的范围,只要我们需要在 Python 和 Rust 端之间通信时就会使用它们。虽然前面的示例与验证和序列化相关,但理论上它可以用于任何事情:错误管理、额外元数据等。

JSON Schema 生成

您可能已经注意到,前面的序列化核心 schema 有一个 return_schema 键。这是因为核心 schema 也用于生成相应的 JSON Schema。

与核心 schema 的生成方式类似,JSON Schema 的生成由 GenerateJsonSchema 类处理。 generate 方法是主要的入口点,并被赋予该模型的核心 schema。

回到我们的 bool 字段示例, bool_schema 方法将被赋予之前生成的 布尔核心 schema,并将返回以下 JSON Schema

{
    {"type": "boolean"}
}

自定义核心 schema 和 JSON schema

虽然 GenerateSchemaGenerateJsonSchema 类处理相应 schema 的创建,但 Pydantic 提供了一种在某些情况下自定义它们的方式,遵循包装器模式。这种自定义通过 __get_pydantic_core_schema____get_pydantic_json_schema__ 方法完成。

为了理解这种包装器模式,我们将以与 Annotated 一起使用的元数据类为例,其中可以使用 __get_pydantic_core_schema__ 方法

from typing import Annotated, Any

from pydantic_core import CoreSchema

from pydantic import GetCoreSchemaHandler, TypeAdapter


class MyStrict:
    @classmethod
    def __get_pydantic_core_schema__(
        cls, source: Any, handler: GetCoreSchemaHandler
    ) -> CoreSchema:
        schema = handler(source)  # (1)!
        schema['strict'] = True
        return schema


class MyGt:
    @classmethod
    def __get_pydantic_core_schema__(
        cls, source: Any, handler: GetCoreSchemaHandler
    ) -> CoreSchema:
        schema = handler(source)  # (2)!
        schema['gt'] = 1
        return schema


ta = TypeAdapter(Annotated[int, MyStrict(), MyGt()])
  1. MyStrict 是第一个应用的注解。此时,schema = {'type': 'int'}
  2. MyGt 是最后一个应用的注解。此时,schema = {'type': 'int', 'strict': True}

GenerateSchema 类为 Annotated[int, MyStrict(), MyGt()] 构建核心 schema 时,它将创建一个 GetCoreSchemaHandler 实例,并将其传递给 MyGt.__get_pydantic_core_schema__ 方法。(1)

  1. 在我们的 Annotated 模式下,GetCoreSchemaHandler 以嵌套方式定义。调用它将递归调用其他 __get_pydantic_core_schema__ 方法,直到它到达 int 注解,此时返回一个简单的 {'type': 'int'} schema。

source 参数取决于核心 schema 生成模式。在 Annotated 的情况下,source 将是正在注解的类型。当定义自定义类型时,source 将是定义 __get_pydantic_core_schema__ 的实际类。

模型验证和序列化

虽然模型定义被限制在 *类* 级别(即定义模型时),但模型验证和序列化发生在 *实例* 级别。这两个概念都在 pydantic-core 中处理(与 Pydantic V1 相比,性能提高了 5 到 20 倍),通过使用之前构建的核心 schema。

pydantic-core 暴露了 SchemaValidatorSchemaSerializer 类来执行这些任务

from pydantic import BaseModel


class Model(BaseModel):
    foo: int


model = Model.model_validate({'foo': 1})  # (1)!
dumped = model.model_dump()  # (2)!
  1. 通过使用 SchemaValidator.validate_python 方法将提供的数据发送到 pydantic-corepydantic-core 将根据模型的核心 schema 验证数据并填充模型的 __dict__ 属性。
  2. 通过使用 SchemaSerializer.to_python 方法将 model 实例发送到 pydantic-corepydantic-core 将读取实例的 __dict__ 属性并构建适当的结果(同样,遵循模型的核心 schema)。