Pydantic根校验器：构建跨字段验证系统

扫描二维码关注或者微信搜一搜：编程智域前端至全栈交流与成长

第一章：根校验器基础

1.1 核心工作机制

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 多字段关联验证

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 分阶段验证流程

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 分布式事务验证

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 动态策略加载

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 递归结构验证

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 异步校验集成

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 校验顺序控制

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 校验结果缓存

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块

错误解决方案速查表

错误信息	原因分析	解决方案
ValidationError: 1 validation error	根校验器未处理可选字段	添加字段存在性检查
ValueError: 循环依赖检测	字段间相互依赖导致死循环	使用pre验证器拆分校验流程
RuntimeError: 异步上下文错误	未正确管理异步资源	使用async with上下文管理器
KeyError: 字段访问异常	未处理字段缺失情况	使用values.get()安全访问

架构原则：根校验器应遵循”单一出口”
原则，每个校验阶段只处理特定类型的验证逻辑。建议将复杂业务规则拆分为多个根校验器，通过pre参数控制执行顺序，构建可维护的验证管道。

余下文章内容请点击跳转至个人博客页面或者扫码关注或者微信搜一搜：编程智域前端至全栈交流与成长，阅读完整的文章：