Pydantic根校验器：构建跨字段验证系统
title: Pydantic根校验器：构建跨字段验证系统
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
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
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
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"]}
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
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
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
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
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
| 错误信息 | 原因分析 | 解决方案 |
|---|---|---|
| ValidationError: 1 validation error | 根校验器未处理可选字段 | 添加字段存在性检查 |
| ValueError: 循环依赖检测 | 字段间相互依赖导致死循环 | 使用pre验证器拆分校验流程 |
| RuntimeError: 异步上下文错误 | 未正确管理异步资源 | 使用async with上下文管理器 |
| KeyError: 字段访问异常 | 未处理字段缺失情况 | 使用values.get()安全访问 |
评论
发表评论