Python Pydantic V2 数据验证实战：用 Rust 内核重新定义数据校验

1
# 正常创建
2
user = User(name="张三", age=28, email="zhangsan@example.com")
3
print(user.name)        # 张三
4
print(user.model_dump()) # 序列化为字典
5

6
# 自动类型转换
7
user2 = User(name="李四", age="25", email="lisi@example.com")
8
print(user2.age)         # 25 (int, 自动从字符串转换)
9
print(type(user2.age))   # <class 'int'>
10

11
# 验证失败
12
try:
13
    User(name="王五", age="not_a_number", email="invalid")
14
except Exception as e:
15
    print(e)
16
    # 2 validation errors for User
17
    # age: Input should be a valid integer ...
18
    # email: value is not a valid email address ...

1
from pydantic import BaseModel, ConfigDict
2

3
class StrictUser(BaseModel):
4
    model_config = ConfigDict(strict=True)
5

6
    name: str
7
    age: int
8

9
# 宽松模式下这没问题
10
user = User(name="张三", age="28")  # OK, "28" → 28
11

12
# 严格模式下直接报错
13
try:
14
    strict_user = StrictUser(name="张三", age="28")
15
except Exception as e:
16
    print(e)  # Input should be a valid integer

1
from pydantic import BaseModel, Field
2

3
class Product(BaseModel):
4
    name: str = Field(
5
        min_length=1,
6
        max_length=100,
7
        description="产品名称"
8
    )
9
    price: float = Field(
10
        gt=0,          # 大于 0
11
        le=999999.99,  # 小于等于 999999.99
12
        description="价格（元）"
13
    )
14
    quantity: int = Field(
15
        ge=0,          # 大于等于 0
16
        default=0,
17
        description="库存数量"
18
    )
19
    tags: list[str] = Field(
20
        default_factory=list,
21
        max_length=10,  # 列表最多 10 个元素
22
        description="产品标签"
23
    )
24
    sku: str = Field(
25
        pattern=r'^[A-Z]{2}-\d{6}$',  # 正则校验
26
        description="SKU 编码，格式如 AB-123456"
27
    )

1
from pydantic import BaseModel, Field
2
from datetime import datetime
3
from enum import Enum
4

5
class OrderStatus(str, Enum):
6
    PENDING = "pending"
7
    PAID = "paid"
8
    SHIPPED = "shipped"
9
    DELIVERED = "delivered"
10
    CANCELLED = "cancelled"
11

12
class Address(BaseModel):
13
    province: str
14
    city: str
15
    district: str
16
    detail: str
17
    zip_code: str = Field(pattern=r'^\d{6}$')
18

19
class OrderItem(BaseModel):
20
    product_id: int
21
    name: str
22
    price: float = Field(gt=0)
23
    quantity: int = Field(ge=1)
24

25
    @property
26
    def subtotal(self) -> float:
27
        return self.price * self.quantity
28

29
class Order(BaseModel):
30
    order_id: str
31
    customer_name: str
32
    shipping_address: Address            # 嵌套模型
33
    items: list[OrderItem]               # 嵌套模型列表
34
    status: OrderStatus = OrderStatus.PENDING  # 枚举
35
    created_at: datetime = Field(default_factory=datetime.now)
36
    note: str | None = None              # 可选字段（Python 3.10+ 语法）
37

38
    @field_validator('items')
39
    @classmethod
40
    def must_have_items(cls, v):
41
        if not v:
42
            raise ValueError('订单至少需要一个商品')
43
        return v
44

45
    @property
46
    def total(self) -> float:
47
        return sum(item.subtotal for item in self.items)

1
order_data = {
2
    "order_id": "ORD-20260510-001",
3
    "customer_name": "张三",
4
    "shipping_address": {
5
        "province": "浙江省",
6
        "city": "杭州市",
7
        "district": "西湖区",
8
        "detail": "文三路 100 号",
9
        "zip_code": "310000"
10
    },
11
    "items": [
12
        {"product_id": 1, "name": "机械键盘", "price": 599.0, "quantity": 1},
13
        {"product_id": 2, "name": "鼠标垫", "price": 49.9, "quantity": 2}
14
    ],
15
    "note": "请在工作日送达"
16
}
17

18
order = Order.model_validate(order_data)
19
print(f"订单总额: ¥{order.total:.2f}")   # ¥698.80
20
print(order.model_dump_json(indent=2))    # 格式化 JSON

1
from pydantic import BaseModel, Field, field_serializer
2
from datetime import datetime
3

4
class Article(BaseModel):
5
    title: str
6
    content: str
7
    author_id: int = Field(exclude=True)  # 序列化时排除
8
    created_at: datetime
9
    internal_score: float = Field(exclude=True)
10

11
    @field_serializer('created_at')
12
    def serialize_datetime(self, value: datetime, _info) -> str:
13
        return value.strftime('%Y年%m月%d日 %H:%M')
14

15
article = Article(
16
    title="Pydantic V2 实战",
17
    content="...",
18
    author_id=42,
19
    created_at=datetime(2026, 5, 10, 14, 30),
20
    internal_score=9.5
21
)
22

23
# 默认序列化（排除标记了 exclude 的字段）
24
print(article.model_dump())
25
# {'title': 'Pydantic V2 实战', 'content': '...', 'created_at': '2026年05月10日 14:30'}
26

27
# 按需包含/排除
28
print(article.model_dump(include={'title', 'created_at'}))
29
# {'title': 'Pydantic V2 实战', 'created_at': '2026年05月10日 14:30'}

1
from fastapi import FastAPI, HTTPException
2
from pydantic import BaseModel, Field, EmailStr
3
from datetime import datetime
4

5
app = FastAPI()
6

7
# 请求模型
8
class UserCreate(BaseModel):
9
    username: str = Field(min_length=3, max_length=20)
10
    email: EmailStr
11
    password: str = Field(min_length=8)
12

13
# 响应模型（不包含密码）
14
class UserResponse(BaseModel):
15
    id: int
16
    username: str
17
    email: str
18
    created_at: datetime
19

20
    model_config = ConfigDict(from_attributes=True)  # 支持从 ORM 对象创建
21

22
# 更新模型（所有字段可选）
23
class UserUpdate(BaseModel):
24
    username: str | None = None
25
    email: EmailStr | None = None
26

27
@app.post("/users", response_model=UserResponse)
28
async def create_user(user: UserCreate):
29
    # FastAPI 自动用 UserCreate 验证请求体
30
    # 如果验证失败，自动返回 422 + 详细错误信息
31
    db_user = fake_db.create(user.model_dump())
32
    return db_user
33

34
@app.patch("/users/{user_id}", response_model=UserResponse)
35
async def update_user(user_id: int, updates: UserUpdate):
36
    # exclude_unset=True 只更新用户实际传入的字段
37
    update_data = updates.model_dump(exclude_unset=True)
38
    if not update_data:
39
        raise HTTPException(400, "没有需要更新的字段")
40
    db_user = fake_db.update(user_id, update_data)
41
    return db_user

1
from pydantic import TypeAdapter
2

3
# 验证简单类型
4
int_adapter = TypeAdapter(int)
5
result = int_adapter.validate_python("42")  # 42
6

7
# 验证复杂类型
8
from typing import List
9
list_adapter = TypeAdapter(List[int])
10
result = list_adapter.validate_python(["1", "2", "3"])  # [1, 2, 3]
11

12
# 验证 Union 类型
13
from typing import Union
14
union_adapter = TypeAdapter(Union[int, str])
15
result = union_adapter.validate_python(42)    # 42
16
result = union_adapter.validate_python("hi")  # "hi"
17

18
# 生成 JSON Schema
19
schema = list_adapter.json_schema()
20
print(schema)  # {'items': {'type': 'integer'}, 'type': 'array'}

1
from pydantic import BaseModel, Field, field_validator, model_validator
2
from pydantic import ConfigDict
3
from pathlib import Path
4
import json
5

6
class DatabaseConfig(BaseModel):
7
    host: str = "localhost"
8
    port: int = Field(default=5432, ge=1, le=65535)
9
    name: str
10
    user: str
11
    password: str = Field(min_length=8)
12
    pool_size: int = Field(default=10, ge=1, le=100)
13
    ssl_mode: str = Field(default="prefer", pattern=r'^(disable|allow|prefer|require)$')
14

15
class RedisConfig(BaseModel):
16
    url: str = "redis://localhost:6379/0"
17
    max_connections: int = Field(default=20, ge=1)
18
    key_prefix: str = "app:"
19

20
class LoggingConfig(BaseModel):
21
    level: str = "INFO"
22
    format: str = "json"
23
    file: Path | None = None
24

25
    @field_validator('level')
26
    @classmethod
27
    def validate_level(cls, v):
28
        allowed = {'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'}
29
        v_upper = v.upper()
30
        if v_upper not in allowed:
31
            raise ValueError(f'日志级别必须是 {allowed} 之一')
32
        return v_upper
33

34
class AppConfig(BaseModel):
35
    model_config = ConfigDict(
36
        extra='forbid',  # 禁止未定义的字段，防止拼写错误
37
    )
38

39
    app_name: str
40
    debug: bool = False
41
    version: str = Field(pattern=r'^\d+\.\d+\.\d+$')
42
    database: DatabaseConfig
43
    redis: RedisConfig = RedisConfig()
44
    logging: LoggingConfig = LoggingConfig()
45

46
    @model_validator(mode='after')
47
    def production_checks(self):
48
        if not self.debug:
49
            if self.database.ssl_mode == 'disable':
50
                raise ValueError('生产环境必须启用数据库 SSL')
51
            if self.logging.level == 'DEBUG':
52
                raise ValueError('生产环境不允许 DEBUG 日志级别')
53
        return self
54

55
def load_config(path: str) -> AppConfig:
56
    """从 JSON 文件加载配置，自动验证"""
57
    with open(path) as f:
58
        data = json.load(f)
59
    return AppConfig.model_validate(data)
60

61
# 使用
62
config = load_config("config.json")
63
print(f"连接数据库: {config.database.host}:{config.database.port}/{config.database.name}")

1
import timeit
2
from pydantic import BaseModel
3

4
class SimpleModel(BaseModel):
5
    name: str
6
    age: int
7
    score: float
8

9
data = {"name": "test", "age": 25, "score": 95.5}
10

11
# 简单模型验证：通常在 1-3 微秒
12
result = timeit.timeit(
13
    lambda: SimpleModel.model_validate(data),
14
    number=100000
15
)
16
print(f"平均耗时: {result/100000*1_000_000:.1f} 微秒")