本节目标：用 Alembic 把数据库结构变更写成有版本号、可回滚、可多人协作的迁移脚本，掌握 autogenerate 与全套升降级命令，并想清演进策略。
适用版本：Python 3.12+（实测 3.14.6）；SQLAlchemy 2.1.4、Alembic 1.20.0
6.3 Alembic 迁移与数据演进
6.1 节的 Base.metadata.create_all() 只能建表，改不了表。真实项目里加一列、改一次类型、回填一批数据，都需要在不丢数据、可回退、可复现的前提下完成。Alembic 就是数据库结构的 Git：每次变更是一个带 revision 编号的脚本，按链式顺序应用或回退。
本机实测版本 Alembic 1.20.0，数据库用 SQLite。无 PostgreSQL 服务端，PG 专属方言相关部分会显式标注未实测。
6.3.1 初始化项目
在项目根目录执行 alembic init，它会生成目录骨架：
alembic init -t generic migrations
生成的结构：
migrations/
env.py # 运行时配置：连数据库、指定 metadata
script.py.mako # 迁移脚本模板
versions/ # 每个迁移脚本一个文件，按 revision 链
alembic.ini # 全局配置，含 sqlalchemy.url
-t generic 是不绑定具体框架的通用模板；若用 FastAPI 项目也推荐它，保持与框架解耦。
6.3.2 配置 env.py
关键改动是让 Alembic 知道「谁是要迁移的目标」。把 target_metadata 指向你的 Base.metadata：
# migrations/env.py
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context
from app_models import Base # 你的模型基类
config = context.config
if config.config_file_name is not None:
fileConfig(config.config_file_name)
target_metadata = Base.metadata
def run_migrations_offline() -> None:
url = config.get_main_option("sqlalchemy.url")
context.configure(url=url, target_metadata=target_metadata, literal_binds=True)
with context.begin_transaction():
context.run_migrations()
def run_migrations_online() -> None:
connectable = engine_from_config(
config.get_section(config.config_ini_section, {}),
prefix="sqlalchemy.", poolclass=pool.NullPool,
)
with connectable.connect() as connection:
context.configure(connection=connection, target_metadata=target_metadata)
with context.begin_transaction():
context.run_migrations()
if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()
要点：
target_metadata = Base.metadata是 autogenerate 的前提，指错了会生成空迁移或误删表。poolclass=pool.NullPool是 Alembic 官方模板的默认，迁移是一次性短连接，不需要池。sqlalchemy.url可以写在alembic.ini，也可以用环境变量覆盖，避免把生产密码提交进仓库。
6.3.3 autogenerate 生成迁移
模型里定义一个 User 后，让 Alembic 对比模型与数据库的差异：
alembic revision --autogenerate -m "create user"
实测输出（关键行）：
INFO [alembic.autogenerate.compare.tables] Detected added table 'user'
Generating .../versions/33415f0a9199_create_user.py ... done
生成的脚本长这样：
"""create user
Revision ID: 33415f0a9199
Revises:
Create Date: 2026-10-09 ...
"""
from alembic import op
import sqlalchemy as sa
revision: str = '33415f0a9199'
down_revision = None
def upgrade() -> None:
op.create_table('user',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('name', sa.String(), nullable=False),
sa.PrimaryKeyConstraint('id'),
)
def downgrade() -> None:
op.drop_table('user')
--autogenerate 只是给你一份草稿，务必逐行 review：它擅长加表加列，但对改类型、重命名、数据回填无能为力，且可能漏掉服务端默认值。草稿里的 # ### commands auto generated ### 注释就是提醒你「这里需要人工确认」。
6.3.4 应用、回退与查看
核心命令就四条：
| 命令 | 作用 |
|---|---|
alembic upgrade head | 应用所有未执行的迁移到最新 |
alembic upgrade +1 / alembic downgrade -1 | 前进 / 回退一步 |
alembic current | 显示数据库当前 revision |
alembic history | 列出迁移链 |
实测应用首个迁移：
INFO [alembic.runtime.migration] Running upgrade -> 33415f0a9199, create user
执行后数据库里出现了两张表——业务表加一张 alembic_version（记录当前 revision）：
['alembic_version', 'user']
6.3.5 演进：加一列再验证
给 User 加一个可空的 email 列，再生成一次迁移：
class User(Base):
__tablename__ = "user"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str]
email: Mapped[str | None]
alembic revision --autogenerate -m "add email"
alembic upgrade head
alembic history
alembic current
实测输出：
INFO [alembic.autogenerate.compare.tables] Detected added column 'user.email'
INFO [alembic.runtime.migration] Running upgrade 33415f0a9199 -> 498b5ecc19de, add email
33415f0a9199 -> 498b5ecc19de (head), add email
<base> -> 33415f0a9199, create user
498b5ecc19de (head)
最终表结构（SQLite 实测）：
CREATE TABLE user (
id INTEGER NOT NULL,
name VARCHAR NOT NULL, email VARCHAR,
PRIMARY KEY (id)
)
history 是从新到旧列出的链，current 显示数据库停在 498b5ecc19de (head)，与模型一致。
6.3.6 演进策略
迁移是不可逆操作的重灾区，策略比命令更重要：
- 加列一律可空或带默认值：
ADD COLUMN ... NOT NULL在已有数据的表上会失败；先加可空列，回填后再收紧。 - 改类型/重命名要手写：autogenerate 常把「重命名」误判成「删一列加一列」，会丢数据。用
op.alter_column/op.rename_table显式表达。 - 数据回填单独写迁移：在
upgrade()里用op.execute("UPDATE ...")或op.get_bind()执行，别混在结构变更里。 - 大表分批：给千万行表加索引或回填时，考虑
CREATE INDEX CONCURRENTLY（PostgreSQL）等在线操作，避免长时间锁表。 - 一次迁移只做一件事：便于回滚和定位；把「建表」和「回填」拆成两个 revision。
- 先降级演练再上线：在预发环境
upgrade后downgrade -1再upgrade，确认双向可用。
6.3.7 与部署流水线的衔接
生产环境应用迁移的常见形态：
# 发布前，在一次性容器里执行（不要放进应用启动流程并发跑）
alembic upgrade head
为什么强调「不要放进应用启动」：多实例同时启动时，upgrade head 会并发争抢，可能两个进程同时跑同一个迁移。正确做法是把它作为独立的发布步骤（如 Kubernetes 的 initContainer 或 CI 的一个 job），执行一次即可。
如果确实要在代码里调用（例如测试 fixture），用编程式 API：
from alembic import command
from alembic.config import Config
cfg = Config("alembic.ini")
command.upgrade(cfg, "head")
测试里也可以用 command.downgrade(cfg, "base") 清库，保证每个用例从干净 schema 起步。
6.3.8 PostgreSQL 差异（本机未实测）
SQLite 不支持 ALTER COLUMN 的部分能力，且默认「非事务性 DDL」。换到 PostgreSQL 时要注意：
- PostgreSQL 支持事务性 DDL，迁移失败可整体回滚，比 SQLite 安全。
- 枚举类型需先
CREATE TYPE，删除时要单独DROP TYPE。 ALTER TABLE ... ALTER COLUMN ... TYPE在大表上可能重写全表，要评估锁窗口。- 方言专属的
JSONB、SERIAL、ON CONFLICT等，本机无 PG 服务端，未实测，正文不给出运行结果。
延伸阅读：Python 数据库与 ORM 完全指南 的迁移章节有更多方言对照。
小结
- Alembic 是数据库结构的版本控制：
alembic init -t generic migrations建骨架，env.py里把target_metadata指向Base.metadata。 alembic revision --autogenerate -m "..."生成迁移草稿，但必须人工 review，它处理不了改类型、重命名和数据回填。- 四条核心命令：
upgrade/downgrade/current/history；Alembic 用一张alembic_version表记录当前 revision。 - 加列要可空或带默认值；改类型、重命名、回填都手写；一次迁移只做一件事。
- 生产把
alembic upgrade head当独立发布步骤执行一次，不要塞进应用启动流程并发跑。 - PostgreSQL 的事务性 DDL 比 SQLite 更安全，但方言专属特性本机未实测，正文已标注。
第 6 章到此收口：6.1 用类型化模型把表映射成对象，6.2 把 session、事务、连接池和 N+1 这些运行时问题讲透，6.3 用 Alembic 让结构变更可追溯可回滚。数据访问层稳了，接下来 7.1 节转向接口契约——用 Pydantic V2 设计 API 模型，把本节讲的 ORM 模型与对外契约彻底分开。
阅读导航：上一节：事务、连接池与 N+1 治理 · 下一节：Pydantic V2 模型设计 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。