Pydantic 与 SQLAlchemy 从浅入深使用指南

1. 基础概念介绍

1.1 Pydantic 是什么？

Pydantic 是一个数据验证和设置管理的 Python 库，主要用途：

• 数据验证
• 序列化和反序列化
• 设置管理
• 自动生成文档

1.2 SQLAlchemy 是什么？

SQLAlchemy 是 Python 的 ORM（对象关系映射）工具：

• 将 Python 类映射到数据库表
• 提供数据库操作接口
• 支持多种数据库后端

2. 基础使用

2.1 安装依赖

pip install pydantic sqlalchemy

2.2 基础模型定义

from pydantic import BaseModel
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# SQLAlchemy 模型
Base = declarative_base()

class UserDB(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50), nullable=False)
    email = Column(String(100), unique=True, index=True)
    age = Column(Integer)

# Pydantic 模型
class UserCreate(BaseModel):
    name: str
    email: str
    age: int

class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    age: int

    class Config:
        orm_mode = True  # 允许从 ORM 对象创建

2.3 基础 CRUD 操作

# 数据库设置
engine = create_engine("sqlite:///./test.db")
Base.metadata.create_all(bind=engine)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def create_user(user_data: UserCreate):
    db = SessionLocal()

    # 将 Pydantic 模型转换为字典
    user_dict = user_data.dict()

    # 创建 SQLAlchemy 模型实例
    db_user = UserDB(**user_dict)

    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    db.close()

    # 将 SQLAlchemy 模型转换为 Pydantic 响应模型
    return UserResponse.from_orm(db_user)

# 使用示例
user_data = UserCreate(name="张三", email="zhangsan@example.com", age=25)
result = create_user(user_data)
print(result)

3. 中级用法

3.1 数据验证和转换

from pydantic import validator, EmailStr
from typing import Optional
from datetime import datetime

class UserAdvanced(BaseModel):
    name: str
    email: EmailStr  # 邮箱格式验证
    age: int
    created_at: Optional[datetime] = None

    @validator('name')
    def name_must_contain_space(cls, v):
        if ' ' not in v:
            raise ValueError('名称必须包含空格')
        return v.title()  # 自动转换为标题格式

    @validator('age')
    def age_must_be_positive(cls, v):
        if v <= 0:
            raise ValueError('年龄必须是正数')
        return v

# 扩展 SQLAlchemy 模型
class UserAdvancedDB(Base):
    __tablename__ = "users_advanced"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50), nullable=False)
    email = Column(String(100), unique=True, index=True)
    age = Column(Integer)
    created_at = Column(DateTime, default=datetime.utcnow)

3.2 关系映射

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship
from typing import List

# SQLAlchemy 关系模型
class PostDB(Base):
    __tablename__ = "posts"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(100), nullable=False)
    content = Column(Text)
    author_id = Column(Integer, ForeignKey("users.id"))

    author = relationship("UserDB", back_populates="posts")

# 在 UserDB 中添加关系
UserDB.posts = relationship("PostDB", back_populates="author")

# Pydantic 关系模型
class PostBase(BaseModel):
    title: str
    content: str

class PostCreate(PostBase):
    author_id: int

class PostResponse(PostBase):
    id: int
    author: UserResponse  # 嵌套关系

    class Config:
        orm_mode = True

class UserWithPosts(UserResponse):
    posts: List[PostResponse] = []

4. 高级用法

4.1 使用 Pydantic 的 Config 类

from pydantic import BaseConfig
from typing import Any, Dict

class CustomConfig(BaseConfig):
    # 允许任意类型（谨慎使用）
    arbitrary_types_allowed = True
    # 使用枚举值
    use_enum_values = True
    # 在验证时也填充字段
    validate_assignment = True
    # 允许字段别名
    allow_population_by_field_name = True

class AdvancedUser(BaseModel):
    class Config(CustomConfig):
        # 表名映射
        orm_mode = True
        # 允许字段别名
        fields = {
            'user_name': 'name',
            'user_email': 'email'
        }

    user_name: str
    user_email: str
    age: int

4.2 复杂验证和依赖注入

from pydantic import Field, root_validator
from fastapi import Depends

class UserWithDependencies(BaseModel):
    name: str = Field(..., min_length=2, max_length=50)
    email: str = Field(..., regex=r'^[\w\.-]+@[\w\.-]+\.\w+$')
    password: str = Field(..., min_length=8)
    confirm_password: str

    @root_validator
    def passwords_match(cls, values):
        password = values.get('password')
        confirm_password = values.get('confirm_password')
        if password != confirm_password:
            raise ValueError('密码不匹配')
        return values

# 数据库依赖注入
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

def create_user_with_deps(
    user: UserWithDependencies,
    db: Session = Depends(get_db)
):
    # 业务逻辑和数据库操作
    db_user = UserDB(**user.dict(exclude={'confirm_password'}))
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return UserResponse.from_orm(db_user)

4.3 高级查询和分页

from typing import Generic, TypeVar, List
from pydantic.generics import GenericModel

T = TypeVar('T')

class PaginatedResponse(GenericModel, Generic[T]):
    items: List[T]
    total: int
    page: int
    size: int
    pages: int

class UserFilters(BaseModel):
    name: Optional[str] = None
    email: Optional[str] = None
    min_age: Optional[int] = None
    max_age: Optional[int] = None

def get_users_with_filters(
    filters: UserFilters,
    page: int = 1,
    size: int = 10,
    db: Session = Depends(get_db)
) -> PaginatedResponse[UserResponse]:
    query = db.query(UserDB)

    # 应用过滤器
    if filters.name:
        query = query.filter(UserDB.name.ilike(f"%{filters.name}%"))
    if filters.email:
        query = query.filter(UserDB.email.ilike(f"%{filters.email}%"))
    if filters.min_age is not None:
        query = query.filter(UserDB.age >= filters.min_age)
    if filters.max_age is not None:
        query = query.filter(UserDB.age <= filters.max_age)

    # 分页
    total = query.count()
    items = query.offset((page - 1) * size).limit(size).all()

    return PaginatedResponse(
        items=[UserResponse.from_orm(item) for item in items],
        total=total,
        page=page,
        size=size,
        pages=(total + size - 1) // size
    )

5. 最佳实践和模式

5.1 工厂模式创建模型

class ModelFactory:
    @staticmethod
    def create_user_model(exclude_fields: set = None):
        class DynamicUser(BaseModel):
            name: str
            email: str
            age: int

            class Config:
                orm_mode = True

        if exclude_fields:
            for field in exclude_fields:
                DynamicUser.__fields__.pop(field, None)

        return DynamicUser

# 使用工厂创建不同的模型
UserCreateModel = ModelFactory.create_user_model()
UserUpdateModel = ModelFactory.create_user_model({'id'})
UserPublicModel = ModelFactory.create_user_model({'email'})

5.2 使用 mixin 类

class TimestampMixin:
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

class BaseMixin(BaseModel):
    id: int
    created_at: datetime
    updated_at: datetime

    class Config:
        orm_mode = True

class UserWithTimestampsDB(Base, TimestampMixin):
    __tablename__ = "users_timestamp"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50))

class UserWithTimestamps(BaseMixin):
    name: str

6. 错误处理和调试

6.1 自定义验证错误

from pydantic import ValidationError

def safe_user_creation(user_data: dict):
    try:
        user = UserCreate(**user_data)
        return create_user(user)
    except ValidationError as e:
        print(f"验证错误: {e}")
        return {"error": "数据验证失败", "details": e.errors()}
    except Exception as e:
        print(f"其他错误: {e}")
        return {"error": "创建用户失败"}

6.2 性能优化

# 使用 selectinload 优化关联查询
from sqlalchemy.orm import selectinload

def get_user_with_posts(user_id: int, db: Session):
    return db.query(UserDB).options(
        selectinload(UserDB.posts)
    ).filter(UserDB.id == user_id).first()

# 批量操作优化
def create_users_batch(users_data: List[UserCreate], db: Session):
    users = [UserDB(**user.dict()) for user in users_data]
    db.bulk_save_objects(users)
    db.commit()
    return users

这个指南从基础到高级涵盖了 Pydantic 和 SQLAlchemy 的结合使用。实际项目中可以根据具体需求选择合适的技术方案。