Pydantic综合指南

📂 所属阶段：第一阶段 — 快速筑基（基础篇）
🔗 相关章节：FastAPI简介与优势 · 环境搭建 · 请求体处理

目录

Pydantic概述

Pydantic是FastAPI的核心组件之一，提供基于类型提示的数据验证和设置管理。它使用Python的类型提示来验证数据，确保数据的类型安全和完整性。

Pydantic核心特性

  1. 类型安全：基于Python类型提示的自动数据验证
  2. 数据转换：自动类型转换和数据清洗
  3. 灵活约束：丰富的验证约束选项
  4. 性能优异：高效的验证算法
  5. 错误信息：详细的验证错误信息

安装与基本使用

# 安装Pydantic
pip install pydantic

# 基础使用示例
from pydantic import BaseModel
from typing import Optional

class User(BaseModel):
    name: str
    age: int
    email: Optional[str] = None

# 数据验证
user = User(name="John", age=30, email="john@example.com")
print(user.model_dump())  # {'name': 'John', 'age': 30, 'email': 'john@example.com'}

# 自动类型转换
user_int_age = User(name="Jane", age="25", email="jane@example.com")  # age会被转换为int
print(user_int_age.age, type(user_int_age.age))  # 25 <class 'int'>

FastAPI中的Pydantic

在FastAPI中，Pydantic主要用于：

  • 请求体验证：验证HTTP请求体数据
  • 查询参数验证：验证URL查询参数
  • 路径参数验证：验证URL路径参数
  • 响应模型：定义API响应数据结构

基础模型定义

简单模型

from pydantic import BaseModel
from typing import Optional

class Person(BaseModel):
    """基础人员模型"""
    name: str
    age: int
    email: Optional[str] = None

# 创建实例
person = Person(name="Alice", age=28, email="alice@example.com")
print(person.name)  # Alice
print(person.model_dump())  # {'name': 'Alice', 'age': 28, 'email': 'alice@example.com'}

字段类型

from pydantic import BaseModel
from typing import List, Dict, Optional, Union
from datetime import datetime
from decimal import Decimal
from uuid import UUID

class ComplexData(BaseModel):
    """复杂数据模型"""
    # 基础类型
    name: str
    age: int
    height: float
    is_student: bool
    
    # 可选类型
    nickname: Optional[str] = None
    
    # 集合类型
    hobbies: List[str] = []
    scores: Dict[str, float] = {}
    
    # 联合类型
    identifier: Union[int, str]
    
    # 特殊类型
    birth_date: datetime
    account_balance: Decimal
    user_id: UUID

# 示例数据
import uuid
from decimal import Decimal

data = {
    "name": "Bob",
    "age": 25,
    "height": 175.5,
    "is_student": True,
    "nickname": "Bobby",
    "hobbies": ["reading", "coding"],
    "scores": {"math": 95.5, "english": 88.0},
    "identifier": "user_123",
    "birth_date": "2000-01-01T00:00:00",
    "account_balance": Decimal("1234.56"),
    "user_id": str(uuid.uuid4())
}

complex_data = ComplexData(**data)
print(complex_data.model_dump())

默认值和工厂函数

from pydantic import BaseModel, Field
from typing import List
from datetime import datetime

class Document(BaseModel):
    """文档模型，演示默认值和工厂函数"""
    title: str
    content: str
    tags: List[str] = Field(default_factory=list)  # 工厂函数创建默认值
    created_at: datetime = Field(default_factory=datetime.utcnow)  # 时间戳
    views: int = 0  # 简单默认值
    metadata: dict = Field(default_factory=dict)  # 字典默认值

# 创建文档
doc = Document(title="Sample Document", content="This is a sample document.")
print(doc.created_at)  # 当前时间
print(doc.views)  # 0
print(doc.tags)  # []

字段验证与约束

使用Field进行字段约束

from pydantic import BaseModel, Field
from typing import Optional
import re

class Product(BaseModel):
    """产品模型，演示各种字段约束"""
    name: str = Field(
        ...,  # ... 表示必填字段
        min_length=3,
        max_length=100,
        description="产品名称，3-100字符"
    )
    description: Optional[str] = Field(
        None,
        max_length=500,
        description="产品描述，最多500字符"
    )
    price: float = Field(
        ...,
        gt=0,  # greater than 0
        le=10000,  # less than or equal to 10000
        description="产品价格，大于0，小于等于10000"
    )
    category: str = Field(
        ...,
        regex=r"^[a-zA-Z_][a-zA-Z0-9_]*$",  # 字母数字下划线，以字母或下划线开头
        description="产品分类，字母数字下划线格式"
    )
    stock: int = Field(
        default=0,
        ge=0,  # greater than or equal to 0
        description="库存数量，非负数"
    )
    rating: float = Field(
        default=0.0,
        ge=0,
        le=5.0,
        description="评分，0-5分"
    )

# 验证实例
try:
    product = Product(
        name="Laptop",
        description="High-performance laptop",
        price=1299.99,
        category="electronics",
        stock=10,
        rating=4.5
    )
    print(product.model_dump())
except Exception as e:
    print(f"验证错误: {e}")

集合类型验证

from pydantic import BaseModel, Field
from typing import List, Set, Tuple
from decimal import Decimal

class CollectionValidation(BaseModel):
    """集合类型验证示例"""
    # 列表验证
    tags: List[str] = Field(
        default=[],
        min_items=0,
        max_items=10,
        description="标签列表，最多10个"
    )
    
    # 集合验证
    unique_ids: Set[int] = Field(
        default=set(),
        description="唯一ID集合"
    )
    
    # 元组验证
    coordinates: Tuple[float, float] = Field(
        default=(0.0, 0.0),
        description="坐标点 (经度, 纬度)"
    )
    
    # 嵌套列表
    matrix: List[List[Decimal]] = Field(
        default=[],
        description="数值矩阵"
    )

# 使用示例
collection = CollectionValidation(
    tags=["tech", "gadgets", "review"],
    unique_ids={1, 2, 3, 4, 5},
    coordinates=(120.12345, 30.12345),
    matrix=[[Decimal("1.5"), Decimal("2.5")], [Decimal("3.5"), Decimal("4.5")]]
)
print(collection.model_dump())

正则表达式验证

from pydantic import BaseModel, Field
import re

class RegexValidation(BaseModel):
    """正则表达式验证示例"""
    # 邮箱验证
    email: str = Field(
        ...,
        regex=r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$",
        description="邮箱地址"
    )
    
    # 电话号码验证
    phone: str = Field(
        ...,
        regex=r"^\+?1?-?\.?\s?\(?(\d{3})\)?[\s\.-]?(\d{3})[\s\.-]?(\d{4})$",
        description="电话号码"
    )
    
    # 用户名验证
    username: str = Field(
        ...,
        regex=r"^[a-zA-Z0-9_]{3,20}$",  # 3-20字符，字母数字下划线
        description="用户名，3-20字符，字母数字下划线"
    )
    
    # 密码强度验证
    password: str = Field(
        ...,
        regex=r"^(?=.*[a-z])(?=.*[A-Z])(?=.*\d)(?=.*[@$!%*?&])[A-Za-z\d@$!%*?&]{8,}$",
        description="密码，至少8位，包含大小写字母、数字和特殊字符"
    )
    
    # URL验证
    website: str = Field(
        default=None,
        regex=r"^https?://(?:[-\w.])+(?:\:[0-9]+)?(?:/(?:[\w/_.])*(?:\?(?:[\w&=%.])*)?(?:\#(?:[\w.])*)?)?$",
        description="网站URL"
    )

# 验证示例
try:
    user_data = RegexValidation(
        email="user@example.com",
        phone="+1-234-567-8900",
        username="user123",
        password="MyPass123!"
    )
    print("验证成功:", user_data.model_dump())
except Exception as e:
    print(f"验证失败: {e}")

自定义验证器

单字段验证器

from pydantic import BaseModel, Field, validator
from typing import Optional
import re

class UserValidation(BaseModel):
    """用户模型，演示自定义验证器"""
    username: str
    email: str
    age: int
    password: str
    confirm_password: str
    
    @validator('username')
    def validate_username(cls, v):
        """验证用户名"""
        if not v:
            raise ValueError('用户名不能为空')
        if len(v) < 3:
            raise ValueError('用户名至少需要3个字符')
        if len(v) > 20:
            raise ValueError('用户名不能超过20个字符')
        if not re.match(r'^[a-zA-Z0-9_]+$', v):
            raise ValueError('用户名只能包含字母、数字和下划线')
        return v.lower()
    
    @validator('email')
    def validate_email(cls, v):
        """验证邮箱"""
        if not v:
            raise ValueError('邮箱不能为空')
        email_regex = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        if not re.match(email_regex, v):
            raise ValueError('邮箱格式不正确')
        return v.lower()
    
    @validator('age')
    def validate_age(cls, v):
        """验证年龄"""
        if v < 0:
            raise ValueError('年龄不能为负数')
        if v > 150:
            raise ValueError('年龄不能超过150岁')
        return v
    
    @validator('password')
    def validate_password(cls, v):
        """验证密码强度"""
        if len(v) < 8:
            raise ValueError('密码至少需要8个字符')
        if not re.search(r'[A-Z]', v):
            raise ValueError('密码必须包含至少一个大写字母')
        if not re.search(r'[a-z]', v):
            raise ValueError('密码必须包含至少一个小写字母')
        if not re.search(r'\d', v):
            raise ValueError('密码必须包含至少一个数字')
        if not re.search(r'[!@#$%^&*(),.?":{}|<>]', v):
            raise ValueError('密码必须包含至少一个特殊字符')
        return v
    
    @validator('confirm_password')
    def passwords_match(cls, v, values):
        """验证密码确认"""
        if 'password' in values and v != values['password']:
            raise ValueError('密码确认不匹配')
        return v

# 测试验证
try:
    user = UserValidation(
        username="john_doe",
        email="john@example.com",
        age=25,
        password="SecurePass123!",
        confirm_password="SecurePass123!"
    )
    print("用户验证成功:", user.model_dump())
except Exception as e:
    print(f"验证错误: {e}")

多字段验证器

from pydantic import BaseModel, Field, validator, root_validator
from datetime import date, datetime
from typing import Optional

class Booking(BaseModel):
    """预订模型，演示多字段验证"""
    customer_name: str
    check_in: date
    check_out: date
    room_type: str
    adults: int
    children: int = 0
    special_requests: Optional[str] = None
    
    @validator('room_type')
    def validate_room_type(cls, v):
        """验证房间类型"""
        allowed_types = ['single', 'double', 'suite', 'deluxe']
        if v.lower() not in allowed_types:
            raise ValueError(f'房间类型必须是 {allowed_types} 之一')
        return v.lower()
    
    @validator('adults')
    def validate_adults(cls, v):
        """验证成人数量"""
        if v < 1:
            raise ValueError('至少需要1名成人')
        if v > 10:
            raise ValueError('成人数量不能超过10人')
        return v
    
    @validator('children')
    def validate_children(cls, v):
        """验证儿童数量"""
        if v < 0:
            raise ValueError('儿童数量不能为负数')
        if v > 10:
            raise ValueError('儿童数量不能超过10人')
        return v
    
    @root_validator
    def validate_booking_logic(cls, values):
        """验证预订逻辑"""
        check_in = values.get('check_in')
        check_out = values.get('check_out')
        adults = values.get('adults')
        children = values.get('children')
        room_type = values.get('room_type')
        
        # 日期验证
        if check_in and check_out:
            if check_in >= check_out:
                raise ValueError('退房日期必须晚于入住日期')
            if check_in < date.today():
                raise ValueError('不能预订过去的日期')
            
            # 预订期限验证
            days = (check_out - check_in).days
            if days > 30:
                raise ValueError('单次预订不能超过30天')
        
        # 人数验证
        total_guests = adults + children
        if room_type and total_guests > 0:
            room_capacity = {
                'single': 2,
                'double': 4, 
                'suite': 6,
                'deluxe': 8
            }
            if total_guests > room_capacity.get(room_type, 2):
                raise ValueError(f'{room_type}房型最多容纳{room_capacity[room_type]}人')
        
        return values

# 测试预订验证
try:
    booking = Booking(
        customer_name="John Smith",
        check_in=date(2024, 6, 1),
        check_out=date(2024, 6, 5),
        room_type="double",
        adults=2,
        children=1
    )
    print("预订验证成功:", booking.model_dump())
except Exception as e:
    print(f"预订验证错误: {e}")

预验证和后验证

from pydantic import BaseModel, Field, validator
from typing import Optional

class PriceCalculator(BaseModel):
    """价格计算器，演示预验证和后验证"""
    base_price: float
    quantity: int
    discount_percent: float = 0.0
    tax_rate: float = 0.0
    shipping_cost: float = 0.0
    
    @validator('base_price', pre=True)
    def validate_base_price_pre(cls, v):
        """预验证：在类型转换前处理"""
        if isinstance(v, str):
            # 移除货币符号和空格
            v = v.replace('$', '').replace(',', '').strip()
        return v
    
    @validator('base_price')
    def validate_base_price(cls, v):
        """后验证：在类型转换后处理"""
        if v <= 0:
            raise ValueError('基础价格必须大于0')
        if v > 1000000:
            raise ValueError('基础价格不能超过1,000,000')
        return round(v, 2)  # 保留两位小数
    
    @validator('quantity')
    def validate_quantity(cls, v):
        """验证数量"""
        if v <= 0:
            raise ValueError('数量必须大于0')
        if v > 10000:
            raise ValueError('数量不能超过10,000')
        return v
    
    @validator('discount_percent')
    def validate_discount(cls, v):
        """验证折扣"""
        if v < 0:
            raise ValueError('折扣不能为负数')
        if v > 100:
            raise ValueError('折扣不能超过100%')
        return round(v, 2)
    
    @validator('tax_rate')
    def validate_tax_rate(cls, v):
        """验证税率"""
        if v < 0:
            raise ValueError('税率不能为负数')
        if v > 100:
            raise ValueError('税率不能超过100%')
        return round(v, 2)
    
    @validator('shipping_cost')
    def validate_shipping(cls, v):
        """验证运费"""
        if v < 0:
            raise ValueError('运费不能为负数')
        return round(v, 2)
    
    def calculate_total(self):
        """计算总价"""
        subtotal = self.base_price * self.quantity
        discount_amount = subtotal * (self.discount_percent / 100)
        discounted_subtotal = subtotal - discount_amount
        tax_amount = discounted_subtotal * (self.tax_rate / 100)
        total = discounted_subtotal + tax_amount + self.shipping_cost
        
        return {
            'subtotal': round(subtotal, 2),
            'discount_amount': round(discount_amount, 2),
            'discounted_subtotal': round(discounted_subtotal, 2),
            'tax_amount': round(tax_amount, 2),
            'shipping_cost': self.shipping_cost,
            'total': round(total, 2)
        }

# 测试价格计算器
try:
    calculator = PriceCalculator(
        base_price="$199.99",
        quantity=2,
        discount_percent=10.0,
        tax_rate=8.5,
        shipping_cost=15.99
    )
    
    result = calculator.calculate_total()
    print("价格计算结果:", result)
except Exception as e:
    print(f"验证错误: {e}")

嵌套模型与复杂结构

嵌套模型定义

from pydantic import BaseModel, Field
from typing import List, Optional
from datetime import datetime

class Address(BaseModel):
    """地址模型"""
    street: str = Field(..., min_length=5, max_length=100)
    city: str = Field(..., min_length=2, max_length=50)
    state: str = Field(..., min_length=2, max_length=50)
    country: str = Field(..., min_length=2, max_length=50)
    postal_code: str = Field(..., regex=r"^[0-9]{5}(-[0-9]{4})?$")

class ContactInfo(BaseModel):
    """联系信息模型"""
    email: str = Field(..., regex=r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$")
    phone: str = Field(..., regex=r"^\+?1?[-.\s]?\(?(\d{3})\)?[-.\s]?(\d{3})[-.\s]?(\d{4})$")
    emergency_contact: Optional[str] = None

class Employee(BaseModel):
    """员工模型，包含嵌套模型"""
    employee_id: int = Field(..., gt=0)
    first_name: str = Field(..., min_length=2, max_length=50)
    last_name: str = Field(..., min_length=2, max_length=50)
    position: str
    department: str
    hire_date: datetime
    salary: float = Field(..., gt=0)
    address: Address
    contact_info: ContactInfo
    skills: List[str] = Field(default=[], max_items=20)
    is_active: bool = True

# 创建嵌套模型实例
address = Address(
    street="123 Main St",
    city="Anytown",
    state="CA",
    country="USA",
    postal_code="12345"
)

contact_info = ContactInfo(
    email="john.doe@company.com",
    phone="+1-555-123-4567"
)

employee = Employee(
    employee_id=1001,
    first_name="John",
    last_name="Doe",
    position="Software Engineer",
    department="Engineering",
    hire_date=datetime(2023, 1, 15),
    salary=75000.00,
    address=address,
    contact_info=contact_info,
    skills=["Python", "FastAPI", "Docker", "AWS"]
)

print("员工信息:", employee.model_dump(mode='json'))

模型继承

from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime

class BaseModelExtended(BaseModel):
    """基础模型，包含通用字段"""
    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: datetime = Field(default_factory=datetime.utcnow)
    is_active: bool = True
    
    class Config:
        # 配置选项
        validate_assignment = True  # 赋值时验证
        use_enum_values = True     # 使用枚举值
        extra = "forbid"           # 禁止额外字段

class User(BaseModelExtended):
    """用户模型，继承基础模型"""
    user_id: int = Field(..., gt=0)
    username: str = Field(..., min_length=3, max_length=50, regex=r"^[a-zA-Z0-9_]+$")
    email: str = Field(..., regex=r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$")
    full_name: Optional[str] = Field(None, max_length=100)
    roles: List[str] = Field(default_factory=list)

class AdminUser(User):
    """管理员用户，继承普通用户"""
    permissions: List[str] = Field(default=["read", "write"])
    is_super_admin: bool = False
    
    @classmethod
    def create_admin(cls, **data):
        """创建管理员的便捷方法"""
        data.setdefault('permissions', ['read', 'write', 'delete', 'admin'])
        data.setdefault('is_super_admin', True)
        return cls(**data)

# 测试模型继承
user = User(
    user_id=1,
    username="johndoe",
    email="john@example.com",
    full_name="John Doe",
    roles=["user", "editor"]
)

admin = AdminUser.create_admin(
    user_id=2,
    username="admin",
    email="admin@example.com",
    full_name="Admin User"
)

print("普通用户:", user.model_dump())
print("管理员用户:", admin.model_dump())

泛型模型

from pydantic import BaseModel
from typing import TypeVar, Generic, List, Optional
from datetime import datetime

T = TypeVar('T')

class ApiResponse(Generic[T], BaseModel):
    """泛型API响应模型"""
    success: bool
    data: Optional[T] = None
    message: Optional[str] = None
    timestamp: datetime = datetime.utcnow()
    error_code: Optional[str] = None

class User(BaseModel):
    """用户模型"""
    id: int
    name: str
    email: str

class Product(BaseModel):
    """产品模型"""
    id: int
    name: str
    price: float

# 使用泛型模型
user_response = ApiResponse[User](
    success=True,
    data=User(id=1, name="John", email="john@example.com"),
    message="用户获取成功"
)

products_response = ApiResponse[List[Product]](
    success=True,
    data=[
        Product(id=1, name="Laptop", price=999.99),
        Product(id=2, name="Mouse", price=29.99)
    ],
    message="产品列表获取成功"
)

print("用户响应:", user_response.model_dump())
print("产品响应:", products_response.model_dump())

模型配置与元数据

模型配置选项

from pydantic import BaseModel, Field, validator
from typing import Optional
from datetime import datetime

class ConfiguredModel(BaseModel):
    """演示各种配置选项的模型"""
    id: int
    name: str = Field(..., alias="full_name")  # 字段别名
    email: str
    created_at: datetime = Field(default_factory=datetime.utcnow)
    
    class Config:
        # 字段别名配置
        allow_population_by_field_name = True  # 允许使用字段名或别名
        # 额外字段处理
        extra = "forbid"  # "allow", "ignore", "forbid"
        # 验证配置
        validate_all = True  # 验证所有字段，包括默认值
        # 赋值验证
        validate_assignment = True
        # 属性类型
        arbitrary_types_allowed = True  # 允许任意类型
        # JSON配置
        json_encoders = {
            datetime: lambda dt: dt.isoformat()
        }
        # Schema配置
        schema_extra = {
            "example": {
                "id": 1,
                "full_name": "John Doe",
                "email": "john@example.com"
            }
        }

# 测试配置
model1 = ConfiguredModel(
    id=1,
    full_name="John Doe",  # 使用别名
    email="john@example.com"
)

model2 = ConfiguredModel(
    id=2,
    name="Jane Doe",  # 使用字段名
    email="jane@example.com"
)

print("模型1:", model1.model_dump())
print("模型2:", model2.model_dump())

字段元数据

from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime

class MetadataModel(BaseModel):
    """演示字段元数据的模型"""
    # 基础元数据
    name: str = Field(
        ...,  # 必填
        title="姓名",
        description="用户的真实姓名",
        min_length=2,
        max_length=50,
        example="张三"
    )
    
    age: int = Field(
        ...,  # 必填
        title="年龄", 
        description="用户的年龄",
        ge=0,  # 大于等于0
        le=150,  # 小于等于150
        example=25
    )
    
    email: str = Field(
        default=None,  # 可选
        title="邮箱",
        description="用户的电子邮箱地址",
        regex=r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$",
        example="user@example.com"
    )
    
    # JSON Schema元数据
    score: float = Field(
        default=0.0,
        title="分数",
        description="用户的评分",
        ge=0.0,
        le=100.0,
        json_schema_extra={
            "unit": "points",
            "precision": 2
        }
    )
    
    # 自定义元数据
    preferences: dict = Field(
        default_factory=dict,
        title="偏好设置",
        description="用户的个性化偏好",
        json_schema_extra={
            "ui_widget": "preferences_editor",
            "category": "user_preferences"
        }
    )
    
    created_at: datetime = Field(
        default_factory=datetime.utcnow,
        title="创建时间",
        description="记录创建的UTC时间"
    )

# 获取模型schema
schema = MetadataModel.model_json_schema()
print("模型Schema:")
for field_name, field_info in schema.get('properties', {}).items():
    print(f"  {field_name}: {field_info}")

# 创建实例
instance = MetadataModel(
    name="李四",
    age=30,
    email="lisi@example.com",
    score=85.5,
    preferences={"theme": "dark", "language": "zh-CN"}
)

print("\n实例数据:", instance.model_dump())

自定义验证错误

from pydantic import BaseModel, Field, ValidationError
from pydantic.error_wrappers import ErrorWrapper
from typing import Optional
import re

class CustomErrorModel(BaseModel):
    """演示自定义验证错误的模型"""
    username: str = Field(
        ...,
        min_length=3,
        max_length=20,
        regex=r"^[a-zA-Z0-9_]+$"
    )
    password: str = Field(..., min_length=8)
    
    class Config:
        error_msg_templates = {
            'value_error.any_str.min_length': '字段值太短，最小长度为 {limit_value} 个字符',
            'value_error.any_str.max_length': '字段值太长，最大长度为 {limit_value} 个字符', 
            'value_error.str.regex': '字段格式不符合要求',
        }

class BusinessRuleModel(BaseModel):
    """业务规则验证模型"""
    account_type: str
    balance: float
    credit_limit: float = 0.0
    
    @classmethod
    def __modify_schema__(cls, field_schema):
        """自定义schema"""
        field_schema['title'] = 'Business Account'
        field_schema['description'] = 'Business account with validation rules'
    
    @validator('balance')
    def validate_balance(cls, v, values):
        """验证余额"""
        account_type = values.get('account_type')
        
        if account_type == 'checking' and v < 0:
            raise ValueError('支票账户余额不能为负数')
        elif account_type == 'savings' and v < 10:
            raise ValueError('储蓄账户余额不能低于10元')
        
        return v
    
    @validator('credit_limit')
    def validate_credit_limit(cls, v, values):
        """验证信用额度"""
        account_type = values.get('account_type')
        balance = values.get('balance', 0)
        
        if account_type == 'credit' and v <= 0:
            raise ValueError('信用卡账户必须设置信用额度')
        elif account_type == 'credit' and balance > v:
            raise ValueError('信用卡余额不能超过信用额度')
        
        return v

# 测试自定义错误
try:
    invalid_user = CustomErrorModel(username="ab", password="1234567")  # 用户名太短，密码太短
except ValidationError as e:
    print("自定义错误信息:")
    for error in e.errors():
        print(f"  位置: {error['loc']}, 信息: {error['msg']}")

try:
    checking_account = BusinessRuleModel(
        account_type='checking',
        balance=-100,  # 支票账户余额为负数
        credit_limit=0
    )
except ValidationError as e:
    print("\n业务规则错误:")
    for error in e.errors():
        print(f"  位置: {error['loc']}, 信息: {error['msg']}")

数据转换与序列化

数据转换

from pydantic import BaseModel, Field, validator
from typing import Optional, Union
from datetime import datetime, date
from decimal import Decimal
import json

class DataConversionModel(BaseModel):
    """演示数据转换的模型"""
    # 字符串到数字转换
    numeric_str: Union[int, float] = Field(..., description="数字字符串自动转换")
    
    # 日期时间转换
    date_str: date = Field(..., description="日期字符串转换")
    datetime_str: datetime = Field(..., description="日期时间字符串转换")
    
    # 布尔值转换
    bool_str: bool = Field(..., description="布尔值字符串转换")
    
    # 数值转换
    decimal_value: Decimal = Field(..., description="浮点数转Decimal")
    
    @validator('numeric_str', pre=True)
    def convert_numeric(cls, v):
        """预转换数值"""
        if isinstance(v, str):
            try:
                if '.' in v:
                    return float(v)
                else:
                    return int(v)
            except ValueError:
                raise ValueError(f'无法转换为数值: {v}')
        return v
    
    @validator('date_str', pre=True)
    def convert_date(cls, v):
        """预转换日期"""
        if isinstance(v, str):
            return datetime.fromisoformat(v.replace('Z', '+00:00')).date()
        return v
    
    @validator('datetime_str', pre=True)
    def convert_datetime(cls, v):
        """预转换日期时间"""
        if isinstance(v, str):
            return datetime.fromisoformat(v.replace('Z', '+00:00'))
        return v
    
    @validator('bool_str', pre=True)
    def convert_bool(cls, v):
        """预转换布尔值"""
        if isinstance(v, str):
            return v.lower() in ('true', '1', 'yes', 'on')
        return bool(v)
    
    @validator('decimal_value', pre=True)
    def convert_decimal(cls, v):
        """预转换Decimal"""
        if not isinstance(v, Decimal):
            return Decimal(str(v))
        return v

# 测试数据转换
conversion_data = {
    "numeric_str": "123.45",  # 字符串转float
    "date_str": "2024-01-15",  # 字符串转date
    "datetime_str": "2024-01-15T10:30:00",  # 字符串转datetime
    "bool_str": "true",  # 字符串转bool
    "decimal_value": 99.99  # 数值转Decimal
}

converted = DataConversionModel(**conversion_data)
print("转换结果:", converted.model_dump())

# 验证转换后的类型
print(f"numeric_str类型: {type(converted.numeric_str)}")
print(f"date_str类型: {type(converted.date_str)}")
print(f"datetime_str类型: {type(converted.datetime_str)}")
print(f"bool_str类型: {type(converted.bool_str)}")
print(f"decimal_value类型: {type(converted.decimal_value)}")

序列化与反序列化

from pydantic import BaseModel, Field
from typing import List, Optional, Dict
from datetime import datetime
import json

class SerializationModel(BaseModel):
    """演示序列化的模型"""
    id: int
    name: str
    metadata: Dict[str, str] = Field(default_factory=dict)
    tags: List[str] = Field(default_factory=list)
    created_at: datetime = Field(default_factory=datetime.utcnow)
    
    class Config:
        # JSON序列化配置
        json_encoders = {
            datetime: lambda dt: dt.isoformat(),
        }
        # Schema配置
        schema_extra = {
            "example": {
                "id": 1,
                "name": "Example Item",
                "metadata": {"key": "value"},
                "tags": ["example", "test"],
                "created_at": "2024-01-15T10:30:00"
            }
        }

# 创建模型实例
model_instance = SerializationModel(
    id=1,
    name="Test Item",
    metadata={"category": "test", "priority": "high"},
    tags=["important", "urgent"]
)

# 序列化为字典
dict_data = model_instance.model_dump()
print("序列化为字典:", dict_data)

# 序列化为JSON字符串
json_data = model_instance.model_dump_json()
print("序列化为JSON:", json_data)

# 从字典反序列化
reconstructed_from_dict = SerializationModel.model_validate(dict_data)
print("从字典重建:", reconstructed_from_dict.model_dump())

# 从JSON字符串反序列化
reconstructed_from_json = SerializationModel.model_validate_json(json_data)
print("从JSON重建:", reconstructed_from_json.model_dump())

# 比较原始和重建的实例
print("原始实例 == 从字典重建:", model_instance == reconstructed_from_dict)
print("原始实例 == 从JSON重建:", model_instance == reconstructed_from_json)

自定义序列化

from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime

class CustomSerializationModel(BaseModel):
    """自定义序列化的模型"""
    id: int
    name: str
    sensitive_data: str = Field(repr=False)  # 不在repr中显示
    internal_flag: bool = Field(default=False, exclude=True)  # 序列化时排除
    created_at: datetime = Field(default_factory=datetime.utcnow)
    
    class Config:
        # 自定义JSON编码器
        json_encoders = {
            datetime: lambda dt: dt.strftime('%Y-%m-%d %H:%M:%S'),
        }
    
    def model_dump(
        self,
        *,
        mode='python',
        include=None,
        exclude=None,
        context=None,
        by_alias: bool = False,
        exclude_unset: bool = False,
        exclude_defaults: bool = False,
        exclude_none: bool = False,
        round_trip: bool = False,
        warnings: bool = True,
    ):
        """自定义dump方法，添加额外处理"""
        # 调用父类方法
        data = super().model_dump(
            mode=mode,
            include=include,
            exclude=exclude,
            context=context,
            by_alias=by_alias,
            exclude_unset=exclude_unset,
            exclude_defaults=exclude_defaults,
            exclude_none=exclude_none,
            round_trip=round_trip,
            warnings=warnings
        )
        
        # 添加自定义处理
        if 'created_at' in data:
            # 格式化时间戳
            if isinstance(data['created_at'], datetime):
                data['timestamp'] = int(data['created_at'].timestamp())
        
        return data
    
    @classmethod
    def serialize_for_api(cls, instance, include_sensitive=False):
        """为API定制的序列化方法"""
        exclude_fields = set()
        if not include_sensitive:
            exclude_fields.add('sensitive_data')
        
        return instance.model_dump(exclude=exclude_fields)

# 测试自定义序列化
custom_model = CustomSerializationModel(
    id=1,
    name="Custom Item",
    sensitive_data="secret information",
    internal_flag=True
)

print("标准序列化:", custom_model.model_dump())
print("包含敏感数据:", CustomSerializationModel.serialize_for_api(custom_model, include_sensitive=True))
print("不包含敏感数据:", CustomSerializationModel.serialize_for_api(custom_model, include_sensitive=False))

# repr显示（不会显示sensitive_data）
print("模型repr:", repr(custom_model))

性能优化技巧

缓存验证结果

from pydantic import BaseModel, Field
from functools import lru_cache
import hashlib

class PerformanceOptimizedModel(BaseModel):
    """性能优化的模型"""
    category: str = Field(..., description="分类，使用缓存验证")
    name: str = Field(..., min_length=1, max_length=100)
    value: float = Field(..., gt=0)
    
    # 预编译的验证逻辑（使用缓存）
    @classmethod
    @lru_cache(maxsize=128)
    def validate_category_cached(cls, category: str) -> bool:
        """缓存分类验证结果"""
        allowed_categories = frozenset([
            'electronics', 'books', 'clothing', 'home', 'sports',
            'beauty', 'toys', 'automotive', 'garden', 'tools'
        ])
        return category.lower() in allowed_categories
    
    @Field_validator('category')  # 注意：这里应该是validator装饰器
    def validate_category(cls, v):
        """使用缓存验证分类"""
        if not cls.validate_category_cached(v.lower()):
            raise ValueError(f'不允许的分类: {v}')
        return v.lower()

# 修复上面的代码错误，使用正确的装饰器
from pydantic import field_validator

class PerformanceOptimizedModel(BaseModel):
    """性能优化的模型"""
    category: str = Field(..., description="分类，使用缓存验证")
    name: str = Field(..., min_length=1, max_length=100)
    value: float = Field(..., gt=0)
    
    # 预编译的验证逻辑（使用缓存）
    @classmethod
    @lru_cache(maxsize=128)
    def validate_category_cached(cls, category: str) -> bool:
        """缓存分类验证结果"""
        allowed_categories = frozenset([
            'electronics', 'books', 'clothing', 'home', 'sports',
            'beauty', 'toys', 'automotive', 'garden', 'tools'
        ])
        return category.lower() in allowed_categories
    
    @field_validator('category')
    def validate_category(cls, v):
        """使用缓存验证分类"""
        if not cls.validate_category_cached(v.lower()):
            raise ValueError(f'不允许的分类: {v}')
        return v.lower()

# 性能测试辅助函数
import time
from typing import List

def performance_test(model_class, test_data_list: List[dict], iterations: int = 1000):
    """性能测试函数"""
    start_time = time.time()
    
    for _ in range(iterations):
        for data in test_data_list:
            try:
                instance = model_class(**data)
                # 访问一个属性以确保完全初始化
                _ = instance.name
            except Exception:
                pass  # 忽略验证错误，只测试性能
    
    end_time = time.time()
    avg_time = (end_time - start_time) / iterations
    print(f"平均每次验证耗时: {avg_time:.6f}秒")
    print(f"总耗时: {end_time - start_time:.4f}秒")
    return avg_time

# 准备测试数据
test_data = [
    {"category": "electronics", "name": "Laptop", "value": 999.99},
    {"category": "books", "name": "Python Guide", "value": 29.99},
    {"category": "clothing", "name": "T-Shirt", "value": 19.99},
] * 10  # 重复数据以增加测试量

print("性能测试结果:")
performance_test(PerformanceOptimizedModel, test_data, 100)

批量验证优化

from pydantic import BaseModel, ValidationError
from typing import List
import time

class BatchValidationModel(BaseModel):
    """用于批量验证的模型"""
    id: int
    name: str
    email: str
    age: int

def batch_validate_sequential(models: List[dict]) -> List[BatchValidationModel]:
    """顺序验证（较慢）"""
    results = []
    errors = []
    
    for i, model_data in enumerate(models):
        try:
            validated = BatchValidationModel(**model_data)
            results.append(validated)
        except ValidationError as e:
            errors.append((i, e))
    
    return results

def batch_validate_with_error_collection(models: List[dict]) -> tuple[List[BatchValidationModel], List[tuple[int, ValidationError]]]:
    """批量验证，收集所有错误"""
    results = []
    errors = []
    
    for i, model_data in enumerate(models):
        try:
            validated = BatchValidationModel(**model_data)
            results.append(validated)
        except ValidationError as e:
            errors.append((i, e))
    
    return results, errors

# 生成测试数据
def generate_test_data(count: int) -> List[dict]:
    """生成测试数据"""
    import random
    import string
    
    data = []
    for i in range(count):
        data.append({
            "id": i + 1,
            "name": f"User{i}",
            "email": f"user{i}@example.com",
            "age": random.randint(18, 80)
        })
    
    # 添加一些无效数据用于测试错误处理
    data.append({
        "id": count + 1,
        "name": "",  # 无效：空名称
        "email": "invalid-email",
        "age": -5  # 无效：负年龄
    })
    
    return data

# 性能比较
test_data = generate_test_data(100)

print("批量验证性能比较:")

# 测试顺序验证
start_time = time.time()
valid_models, validation_errors = batch_validate_with_error_collection(test_data)
sequential_time = time.time() - start_time

print(f"顺序验证耗时: {sequential_time:.4f}秒")
print(f"有效模型数量: {len(valid_models)}")
print(f"验证错误数量: {len(validation_errors)}")

验证器优化

import re
from pydantic import BaseModel, Field, field_validator
from typing import Optional

# 预编译正则表达式
EMAIL_REGEX = re.compile(r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$")
PHONE_REGEX = re.compile(r"^\+?1?[-.\s]?\(?(\d{3})\)?[-.\s]?(\d{3})[-.\s]?(\d{4})$")

class OptimizedValidatorModel(BaseModel):
    """优化验证器的模型"""
    email: str = Field(..., description="邮箱地址")
    phone: Optional[str] = Field(None, description="电话号码")
    username: str = Field(..., description="用户名")
    
    @field_validator('email')
    def validate_email_optimized(cls, v):
        """优化的邮箱验证"""
        if not EMAIL_REGEX.match(v):
            raise ValueError('无效的邮箱格式')
        return v.lower()
    
    @field_validator('phone', mode='before')  # 使用pre验证模式
    def validate_phone_optimized(cls, v):
        """优化的电话验证"""
        if v is None:
            return v
        # 快速长度检查
        if len(v) < 10 or len(v) > 20:
            raise ValueError('电话号码长度无效')
        # 正则验证
        if not PHONE_REGEX.match(v):
            raise ValueError('无效的电话号码格式')
        return v
    
    @field_validator('username')
    def validate_username_optimized(cls, v):
        """优化的用户名验证"""
        # 快速检查
        if not v or len(v) < 3 or len(v) > 20:
            raise ValueError('用户名长度必须在3-20字符之间')
        
        # 检查是否只包含允许的字符
        if not v.replace('_', '').replace('.', '').isalnum():
            raise ValueError('用户名只能包含字母、数字、点和下划线')
        
        return v.lower()

# 测试优化后的验证器
try:
    optimized_instance = OptimizedValidatorModel(
        email="test@example.com",
        phone="+1-555-123-4567",
        username="test.user_123"
    )
    print("优化验证器测试成功:", optimized_instance.model_dump())
except Exception as e:
    print(f"优化验证器测试失败: {e}")

错误处理与调试

详细错误信息处理

from pydantic import BaseModel, Field, ValidationError
from typing import Optional
import traceback

class ErrorHandlingModel(BaseModel):
    """用于错误处理演示的模型"""
    name: str = Field(..., min_length=2, max_length=50)
    age: int = Field(..., ge=0, le=150)
    email: Optional[str] = Field(None, regex=r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$")

def handle_validation_errors(data: dict):
    """处理验证错误的函数"""
    try:
        instance = ErrorHandlingModel(**data)
        return instance, None
    except ValidationError as e:
        # 详细错误分析
        errors = []
        for error in e.errors():
            error_info = {
                'location': ' -> '.join(str(loc) for loc in error['loc']),
                'message': error['msg'],
                'type': error['type'],
                'input': error.get('input'),
                'ctx': error.get('ctx', {})
            }
            errors.append(error_info)
        
        return None, errors

# 测试错误处理
test_invalid_data = {
    "name": "A",  # 太短
    "age": 200,   # 超出范围
    "email": "invalid-email"  # 格式错误
}

instance, errors = handle_validation_errors(test_invalid_data)

if errors:
    print("验证错误详情:")
    for i, error in enumerate(errors, 1):
        print(f"{i}. 位置: {error['location']}")
        print(f"   信息: {error['message']}")
        print(f"   类型: {error['type']}")
        print(f"   输入值: {error['input']}")
        print()

调试工具

from pydantic import BaseModel, Field, field_validator
from typing import Optional
import logging
from contextlib import contextmanager

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class DebuggingModel(BaseModel):
    """用于调试的模型"""
    name: str = Field(..., min_length=2)
    value: float = Field(..., gt=0)
    
    @field_validator('name')
    def debug_name_validation(cls, v):
        """带调试信息的验证器"""
        logger.info(f"验证名称: {v}")
        if len(v) < 2:
            logger.warning(f"名称太短: {v}")
            raise ValueError('名称太短')
        logger.info(f"名称验证通过: {v}")
        return v
    
    @field_validator('value')
    def debug_value_validation(cls, v):
        """带调试信息的值验证"""
        logger.info(f"验证值: {v}")
        if v <= 0:
            logger.error(f"值无效: {v}")
            raise ValueError('值必须大于0')
        logger.info(f"值验证通过: {v}")
        return v
    
    def __init__(self, **data):
        logger.info(f"创建模型实例，输入数据: {data}")
        super().__init__(**data)
        logger.info(f"模型实例创建成功: {self}")

# 测试调试功能
try:
    debug_instance = DebuggingModel(name="Test", value=10.5)
    print("调试模型创建成功")
except Exception as e:
    print(f"调试模型创建失败: {e}")

# 自定义调试上下文管理器
@contextmanager
def validation_debug(model_class, data):
    """验证调试上下文管理器"""
    logger.info(f"开始验证 {model_class.__name__} 模型")
    logger.info(f"输入数据: {data}")
    
    try:
        instance = model_class(**data)
        logger.info(f"验证成功: {instance}")
        yield instance
    except Exception as e:
        logger.error(f"验证失败: {e}")
        logger.error(f"错误类型: {type(e).__name__}")
        logger.error(f"完整错误: {traceback.format_exc()}")
        raise
    finally:
        logger.info(f"验证结束 {model_class.__name__}")

# 使用调试上下文
test_data = {"name": "ValidName", "value": 100.0}
with validation_debug(DebuggingModel, test_data) as instance:
    print(f"在上下文中: {instance}")

高级验证模式

条件验证

from pydantic import BaseModel, Field, field_validator, model_validator
from typing import Optional, Literal
from datetime import date

class ConditionalValidationModel(BaseModel):
    """条件验证模型"""
    user_type: Literal['admin', 'regular', 'guest']
    age: Optional[int] = None
    license_number: Optional[str] = None
    admin_level: Optional[int] = None
    membership_valid_until: Optional[date] = None
    
    @model_validator(mode='after')
    def conditional_validation(self):
        """条件验证逻辑"""
        if self.user_type == 'admin':
            if self.admin_level is None:
                raise ValueError('管理员必须指定级别')
            if self.admin_level < 1 or self.admin_level > 5:
                raise ValueError('管理员级别必须在1-5之间')
        
        if self.user_type == 'regular':
            if self.age is None:
                raise ValueError('普通用户必须提供年龄')
            if self.age < 13:
                raise ValueError('普通用户年龄必须至少13岁')
        
        if self.user_type == 'guest':
            if self.membership_valid_until is not None:
                raise ValueError('访客用户不能有会员到期时间')
        
        # 驾驶员许可证验证
        if self.age is not None and self.age >= 16:
            if self.license_number is not None:
                # 验证驾照号码格式
                if not self.license_number.startswith('DL') or len(self.license_number) != 10:
                    raise ValueError('驾照号码格式无效（应为DL+8位数字）')
        
        return self

# 测试条件验证
try:
    admin_user = ConditionalValidationModel(
        user_type='admin',
        admin_level=3
    )
    print("管理员用户创建成功:", admin_user.model_dump())
except Exception as e:
    print(f"管理员用户创建失败: {e}")

try:
    regular_user = ConditionalValidationModel(
        user_type='regular',
        age=25,
        license_number='DL12345678'
    )
    print("普通用户创建成功:", regular_user.model_dump())
except Exception as e:
    print(f"普通用户创建失败: {e}")

try:
    guest_user = ConditionalValidationModel(
        user_type='guest'
    )
    print("访客用户创建成功:", guest_user.model_dump())
except Exception as e:
    print(f"访客用户创建失败: {e}")

动态验证

from pydantic import BaseModel, Field, field_validator
from typing import Any, Dict, Optional
import inspect

class DynamicValidationModel(BaseModel):
    """动态验证模型"""
    config: Dict[str, Any] = Field(default_factory=dict)
    
    @field_validator('config')
    def validate_dynamic_config(cls, v):
        """动态配置验证"""
        if not isinstance(v, dict):
            raise ValueError('配置必须是字典格式')
        
        # 定义验证规则
        validation_rules = {
            'database_url': {
                'required': True,
                'type': str,
                'pattern': r'^[a-z]+://.*'
            },
            'max_connections': {
                'required': False,
                'type': int,
                'min': 1,
                'max': 100
            },
            'debug_mode': {
                'required': False,
                'type': bool
            },
            'allowed_hosts': {
                'required': False,
                'type': list,
                'item_type': str
            }
        }
        
        errors = []
        
        for field_name, rules in validation_rules.items():
            value = v.get(field_name)
            
            # 检查必需字段
            if rules.get('required', False) and value is None:
                errors.append(f'缺少必需字段: {field_name}')
                continue
            
            if value is not None:
                # 类型检查
                expected_type = rules.get('type')
                if expected_type and not isinstance(value, expected_type):
                    errors.append(f'{field_name} 类型错误，期望 {expected_type.__name__}，实际 {type(value).__name__}')
                
                # 模式检查
                pattern = rules.get('pattern')
                if pattern and isinstance(value, str):
                    import re
                    if not re.match(pattern, value):
                        errors.append(f'{field_name} 格式不符合要求: {pattern}')
                
                # 数值范围检查
                if isinstance(value, (int, float)):
                    min_val = rules.get('min')
                    max_val = rules.get('max')
                    if min_val is not None and value < min_val:
                        errors.append(f'{field_name} 值 {value} 小于最小值 {min_val}')
                    if max_val is not None and value > max_val:
                        errors.append(f'{field_name} 值 {value} 大于最大值 {max_val}')
                
                # 列表项类型检查
                if isinstance(value, list) and rules.get('item_type'):
                    item_type = rules['item_type']
                    for i, item in enumerate(value):
                        if not isinstance(item, item_type):
                            errors.append(f'{field_name}[{i}] 项类型错误，期望 {item_type.__name__}')
        
        if errors:
            raise ValueError('; '.join(errors))
        
        return v

# 测试动态验证
try:
    valid_config = DynamicValidationModel(config={
        'database_url': 'postgresql://localhost/mydb',
        'max_connections': 20,
        'debug_mode': True,
        'allowed_hosts': ['localhost', '127.0.0.1']
    })
    print("动态配置验证成功:", valid_config.model_dump())
except Exception as e:
    print(f"动态配置验证失败: {e}")

实际应用案例

API请求模型

from pydantic import BaseModel, Field, field_validator
from typing import Optional, List, Dict, Any
from datetime import datetime
import re

class APIRequestModel(BaseModel):
    """API请求模型示例"""
    # 身份验证
    api_key: str = Field(..., min_length=32, max_length=64, description="API密钥")
    user_id: Optional[int] = Field(None, gt=0, description="用户ID")
    
    # 业务数据
    action: str = Field(..., description="API动作")
    parameters: Dict[str, Any] = Field(default_factory=dict, description="请求参数")
    metadata: Dict[str, str] = Field(default_factory=dict, description="元数据")
    
    # 时间戳
    timestamp: datetime = Field(default_factory=datetime.utcnow, description="请求时间戳")
    
    # 限制和配额
    rate_limit_window: int = Field(default=60, ge=1, le=3600, description="速率限制窗口（秒）")
    priority: int = Field(default=1, ge=1, le=5, description="请求优先级")
    
    @field_validator('api_key')
    def validate_api_key(cls, v):
        """验证API密钥格式"""
        # 通常API密钥是字母数字混合，长度固定
        if not re.match(r'^[A-Za-z0-9]{32,64}$', v):
            raise ValueError('API密钥格式无效')
        return v
    
    @field_validator('action')
    def validate_action(cls, v):
        """验证允许的操作"""
        allowed_actions = {
            'create_user', 'update_user', 'delete_user',
            'get_user', 'list_users', 'search_users',
            'create_order', 'update_order', 'get_order',
            'process_payment', 'refund_payment'
        }
        if v not in allowed_actions:
            raise ValueError(f'不允许的操作: {v}')
        return v
    
    @field_validator('parameters')
    def validate_parameters(cls, v):
        """验证参数安全性"""
        dangerous_patterns = [
            r'<script', r'javascript:', r'on\w+\s*=',
            r'eval\s*\(', r'exec\s*\(', r'__proto__'
        ]
        
        def check_dict(d, path=''):
            for key, value in d.items():
                current_path = f"{path}.{key}" if path else str(key)
                
                if isinstance(value, str):
                    for pattern in dangerous_patterns:
                        if re.search(pattern, value, re.IGNORECASE):
                            raise ValueError(f'参数 {current_path} 包含危险内容')
                elif isinstance(value, dict):
                    check_dict(value, current_path)
        
        if v:
            check_dict(v)
        
        return v

# 测试API请求模型
try:
    api_request = APIRequestModel(
        api_key='a1b2c3d4e5f6g7h8i9j0k1l2m3n4o5p6',
        user_id=12345,
        action='get_user',
        parameters={'user_id': 12345},
        metadata={'source': 'mobile_app', 'version': '1.0.0'},
        priority=3
    )
    print("API请求验证成功:", api_request.model_dump())
except Exception as e:
    print(f"API请求验证失败: {e}")

配置管理模型

from pydantic import BaseModel, Field, field_validator
from typing import Optional, List, Union
from pathlib import Path
import os

class DatabaseConfig(BaseModel):
    """数据库配置"""
    url: str = Field(..., description="数据库连接URL")
    pool_size: int = Field(default=5, ge=1, le=100, description="连接池大小")
    echo: bool = Field(default=False, description="是否打印SQL语句")
    timeout: float = Field(default=30.0, gt=0, description="连接超时时间")

class CacheConfig(BaseModel):
    """缓存配置"""
    backend: str = Field(default='memory', description="缓存后端")
    ttl: int = Field(default=300, gt=0, description="缓存生存时间（秒）")
    redis_url: Optional[str] = Field(None, description="Redis连接URL")

class AppConfig(BaseModel):
    """应用配置"""
    app_name: str = Field(default='MyApp', description="应用名称")
    debug: bool = Field(default=False, description="调试模式")
    host: str = Field(default='127.0.0.1', description="监听主机")
    port: int = Field(default=8000, ge=1, le=65535, description="监听端口")
    workers: int = Field(default=1, ge=1, le=32, description="工作进程数")
    
    database: DatabaseConfig = Field(default_factory=DatabaseConfig, description="数据库配置")
    cache: CacheConfig = Field(default_factory=CacheConfig, description="缓存配置")
    
    allowed_hosts: List[str] = Field(default=['localhost'], description="允许的主机列表")
    cors_origins: List[str] = Field(default=['*'], description="CORS来源列表")
    
    secret_key: str = Field(..., min_length=32, description="密钥")
    algorithm: str = Field(default='HS256', description="加密算法")
    
    @field_validator('host')
    def validate_host(cls, v):
        """验证主机地址"""
        if v not in ['localhost', '127.0.0.1', '0.0.0.0'] and not v.replace('.', '').isdigit():
            # 如果不是本地地址，验证是否为有效域名
            import re
            domain_pattern = r'^([a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z]{2,}$'
            if not re.match(domain_pattern, v):
                raise ValueError('无效的主机地址')
        return v
    
    @field_validator('secret_key')
    def validate_secret_key(cls, v):
        """验证密钥强度"""
        if len(v) < 32:
            raise ValueError('密钥长度至少32字符')
        # 检查是否包含足够的随机性
        if len(set(v)) < 10:
            raise ValueError('密钥随机性不足')
        return v
    
    @classmethod
    def from_env(cls):
        """从环境变量创建配置"""
        return cls(
            app_name=os.getenv('APP_NAME', 'MyApp'),
            debug=os.getenv('DEBUG', 'false').lower() == 'true',
            host=os.getenv('HOST', '127.0.0.1'),
            port=int(os.getenv('PORT', '8000')),
            workers=int(os.getenv('WORKERS', '1')),
            secret_key=os.getenv('SECRET_KEY', ''),
            database=DatabaseConfig(
                url=os.getenv('DATABASE_URL', 'sqlite:///./test.db'),
                pool_size=int(os.getenv('DB_POOL_SIZE', '5')),
                echo=os.getenv('DB_ECHO', 'false').lower() == 'true',
                timeout=float(os.getenv('DB_TIMEOUT', '30.0'))
            ),
            cache=CacheConfig(
                backend=os.getenv('CACHE_BACKEND', 'memory'),
                ttl=int(os.getenv('CACHE_TTL', '300')),
                redis_url=os.getenv('REDIS_URL')
            )
        )

# 创建配置实例
config = AppConfig(
    app_name='ProductionApp',
    debug=False,
    host='0.0.0.0',
    port=8000,
    workers=4,
    secret_key='very_long_and_random_secret_key_that_meets_requirements',
    allowed_hosts=['example.com', 'api.example.com'],
    database=DatabaseConfig(
        url='postgresql://user:pass@localhost/proddb',
        pool_size=10,
        echo=False
    )
)

print("应用配置:", config.model_dump())

相关教程

使用Pydantic进行数据验证时，建议明确定义字段约束、使用类型提示、合理使用验证器，并在生产环境中进行充分的错误处理。 对于高频验证场景，考虑使用缓存验证结果、预编译正则表达式和批量验证来优化性能。

总结

Pydantic提供了强大而灵活的数据验证系统：

  1. 类型安全：基于Python类型提示的自动验证
  2. 灵活约束：丰富的字段约束选项
  3. 自定义验证：支持自定义验证逻辑
  4. 嵌套模型：支持复杂数据结构
  5. 性能优化：高效的验证算法
  6. 错误处理：详细的错误信息

掌握Pydantic的使用对于构建健壮的FastAPI应用至关重要。