Pydantic Mixin：构建可组合的验证系统体系
title: Pydantic Mixin：构建可组合的验证系统体系
class TimestampMixin(BaseModel):
created_at: datetime = Field(default_factory=datetime.now)
updated_at: datetime = Field(default_factory=datetime.now)
class UserBase(BaseModel):
name: str
email: str
class UserWithTime(UserBase, TimestampMixin):
pass
user = UserWithTime(name="John", email="john@example.com")
print(user.created_at) # 自动生成时间戳
class PhoneValidationMixin(BaseModel):
@validator("phone")
def validate_phone_format(cls, v):
if not re.match(r"^\+?[1-9]\d{1,14}$", v):
raise ValueError("国际电话号码格式错误")
return v
class ContactForm(PhoneValidationMixin, BaseModel):
name: str
phone: str
class GeoValidationMixin(BaseModel):
@validator("latitude")
def validate_lat(cls, v):
if not -90 <= v <= 90:
raise ValueError("纬度值越界")
return v
class LocationModel(GeoValidationMixin, PhoneValidationMixin):
address: str
latitude: float
longitude: float
contact_phone: str
def create_dynamic_model(*mixins):
class DynamicModel(BaseModel):
class Config:
extra = "forbid"
for mixin in reversed(mixins):
DynamicModel = type(
f"{mixin.__name__}Model",
(mixin, DynamicModel),
{}
)
return DynamicModel
# 动态创建模型
SecurityModel = create_dynamic_model(TimestampMixin, PhoneValidationMixin)
from pydantic import BaseModel, validator
class PluginMixin(BaseModel):
@classmethod
def inject_validator(cls, field: str):
def decorator(func):
setattr(cls, f"validate_{field}", classmethod(func))
return func
return decorator
class ExtensibleModel(PluginMixin):
name: str
@ExtensibleModel.inject_validator("name")
def validate_name(cls, v):
if len(v) < 2:
raise ValueError("名称过短")
return v
class CoreValidationMixin(BaseModel):
@classmethod
def validate_all(cls, values):
values = super().validate_all(values)
if "prohibited_word" in str(values):
raise ValueError("包含禁用内容")
return values
class UserServiceModel(CoreValidationMixin, BaseModel):
username: str
content: str
class OrderServiceModel(CoreValidationMixin, BaseModel):
order_id: str
description: str
class TransactionMixin(BaseModel):
amount: float
@classmethod
def __get_validators__(cls):
yield cls.validate_transaction_chain
@classmethod
def validate_transaction_chain(cls, values):
if "previous_hash" in values and not verify_chain(values):
raise ValueError("交易链验证失败")
return values
class BitcoinTransaction(TransactionMixin):
wallet_address: str
previous_hash: Optional[str]
class ConflictMixinA(BaseModel):
@validator("id")
def validate_a(cls, v):
return v
class ConflictMixinB(BaseModel):
@validator("id")
def validate_b(cls, v):
return v
class ResolutionModel(ConflictMixinB, ConflictMixinA):
id: str
# 实际生效的校验器：ConflictMixinB.validate_b
class CachedValidationMixin(BaseModel):
_validator_cache = {}
@classmethod
def validate(cls, value):
cache_key = hash(frozenset(value.items()))
if cache_key in cls._validator_cache:
return cls._validator_cache[cache_key]
result = super().validate(value)
cls._validator_cache[cache_key] = result
return result
| 错误信息 | 原因分析 | 解决方案 |
|---|---|---|
| ValidationError: multiple validators | Mixin校验方法冲突 | 调整Mixin类继承顺序 |
| AttributeError: validator not found | 动态注入失效 | 检查元类注入逻辑 |
| ValueError: recursion detected | 循环校验依赖 | 使用@root_validator重构逻辑 |
| TypeError: invalid validator | 非类方法校验器 | 添加@classmethod装饰器 |
评论
发表评论