数据类与 Pydantic
dataclass与Pydantic模型
前置知识
- Python 与 Django:建议先完成前一篇的学习
学习目标
- 掌握「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 模块的核心设计目标:
- 标准库内置,无需第三方依赖。
- 类型注解驱动,字段通过
name: type声明。 - 可配置生成方法:通过
init、repr、eq、order、unsafe_hash、frozen参数控制。 - 不引入运行时类型检查,类型注解仅为 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 类型注解的数据验证库”。核心创新:
- 类型即校验:
name: str自动校验赋值是否为str,否则抛出ValidationError。 - JSON 原生支持:
parse_raw()、.json()直接处理 JSON 字符串。 - 嵌套模型:模型字段可以是另一个模型,递归校验。
- 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 正式发布,进行了几乎完全的重写:
- pydantic-core:用 Rust 实现的验证内核,性能提升 5-50 倍。
- API 重命名:
dict()→model_dump()、parse_obj()→model_validate()、@validator→field_validator。 ConfigDict替代内部Config类。Annotated优先:推荐使用Annotated[int, Field(ge=0)]而非int = Field(ge=0)。- 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 数据建模库的演进反映了三个哲学转向:
- 从手工到自动(1991-2017):消除样板代码,
__init__/__repr__自动生成。 - 从静态到运行时(2018-2022):类型注解从 hint 升级为可执行校验。
- 从 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)的形式化
一个数据类 可形式化为五元组:
其中:
- :字段集合(fields),每个字段 是三元组 。
- :生成方法集合(methods),如
__init__、__repr__、__eq__、__hash__。 - :验证器集合(validators),可为空(dataclass 默认无验证)。
- :序列化器集合(serializers),可为空。
- :配置(config),如
frozen、slots、kw_only。
2.2 类型验证的形式化
类型验证函数 接受值 与类型 ,返回验证结果:
关键性质:Pydantic 的 是强制转换(coercion)而非严格相等(strict equality)。例如:
而严格模式(strict=True)下:
2.3 nominal typing 与 structural typing
Nominal typing(名义类型):类型等价基于类型名称。 是 的子类型当且仅当 显式声明继承 。
Structural typing(结构类型):类型等价基于结构。 是 的子类型当且仅当 拥有 所有字段与方法(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:
不可变性带来的关键性质:
- 线程安全:无需锁即可在多线程间共享。
- 可哈希:可作为字典键、集合元素(
__hash__基于 计算)。 - 引用透明(referential transparency):相同输入始终产生相同输出,便于推理。
2.5 slots 的内存模型
slots=True 使 dataclass 使用 __slots__ 替代 __dict__。形式化地,普通类的实例内存布局:
slots 类的实例内存布局:
slots 省去了 dict 哈希表,每个字段访问从 哈希查找变为 偏移量访问。内存节省:
对于百万级实例,slots 可节省数百 MB 内存。
3. 理论推导与原理解析
3.1 @dataclass 的代码生成
@dataclass 装饰器本质是一个编译期(导入时)代码生成器。它读取类的 __annotations__,生成 __init__、__repr__、__eq__ 等方法并绑定到类。形式化地:
其中 __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.items.append(1) 会修改 b.items,造成隐蔽 bug。field(default_factory=list) 的修复:
3.3 Pydantic 验证的多阶段流水线
Pydantic v2 的验证过程分为多个阶段:
- 解析阶段:从 dict/JSON/ORM 对象提取原始数据。
- 类型转换阶段(coercion):将原始值转换为目标类型(
"42"→42)。 - 约束校验阶段:检查
Field(ge=0, le=150)等约束。 - 字段验证器阶段:运行
@field_validator装饰的函数。 - 模型验证器阶段:运行
@model_validator(mode='after')装饰的函数。 - 模型构建:实例化模型对象。
形式化地:
3.4 Pydantic v2 性能优势:Rust 内核
Pydantic v1 的验证循环用 Python 实现,每次字段访问都经过 Python 解释器。Pydantic v2 将验证逻辑编译为 Rust 函数(pydantic-core),仅边界处经过 Python-Rust FFI。
性能模型:
其中 ,,。对 10 字段模型:
实测 10 倍以上加速。
3.5 序列化的对称性
理想的序列化应满足往返一致性(round-trip consistency):
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
| 特性 | dataclass | attrs | Pydantic v2 | msgspec |
|---|---|---|---|---|
| 标准库 | 是 | 否 | 否 | 否 |
| 类型校验 | 否(仅 hint) | 可选(validator) | 是(强制) | 是(强制) |
| 性能(10字段实例化) | ~1μs | ~1.5μs | ~3μs | ~0.5μs |
| 序列化(JSON) | 需手动 | 需手动 | 内置(Rust) | 内置(C) |
| 嵌套模型 | 需手动 | 需手动 | 是 | 是 |
| 不可变 | frozen=True | @frozen | ConfigDict(frozen=True) | frozen=True |
| slots | slots=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 API | v2 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 | `X | None = None` |
Field(..., env=) | 移除(用 pydantic-settings) | 环境变量 |
.copy() | .model_copy() | 复制 |
update_forward_refs() | model_rebuild() | 前向引用 |
5.3 跨语言对比
| 语言 | 等价方案 | 特点 |
|---|---|---|
| Python | dataclasses / Pydantic | 类型注解驱动,运行时校验可选 |
| JavaScript/TypeScript | class-validator / zod / io-ts | 装饰器或运行时 schema |
| Go | struct + tags | struct tag 驱动,encoding/json 内置 |
| Rust | serde derive | 编译期生成,零运行时开销 |
| Java | Lombok / Record | 注解处理器,Record 是 Java 14+ 标准方案 |
| Swift | Codable | 协议驱动,编译期生成 |
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 的字段验证器执行顺序:
mode="before"验证器(在类型转换前)- 类型转换与约束校验
mode="after"验证器(在类型转换后)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-timedict[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
- Smith, E. V. (2017). PEP 557: Data Classes. Python Enhancement Proposals. https://peps.python.org/pep-0557/
- van Rossum, G., Lehtosalo, J., & Langa, Ł. (2014). PEP 484: Type Hints. Python Enhancement Proposals. https://peps.python.org/pep-0484/
- Levy, S. (2015). PEP 526: Syntax for Variable Annotations. Python Enhancement Proposals. https://peps.python.org/pep-0526/
- Bond, M. (2019). PEP 544: Protocols: Structural subtyping. Python Enhancement Proposals. https://peps.python.org/pep-0544/
11.2 官方文档
- Python Software Foundation. (2024). dataclasses — Data Classes. Python 3.12 Documentation. https://docs.python.org/3/library/dataclasses.html
- Pydantic Team. (2024). Pydantic v2 Documentation. https://docs.pydantic.dev/latest/
- Schlawack, H. (2024). attrs Documentation. https://www.attrs.org/
- Crist-Haruf, J. (2024). msgspec Documentation. https://jcristharif.com/msgspec/
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 工程实践
- Colvin, S. (2023). Pydantic V2: A Complete Rewrite. Pydantic Blog. https://pydantic.dev/articles/pydantic-v2-final
- Ramírez, S. (2023). FastAPI Documentation. https://fastapi.tiangolo.com/
- Schlawack, H. (2022). Why I Don’t Like Data Classes. Hynek Schlawack Blog. https://hynek.me/articles/why-i-dont-like-dataclasses/
11.5 性能基准
- Pydantic Team. (2024). Pydantic V2 Benchmarks. https://pydantic.dev/articles/pydantic-v2-performance
- Crist-Haruf, J. (2023). msgspec vs Pydantic vs dataclasses Benchmark. https://jcristharif.com/msgspec/benchmarks.html
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 参数速查
| 参数 | 默认值 | 说明 |
|---|---|---|
init | True | 生成 __init__ |
repr | True | 生成 __repr__ |
eq | True | 生成 __eq__ |
order | False | 生成 __lt__/__le__/__gt__/__ge__ |
unsafe_hash | False | 强制生成 __hash__(不安全) |
frozen | False | 不可变 |
slots | False(3.10+) | 使用 __slots__ |
kw_only | False(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 常用选项
| 选项 | 默认值 | 说明 |
|---|---|---|
strict | False | 严格模式(不强制转换) |
frozen | False | 不可变 |
extra | "ignore" | 处理多余字段:"ignore"/"allow"/"forbid" |
validate_assignment | False | 属性赋值时校验 |
use_enum_values | False | 序列化时用枚举值 |
str_strip_whitespace | False | 去除字符串首尾空格 |
str_to_lower | False | 字符串转小写 |
str_to_upper | False | 字符串转大写 |
from_attributes | False | 从对象属性创建(ORM 互操作) |
populate_by_name | False | 允许字段别名与原名都可用 |
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 团队代码规范
- 优先使用
Annotated写法:Annotated[int, Field(gt=0)]而非int = Field(gt=0)。 - 值对象用
frozen=True, slots=True:保证不可变与内存高效。 - 领域模型用
BaseModel:享受校验与序列化。 - 配置用
pydantic-settings:统一管理环境变量。 - 避免深层嵌套:超过 3 层考虑拆分。
- 自定义类型集中管理:放在
types.py或schemas/types.py。 - 测试覆盖率 >= 90%:覆盖校验失败路径。
- CI 集成
pydantic.mypy插件:静态检查 Pydantic 模型。
13.6 常见错误码对照
| 错误类型 | 触发场景 | 解决方案 |
|---|---|---|
ValidationError | 校验失败 | 检查字段类型与约束 |
FrozenInstanceError | 修改 frozen 实例 | 使用 object.__setattr__ 或去掉 frozen |
PydanticUndefinedError | 访问未设置的字段 | 检查 Field(default=...) |
ConfigError | 配置错误 | 检查 model_config |
SchemaError | schema 编译失败 | 检查类型注解与 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)