sqlmodel-expert

SQLModel 专家

高级 SQLModel 模式以及适用于生产数据库的全面 Alembic 迁移。

快速入门

定义基本模型

from sqlmodel import Field, SQLModel
from typing import Optional
from datetime import datetime

class Task(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str = Field(index=True)
    description: Optional[str] = None
    completed: bool = Field(default=False)
    created_at: datetime = Field(default_factory=datetime.utcnow)

初始化数据库

# Using provided script
python scripts/init_db.py --url postgresql://user:pass@localhost/db

# Or manually
from sqlmodel import create_engine
engine = create_engine("postgresql://user:pass@localhost/db")
SQLModel.metadata.create_all(engine)

创建迁移

# Using provided helper script
./scripts/migrate.sh create "add user table"

# Or directly with Alembic
alembic revision --autogenerate -m "add user table"
alembic upgrade head

核心主题

1. 高级模型模式

参见： references/advanced-models.md

• 关系：一对多、多对多、自引用
• 继承：单表继承、联接表继承、多态
• 验证：Pydantic 验证器、自定义约束
• 混入类：时间戳、软删除、可复用模式
• 字段类型：枚举、JSON、数组、自定义类型
• 索引：单列、复合、部分索引
• 约束：唯一约束、检查约束、外键级联

2. 全面迁移

参见： references/migrations.md

• Alembic 设置：SQLModel 的配置和 env.py
• 创建迁移：自动生成与手动创建
• 架构变更：添加/删除列、重命名、更改类型
• 数据迁移：复杂的数据转换
• 生产工作流：零停机迁移
• 回滚策略：安全的降级模式
• 故障排除：常见问题及解决方案

3. 查询优化

参见： references/queries-optimization.md

• N+1 问题：使用预先加载解决
• 查询模式：联接、聚合、子查询
• 性能：索引、批量操作、性能分析
• 高级查询：窗口函数、CTE
• 批量操作：大规模插入、更新、删除
• 测试：查询计数、执行计划分析

常见模式

一对多关系

from typing import List
from sqlmodel import Field, Relationship, SQLModel

class Team(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str

    # One team has many heroes
    heroes: List["Hero"] = Relationship(back_populates="team")

class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    team_id: Optional[int] = Field(foreign_key="team.id")

    # Many heroes belong to one team
    team: Optional[Team] = Relationship(back_populates="heroes")

使用关联表实现多对多关系

class HeroTeamLink(SQLModel, table=True):
    hero_id: int = Field(foreign_key="hero.id", primary_key=True)
    team_id: int = Field(foreign_key="team.id", primary_key=True)
    joined_at: datetime = Field(default_factory=datetime.utcnow)

class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    teams: List["Team"] = Relationship(
        back_populates="heroes",
        link_model=HeroTeamLink
    )

class Team(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    heroes: List[Hero] = Relationship(
        back_populates="teams",
        link_model=HeroTeamLink
    )

解决 N+1 查询问题

from sqlalchemy.orm import selectinload

# BAD - N+1 queries
users = session.exec(select(User)).all()
for user in users:
    posts = user.posts  # Each triggers a query!

# GOOD - Eager loading (2 queries total)
statement = select(User).options(selectinload(User.posts))
users = session.exec(statement).all()
for user in users:
    posts = user.posts  # No additional query!

创建迁移

# 1. Modify your model
class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    email: str
    phone: str  # New field added

# 2. Generate migration
# alembic revision --autogenerate -m "add phone to user"

# 3. Review generated migration
def upgrade() -> None:
    op.add_column('user', sa.Column('phone', sa.String(), nullable=True))

def downgrade() -> None:
    op.drop_column('user', 'phone')

# 4. Apply migration
# alembic upgrade head

迁移辅助脚本

初始化数据库

python scripts/init_db.py --url postgresql://user:pass@localhost/db

迁移操作

./scripts/migrate.sh init              # Initialize Alembic
./scripts/migrate.sh create "message"  # Create migration
./scripts/migrate.sh upgrade           # Apply migrations
./scripts/migrate.sh downgrade         # Rollback one
./scripts/migrate.sh current           # Show current
./scripts/migrate.sh history           # Show history
./scripts/migrate.sh test              # Test up & down

示例模型

将 assets/example-models.py 中的示例模型用作模板：

• 带有时间戳混入类的 User 模型
• 带有枚举和关系的 Task 模型
• 带有多对多关系的 Team 模型
• 使用关联表的标签系统
• 独立的读取/写入/更新模型

复制到你的项目中：

cp assets/example-models.py your-project/app/models.py

最佳实践检查清单

模型设计

• 为所有字段使用类型提示
• 分离读取/写入/更新模型
• 使用混入类添加通用字段（时间戳、软删除）
• 为外键和经常查询的列定义索引
• 对受限选项使用枚举
• 使用 Pydantic 验证器实施适当的验证

关系

• 对双向关系使用 back_populates
• 为多对多关系创建显式关联表
• 考虑级联删除行为
• 使用预先加载避免 N+1 查询
• 为外键列添加索引

迁移

• 始终检查自动生成的迁移
• 每个迁移只包含一项逻辑变更
• 测试升级和降级
• 使用描述性迁移名称
• 切勿编辑已应用的迁移
• 更改架构时添加数据迁移
• 在生产迁移前备份数据库

查询优化

• 对关系使用预先加载（selectinload）
• 只选择所需列
• 为 WHERE/ORDER BY 列添加索引
• 使用批量操作代替循环
• 分析慢查询
• 使用连接池

故障排除指南

迁移问题

问题：Alembic 未检测到模型变更

# Solution: Ensure models are imported in env.py
from app.models import User, Task, Team  # Import all models
target_metadata = SQLModel.metadata

问题：迁移失败

# Check current state
alembic current

# Manually fix issue, then stamp
alembic stamp head

# Or downgrade and retry
alembic downgrade -1
alembic upgrade head

查询性能

问题：查询缓慢

# Enable query logging
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)

# Use EXPLAIN ANALYZE
explain = session.exec(text("EXPLAIN ANALYZE SELECT ...")).all()

# Profile queries
# See references/queries-optimization.md for detailed patterns

问题：N+1 查询

# Use selectinload
statement = select(User).options(selectinload(User.posts))

# Or joinedload
from sqlalchemy.orm import joinedload
statement = select(User).options(joinedload(User.posts))

生产工作流

开发

1. 修改 SQLModel 模型
2. 生成迁移：./scripts/migrate.sh create "description"
3. 检查生成的迁移文件
4. 测试迁移：./scripts/migrate.sh test
5. 提交迁移文件

预发布

1. 部署应用代码
2. 运行迁移：alembic upgrade head
3. 验证数据完整性
4. 测试应用

生产环境

1. 备份数据库：pg_dump mydb > backup.sql
2. 在维护窗口期间部署
3. 运行迁移：alembic upgrade head
4. 监控日志和指标
5. 验证应用功能

零停机迁移策略

对于大型生产数据库：

# Phase 1: Add new column (nullable)
def upgrade():
    op.add_column('user', sa.Column('new_email', sa.String(), nullable=True))

# Deploy app version that writes to both columns

# Phase 2: Backfill data
def upgrade():
    op.execute("UPDATE user SET new_email = email WHERE new_email IS NULL")

# Phase 3: Make non-nullable
def upgrade():
    op.alter_column('user', 'new_email', nullable=False)

# Deploy app version that reads from new column

# Phase 4: Drop old column
def upgrade():
    op.drop_column('user', 'email')

其他资源

• 高级模式：请参阅 references/advanced-models.md，了解继承、多态和复合键
• 迁移指南：请参阅 references/migrations.md，掌握 Alembic
• 查询优化：请参阅 references/queries-optimization.md，了解性能调优

此技能提供专业 SQLModel 开发和数据库管理所需的一切。