SQLModel Expert

SQLModel 专家指南

Advanced SQLModel patterns and comprehensive Alembic migrations for production databases.

面向生产数据库的进阶SQLModel模式与全面Alembic迁移方案。

Quick Start

快速开始

Define a Basic Model

定义基础模型

python

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)

python

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)

Initialize Database

初始化数据库

bash

undefined

bash

undefined

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)

undefined

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

undefined

Create Migration

创建迁移

bash

undefined

bash

undefined

Using provided helper script

使用提供的辅助脚本

./scripts/migrate.sh create "add user table"

Or directly with Alembic

或直接使用Alembic

alembic revision --autogenerate -m "add user table" alembic upgrade head

undefined

alembic revision --autogenerate -m "add user table" alembic upgrade head

undefined

Core Topics

核心主题

1. Advanced Model Patterns

1. 进阶模型模式

See: references/advanced-models.md

Relationships: One-to-many, many-to-many, self-referential
Inheritance: Single table, joined table, polymorphism
Validation: Pydantic validators, custom constraints
Mixins: Timestamp, soft delete, reusable patterns
Field Types: Enums, JSON, arrays, custom types
Indexes: Single, composite, partial indexes
Constraints: Unique, check, foreign key cascades

参考： references/advanced-models.md

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

2. Comprehensive Migrations

2. 全面迁移方案

See: references/migrations.md

Alembic Setup: Configuration, env.py for SQLModel
Creating Migrations: Autogenerate vs manual
Schema Changes: Add/drop columns, rename, change types
Data Migrations: Complex data transformations
Production Workflow: Zero-downtime migrations
Rollback Strategies: Safe downgrade patterns
Troubleshooting: Common issues and solutions

参考： references/migrations.md

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

3. Query Optimization

3. 查询优化

See: references/queries-optimization.md

N+1 Problem: Solutions with eager loading
Query Patterns: Joins, aggregations, subqueries
Performance: Indexes, batch operations, profiling
Advanced Queries: Window functions, CTEs
Bulk Operations: Insert, update, delete at scale
Testing: Query counting, explain analyze

参考： references/queries-optimization.md

N+1问题：预加载解决方案
查询模式：连接、聚合、子查询
性能优化：索引、批量操作、性能分析
进阶查询：窗口函数、公共表表达式(CTE)
批量操作：大规模插入、更新、删除
测试验证：查询计数、执行计划分析

Common Patterns

常见模式

One-to-Many Relationship

一对多关系

python

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")

python

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

    # 一个团队对应多个英雄
    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")

    # 多个英雄属于一个团队
    team: Optional[Team] = Relationship(back_populates="heroes")

Many-to-Many with Link Table

带关联表的多对多关系

python

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
    )

python

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
    )

Solving N+1 Query Problem

解决N+1查询问题

python

from sqlalchemy.orm import selectinload

python

from sqlalchemy.orm import selectinload

BAD - N+1 queries

错误示例 - 产生N+1查询

users = session.exec(select(User)).all() for user in users: posts = user.posts # Each triggers a query!

users = session.exec(select(User)).all() for user in users: posts = user.posts # 每个用户都会触发一次查询！

GOOD - Eager loading (2 queries total)

正确示例 - 预加载（总共2次查询）

statement = select(User).options(selectinload(User.posts)) users = session.exec(statement).all() for user in users: posts = user.posts # No additional query!

undefined

statement = select(User).options(selectinload(User.posts)) users = session.exec(statement).all() for user in users: posts = user.posts # 无需额外查询！

undefined

Creating a Migration

创建迁移

python

undefined

python

undefined

1. Modify your model

1. 修改模型

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

class User(SQLModel, table=True): id: Optional[int] = Field(default=None, primary_key=True) email: str phone: str # 新增字段

2. Generate migration

2. 生成迁移

alembic revision --autogenerate -m "add phone to user"

3. Review generated migration

3. 检查生成的迁移文件

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

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

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

4. 应用迁移

alembic upgrade head

undefined

undefined

Migration Helper Scripts

迁移辅助脚本

Initialize Database

初始化数据库

bash

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

bash

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

Migration Operations

迁移操作

bash

./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

bash

./scripts/migrate.sh init              # 初始化Alembic
./scripts/migrate.sh create "message"  # 创建迁移
./scripts/migrate.sh upgrade           # 应用迁移
./scripts/migrate.sh downgrade         # 回滚一个版本
./scripts/migrate.sh current           # 查看当前版本
./scripts/migrate.sh history           # 查看版本历史
./scripts/migrate.sh test              # 测试升级与降级

Example Models

示例模型

Use the example models in

assets/example-models.py

as templates:

User model with timestamp mixin
Task model with enums and relationships
Team model with many-to-many
Tag system with link tables
Separate read/write/update models

Copy to your project:

bash

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

可使用

assets/example-models.py

中的示例模型作为模板：

带时间戳混合类的User模型
带枚举与关系的Task模型
带多对多关系的Team模型
带关联表的标签系统
分离的读/写/更新模型

复制到你的项目：

bash

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

Best Practices Checklist

最佳实践清单

Model Design

模型设计

Use type hints for all fields
Separate read/write/update models
Use mixins for common fields (timestamps, soft delete)
Define indexes on foreign keys and frequently queried columns
Use enums for constrained choices
Implement proper validation with Pydantic validators

为所有字段使用类型提示
分离读/写/更新模型
为通用字段使用混合类（时间戳、软删除）
为外键与频繁查询的列定义索引
对受限选项使用枚举
使用Pydantic验证器实现合理验证

Relationships

关系配置

Use
```
back_populates
```
for bidirectional relationships
Create explicit link tables for many-to-many
Consider cascade delete behavior
Use eager loading to prevent N+1 queries
Index foreign key columns

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

Migrations

迁移管理

Always review autogenerated migrations
One logical change per migration
Test both upgrade and downgrade
Use descriptive migration names
Never edit applied migrations
Add data migrations when changing schemas
Backup database before production migrations

始终检查自动生成的迁移文件
每次迁移对应一个逻辑变更
测试升级与降级流程
使用描述性的迁移名称
不要修改已应用的迁移
变更架构时添加数据迁移
生产环境迁移前备份数据库

Query Optimization

查询优化

Use eager loading (selectinload) for relationships
Select only needed columns
Use indexes for WHERE/ORDER BY columns
Batch operations instead of loops
Profile slow queries
Use connection pooling

Troubleshooting Guide

问题排查指南

Migration Issues

迁移问题

Problem: Alembic doesn't detect model changes

python

undefined

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

python

undefined

Solution: Ensure models are imported in env.py

解决方案：确保在env.py中导入所有模型

from app.models import User, Task, Team # Import all models target_metadata = SQLModel.metadata


**Problem**: Failed migration
```bash

from app.models import User, Task, Team # 导入所有模型 target_metadata = SQLModel.metadata


**问题**：迁移执行失败
```bash

Check current state

查看当前状态

alembic current

Manually fix issue, then stamp

手动修复问题后标记当前版本

alembic stamp head

Or downgrade and retry

或回滚后重试

alembic downgrade -1 alembic upgrade head

undefined

alembic downgrade -1 alembic upgrade head

undefined

Query Performance

查询性能问题

Problem: Slow queries

python

undefined

问题：查询速度慢

python

undefined

Enable query logging

启用查询日志

logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)

Use EXPLAIN ANALYZE

使用EXPLAIN ANALYZE分析

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

Profile queries

查询性能分析详情请参考：references/queries-optimization.md

See references/queries-optimization.md for detailed patterns

—


**Problem**: N+1 queries
```python


**问题**：出现N+1查询
```python

Use selectinload

使用selectinload预加载

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

Or joinedload

或使用joinedload

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

undefined

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

undefined

Production Workflow

生产环境流程

Development

开发阶段

Modify SQLModel models

Generate migration:

./scripts/migrate.sh create "description"

Review generated migration file
Test migration:
```
./scripts/migrate.sh test
```
Commit migration file

修改SQLModel模型

生成迁移：

./scripts/migrate.sh create "description"

检查生成的迁移文件
测试迁移：
```
./scripts/migrate.sh test
```
提交迁移文件

Staging

预发布阶段

Deploy application code
Run migrations:
```
alembic upgrade head
```
Verify data integrity
Test application

部署应用代码
执行迁移：
```
alembic upgrade head
```
验证数据完整性
测试应用功能

Production

生产阶段

Backup database:
```
pg_dump mydb > backup.sql
```
Deploy in maintenance window
Run migrations:
```
alembic upgrade head
```
Monitor logs and metrics
Verify application functionality

备份数据库：
```
pg_dump mydb > backup.sql
```
在维护窗口部署
执行迁移：
```
alembic upgrade head
```
监控日志与指标
验证应用功能

Zero-Downtime Migration Strategy

零停机迁移策略

For large production databases:

python

undefined

针对大型生产数据库：

python

undefined

Phase 1: Add new column (nullable)

阶段1：添加新列（可空）

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

阶段2：回填数据

def upgrade(): op.execute("UPDATE user SET new_email = email WHERE new_email IS NULL")

Phase 3: Make non-nullable

阶段3：设置为非空

def upgrade(): op.alter_column('user', 'new_email', nullable=False)

Deploy app version that reads from new column

部署读取新列的应用版本

Phase 4: Drop old column

阶段4：删除旧列

def upgrade(): op.drop_column('user', 'email')

undefined

def upgrade(): op.drop_column('user', 'email')

undefined

Additional Resources

额外资源

Advanced Patterns: See references/advanced-models.md for inheritance, polymorphism, composite keys
Migration Guide: See references/migrations.md for Alembic mastery
Query Optimization: See references/queries-optimization.md for performance tuning

This skill provides everything needed for professional SQLModel development and database management.

进阶模式：参考references/advanced-models.md了解继承、多态、复合键
迁移指南：参考references/migrations.md掌握Alembic
查询优化：参考references/queries-optimization.md进行性能调优

本指南提供了专业SQLModel开发与数据库管理所需的全部内容。

sqlmodel-expert

Original

Translation

SQLModel Expert

SQLModel 专家指南

Quick Start

快速开始

Define a Basic Model

定义基础模型

Initialize Database

初始化数据库

Using provided script

使用提供的脚本

Or manually

或手动执行

Create Migration

创建迁移

Using provided helper script

使用提供的辅助脚本

Or directly with Alembic

或直接使用Alembic

Core Topics

核心主题

1. Advanced Model Patterns

1. 进阶模型模式

2. Comprehensive Migrations

2. 全面迁移方案

3. Query Optimization

3. 查询优化

Common Patterns

常见模式

One-to-Many Relationship

一对多关系

Many-to-Many with Link Table

带关联表的多对多关系

Solving N+1 Query Problem

解决N+1查询问题

BAD - N+1 queries

错误示例 - 产生N+1查询

GOOD - Eager loading (2 queries total)

正确示例 - 预加载（总共2次查询）

Creating a Migration

创建迁移

1. Modify your model

1. 修改模型

2. Generate migration

2. 生成迁移

alembic revision --autogenerate -m "add phone to user"

alembic revision --autogenerate -m "add phone to user"

3. Review generated migration

3. 检查生成的迁移文件

4. Apply migration

4. 应用迁移

alembic upgrade head

alembic upgrade head

Migration Helper Scripts

迁移辅助脚本

Initialize Database

初始化数据库

Migration Operations

迁移操作

Example Models

示例模型

Best Practices Checklist

最佳实践清单

Model Design

模型设计

Relationships

关系配置

Migrations

迁移管理

Query Optimization

查询优化

Troubleshooting Guide

问题排查指南

Migration Issues

迁移问题

Solution: Ensure models are imported in env.py

解决方案：确保在env.py中导入所有模型

Check current state