第一章：根校验器基础
1.1 核心工作机制
PYTHON

from pydantic import BaseModel, root_validator


class OrderValidator(BaseModel):
    price: float
    quantity: int
    total: float

    @root_validator(pre=True)
    def calculate_total(cls, values):
        if "price" in values and "quantity" in values:
            values["total"] = values["price"] * values["quantity"]
        return values


# 自动计算总价
print(OrderValidator(price=9.99, quantity=3).total)  # 29.97

根校验器特性：

可访问所有字段值
支持pre/post两种模式
可修改整个values字典
支持多层级校验流程
第二章：复杂业务规则
2.1 多字段关联验证
PYTHON

class FlightBooking(BaseModel):
    departure: datetime
    arrival: datetime
    passengers: int

    @root_validator
    def check_flight_rules(cls, values):
        if values["arrival"] <= values["departure"]:
            raise ValueError("到达时间必须晚于出发时间")

        if values["passengers"] > 6:
            values["insurance"] = "required"
        return values

2.2 分阶段验证流程
PYTHON

class MultiStepForm(BaseModel):
    email: Optional[str]
    password: Optional[str]
    token: Optional[str]

    @root_validator(pre=True)
    def validate_stage1(cls, values):
        if not values.get("email"):
            raise ValueError("需要先完成邮箱验证")
        return values

    @root_validator(pre=True)
    def validate_stage2(cls, values):
        if "email" in values and not values.get("token"):
            raise ValueError("需要短信验证码")
        return values

第三章：企业级验证模式
3.1 分布式事务验证
PYTHON

class TransactionValidator(BaseModel):
    account_id: str
    amount: float
    currency: str

    @root_validator
    def check_balance(cls, values):
        # 调用外部微服务接口
        balance = get_account_balance(values["account_id"])
        if balance < values["amount"]:
            raise ValueError("账户余额不足")
        return values | {"new_balance": balance - values["amount"]}

3.2 动态策略加载
PYTHON

class StrategyValidator(BaseModel):
    config: dict
    data: dict

    @root_validator
    def load_validation_strategy(cls, values):
        strategy = values["config"].get("validation_strategy")
        if strategy == "strict":
            values["data"] = StrictPolicy().validate(values["data"])
        elif strategy == "relaxed":
            values["data"] = RelaxedPolicy().validate(values["data"])
        return values

第四章：高级验证技术
4.1 递归结构验证
PYTHON

class TreeNode(BaseModel):
    name: str
    children: list["TreeNode"]

    @root_validator
    def check_duplicates(cls, values):
        seen = set()

        def traverse(node):
            if node.name in seen:
                raise ValueError("发现重复节点")
            seen.add(node.name)
            for child in node.children:
                traverse(child)

        traverse(values["self"])
        return values

4.2 异步校验集成
PYTHON

import asyncio


class AsyncValidator(BaseModel):
    url: str

    @root_validator
    async def check_url_availability(cls, values):
        async with aiohttp.ClientSession() as session:
            async with session.head(values["url"]) as resp:
                if resp.status >= 400:
                    raise ValueError("资源不可用")
        return values

第五章：错误处理与优化
5.1 校验顺序控制
PYTHON

class OrderedValidation(BaseModel):
    phase: int
    status: str

    @root_validator(pre=True, skip_on_failure=True)
    def validate_phase1(cls, values):
        if values.get("phase") < 1:
            raise ValueError("初始阶段验证失败")
        return values

    @root_validator
    def validate_phase2(cls, values):
        if values["status"] == "error" and values["phase"] > 1:
            raise ValueError("阶段冲突")
        return values

5.2 校验结果缓存
PYTHON

class CachedValidator(BaseModel):
    _cache = {}

    @root_validator
    def cache_validation_result(cls, values):
        cache_key = hash(frozenset(values.items()))
        if cache_key in cls._cache:
            return cls._cache[cache_key]

        # 执行复杂校验逻辑
        processed = complex_validation(values)
        cls._cache[cache_key] = processed
        return processed

课后Quiz
Q1：pre-root校验器的执行时机是？
A) 在所有字段校验之后
B) 在字段校验之前
C) 仅在第一次校验时

Q2：处理异步验证的正确方式是？

使用async/await
创建新线程
调用外部服务
Q3：校验顺序控制的推荐方法是？

使用skip_on_failure参数
调整字段定义顺序
使用try/except块
错误解决方案速查表

来源：https://blog.cmdragon.cn/posts/60d359baeb6c/