Python dataclasses + Pydantic 实战：构建类型安全的 API 数据层

1
from dataclasses import dataclass
2
from typing import Optional
3

4
@dataclass
5
class UserProfile:
6
    name: str
7
    age: int
8
    email: Optional[str] = None
9

10
@dataclass
11
class UserResponse:
12
    user: UserProfile
13

14
def process_user_response(data: dict) -> str:
15
    # 需要手动从 dict 构造 dataclass
16
    profile_data = data["user"]["profile"]
17
    user = UserResponse(
18
        user=UserProfile(
19
            name=profile_data["name"],
20
            age=int(profile_data["age"]),  # 还是要手动转换
21
            email=profile_data.get("email"),
22
        )
23
    )
24
    return f"{user.user.name} ({user.user.age}) - {user.user.email or 'unknown'}"

1
from pydantic import BaseModel, Field, EmailStr
2
from typing import Optional
3

4
class UserProfile(BaseModel):
5
    name: str = Field(min_length=1, max_length=100)
6
    age: int = Field(ge=0, le=150)
7
    email: Optional[EmailStr] = None
8

9
class UserResponse(BaseModel):
10
    user: UserProfile
11

12
def process_user_response(data: dict) -> str:
13
    # 一行搞定：解析 + 验证 + 类型转换
14
    user = UserResponse.model_validate(data)
15
    return f"{user.user.name} ({user.user.age}) - {user.user.email or 'unknown'}"
16

17
# 验证示例
18
process_user_response({
19
    "user": {
20
        "profile": {  # 等等，字段名不匹配！
21
            "name": "张三",
22
            "age": "25",  # 字符串，会自动转 int
23
            "email": "not-an-email"  # 会报验证错误
24
        }
25
    }
26
})
27
# ValidationError: 1 validation error for UserResponse
28
# user.profile -> Field required

1
from pydantic import BaseModel, Field, field_validator, model_validator
2
from typing import Optional
3
from datetime import date
4
from enum import Enum
5

6
class WeatherCondition(str, Enum):
7
    SUNNY = "晴"
8
    CLOUDY = "多云"
9
    RAINY = "雨"
10
    SNOWY = "雪"
11

12
class ForecastDay(BaseModel):
13
    date: date
14
    high: int = Field(ge=-60, le=60, description="最高温度(℃)")
15
    low: int = Field(ge=-60, le=60, description="最低温度(℃)")
16
    condition: WeatherCondition
17

18
    @field_validator("high")
19
    @classmethod
20
    def high_must_be_ge_low(cls, v: int, info) -> int:
21
        # 注意：field_validator 只能访问当前字段
22
        # 跨字段验证需要用 model_validator
23
        return v
24

25
class WeatherResponse(BaseModel):
26
    city: str = Field(min_length=1, max_length=50)
27
    temperature: float = Field(ge=-60, le=60)
28
    humidity: int = Field(ge=0, le=100, description="湿度百分比")
29
    forecast: list[ForecastDay] = Field(min_length=1, max_length=15)
30
    aqi: Optional[int] = Field(default=None, ge=0, le=500)
31
    warning: Optional[str] = None
32

33
    @model_validator(mode="after")
34
    def validate_humidity_temperature_correlation(self):
35
        """跨字段验证：高温低湿是正常组合"""
36
        if self.temperature > 35 and self.humidity > 80:
37
            # 高温高湿 = 桑拿天，记录警告但不拒绝
38
            import warnings
39
            warnings.warn(
40
                f"高温高湿警告: {self.city} {self.temperature}℃/{self.humidity}%",
41
                UserWarning
42
            )
43
        return self
44

45
    @property
46
    def is_comfortable(self) -> bool:
47
        """计算体感舒适度（简单模型）"""
48
        return (
49
            18 <= self.temperature <= 28
50
            and 40 <= self.humidity <= 70
51
            and (self.aqi or 0) < 100
52
        )
53

54
    def model_post_init(self, __context):
55
        """模型初始化后的钩子"""
56
        # 可以在这里做日志、缓存等后置操作
57
        pass

1
import httpx
2

3
async def fetch_weather(city: str) -> WeatherResponse:
4
    """从 API 获取天气数据并自动验证"""
5
    async with httpx.AsyncClient() as client:
6
        resp = await client.get(
7
            f"https://api.weather.example.com/v1/{city}",
8
            headers={"Authorization": "Bearer YOUR_TOKEN"}
9
        )
10
        resp.raise_for_status()
11
        # 一行：dict → 验证 → 模型对象
12
        return WeatherResponse.model_validate(resp.json())
13

14
# 序列化回 JSON（给前端用）
15
weather = await fetch_weather("上海")
16
json_str = weather.model_dump_json(indent=2)
17

18
# 只输出部分字段（给移动端用）
19
lite = weather.model_dump(
20
    include={"city", "temperature", "forecast"}
21
)

1
from pydantic import BaseModel, ConfigDict, AliasPath, AliasChoices
2

3
class LegacyAPIResponse(BaseModel):
4
    model_config = ConfigDict(
5
        # 忽略多余字段（API 新增字段不会报错）
6
        extra="ignore",
7
        # 字段名不匹配时尝试多种别名
8
        populate_by_name=True,
9
    )
10

11
    # 使用 AliasPath 处理嵌套字段
12
    user_name: str = Field(
13
        validation_alias=AliasPath("data", "user_info", "username")
14
    )
15
    # 使用 AliasChoices 处理字段名变化（API 改版兼容）
16
    user_id: str = Field(
17
        validation_alias=AliasChoices("user_id", "userId", "uid")
18
    )
19
    # 旧字段已废弃，但兼容
20
    deprecated_field: Optional[str] = Field(
21
        default=None,
22
        validation_alias=AliasChoices("old_field", "legacy_field"),
23
        description="已废弃，请使用新字段"
24
    )

1
from pydantic import BaseModel
2

3
class User(BaseModel):
4
    id: int
5
    name: str
6

7
# ❌ 默认不可 hash，不能作为 dict key 或 set 元素
8
user = User(id=1, name="张三")
9
cache = {user: "cached_data"}  # TypeError: unhashable type
10

11
# ✅ 方案一： frozen=True 使其可 hash
12
class UserHashable(BaseModel):
13
    model_config = {"frozen": True}
14
    id: int
15
    name: str
16

17
# ✅ 方案二：用 id 作为 key
18
cache = {user.id: "cached_data"}

1
class Parent(BaseModel):
2
    name: str
3
    children: list[Child]  # Child 引用了 Parent
4

5
class Child(BaseModel):
6
    name: str
7
    parent: Parent  # 循环引用！
8

9
# model_validate 时会无限递归 → RecursionError
10
# 解决：使用 Optional + 延迟引用
11
from __future__ import annotations
12
from typing import Optional
13

14
class Parent(BaseModel):
15
    name: str
16
    children: list[Child] = []
17

18
class Child(BaseModel):
19
    name: str
20
    parent: Optional[Parent] = None  # 设为 Optional 打破循环