Pydantic多态模型:用鉴别器构建类型安全的API接口
title: Pydantic多态模型:用鉴别器构建类型安全的API接口
class Payment(BaseModel):
amount: float
currency: str = "USD"
class CreditCardPayment(Payment):
card_number: str
expiry_date: str
class AlipayPayment(Payment):
account_id: str
auth_code: str
# 反模式:使用条件判断路由类型
def process_payment(data: dict):
if "card_number" in data:
return CreditCardPayment(**data)
elif "account_id" in data:
return AlipayPayment(**data)
else:
raise ValueError("未知支付类型")
from pydantic import BaseModel, Field
class Animal(BaseModel):
type: str = Field(..., alias="_type", discriminator="animal_type")
class Dog(Animal):
animal_type: Literal["dog"] = "dog"
breed: str
class Cat(Animal):
animal_type: Literal["cat"] = "cat"
lives_left: int
# 自动解析示例
data = {"_type": "dog", "breed": "Golden Retriever"}
animal = Animal.parse_obj(data) # 自动实例化为Dog类型
from pydantic import create_model
vehicle_models = {
"car": create_model("Car", speed=(float, ...)),
"plane": create_model("Plane", altitude=(float, ...))
}
class Vehicle(BaseModel):
vehicle_type: str = Field(..., discriminator="vehicle_type")
__root__: Union[tuple(vehicle_models.values())] # 动态联合类型
class Product(BaseModel):
category: str = Field(..., discriminator="product_category")
class Book(Product):
product_category: Literal["book"] = "book"
author: str
pages: int
class EBook(Book):
format: str = Field(..., discriminator="file_format")
class PDF(EBook):
file_format: Literal["pdf"] = "pdf"
dpi: int
class EPUB(EBook):
file_format: Literal["epub"] = "epub"
reflowable: bool
from pydantic import validator
class Media(BaseModel):
media_type: str = Field(..., discriminator="media_kind")
content_type: str = Field(..., discriminator="mime_type")
class Video(Media):
media_kind: Literal["video"] = "video"
mime_type: Literal["video/mp4"] = "video/mp4"
resolution: str
# 自动处理双鉴别字段
data = {
"media_type": "video",
"mime_type": "video/mp4",
"resolution": "1080p"
}
media = Media.parse_obj(data) # 精确匹配Video类型
class ApiResponse(BaseModel):
status: Literal["success", "error"]
data: Union[UserResponse, ErrorResponse] = Field(...,
discriminator="response_type"
)
class UserResponse(BaseModel):
response_type: Literal["user"] = "user"
id: int
name: str
class ErrorResponse(BaseModel):
response_type: Literal["error"] = "error"
code: int
message: str
class KafkaMessage(BaseModel):
event_type: str = Field(..., discriminator="event_category")
timestamp: datetime = Field(default_factory=datetime.now)
class OrderCreated(KafkaMessage):
event_category: Literal["order_created"] = "order_created"
order_id: str
amount: float
class PaymentFailed(KafkaMessage):
event_category: Literal["payment_failed"] = "payment_failed"
error_code: int
retry_count: int
try:
Animal.parse_obj({"_type": "fish"})
except ValidationError as e:
print(e.json())
"""
[
{
"loc": ["_type"],
"msg": "No match for discriminator 'animal_type'
and value 'fish'",
"type": "value_error.discriminator.not_found"
}
]
"""
from pydantic import BaseModel, ConfigDict
class OptimizedModel(BaseModel):
model_config = ConfigDict(
from_attributes=True,
revalidate_instances="always"
)
__slots__ = ("__weakref__",) # 减少内存占用
错误信息 | 原因分析 | 解决方案 |
---|---|---|
discriminator.not_found | 未注册子模型类型 | 更新Union联合类型定义 |
value_error.union.invalid | 类型匹配顺序错误 | 调整Union类型顺序 |
validation_error.missing | 鉴别器字段缺失 | 添加必需鉴别字段 |
type_error.invalid_generic | 动态模型未正确注册 | 使用create_model显式创建 |
评论
发表评论