Pydantic综合指南：构建类型安全的Python应用

📂 所属阶段：第一阶段 — 快速筑基（基础篇）
🔗 相关章节：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'>

基础模型定义

简单模型

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

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

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)  # 字典默认值

# 创建实例
person = Person(name="Alice", age=28, email="alice@example.com")
doc = Document(title="Sample Document", content="This is a sample document.")

print(person.name)  # Alice
print(doc.created_at)  # 当前时间

字段验证与约束

使用Field进行字段约束

from pydantic import BaseModel, Field
from typing import Optional

class Product(BaseModel):
    """产品模型，演示各种字段约束"""
    name: str = Field(..., min_length=3, max_length=100, description="产品名称")
    price: float = Field(..., gt=0, le=10000, description="产品价格")
    category: str = Field(..., pattern=r"^[a-zA-Z_][a-zA-Z0-9_]*$", description="产品分类")
    stock: int = Field(default=0, ge=0, description="库存数量")
    rating: float = Field(default=0.0, ge=0, le=5.0, description="评分")

# 验证实例
try:
    product = Product(
        name="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_validator
import re

class UserValidation(BaseModel):
    """用户模型，演示自定义验证器"""
    username: str
    email: str
    password: str
    confirm_password: str
    
    @field_validator('username')
    def validate_username(cls, v):
        """验证用户名"""
        if len(v) < 3 or len(v) > 20:
            raise ValueError('用户名长度必须在3-20字符之间')
        if not re.match(r'^[a-zA-Z0-9_]+$', v):
            raise ValueError('用户名只能包含字母、数字和下划线')
        return v.lower()
    
    @field_validator('email')
    def validate_email(cls, v):
        """验证邮箱"""
        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()
    
    @field_validator('confirm_password')
    def passwords_match(cls, v, info):
        """验证密码确认"""
        if 'password' in info.data and v != info.data['password']:
            raise ValueError('密码确认不匹配')
        return v

多字段验证器

from pydantic import BaseModel, model_validator
from datetime import date
from typing import Optional

class Booking(BaseModel):
    """预订模型，演示多字段验证"""
    customer_name: str
    check_in: date
    check_out: date
    room_type: str
    adults: int
    children: int = 0
    
    @model_validator(mode='after')
    def validate_booking_logic(self):
        """验证预订逻辑"""
        # 日期验证
        if self.check_in >= self.check_out:
            raise ValueError('退房日期必须晚于入住日期')
        if self.check_in < date.today():
            raise ValueError('不能预订过去的日期')
        
        # 人数验证
        total_guests = self.adults + self.children
        room_capacity = {'single': 2, 'double': 4, 'suite': 6, 'deluxe': 8}
        if total_guests > room_capacity.get(self.room_type, 2):
            raise ValueError(f'{self.room_type}房型最多容纳{room_capacity[self.room_type]}人')
        
        return self

嵌套模型与复杂结构

嵌套模型定义

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)
    country: str = Field(..., min_length=2, max_length=50)
    postal_code: str = Field(..., pattern=r"^[0-9]{5}(-[0-9]{4})?$")

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
    hire_date: datetime
    address: Address
    skills: List[str] = Field(default=[], max_items=20)

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

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

print("员工信息:", employee.model_dump())

模型继承

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
        extra = "forbid"

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

class AdminUser(User):
    """管理员用户，继承普通用户"""
    permissions: List[str] = Field(default=["read", "write"])
    is_super_admin: bool = False

数据转换与序列化

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

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_encoders = {
            datetime: lambda dt: dt.isoformat(),
        }

# 创建模型实例
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())

实际应用案例

API请求模型

from pydantic import BaseModel, Field, field_validator
from typing import Optional, 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="请求参数")
    timestamp: datetime = Field(default_factory=datetime.utcnow, description="请求时间戳")
    priority: int = Field(default=1, ge=1, le=5, description="请求优先级")
    
    @field_validator('api_key')
    def validate_api_key(cls, v):
        """验证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'
        }
        if v not in allowed_actions:
            raise ValueError(f'不允许的操作: {v}')
        return v

配置管理模型

from pydantic import BaseModel, Field
from typing import Optional, List
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语句")

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="监听端口")
    database: DatabaseConfig = Field(default_factory=DatabaseConfig, description="数据库配置")
    secret_key: str = Field(..., min_length=32, description="密钥")
    
    @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')),
            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'
            )
        )

相关教程

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

总结

Pydantic提供了强大而灵活的数据验证系统，通过类型提示确保数据安全，支持自定义验证逻辑和复杂嵌套结构。掌握Pydantic的使用对于构建健壮的FastAPI应用至关重要。