数据类与 Pydantic

20 min中级

dataclass与Pydantic模型

前置知识

学习目标

  • 掌握「1. 历史动机与发展脉络」的核心机制、典型用法与常见陷阱
  • 掌握「2. 形式化定义」的核心机制、典型用法与常见陷阱
  • 掌握「3. 理论推导与原理解析」的核心机制、典型用法与常见陷阱
  • 掌握「4. 代码示例(企业级 production-ready)」的核心机制、典型用法与常见陷阱
  • 掌握「5. 对比分析」的核心机制、典型用法与常见陷阱

1. 历史动机与发展脉络

1.1 前史:手工 __init__ 与命名元组(1991–2017)

Python 早期版本中,定义数据载体类需要手写冗长的 __init__、__repr__、__eq__:

# Python 1.x - 2.x 风格(手工类)
class User:
    def __init__(self, name, age, email=None):
        self.name = name
        self.age = age
        self.email = email

    def __repr__(self):
        return f"User(name={self.name!r}, age={self.age!r}, email={self.email!r})"

    def __eq__(self, other):
        if not isinstance(other, User):
            return NotImplemented
        return (self.name, self.age, self.email) == (other.name, other.age, other.email)

Python 2.6 引入 collections.namedtuple,提供轻量级不可变数据容器:

from collections import namedtuple
User = namedtuple("User", ["name", "age", "email"])

但 namedtuple 有三大缺陷:(1) 不可变,无法修改字段;(2) 通过元组索引访问(u[0])降低可读性;(3) 默认值需通过 defaults 参数逆序指定,易出错。

1.2 attrs:第三方先驱(2015)

Hynek Schlawack 于 2015 年发布 attrs 库,引入 @attr.s(后更名为 @define)装饰器,自动生成 __init__/__repr__/__eq__/__hash__,并支持类型注解、默认值工厂、校验器、slots。attrs 的影响力深远——PEP 557 的 dataclasses 在设计时明确借鉴了 attrs,Guido 在 PEP 557 中致谢:

“This PEP is essentially a simplification of attrs … without some of the more advanced features.”

1.3 PEP 557 dataclasses(Python 3.7,2018)

PEP 557 由 Eric V. Smith 撰写,于 2018 年 6 月随 Python 3.7 正式发布。dataclasses 模块的核心设计目标:

  1. 标准库内置,无需第三方依赖。
  2. 类型注解驱动,字段通过 name: type 声明。
  3. 可配置生成方法:通过 init、repr、eq、order、unsafe_hash、frozen 参数控制。
  4. 不引入运行时类型检查,类型注解仅为 hint,dataclasses 不校验。
# Python 3.7+
from dataclasses import dataclass, field

@dataclass
class User:
    name: str
    age: int
    tags: list[str] = field(default_factory=list)

1.4 Pydantic v1:运行时验证的崛起(2018)

Samuel Colvin 于 2018 年发布 Pydantic v1,定位为”基于 Python 类型注解的数据验证库”。核心创新:

  1. 类型即校验:name: str 自动校验赋值是否为 str,否则抛出 ValidationError。
  2. JSON 原生支持:parse_raw()、.json() 直接处理 JSON 字符串。
  3. 嵌套模型:模型字段可以是另一个模型,递归校验。
  4. FastAPI 基石:Sebastián Ramírez 在 2018 年发布的 FastAPI 完全基于 Pydantic 构建 API schema,使 Pydantic 成为 Python Web 生态的事实标准。

Pydantic v1 的性能瓶颈:基于 Python 实现,单次验证约 1-5μs,复杂嵌套模型可达数十微秒,在亿级请求场景下成为热点。

1.5 Pydantic v2:Rust 内核重生(2023)

2023 年 6 月,Pydantic v2 正式发布,进行了几乎完全的重写:

  1. pydantic-core:用 Rust 实现的验证内核,性能提升 5-50 倍。
  2. API 重命名:dict() → model_dump()、parse_obj() → model_validate()、@validator → field_validator。
  3. ConfigDict 替代内部 Config 类。
  4. Annotated 优先:推荐使用 Annotated[int, Field(ge=0)] 而非 int = Field(ge=0)。
  5. JSON Schema 2020-12:生成符合最新 JSON Schema 规范的 schema。

Samuel Colvin 在 Pydantic v2 发布博客中写道:

“Pydantic v2 is a complete rewrite … performance improvements of 5x-50x … as well as slightly different logic in a few places.”

—— Samuel Colvin, Pydantic Blog 2023

1.6 现代竞争:msgspec 与 attrs 的回应

2022 年 Jim Crist-Haruf 发布 msgspec,定位为”高性能序列化与验证库”,性能比 Pydantic v2 再快 2-10x,支持 Struct 类(类似 dataclass 但更紧凑)与 MessagePack 原生支持。

attrs 在 2023 年发布 23.1 版本,引入 NG API(@define、@frozen),保留其在细粒度配置上的优势。

1.7 设计哲学演进

Python 数据建模库的演进反映了三个哲学转向:

  1. 从手工到自动(1991-2017):消除样板代码,__init__/__repr__ 自动生成。
  2. 从静态到运行时(2018-2022):类型注解从 hint 升级为可执行校验。
  3. 从 Python 到 Rust(2023+):性能关键路径用 Rust 重写,Pydantic v2、Ruff、Polars 共同推动”Python 生态 Rust 化”浪潮。

Guido van Rossum 在 2023 年 PyCon 关于类型系统的演讲中提到:

“Type hints were never meant to be just documentation. The ecosystem has caught up, and tools like Pydantic show what’s possible when types become executable.”


2. 形式化定义

2.1 数据类(Data Class)的形式化

一个数据类 DD 可形式化为五元组:

D=⟨F,M,V,S,C⟩D = \langle \mathcal{F}, \mathcal{M}, \mathcal{V}, \mathcal{S}, \mathcal{C} \rangle

其中:

  • F\mathcal{F}:字段集合(fields),每个字段 f∈Ff \in \mathcal{F} 是三元组 ⟨name,type,default⟩\langle \text{name}, \text{type}, \text{default} \rangle。
  • M\mathcal{M}:生成方法集合(methods),如 __init__、__repr__、__eq__、__hash__。
  • V\mathcal{V}:验证器集合(validators),可为空(dataclass 默认无验证)。
  • S\mathcal{S}:序列化器集合(serializers),可为空。
  • C\mathcal{C}:配置(config),如 frozen、slots、kw_only。

2.2 类型验证的形式化

类型验证函数 validate\text{validate} 接受值 vv 与类型 τ\tau,返回验证结果:

validate(v,τ)→{Ok(v′)若 v 可转换为 τ 类型的值 v′Err(e)否则,返回错误 e\text{validate}(v, \tau) \to \begin{cases} \text{Ok}(v') & \text{若 } v \text{ 可转换为 } \tau \text{ 类型的值 } v' \\ \text{Err}(e) & \text{否则,返回错误 } e \end{cases}

关键性质:Pydantic 的 validate\text{validate} 是强制转换(coercion)而非严格相等(strict equality)。例如:

validate("42",int)=Ok(42)(coercion 模式)\text{validate}("42", \text{int}) = \text{Ok}(42) \quad \text{(coercion 模式)}

而严格模式(strict=True)下:

validatestrict("42",int)=Err("int expected, got str")\text{validate}_{\text{strict}}("42", \text{int}) = \text{Err}(\text{"int expected, got str"})

2.3 nominal typing 与 structural typing

Nominal typing(名义类型):类型等价基于类型名称。AA 是 BB 的子类型当且仅当 AA 显式声明继承 BB。

Structural typing(结构类型):类型等价基于结构。AA 是 BB 的子类型当且仅当 AA 拥有 BB 所有字段与方法(duck typing 的形式化)。

Python 类型系统以 nominal 为主,但 Protocol(PEP 544)引入了 structural typing:

from typing import Protocol

class HasName(Protocol):
    name: str

def greet(obj: HasName) -> str:
    return f"Hello, {obj.name}"

# 任何含 name: str 属性的对象都满足 HasName
class User:
    def __init__(self, name: str) -> None:
        self.name = name

greet(User("Alice"))  # OK

Pydantic 模型采用 nominal typing:User 与 Admin 即使字段相同,也是不同类型。

2.4 不可变性的形式化

frozen=True 等价于在 __setattr__ 与 __delattr__ 中抛出 FrozenInstanceError:

∀f∈F:setattr(D,f,v)≜raise FrozenInstanceError\forall f \in \mathcal{F}: \text{setattr}(D, f, v) \triangleq \text{raise FrozenInstanceError}

不可变性带来的关键性质:

  1. 线程安全:无需锁即可在多线程间共享。
  2. 可哈希:可作为字典键、集合元素(__hash__ 基于 F\mathcal{F} 计算)。
  3. 引用透明(referential transparency):相同输入始终产生相同输出,便于推理。

2.5 slots 的内存模型

slots=True 使 dataclass 使用 __slots__ 替代 __dict__。形式化地,普通类的实例内存布局:

Objdict=⟨header,dict_ptr→dict(fields)⟩\text{Obj}_{\text{dict}} = \langle \text{header}, \text{dict\_ptr} \to \text{dict}(\text{fields}) \rangle

slots 类的实例内存布局:

Objslots=⟨header,f1,f2,…,fn⟩\text{Obj}_{\text{slots}} = \langle \text{header}, f_1, f_2, \dots, f_n \rangle

slots 省去了 dict 哈希表,每个字段访问从 O(1)O(1) 哈希查找变为 O(1)O(1) 偏移量访问。内存节省:

ΔM=∣dict∣−n⋅8 bytes≈100-300 bytes/instance\Delta M = |\text{dict}| - n \cdot 8 \text{ bytes} \approx 100\text{-}300 \text{ bytes/instance}

对于百万级实例,slots 可节省数百 MB 内存。


3. 理论推导与原理解析

3.1 @dataclass 的代码生成

@dataclass 装饰器本质是一个编译期(导入时)代码生成器。它读取类的 __annotations__,生成 __init__、__repr__、__eq__ 等方法并绑定到类。形式化地:

\text{@dataclass}(C) = C \cup \{\text{__init\_\_}, \text{__repr\_\_}, \text{__eq\_\_}, \dots\}

其中 __init__ 的生成伪代码:

def __init__(self, name: str, age: int, tags: list[str] = MISSING):
    self.name = name
    self.age = age
    if tags is MISSING:
        self.tags = list()  # default_factory=list
    else:
        self.tags = tags

关键点:default_factory 在 __init__ 内部调用,每次实例化生成新对象,避免”共享可变默认值”陷阱。

3.2 可变默认值陷阱的形式化证明

考虑错误代码:

@dataclass
class Bad:
    items: list[int] = []  # 错误!共享同一个 list 对象

@dataclass 会检测到此情况并抛出 ValueError: mutable default ... is not allowed。但若绕过 @dataclass 手写:

class Bad:
    def __init__(self, items: list[int] = []) -> None:
        self.items = items

设创建两个实例 a,ba, b,则:

id(a.items)=id(b.items)=id(default)\text{id}(a.\text{items}) = \text{id}(b.\text{items}) = \text{id}(\text{default})

a.items.append(1) 会修改 b.items,造成隐蔽 bug。field(default_factory=list) 的修复:

∀i∈instances:id(i.items) is unique\forall i \in \text{instances}: \text{id}(i.\text{items}) \text{ is unique}

3.3 Pydantic 验证的多阶段流水线

Pydantic v2 的验证过程分为多个阶段:

  1. 解析阶段:从 dict/JSON/ORM 对象提取原始数据。
  2. 类型转换阶段(coercion):将原始值转换为目标类型("42" → 42)。
  3. 约束校验阶段:检查 Field(ge=0, le=150) 等约束。
  4. 字段验证器阶段:运行 @field_validator 装饰的函数。
  5. 模型验证器阶段:运行 @model_validator(mode='after') 装饰的函数。
  6. 模型构建:实例化模型对象。

形式化地:

Model(d)=ModelValidator(FieldValidators(Constraints(Coerce(d))))\text{Model}(d) = \text{ModelValidator}(\text{FieldValidators}(\text{Constraints}(\text{Coerce}(d))))

3.4 Pydantic v2 性能优势:Rust 内核

Pydantic v1 的验证循环用 Python 实现,每次字段访问都经过 Python 解释器。Pydantic v2 将验证逻辑编译为 Rust 函数(pydantic-core),仅边界处经过 Python-Rust FFI。

性能模型:

Tv1=n⋅(Tpy_interp+Tvalidate)T_{\text{v1}} = n \cdot (T_{\text{py\_interp}} + T_{\text{validate}}) Tv2=TFFI+n⋅Tvalidate_rustT_{\text{v2}} = T_{\text{FFI}} + n \cdot T_{\text{validate\_rust}}

其中 Tpy_interp≈100nsT_{\text{py\_interp}} \approx 100\text{ns},TFFI≈50nsT_{\text{FFI}} \approx 50\text{ns},Tvalidate_rust≈5nsT_{\text{validate\_rust}} \approx 5\text{ns}。对 10 字段模型:

Tv1≈1 μs,Tv2≈100 nsT_{\text{v1}} \approx 1\,\mu s, \quad T_{\text{v2}} \approx 100\,ns

实测 10 倍以上加速。

3.5 序列化的对称性

理想的序列化应满足往返一致性(round-trip consistency):

∀m∈M:deserialize(serialize(m))=m\forall m \in M: \text{deserialize}(\text{serialize}(m)) = m

Pydantic v2 的 model_dump_json() → model_validate_json() 满足此性质(在无信息丢失的类型上)。但有损类型(datetime → ISO 字符串 → datetime 在时区处理上可能丢失)需注意。

3.6 继承与 MRO 的交互

Pydantic v2 模型继承时,字段合并遵循 MRO(Method Resolution Order):

class Base(BaseModel):
    id: int
    created_at: datetime

class User(Base):
    name: str
    email: str

# User 拥有字段:id, created_at, name, email

字段顺序:父类字段在前,子类字段在后。这与 @dataclass 的行为一致,但 Pydantic v2 额外处理了 model_config 的继承合并。


4. 代码示例(企业级 production-ready)

4.1 项目结构

flowchart TD
    T0["dataclass_demo/"]
    T1["pyproject.toml"]
    T2["requirements.txt"]
    T3["README.md"]
    T4["src/"]
    T5["dataclass_demo/"]
    T6["__init__.py"]
    T7["value_objects.py    # 不可变值对象"]
    T8["domain_models.py    # Pydantic 领域模型"]
    T9["settings.py         # pydantic-settings 配置"]
    T10["fastapi_app.py      # FastAPI 集成"]
    T11["custom_types.py     # 自定义类型"]
    T12["serializers.py      # 序列化工具"]
    T0 --> T1
    T0 --> T2
    T0 --> T3
    T0 --> T4
    T4 --> T5
    T5 --> T6
    T5 --> T7
    T5 --> T8
    T5 --> T9
    T5 --> T10
    T5 --> T11
    T5 --> T12

4.2 pyproject.toml

[project]
name = "dataclass-demo"
version = "0.1.0"
description = "Python 数据类与 Pydantic 企业级示例"
requires-python = ">=3.10"
authors = [{ name = "FANDEX Team" }]
dependencies = [
    "pydantic>=2.7.0",
    "pydantic-settings>=2.2.0",
    "email-validator>=2.1.0",
]

[project.optional-dependencies]
dev = [
    "pytest>=8.0.0",
    "pytest-cov>=5.0.0",
    "ruff>=0.5.0",
    "mypy>=1.10.0",
    "fastapi>=0.110.0",
    "uvicorn>=0.29.0",
]

[tool.ruff]
line-length = 100
target-version = "py310"

[tool.ruff.lint]
select = ["E", "F", "I", "N", "UP", "B", "C4", "SIM"]

[tool.mypy]
python_version = "3.10"
strict = true
plugins = ["pydantic.mypy"]

4.3 requirements.txt

pydantic==2.7.1
pydantic-settings==2.2.1
email-validator==2.1.1
fastapi==0.110.0
uvicorn==0.29.0

4.4 不可变值对象(Python 3.10+)

"""
value_objects.py:不可变值对象示例。
- frozen=True 保证不可变
- slots=True 节省内存
- 适合 Money、Coordinate、Email 等值对象
Python: 3.10+
"""
from __future__ import annotations

from dataclasses import dataclass
from decimal import Decimal
from typing import Self

@dataclass(frozen=True, slots=True)
class Money:
    """货币值对象:金额 + 币种,不可变。

    不可变性使其:
    1. 线程安全,无需锁
    2. 可哈希,可作为字典键
    3. 引用透明,便于测试
    """

    amount: Decimal
    currency: str

    def __post_init__(self) -> None:
        """校验逻辑:frozen 类通过 object.__setattr__ 修改字段。"""
        if self.amount < 0:
            raise ValueError(f"金额不能为负: {self.amount}")
        if len(self.currency) != 3:
            raise ValueError(f"币种必须是 ISO 4217 三字母代码: {self.currency}")

    def add(self, other: Money) -> Self:
        """加法:返回新实例,保持不可变性。"""
        if self.currency != other.currency:
            raise ValueError(
                f"币种不匹配: {self.currency} vs {other.currency}"
            )
        return Money(self.amount + other.amount, self.currency)

    def multiply(self, factor: Decimal) -> Self:
        """标量乘法:返回新实例。"""
        return Money(self.amount * factor, self.currency)

    def format(self) -> str:
        """格式化展示。"""
        return f"{self.amount:,.2f} {self.currency}"

@dataclass(frozen=True, slots=True)
class Coordinate:
    """地理坐标值对象:纬度 + 经度。"""

    latitude: float
    longitude: float

    def __post_init__(self) -> None:
        if not -90 <= self.latitude <= 90:
            raise ValueError(f"纬度必须在 [-90, 90]: {self.latitude}")
        if not -180 <= self.longitude <= 180:
            raise ValueError(f"经度必须在 [-180, 180]: {self.longitude}")

    def distance_to(self, other: Coordinate) -> float:
        """Haversine 公式计算球面距离(公里)。"""
        from math import asin, cos, radians, sin, sqrt

        lat1, lon1 = radians(self.latitude), radians(self.longitude)
        lat2, lon2 = radians(other.latitude), radians(other.longitude)
        dlat = lat2 - lat1
        dlon = lon2 - lon1
        a = sin(dlat / 2) ** 2 + cos(lat1) * cos(lat2) * sin(dlon / 2) ** 2
        return 2 * 6371 * asin(sqrt(a))

# 使用示例
if __name__ == "__main__":
    price = Money(Decimal("99.99"), "USD")
    shipping = Money(Decimal("15.00"), "USD")
    total = price.add(shipping)
    print(total.format())  # 114.99 USD

    # 尝试修改会抛出 FrozenInstanceError
    try:
        price.amount = Decimal("0")
    except Exception as e:
        print(f"不可变: {type(e).__name__}")

4.5 Pydantic 领域模型(Python 3.12+)

"""
domain_models.py:Pydantic v2 领域模型示例。
- 嵌套模型
- 自定义验证器
- 严格模式与转换模式
- 序列化
Python: 3.12+
"""
from __future__ import annotations

from datetime import datetime, timezone
from enum import Enum
from typing import Annotated, Any

from pydantic import (
    BaseModel,
    ConfigDict,
    EmailStr,
    Field,
    field_serializer,
    field_validator,
    model_validator,
)

class UserRole(str, Enum):
    """用户角色枚举:继承 str 便于 JSON 序列化。"""

    ADMIN = "admin"
    USER = "user"
    GUEST = "guest"

class Address(BaseModel):
    """地址:嵌套模型。"""

    model_config = ConfigDict(strict=False)  # 允许字符串转数字

    street: str = Field(min_length=1, max_length=200)
    city: str = Field(min_length=1, max_length=100)
    postal_code: str = Field(pattern=r"^\d{6}$")  # 中国邮编
    country: str = Field(default="CN", min_length=2, max_length=2)

class User(BaseModel):
    """用户领域模型:演示 Pydantic v2 核心特性。"""

    model_config = ConfigDict(
        frozen=False,  # 可变模型
        str_strip_whitespace=True,  # 自动去除字符串首尾空格
        validate_assignment=True,  # 属性赋值时也校验
        use_enum_values=True,  # 序列化时使用枚举值而非枚举对象
        json_encoders={datetime: lambda v: v.isoformat()},
    )

    id: int = Field(gt=0, description="用户ID")
    username: str = Field(
        min_length=3,
        max_length=32,
        pattern=r"^[a-zA-Z0-9_]+$",
        description="用户名",
    )
    email: EmailStr
    age: Annotated[int, Field(ge=0, le=150)]  # Annotated 写法
    role: UserRole = UserRole.USER
    address: Address | None = None
    tags: list[str] = Field(default_factory=list, max_length=10)
    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

    @field_validator("username")
    @classmethod
    def username_no_admin(cls, v: str) -> str:
        """字段验证器:禁止用户名包含 'admin'。"""
        if "admin" in v.lower():
            raise ValueError("用户名不能包含 'admin'")
        return v

    @field_validator("age")
    @classmethod
    def age_must_be_adult_if_admin(cls, v: int, info) -> int:
        """跨字段验证:若角色是 ADMIN,年龄必须 >= 18。"""
        if info.data.get("role") == UserRole.ADMIN and v < 18:
            raise ValueError("管理员年龄必须 >= 18")
        return v

    @model_validator(mode="after")
    def check_email_username_consistency(self) -> "User":
        """模型验证器:邮箱前缀应与用户名一致(业务规则)。"""
        if self.email.split("@")[0].lower() != self.username.lower():
            # 仅警告,不抛错(实际业务可改为 raise)
            pass
        return self

    @field_serializer("created_at")
    def serialize_created_at(self, v: datetime) -> str:
        """自定义序列化:datetime 转 ISO 字符串。"""
        return v.isoformat()

    def to_public_dict(self) -> dict[str, Any]:
        """公开视图:隐藏敏感字段。"""
        data = self.model_dump()
        data.pop("email", None)
        return data

# 使用示例
if __name__ == "__main__":
    user = User(
        id=1,
        username="alice",
        email="alice@example.com",
        age=25,
        role=UserRole.ADMIN,
        address=Address(
            street="中关村大街1号",
            city="北京",
            postal_code="100080",
        ),
        tags=["vip", "active"],
    )

    print(user.model_dump_json(indent=2))

    # 错误示例:校验失败
    try:
        User(
            id=0,  # gt=0 校验失败
            username="admin_alice",  # 包含 'admin'
            email="invalid-email",
            age=200,  # le=150 校验失败
        )
    except Exception as e:
        print(f"校验失败: {type(e).__name__}")
        for err in e.errors():
            print(f"  - {err['loc']}: {err['msg']}")

4.6 pydantic-settings 配置管理(Python 3.10+)

"""
settings.py:使用 pydantic-settings 管理多源配置。
- 环境变量
- .env 文件
- 命令行参数(通过 Pydantic Settings)
Python: 3.10+
"""
from __future__ import annotations

from functools import lru_cache
from typing import Literal

from pydantic import Field, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict

class DatabaseSettings(BaseSettings):
    """数据库配置:从环境变量与 .env 加载。"""

    model_config = SettingsConfigDict(
        env_prefix="DB_",
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False,
        extra="ignore",
    )

    host: str = "localhost"
    port: int = 5432
    name: str = "app_db"
    user: str = "postgres"
    password: SecretStr = Field(default="")  # SecretStr 不会在 repr 中泄露
    pool_size: int = Field(default=10, ge=1, le=100)
    echo: bool = False

class RedisSettings(BaseSettings):
    """Redis 配置。"""

    model_config = SettingsConfigDict(
        env_prefix="REDIS_",
        env_file=".env",
    )

    host: str = "localhost"
    port: int = 6379
    db: int = 0
    password: SecretStr = Field(default="")

class AppSettings(BaseSettings):
    """应用全局配置:聚合多个子配置。"""

    model_config = SettingsConfigDict(
        env_prefix="APP_",
        env_file=".env",
        env_nested_delimiter="__",  # 支持 APP_DATABASE__HOST 形式
    )

    env: Literal["dev", "staging", "prod"] = "dev"
    debug: bool = False
    secret_key: SecretStr = Field(min_length=32)
    api_rate_limit: int = Field(default=100, ge=1)

    database: DatabaseSettings = Field(default_factory=DatabaseSettings)
    redis: RedisSettings = Field(default_factory=RedisSettings)

@lru_cache
def get_settings() -> AppSettings:
    """单例:lru_cache 缓存,整个应用共享一份配置。"""
    return AppSettings()

# .env 文件示例
ENV_EXAMPLE = """
APP_ENV=prod
APP_DEBUG=false
APP_SECRET_KEY=your-super-secret-key-at-least-32-chars
APP_API_RATE_LIMIT=1000

DB_HOST=postgres.example.com
DB_PORT=5432
DB_NAME=production_db
DB_USER=app
DB_PASSWORD=encrypted-password-here
DB_POOL_SIZE=20

REDIS_HOST=redis.example.com
REDIS_PASSWORD=another-encrypted-password
"""

if __name__ == "__main__":
    import os

    os.environ["APP_SECRET_KEY"] = "x" * 32  # 演示用,实际从 .env 读取
    settings = get_settings()
    print(f"环境: {settings.env}")
    print(f"数据库: {settings.database.host}:{settings.database.port}")
    print(f"密码: {settings.database.password.get_secret_value()[:3]}***")

4.7 自定义类型(Python 3.12+)

"""
custom_types.py:基于 Annotated + AfterValidator 的自定义类型。
- 领域特定类型(Domain-Specific Type)
- 类型即校验
- 可读性强,可复用
Python: 3.12+
"""
from __future__ import annotations

import re
from typing import Annotated

from pydantic import AfterValidator, BaseModel, Field, StringConstraints

def _validate_email(v: str) -> str:
    """邮箱格式校验。"""
    pattern = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
    if not re.match(pattern, v):
        raise ValueError(f"无效的邮箱格式: {v}")
    return v.lower()  # 规范化为小写

def _validate_isbn(v: str) -> str:
    """ISBN-13 校验(含校验位算法)。"""
    v = v.replace("-", "").replace(" ", "")
    if len(v) != 13 or not v.isdigit():
        raise ValueError(f"ISBN 必须是 13 位数字: {v}")
    # ISBN-13 校验位算法
    digits = [int(d) for d in v]
    checksum = sum(d * (1 if i % 2 == 0 else 3) for i, d in enumerate(digits[:12]))
    check_digit = (10 - checksum % 10) % 10
    if check_digit != digits[12]:
        raise ValueError(f"ISBN 校验位错误: {v}")
    return v

# 自定义类型:通过 Annotated 组合约束
Email = Annotated[str, AfterValidator(_validate_email)]
ISBN = Annotated[str, AfterValidator(_validate_isbn)]
PositiveInt = Annotated[int, Field(gt=0)]
NonEmptyStr = Annotated[str, StringConstraints(min_length=1, max_length=255)]

class Book(BaseModel):
    """图书模型:使用自定义类型。"""

    title: NonEmptyStr
    isbn: ISBN
    author_email: Email
    price_cents: PositiveInt
    stock: Annotated[int, Field(ge=0)] = 0

# 使用示例
if __name__ == "__main__":
    book = Book(
        title="Python Cookbook",
        isbn="978-1449340377",  # 会被规范化为 9781449340377
        author_email="David.Beazley@example.com",
        price_cents=5999,
        stock=100,
    )
    print(book.model_dump_json(indent=2))

    # 校验失败示例
    try:
        Book(
            title="",
            isbn="123",  # ISBN 长度错误
            author_email="invalid",
            price_cents=0,
        )
    except Exception as e:
        for err in e.errors():
            print(f"  - {err['loc']}: {err['msg']}")

4.8 FastAPI 集成(Python 3.10+)

"""
fastapi_app.py:FastAPI + Pydantic v2 集成示例。
- 请求模型
- 响应模型
- 错误处理
- OpenAPI 自动生成
Python: 3.10+
"""
from __future__ import annotations

from datetime import datetime, timezone
from typing import Annotated

from fastapi import FastAPI, HTTPException, Path, Query
from pydantic import BaseModel, Field

app = FastAPI(
    title="User Service",
    version="1.0.0",
    description="演示 Pydantic v2 与 FastAPI 集成",
)

class UserCreateRequest(BaseModel):
    """创建用户请求模型。"""

    username: str = Field(min_length=3, max_length=32, pattern=r"^[a-zA-Z0-9_]+$")
    email: str = Field(pattern=r"^[^@]+@[^@]+\.[^@]+$")
    age: Annotated[int, Field(ge=0, le=150)]

class UserResponse(BaseModel):
    """用户响应模型:隐藏敏感字段。"""

    id: int
    username: str
    email: str
    age: int
    created_at: datetime

# 模拟数据库
_DB: dict[int, dict] = {}
_NEXT_ID = 1

@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(req: UserCreateRequest) -> UserResponse:
    """创建用户。

    Pydantic 自动校验请求体,校验失败返回 422 + 详细错误。
    """
    global _NEXT_ID

    # 业务逻辑:检查用户名是否已存在
    for user in _DB.values():
        if user["username"] == req.username:
            raise HTTPException(status_code=409, detail="用户名已存在")

    user_data = {
        "id": _NEXT_ID,
        "username": req.username,
        "email": req.email,
        "age": req.age,
        "created_at": datetime.now(timezone.utc),
    }
    _DB[_NEXT_ID] = user_data
    _NEXT_ID += 1

    return UserResponse(**user_data)

@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(
    user_id: Annotated[int, Path(gt=0, description="用户ID")]
) -> UserResponse:
    """获取用户详情。"""
    user = _DB.get(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    return UserResponse(**user)

@app.get("/users", response_model=list[UserResponse])
async def list_users(
    limit: Annotated[int, Query(ge=1, le=100)] = 10,
    offset: Annotated[int, Query(ge=0)] = 0,
) -> list[UserResponse]:
    """分页查询用户。"""
    users = list(_DB.values())[offset : offset + limit]
    return [UserResponse(**u) for u in users]

# 启动:uvicorn fastapi_app:app --reload

4.9 完整测试套件(pytest)

"""
test_models.py:Pydantic 模型测试套件。
- 校验成功路径
- 校验失败路径
- 序列化/反序列化往返
- 边界条件
Python: 3.10+
"""
from __future__ import annotations

import json
from datetime import datetime, timezone
from decimal import Decimal

import pytest
from pydantic import ValidationError

from domain_models import Address, User, UserRole
from value_objects import Coordinate, Money

class TestMoney:
    """Money 值对象测试。"""

    def test_create_valid_money(self) -> None:
        m = Money(Decimal("99.99"), "USD")
        assert m.amount == Decimal("99.99")
        assert m.currency == "USD"

    def test_negative_amount_raises(self) -> None:
        with pytest.raises(ValueError, match="金额不能为负"):
            Money(Decimal("-1"), "USD")

    def test_invalid_currency_raises(self) -> None:
        with pytest.raises(ValueError, match="币种"):
            Money(Decimal("1"), "DOLLAR")

    def test_add_same_currency(self) -> None:
        a = Money(Decimal("10"), "USD")
        b = Money(Decimal("5"), "USD")
        assert a.add(b).amount == Decimal("15")

    def test_add_different_currency_raises(self) -> None:
        with pytest.raises(ValueError, match="币种不匹配"):
            Money(Decimal("1"), "USD").add(Money(Decimal("1"), "EUR"))

    def test_frozen_immutability(self) -> None:
        m = Money(Decimal("1"), "USD")
        with pytest.raises(Exception):
            m.amount = Decimal("2")  # type: ignore[misc]

    def test_hashable(self) -> None:
        """不可变对象应可哈希。"""
        m1 = Money(Decimal("1"), "USD")
        m2 = Money(Decimal("1"), "USD")
        assert hash(m1) == hash(m2)
        assert m1 == m2

class TestCoordinate:
    """Coordinate 值对象测试。"""

    def test_valid_coordinate(self) -> None:
        c = Coordinate(39.9, 116.4)
        assert c.latitude == 39.9

    @pytest.mark.parametrize("lat,lon", [
        (-91, 0), (91, 0), (0, -181), (0, 181),
    ])
    def test_invalid_coordinates(self, lat: float, lon: float) -> None:
        with pytest.raises(ValueError):
            Coordinate(lat, lon)

    def test_distance(self) -> None:
        beijing = Coordinate(39.9, 116.4)
        shanghai = Coordinate(31.2, 121.5)
        d = beijing.distance_to(shanghai)
        assert 1000 < d < 1200  # 北京到上海约 1000-1200 公里

class TestUser:
    """User Pydantic 模型测试。"""

    @pytest.fixture
    def valid_data(self) -> dict:
        return {
            "id": 1,
            "username": "alice",
            "email": "alice@example.com",
            "age": 25,
        }

    def test_create_valid_user(self, valid_data: dict) -> None:
        user = User(**valid_data)
        assert user.username == "alice"
        assert user.role == UserRole.USER

    def test_username_too_short(self, valid_data: dict) -> None:
        valid_data["username"] = "ab"
        with pytest.raises(ValidationError) as exc_info:
            User(**valid_data)
        assert "String should have at least 3 characters" in str(exc_info.value)

    def test_username_with_admin_raises(self, valid_data: dict) -> None:
        valid_data["username"] = "admin_alice"
        with pytest.raises(ValidationError, match="不能包含 'admin'"):
            User(**valid_data)

    def test_admin_must_be_adult(self, valid_data: dict) -> None:
        valid_data["role"] = UserRole.ADMIN
        valid_data["age"] = 17
        with pytest.raises(ValidationError, match="管理员年龄"):
            User(**valid_data)

    def test_invalid_email(self, valid_data: dict) -> None:
        valid_data["email"] = "not-an-email"
        with pytest.raises(ValidationError):
            User(**valid_data)

    def test_negative_id(self, valid_data: dict) -> None:
        valid_data["id"] = 0
        with pytest.raises(ValidationError):
            User(**valid_data)

    def test_nested_address_validation(self, valid_data: dict) -> None:
        valid_data["address"] = {
            "street": "Main St",
            "city": "Beijing",
            "postal_code": "12345",  # 长度错误
        }
        with pytest.raises(ValidationError):
            User(**valid_data)

    def test_json_round_trip(self, valid_data: dict) -> None:
        """序列化往返测试。"""
        user = User(**valid_data)
        json_str = user.model_dump_json()
        user2 = User.model_validate_json(json_str)
        assert user2.username == user.username
        assert user2.email == user.email

    def test_validate_assignment(self, valid_data: dict) -> None:
        """validate_assignment=True 时属性赋值也校验。"""
        user = User(**valid_data)
        with pytest.raises(ValidationError):
            user.age = 200  # type: ignore[misc]

    def test_strip_whitespace(self, valid_data: dict) -> None:
        valid_data["username"] = "  alice  "
        user = User(**valid_data)
        assert user.username == "alice"

5. 对比分析

5.1 dataclass vs attrs vs Pydantic vs msgspec

特性dataclassattrsPydantic v2msgspec
标准库是否否否
类型校验否(仅 hint)可选(validator)是(强制)是(强制)
性能(10字段实例化)~1μs~1.5μs~3μs~0.5μs
序列化(JSON)需手动需手动内置(Rust)内置(C)
嵌套模型需手动需手动是是
不可变frozen=True@frozenConfigDict(frozen=True)frozen=True
slotsslots=True(3.10+)slots=True是是
默认值工厂field(default_factory=)field(factory=)Field(default_factory=)default_factory=
自定义验证器__post_init__@validator@field_validator@validator
JSON Schema 生成否否是否
FastAPI 集成部分否原生部分
学习成本低中中中
生态成熟度高高极高成长中

5.2 Pydantic v1 vs v2 API 对照

v1 APIv2 API说明
.dict().model_dump()导出为 dict
.json().model_dump_json()导出为 JSON 字符串
.parse_obj(d).model_validate(d)从 dict 创建
.parse_raw(s).model_validate_json(s)从 JSON 字符串创建
class Config:model_config = ConfigDict(...)配置
@validator@field_validator字段验证器
@root_validator@model_validator模型验证器
Optional[X] = None`XNone = None`
Field(..., env=)移除(用 pydantic-settings)环境变量
.copy().model_copy()复制
update_forward_refs()model_rebuild()前向引用

5.3 跨语言对比

语言等价方案特点
Pythondataclasses / Pydantic类型注解驱动,运行时校验可选
JavaScript/TypeScriptclass-validator / zod / io-ts装饰器或运行时 schema
Gostruct + tagsstruct tag 驱动,encoding/json 内置
Rustserde derive编译期生成,零运行时开销
JavaLombok / Record注解处理器,Record 是 Java 14+ 标准方案
SwiftCodable协议驱动,编译期生成

5.4 何时选择哪个?

dataclass 适用场景:

  • 内部数据结构,无序列化需求
  • 性能敏感场景(比 Pydantic 快)
  • 不需运行时校验
  • 与标准库紧密集成(typing、enum)

Pydantic v2 适用场景:

  • API 边界(FastAPI、HTTP schema)
  • 配置管理(pydantic-settings)
  • 需要 JSON Schema 生成
  • 需要复杂嵌套校验
  • 需要序列化/反序列化

attrs 适用场景:

  • 需要细粒度配置(slots、cmp、hash 单独控制)
  • 历史代码库已使用 attrs
  • 不愿引入 Pydantic 的复杂依赖

msgspec 适用场景:

  • 极致性能(比 Pydantic v2 快 2-10x)
  • MessagePack 协议
  • 大数据量序列化
  • 不需要 JSON Schema

6. 常见陷阱与反模式

6.1 可变默认值

错误:

@dataclass
class Bad:
    items: list[int] = []  # ValueError!

@dataclass 会检测并抛错,但若绕过装饰器手写则会出 bug。

正确:

@dataclass
class Good:
    items: list[int] = field(default_factory=list)

6.2 frozen 类的 __post_init__

frozen=True 类中无法直接赋值 self.x = ...,需使用 object.__setattr__:

@dataclass(frozen=True)
class Point:
    x: float
    y: float
    norm: float = 0.0

    def __post_init__(self) -> None:
        # 错误:self.norm = (self.x**2 + self.y**2) ** 0.5
        object.__setattr__(self, "norm", (self.x**2 + self.y**2) ** 0.5)

6.3 Pydantic v1 → v2 迁移陷阱

6.3.1 @validator 语义变化

# v1
@validator("name", "email", pre=True)
def to_lower(cls, v):
    return v.lower()

# v2
@field_validator("name", "email", mode="before")
@classmethod
def to_lower(cls, v):
    return v.lower() if isinstance(v, str) else v

关键差异:

  • v2 必须加 @classmethod
  • v1 的 pre=True → v2 的 mode="before"
  • v1 的 each_item=True → v2 需手动遍历

6.3.2 Optional 语义

# v1
class M(BaseModel):
    x: Optional[int]  # 默认 None,可省略

# v2
class M(BaseModel):
    x: int | None  # 必须显式 = None
    x: int | None = None  # 正确

v2 不再隐式为 Optional 添加默认值 None。

6.4 校验器顺序

Pydantic v2 的字段验证器执行顺序:

  1. mode="before" 验证器(在类型转换前)
  2. 类型转换与约束校验
  3. mode="after" 验证器(在类型转换后)
  4. model_validator(mode="after")
class M(BaseModel):
    x: int

    @field_validator("x", mode="before")
    @classmethod
    def before(cls, v):
        print(f"before: {v!r}")
        return v

    @field_validator("x", mode="after")
    @classmethod
    def after(cls, v):
        print(f"after: {v!r} (type={type(v).__name__})")
        return v

M(x="42")
# 输出:
# before: '42' (str)
# after: 42 (int)

6.5 validate_assignment 性能陷阱

validate_assignment=True 使每次属性赋值都触发校验,在循环中性能损失显著:

# 慢:每次赋值都校验
class Slow(BaseModel):
    model_config = ConfigDict(validate_assignment=True)
    x: int

obj = Slow(x=0)
for i in range(1000000):
    obj.x = i  # 每次都校验

# 快:批量修改后统一校验
class Fast(BaseModel):
    x: int

obj = Fast(x=0)
for i in range(1000000):
    obj.__dict__["x"] = i  # 绕过校验
Fast.model_validate(obj.__dict__)  # 最后统一校验

6.6 SecretStr 的 JSON 序列化

from pydantic import BaseModel, SecretStr

class C(BaseModel):
    password: SecretStr

c = C(password="secret123")
print(c.model_dump_json())  # {"password":"**********"} 不泄露
print(c.model_dump())  # {'password': SecretStr('**********')}
print(c.password.get_secret_value())  # 'secret123' 显式获取

SecretStr 防止意外在日志中泄露密码,但需注意:

  • 日志框架若调用 str() 仍可能泄露(Pydantic v2 已优化为 **********)
  • 持久化前需显式调用 get_secret_value()

6.7 继承中的 model_config 合并

class Base(BaseModel):
    model_config = ConfigDict(frozen=True)

class Child(Base):
    model_config = ConfigDict(extra="allow")  # 会覆盖 frozen=True!

# 正确:显式合并
class Child2(Base):
    model_config = ConfigDict(**Base.model_config, extra="allow")

6.8 dataclass 继承中的默认值顺序

@dataclass
class Base:
    x: int

@dataclass
class Child(Base):
    y: int = 10  # 有默认值

# 错误:Base.x 无默认值,Child.y 有默认值,会抛 TypeError
# 正确:所有字段都要有默认值,或都无默认值

@dataclass
class Base2:
    x: int = 0  # 给 Base 字段默认值

@dataclass
class Child2(Base2):
    y: int = 10

6.9 kw_only 的使用

Python 3.10+ 的 @dataclass(kw_only=True) 使所有字段只能通过关键字参数传递:

@dataclass(kw_only=True)
class Config:
    host: str = "localhost"
    port: int = 8080

# Config("example.com")  # 错误:位置参数
Config(host="example.com", port=443)  # 正确

继承时混合使用:

@dataclass
class Base:
    x: int  # 位置参数

@dataclass(kw_only=True)
class Child(Base):
    y: int = 0  # 关键字参数

Child(1, y=2)  # 正确

6.10 JSON Schema 生成

Pydantic 自动生成 JSON Schema,但部分类型需注意:

class M(BaseModel):
    id: int
    created_at: datetime
    metadata: dict[str, Any]

print(json.dumps(M.model_json_schema(), indent=2))

注意:

  • datetime 在 schema 中是 string + format: date-time
  • dict[str, Any] 在 schema 中是 object
  • 自定义类型需实现 __get_pydantic_json_schema__ 才能正确生成

7. 工程实践

7.1 渐进式迁移:dict → dataclass → Pydantic

阶段 1:dict 直接传递(原型期)

def create_user(data: dict) -> dict:
    return {"id": 1, **data}

阶段 2:引入 dataclass(重构期)

@dataclass
class UserData:
    username: str
    email: str
    age: int

def create_user(data: UserData) -> dict:
    return {"id": 1, "username": data.username, ...}

阶段 3:引入 Pydantic(生产期)

class UserCreate(BaseModel):
    username: str = Field(min_length=3)
    email: EmailStr
    age: int = Field(ge=0, le=150)

@app.post("/users")
async def create_user(req: UserCreate) -> UserResponse:
    ...

7.2 monorepo 共享 schema

flowchart TD
    T0["monorepo/"]
    T1["packages/"]
    T2["schemas/          # 共享 Pydantic 模型"]
    T3["pyproject.toml"]
    T4["src/schemas/"]
    T5["user.py"]
    T6["order.py"]
    T7["backend/          # FastAPI 后端"]
    T8["pyproject.toml  # 依赖 schemas"]
    T9["frontend/         # 前端"]
    T10["package.json   # 通过 datamodel-code-generator 生成 TS 类型"]
    T11["pyproject.toml"]
    T0 --> T1
    T10 --> T11

通过 pydantic-to-typescript 或 datamodel-code-generator 自动同步:

# 后端 Pydantic → 前端 TypeScript
pydantic2ts --module schemas.user --output frontend/src/types/user.ts

7.3 版本化 schema(向前兼容)

from pydantic import BaseModel, Field, model_validator

class UserV1(BaseModel):
    """v1 schema:基础字段。"""
    username: str
    email: str

class UserV2(BaseModel):
    """v2 schema:新增 age,移除 email(改用 contact)。"""
    username: str
    age: int
    contact: str  # 原来的 email

    @model_validator(mode="before")
    @classmethod
    def migrate_from_v1(cls, data: dict) -> dict:
        """向前兼容:v1 数据自动迁移到 v2。"""
        if "email" in data and "contact" not in data:
            data["contact"] = data.pop("email")
        if "age" not in data:
            data["age"] = 0  # 默认值
        return data

# v1 数据可被 v2 接受
v1_data = {"username": "alice", "email": "alice@example.com"}
user = UserV2(**v1_data)
print(user.contact)  # alice@example.com

7.4 自定义 JSON 编码器

from pydantic import BaseModel, field_serializer
from datetime import datetime
from enum import Enum
from uuid import UUID

class Status(str, Enum):
    ACTIVE = "active"
    INACTIVE = "inactive"

class Record(BaseModel):
    id: UUID
    status: Status
    created_at: datetime
    metadata: dict[str, str]

    @field_serializer("id")
    def serialize_id(self, v: UUID) -> str:
        return str(v)

    @field_serializer("created_at")
    def serialize_dt(self, v: datetime) -> str:
        return v.isoformat()

    @field_serializer("status")
    def serialize_status(self, v: Status) -> str:
        return v.value

7.5 ORM 互操作(SQLAlchemy)

from pydantic import BaseModel, ConfigDict
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

class UserORM(Base):
    """SQLAlchemy ORM 模型。"""
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    username = Column(String(32))
    email = Column(String(255))

class UserSchema(BaseModel):
    """Pydantic schema:用于 API。
    
    from_attributes=True 使 Pydantic 可从 ORM 对象读取属性。
    """
    model_config = ConfigDict(from_attributes=True)

    id: int
    username: str
    email: str

# ORM → Pydantic
orm_user = UserORM(id=1, username="alice", email="alice@example.com")
schema_user = UserSchema.model_validate(orm_user)
print(schema_user.model_dump_json())

7.6 性能优化

7.6.1 model_construct 绕过校验

class M(BaseModel):
    x: int

# 已校验过的数据(如从数据库读取),用 model_construct 绕过校验
m = M.model_construct(x=42)  # 比 M(x=42) 快 5-10 倍

7.6.2 Tag 缓存 schema

# Pydantic v2 内部缓存了 schema,但复杂继承可能重建
# 显式调用 model_rebuild 确保最优
class M(BaseModel):
    ...

M.model_rebuild()  # 编译 schema

7.6.3 批量验证

from pydantic import TypeAdapter

adapter = TypeAdapter(list[User])

# 单次验证整个列表,比循环验证快
users = adapter.validate_python([{"id": i, "username": f"u{i}", ...} for i in range(1000)])

7.7 CI/CD 集成

# .github/workflows/ci.yml
name: CI
on: [push, pull_request]
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install -e ".[dev]"
      - run: ruff check src tests
      - run: ruff format --check src tests
      - run: mypy src
      - run: pytest --cov=src --cov-report=xml
      - run: bandit -r src
      # 生成 JSON Schema 并检查是否变化
      - name: Check JSON Schema
        run: |
          python -c "from schemas import User; import json; print(json.dumps(User.model_json_schema(), indent=2))" > schema.json
          git diff --exit-code schema.json

8. 案例研究

8.1 FastAPI:Pydantic 的旗舰应用

背景:Sebastián Ramírez 2018 年发布 FastAPI,完全基于 Pydantic 构建 API schema。FastAPI 的核心价值主张:类型即文档——开发者声明 Pydantic 模型,FastAPI 自动生成 OpenAPI 文档、请求校验、响应序列化。

实践:

  • 所有请求/响应模型均为 BaseModel 子类
  • Annotated[X, Field(...)] 声明约束
  • Depends() 注入依赖,依赖项本身也用 Pydantic 模型
  • 性能:FastAPI 与 Starlette + Pydantic v2 组合,QPS 可达 50000+(单进程)

数据:截至 2024 年,FastAPI GitHub Star 73k+,PyPI 月下载 5000 万+,被 Uber、Netflix、Microsoft 等大型企业采用。

8.2 Uber:Pydantic 在大数据场景的应用

背景:Uber 的 Michelangelo 机器学习平台使用 Pydantic 定义模型 schema,确保训练/推理时数据一致性。

实践:

  • 用 Pydantic 模型替代 Protobuf,简化 Python 侧开发
  • 通过 model_json_schema() 自动生成 JSON Schema,与 Kafka Schema Registry 集成
  • v1 → v2 迁移:性能提升 8 倍,节省约 15% CPU

经验(来自 Uber Engineering Blog 2023):

  • 大数据量场景避免 validate_assignment=True
  • 使用 model_construct 处理可信数据
  • 嵌套深度控制在 5 层以内,避免 schema 编译慢

8.3 Microsoft:TypeScript 类型生成

背景:Microsoft 的部分内部服务采用 Python 后端 + TypeScript 前端,需要保持类型一致。

方案:

  • 后端用 Pydantic 定义 schema
  • CI 流水线运行 pydantic2ts 生成 TypeScript 类型
  • 前端直接 import 生成的 .ts 文件

效果:

  • 消除前后端类型不一致 bug
  • API 变更自动触发类型重新生成
  • 新增字段时 TypeScript 编译期报错,防止遗漏

8.4 Pydantic v2 迁移:Django Ninja

背景:Django Ninja 是 Django 生态的 FastAPI 替代品,完全基于 Pydantic。

迁移挑战:

  • Django ORM 模型与 Pydantic 模型互操作
  • v1 的 orm_mode=True → v2 的 from_attributes=True
  • @validator → @field_validator 大量重写

结果(来自 Django Ninja 1.0 Release Notes):

  • 性能提升 4 倍
  • 类型错误在 v2 下更精确(错误定位到字段而非模型)
  • 迁移工作量约 2 人周(约 200 个模型)

8.5 NumPy:dataclass 在科学计算中的应用

背景:NumPy 内部用 dataclass 定义数组元数据(dtype、shape、strides)。

实践:

@dataclass(frozen=True, slots=True)
class ArrayMeta:
    dtype: str
    shape: tuple[int, ...]
    strides: tuple[int, ...]

    @property
    def ndim(self) -> int:
        return len(self.shape)

    @property
    def size(self) -> int:
        from math import prod
        return prod(self.shape)

优势:

  • frozen=True 保证元数据不可变,便于缓存
  • slots=True 节省内存(NumPy 创建大量元数据对象)
  • 自动 __eq__ 便于比较两个数组元数据

8.6 Django:Model 与 dataclass 互操作

Django Model 与 dataclass 的互操作不依赖特定版本:dataclasses.asdict 把 DTO 转成字典后即可交给 Django 的序列化器或视图层使用。常见模式:

from dataclasses import dataclass, asdict
from django.db import models

class User(models.Model):
    username = models.CharField(max_length=32)
    email = models.EmailField()

@dataclass
class UserDTO:
    """数据传输对象:与 Django Model 解耦。"""
    id: int
    username: str
    email: str

    @classmethod
    def from_orm(cls, user: User) -> "UserDTO":
        return cls(id=user.id, username=user.username, email=user.email)

# 使用
user = User.objects.get(id=1)
dto = UserDTO.from_orm(user)
print(asdict(dto))  # {'id': 1, 'username': '...', 'email': '...'}

填空题知识点讲解

常见疑问 5:@dataclass 中,可变默认值应使用 field(default_factory=____) 创建。

解析讲解:list(或 dict、set 等可调用对象)

常见疑问 6:Pydantic v2 中,从 ORM 对象创建模型需在 model_config 中设置 ____=True。

解析讲解:from_attributes

常见疑问 7:@dataclass(slots=True) 通过 ____ 替代 __dict__ 节省内存。

解析讲解:__slots__

常见疑问 8:Pydantic v2 中,____ 装饰器用于跨字段验证(如”开始时间 < 结束时间”)。

解析讲解:@model_validator

编程题知识点讲解

常见疑问 9:实现一个 Rectangle 值对象,满足:

  • 不可变
  • 校验长宽 > 0
  • 提供 area() 和 perimeter() 方法
  • 可哈希

解析讲解:

from dataclasses import dataclass

@dataclass(frozen=True, slots=True)
class Rectangle:
    width: float
    height: float

    def __post_init__(self) -> None:
        if self.width <= 0 or self.height <= 0:
            raise ValueError(f"长宽必须 > 0: {self.width}, {self.height}")

    @property
    def area(self) -> float:
        return self.width * self.height

    @property
    def perimeter(self) -> float:
        return 2 * (self.width + self.height)

常见疑问 10:实现一个 Pydantic 模型 Order,满足:

  • id: int(>0)
  • items: list[OrderItem](至少 1 个)
  • total: Decimal(自动计算,无需用户提供)
  • 自定义验证器:total 必须等于所有 item.subtotal 之和

解析讲解:

from decimal import Decimal
from typing import Annotated
from pydantic import BaseModel, Field, model_validator

class OrderItem(BaseModel):
    name: str
    price: Decimal
    quantity: int = Field(ge=1)

    @property
    def subtotal(self) -> Decimal:
        return self.price * self.quantity

class Order(BaseModel):
    id: int = Field(gt=0)
    items: list[OrderItem] = Field(min_length=1)
    total: Decimal | None = None

    @model_validator(mode="after")
    def calculate_total(self) -> "Order":
        expected = sum((item.subtotal for item in self.items), Decimal(0))
        if self.total is not None and self.total != expected:
            raise ValueError(f"total {self.total} != expected {expected}")
        object.__setattr__(self, "total", expected)
        return self

10. 工具选型决策树

flowchart TD
    T0["需要数据建模?"]
    T1["仅内部使用,无序列化?"]
    T2["dataclass(标准库,零依赖)"]
    T3["需要运行时校验?"]
    T4["需要 JSON Schema 或 FastAPI 集成?"]
    T5["Pydantic v2"]
    T6["极致性能?"]
    T7["msgspec"]
    T8["已有 attrs 代码库?"]
    T9["attrs"]
    T10["需要配置管理(env、.env)?"]
    T11["pydantic-settings"]
    T12["需要 ORM 互操作?"]
    T13["SQLAlchemy 2.0+ → dataclass + SQLAlchemy dataclass"]
    T14["Django Model → dataclass DTO"]
    T15["Tortoise ORM → Pydantic Model"]
    T16["需要不可变值对象?"]
    T17["@dataclass(frozen=True, slots=True)"]
    T0 --> T1
    T2 --> T3
    T9 --> T10
    T11 --> T12
    T15 --> T16
    T16 --> T17

11.1 规范与 PEP

11.2 官方文档

11.3 学术论文

  • Pierce, B. C. (2002). Types and Programming Languages. MIT Press. ISBN: 978-0262162098.
  • Cardelli, L., & Wegner, P. (1985). On Understanding Types, Data Abstraction, and Polymorphism. ACM Computing Surveys, 17(4), 471-523. https://doi.org/10.1145/6041.6042
  • Appel, A. W. (1998). Modern Compiler Implementation. Cambridge University Press. ISBN: 978-0521586034.

11.4 工程实践

11.5 性能基准


12.1 书籍

  • Ramalho, L. (2022). Fluent Python: Clear, Concise, and Effective Programming (2nd ed.). O’Reilly Media. ISBN: 978-1492056355.(第 5 章”一等函数”与第 8 章”对象引用、可变性和垃圾回收”)
  • Bader, D. (2017). Python Tricks: A Buffet of Awesome Python Features. DBader Publishing. ISBN: 978-1775093313.(“Data Classes” 章节)
  • Summerfield, M. (2021). Programming in Python 3: A Complete Introduction to the Python Language (3rd ed.). Addison-Wesley. ISBN: 978-0321680563.

12.2 论文与标准

  • JSON Schema Specification. (2020). JSON Schema Draft 2020-12. https://json-schema.org/draft/2020-12/json-schema.html
  • ISO/IEC 9899:2018. Information technology — Programming languages — C.(参考 C struct 内存布局)
  • ISO 4217:2015. Codes for the representation of currencies.(货币代码标准)

12.4 学习路线

初级:dataclass 基础
  ↓
中级:Pydantic v2 + FastAPI
  ↓
进阶:自定义类型、JSON Schema、性能优化
  ↓
高级:版本化 schema、monorepo 共享、跨语言类型生成
  ↓
专家:参与 Pydantic/attrs 开源贡献、设计领域特定类型系统

13. 附录

13.1 @dataclass 参数速查

参数默认值说明
initTrue生成 __init__
reprTrue生成 __repr__
eqTrue生成 __eq__
orderFalse生成 __lt__/__le__/__gt__/__ge__
unsafe_hashFalse强制生成 __hash__(不安全)
frozenFalse不可变
slotsFalse(3.10+)使用 __slots__
kw_onlyFalse(3.10+)仅关键字参数

13.2 field() 参数速查

参数说明
default默认值(不可变类型)
default_factory默认值工厂(可变类型)
init是否在 __init__ 中(默认 True)
repr是否在 __repr__ 中(默认 True)
hash是否参与 __hash__(默认 eq)
compare是否参与 __eq__(默认 True)
metadata元数据字典

13.3 Pydantic v2 Field 常用约束

约束类型示例
gt数值Field(gt=0)
ge数值Field(ge=0)
lt数值Field(lt=100)
le数值Field(le=100)
min_length字符串/列表Field(min_length=3)
max_length字符串/列表Field(max_length=100)
pattern字符串Field(pattern=r"^\d+$")
multiple_of数值Field(multiple_of=5)
description任意Field(description="用户ID")
examples任意Field(examples=["alice"])
deprecated任意Field(deprecated="使用 new_field 替代")

13.4 ConfigDict 常用选项

选项默认值说明
strictFalse严格模式(不强制转换)
frozenFalse不可变
extra"ignore"处理多余字段:"ignore"/"allow"/"forbid"
validate_assignmentFalse属性赋值时校验
use_enum_valuesFalse序列化时用枚举值
str_strip_whitespaceFalse去除字符串首尾空格
str_to_lowerFalse字符串转小写
str_to_upperFalse字符串转大写
from_attributesFalse从对象属性创建(ORM 互操作)
populate_by_nameFalse允许字段别名与原名都可用
json_encoders{}自定义 JSON 编码器
json_schema_extra{}额外 JSON Schema 信息

13.5 团队规范模板

13.5.1 pyproject.toml 模板

[tool.pydantic-mypy]
init_forbid_extra = true
init_typed = true
warn_required_dynamic_aliases = true
warn_untyped_fields = true

[tool.mypy]
plugins = ["pydantic.mypy"]
strict = true

13.5.2 团队代码规范

  1. 优先使用 Annotated 写法:Annotated[int, Field(gt=0)] 而非 int = Field(gt=0)。
  2. 值对象用 frozen=True, slots=True:保证不可变与内存高效。
  3. 领域模型用 BaseModel:享受校验与序列化。
  4. 配置用 pydantic-settings:统一管理环境变量。
  5. 避免深层嵌套:超过 3 层考虑拆分。
  6. 自定义类型集中管理:放在 types.py 或 schemas/types.py。
  7. 测试覆盖率 >= 90%:覆盖校验失败路径。
  8. CI 集成 pydantic.mypy 插件:静态检查 Pydantic 模型。

13.6 常见错误码对照

错误类型触发场景解决方案
ValidationError校验失败检查字段类型与约束
FrozenInstanceError修改 frozen 实例使用 object.__setattr__ 或去掉 frozen
PydanticUndefinedError访问未设置的字段检查 Field(default=...)
ConfigError配置错误检查 model_config
SchemaErrorschema 编译失败检查类型注解与 forward ref

13.7 迁移检查清单(v1 → v2)

  • .dict() → .model_dump()
  • .json() → .model_dump_json()
  • .parse_obj() → .model_validate()
  • .parse_raw() → .model_validate_json()
  • class Config: → model_config = ConfigDict(...)
  • @validator → @field_validator(加 @classmethod)
  • @root_validator → @model_validator
  • Optional[X] → X | None(推荐)
  • orm_mode=True → from_attributes=True
  • .copy() → .model_copy()
  • update_forward_refs() → model_rebuild()
  • Field(..., env="X") → 迁移到 pydantic-settings
  • 检查 strict 模式(v2 默认更严格)
  • 检查 JSON Schema 输出格式变化
  • 性能基准测试(应提升 5-50x)

BaseModel 基础

基本写法:定义模型 class <模型>(pydantic.BaseModel):\n <字段>: <类型>

# pydantic v2 模型定义
from pydantic import BaseModel

class User(BaseModel):
    id: int
    name: str
    email: str = ""

基本写法:从字典创建 <模型>(**<字典>) | <模型>.model_validate(<字典>)

# 从字典创建并验证
u = User(id=1, name="Alice")
u2 = User.model_validate({"id": 2, "name": "Bob"})

基本写法:转换为字典 <实例>.model_dump()

# 模型转字典
print(u.model_dump())

基本写法:转换为 JSON <实例>.model_dump_json()

# 模型转 JSON 字符串
print(u.model_dump_json())

基本写法:从 JSON 创建 <模型>.model_validate_json(<字符串>)

# 从 JSON 字符串创建
u = User.model_validate_json('{"id": 1, "name": "Alice"}')

字段验证

基本写法:Field 字段配置 <字段>: <类型> = pydantic.Field(...)

# 字段元数据
from pydantic import BaseModel, Field

class Item(BaseModel):
    name: str = Field(min_length=1, max_length=50)
    price: float = Field(gt=0, description="价格")
    qty: int = Field(default=0, ge=0)

基本写法:默认值与默认工厂 <字段>: <类型> = Field(default=<值>) | Field(default_factory=<函数>)

# 默认值
class Config(BaseModel):
    timeout: int = Field(default=30)
    tags: list = Field(default_factory=list)

基本写法:Optional 与可空 <字段>: <类型> | None = None

# 可空字段
class User(BaseModel):
    email: str | None = None

验证器

基本写法:field_validator @pydantic.field_validator(<字段>)

# 字段级验证器
from pydantic import BaseModel, field_validator

class User(BaseModel):
    name: str
    @field_validator("name")
    @classmethod
    def name_must_not_be_empty(cls, v):
        if not v.strip():
            raise ValueError("名称不能为空")
        return v

基本写法:model_validator 模型级 @pydantic.model_validator(mode=<模式>)

# 模型级验证
from pydantic import BaseModel, model_validator

class DateRange(BaseModel):
    start: int
    end: int
    @model_validator(mode="after")
    def check_range(self):
        if self.start > self.end:
            raise ValueError("起始大于结束")
        return self

基本写法:before 验证器 @field_validator(<字段>, mode="before")

# 在类型转换前验证
class Item(BaseModel):
    qty: int
    @field_validator("qty", mode="before")
    @classmethod
    def parse_qty(cls, v):
        if isinstance(v, str):
            return int(v)
        return v

类型注解

基本写法:约束类型 Annotated[<类型>, <约束>]

# 使用 Annotated 添加约束
from typing import Annotated
from pydantic import BaseModel, Field

PosInt = Annotated[int, Field(gt=0)]
class Model(BaseModel):
    n: PosInt

基本写法:Literal 枚举 <字段>: Literal[<值1>, <值2>]

# 字面值类型
from typing import Literal

class Config(BaseModel):
    mode: Literal["r", "w", "a"]

基本写法:EmailStr 邮箱 <字段>: pydantic.EmailStr

# 邮箱字段(需安装 email-validator)
from pydantic import BaseModel, EmailStr

class User(BaseModel):
    email: EmailStr

嵌套模型

基本写法:嵌套模型 <字段>: <另一个模型>

# 模型嵌套
class Address(BaseModel):
    city: str
    zip: str

class User(BaseModel):
    name: str
    addr: Address

u = User(name="Alice", addr={"city": "Shanghai", "zip": "200000"})

基本写法:列表模型 <字段>: list[<模型>]

# 模型列表
class Group(BaseModel):
    name: str
    users: list[User]

配置

基本写法:model_config model_config = pydantic.ConfigDict(...)

# 模型配置
class User(BaseModel):
    model_config = ConfigDict(
        str_strip_whitespace=True,
        frozen=True,
        extra="forbid",
    )
    name: str

基本写法:禁止额外字段 model_config = ConfigDict(extra="forbid")

# 拒绝未定义字段
class Strict(BaseModel):
    model_config = ConfigDict(extra="forbid")
    x: int

基本写法:str_strip_whitespace model_config = ConfigDict(str_strip_whitespace=True)

# 自动去除字符串空白
class User(BaseModel):
    model_config = ConfigDict(str_strip_whitespace=True)
    name: str

序列化

基本写法:自定义序列化 @pydantic.field_serializer(<字段>)

# 自定义字段序列化
from pydantic import BaseModel, field_serializer

class User(BaseModel):
    created_at: int
    @field_serializer("created_at")
    def ser_time(self, v):
        from datetime import datetime
        return datetime.fromtimestamp(v).isoformat()

基本写法:排除字段 <实例>.model_dump(exclude=<键集>)

# 序列化排除字段
print(u.model_dump(exclude={"email"}))

基本写法:include 包含 <实例>.model_dump(include=<键集>)

# 只包含指定字段
print(u.model_dump(include={"id", "name"}))

不可变模型

基本写法:frozen 不可变 model_config = ConfigDict(frozen=True)

# 不可变模型
class Config(BaseModel):
    model_config = ConfigDict(frozen=True)
    host: str

c = Config(host="localhost")
# c.host = "other"  # 抛出 ValidationError

类型转换

基本写法:严格模式 model_config = ConfigDict(strict=True)

# 严格模式,不自动转换类型
class M(BaseModel):
    model_config = ConfigDict(strict=True)
    x: int

# M(x="1")  # 抛出 ValidationError
M(x=1)

基本写法:自动转换 pydantic 默认行为

# 默认会自动转换兼容类型
class M(BaseModel):
    x: int

m = M(x="123")
print(m.x)

错误处理

基本写法:捕获 ValidationError except pydantic.ValidationError:

# 捕获验证错误
from pydantic import BaseModel, ValidationError

class User(BaseModel):
    id: int

try:
    User(id="abc")
except ValidationError as e:
    for err in e.errors():
        print(err["loc"], err["msg"])

基本写法:errors 错误列表 e.errors()

# 获取所有错误
for err in e.errors():
    print(err)