第26章：Alembic 进阶——分支、数据迁移与多人协作

davidwang456

417人浏览 · 2026-08-19 09:56:07

davidwang456 · 2026-08-19 09:56:07 发布

一、项目背景

“我本地跑的迁移脚本，在测试环境报错：‘Can’t locate revision identified by xyz’——怎么回事？”

周三上午，星云电商的测试环境 CI 流水线全线飘红。原因是两位开发分别在自己的分支上为订单表新增了字段：小李加了一个 discount_amount 列，小王加了一个 coupon_code 列。两人各自生成了 Alembic migration，都在本地和本地测试环境跑通了。但当两人的代码合并到 main 分支时，Alembic 发现了两条 head 指向不同的母版本——也就是所谓的"双头"（Multiple Heads）。

运维尝试先跑小李的迁移再跑小王的，结果小王的迁移因为依赖的 alembic_version 表状态与预期不一致而报错。CI 管道堵了三个小时，最后是大师手动合并了两条迁移链才解决。

这暴露了 Alembic 在团队协作中的典型问题：多人同时修改数据库 Schema 时，merge conflict 不是发生在代码层（Git），而是发生在迁移历史层（Alembic）。Git 合并了 Python 代码，但 Alembic 的迁移链仍然是两条分叉——需要手动创建 merge point。

另一个常见场景是数据迁移。某次上线需要把订单的 status 字段从 '0'/'1'/'2'（数字字符串）改为 'pending'/'paid'/'shipped'（语义化字符串）。这是一次 Schema 变更 + 数据回填的组合操作——先加新列、回填数据、再删旧列。如果只做 Schema 迁移而忘记数据回填，老数据的状态字段就会是 NULL。

本章将深入 Alembic 进阶话题：多 head 合并策略、数据迁移与 Schema 迁移拆分、在线 DDL 风险控制、以及生产变更窗口管理。

二、项目设计

场景：CI 故障修复后，大师召开了一次"迁移规范"专题会。白板上画着两条分叉的迁移链，最终汇入一个 merge point。

小胖：“上次双头故障太吓人了——我跟小王的迁移互相不认识对方，一合进去就炸。这以后每次多人改表都要串行排队？”

大师：“不用串行排队，但要学会处理分支。你们各自从同一个 head 出发生成各自的迁移——这是正常的并行开发。问题在于合并后没有创建 merge revision——一个同时指向两条父链的’汇合点’。”

小胖：“merge revision 听起来像 Git 的 merge commit？”

大师：“技术映射：Alembic 的 merge revision = Git 的 merge commit（将两条分支合二为一）；revision = Git commit（一次变更快照）；head = Git HEAD（当前最新版本）。”

大师：“来看具体操作——”

# 1. 查看当前状态——发现有多个 head
alembic heads
# 输出：
# a1b2c3d4 (小李的: add discount_amount)
# e5f6g7h8 (小王的: add coupon_code)
# → 两条 head！

# 2. 创建 merge revision——将两条 head 合为一条
alembic merge a1b2c3d4 e5f6g7h8 -m "merge discount and coupon"

# 3. 现在只有一个 head 了
alembic heads
# 输出：i9j0k1l2 (head) (merge discount and coupon)

小白：“merge revision 里面有什么？会生成新的表变更 SQL 吗？”

大师：“merge revision 的 upgrade() 和 downgrade() 是空函数——它不产生 DDL，只是一个指针，告诉 Alembic ‘这两条历史链都已经被应用了’。它的作用纯粹是让迁移链重新变成线性，后续的迁移只需要一个父节点。”

小胖：“技术映射：merge revision = 收费站合并路口——两条路（小李和小王的迁移）并成一条主路（后续迁移的共同起点）。”

小白：“那数据迁移呢——比如改状态枚举值、回填新列的默认值。这些应该放在 Schema 迁移里一起跑，还是分开跑？”

大师："这是最容易踩坑的地方。原则是：Schema 迁移和数据迁移要分开。Schema 迁移在 upgrade() 中是 DDL（ALTER TABLE ADD COLUMN），是事务性的——失败可以回滚。但数据迁移（UPDATE 百万行）会持有行锁、可能超时、失败后回滚代价大。拆成两步：

# revision_1: Schema 迁移（加列）
def upgrade():
    op.add_column('orders', sa.Column('discount_amount', sa.Numeric(12,2), nullable=True))
    op.add_column('orders', sa.Column('status_v2', sa.String(20), nullable=True))

# revision_2: 数据迁移（回填+转换）
def upgrade():
    # 回填默认值
    op.execute("UPDATE orders SET discount_amount = 0 WHERE discount_amount IS NULL")
    # 状态转换
    op.execute("UPDATE orders SET status_v2 = CASE status "
               "WHEN '0' THEN 'pending' WHEN '1' THEN 'paid' "
               "WHEN '2' THEN 'shipped' END")

小白：“那如果数据迁移失败了怎么办？Schema 已经改了——回不去了？”

大师：“这就是在线 DDL 的另一个关键技巧——先加列（nullable），后回填，再加 NOT NULL 约束。即使回填失败，新列是 nullable 的，不影响业务。回填成功后再加约束。如果要回滚，降级脚本也应该能处理。”

小胖：“技术映射：Schema 迁移 = 盖房子框架（可逆）；数据迁移 = 搬家具（耗时且有风险）；先建房再搬家具 = 安全变更顺序。”

大师：“最后一个要点——生产变更窗口。ALTER TABLE ... ADD COLUMN 在 PostgreSQL 上是轻量操作，不需要重写全表。但 ALTER TABLE ... ALTER COLUMN ... TYPE 可能触发全表重写——生产环境千万不能直接在高峰期跑。”

三、项目实战

实战目标

模拟两人修改同一模型产生双 head，创建 merge revision 合并；实现"状态枚举值回填"的数据迁移；配置在线 DDL 安全步骤。

步骤一：环境准备与基线

# 首先初始化 Alembic 环境（假设已有项目）
# alembic init alembic
# 编辑 alembic.ini 和 alembic/env.py 配置数据库连接

# 初始模型——创建 orders 表
from sqlalchemy import create_engine, String, Integer, Numeric, DateTime
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

# 基线 migration：
# alembic revision --autogenerate -m "create orders table"
# 生成的 upgrade():
# def upgrade():
#     op.create_table('orders',
#         sa.Column('id', sa.Integer(), nullable=False),
#         sa.Column('order_no', sa.String(32), nullable=False),
#         sa.Column('status', sa.String(2), nullable=False, server_default='0'),
#         sa.Column('total_amount', sa.Numeric(12,2), nullable=False),
#         sa.PrimaryKeyConstraint('id')
#     )

步骤二：模拟双头场景

"""ch26_alembic_advanced.py —— Alembic 进阶实战"""

# =============================================
# 场景 A：小李的迁移——新增 discount_amount 列
# =============================================

# 小李的模型：
class OrderA(Base):
    __tablename__ = "orders"
    id: Mapped[int] = mapped_column(primary_key=True)
    order_no: Mapped[str] = mapped_column(String(32))
    status: Mapped[str] = mapped_column(String(2))
    total_amount: Mapped[float] = mapped_column(Numeric(12, 2))
    discount_amount: Mapped[float | None] = mapped_column(Numeric(12, 2), nullable=True)  # 新增

# 小李执行：alembic revision --autogenerate -m "add discount_amount"
# 生成 migration_lee.py：
"""
revision: str = 'a1b2c3d4'
down_revision: str | None = '0001_base'
branch_labels: str | None = None
depends_on: str | None = None

def upgrade():
    op.add_column('orders', sa.Column('discount_amount', sa.Numeric(12,2), nullable=True))

def downgrade():
    op.drop_column('orders', 'discount_amount')
"""

# =============================================
# 场景 B：小王的迁移——新增 coupon_code 列 + 索引
# =============================================

# 小王的模型（与小李从同一基线下分叉）：
class OrderB(Base):
    __tablename__ = "orders"
    id: Mapped[int] = mapped_column(primary_key=True)
    order_no: Mapped[str] = mapped_column(String(32))
    status: Mapped[str] = mapped_column(String(2))
    total_amount: Mapped[float] = mapped_column(Numeric(12, 2))
    coupon_code: Mapped[str | None] = mapped_column(String(32), nullable=True)  # 新增

# 小王执行：alembic revision --autogenerate -m "add coupon_code"
# 生成 migration_wang.py：
"""
revision: str = 'e5f6g7h8'
down_revision: str | None = '0001_base'
branch_labels: str | None = None
depends_on: str | None = None

def upgrade():
    op.add_column('orders', sa.Column('coupon_code', sa.String(32), nullable=True))
    op.create_index('idx_orders_coupon', 'orders', ['coupon_code'])

def downgrade():
    op.drop_index('idx_orders_coupon')
    op.drop_column('orders', 'coupon_code')
"""

# =============================================
# 合并后状态：两个 head 同时存在
# $ alembic heads
# a1b2c3d4 (小李: add discount_amount) (head)
# e5f6g7h8 (小王: add coupon_code) (head)

步骤三：创建 merge revision 解决双头

# =============================================
# 解决双头：创建 merge revision
# =============================================

# 命令：alembic merge a1b2c3d4 e5f6g7h8 -m "merge discount and coupon"
# 或手动创建：
"""
revision: str = 'm9n0o1p2'
down_revision: tuple[str, str] | None = ('a1b2c3d4', 'e5f6g7h8')
branch_labels: str | None = None
depends_on: str | None = None

def upgrade():
    pass  # merge revision 不需要 DDL

def downgrade():
    pass  # merge revision 不需要 DDL
"""

# 验证线性：
# $ alembic heads
# m9n0o1p2 (head) → 只有一个 head 了

# $ alembic history
# 0001_base → a1b2c3d4 (add discount) ──┐
#            → e5f6g7h8 (add coupon)   ──┤
#                                        ├→ m9n0o1p2 (merge) (head)

步骤四：数据迁移——状态枚举值回填

# =============================================
# 实战：状态枚举值迁移（数字 → 语义化字符串）
# =============================================

# 步骤 1/3：Schema 迁移——加新列（nullable）
# alembic revision -m "add status_v2 column"
"""
def upgrade():
    # 加新列，允许 NULL
    op.add_column('orders', sa.Column('status_v2', sa.String(20), nullable=True))

def downgrade():
    op.drop_column('orders', 'status_v2')
"""

# 步骤 2/3：数据迁移——回填状态值
# alembic revision -m "backfill status_v2"
"""
from alembic import op
import sqlalchemy as sa

# 定义临时表结构用于数据迁移（避免依赖最新模型）
orders_table = sa.table(
    'orders',
    sa.column('id', sa.Integer),
    sa.column('status', sa.String(2)),
    sa.column('status_v2', sa.String(20)),
)

def upgrade():
    # 分批更新以减小锁影响，commit 分批提交
    conn = op.get_bind()

    # 使用批量 UPDATE（PostgreSQL 示例）
    conn.execute(
        orders_table.update().where(orders_table.c.status == '0').values(status_v2='pending')
    )
    conn.execute(
        orders_table.update().where(orders_table.c.status == '1').values(status_v2='paid')
    )
    conn.execute(
        orders_table.update().where(orders_table.c.status == '2').values(status_v2='shipped')
    )
    conn.execute(
        orders_table.update().where(orders_table.c.status == '3').values(status_v2='cancelled')
    )

def downgrade():
    op.execute("UPDATE orders SET status_v2 = NULL")
"""

# 步骤 3/3：Schema 迁移——删旧列/加 NOT NULL
# alembic revision -m "cleanup old status and add constraint"
"""
def upgrade():
    # 数据已回填完成，可以加 NOT NULL 约束（先 CREATE INDEX）
    op.alter_column('orders', 'status_v2', nullable=False)
    # 删除旧列（先备份，观察几天再执行）
    # op.drop_column('orders', 'status')  # 生产环境延迟执行

def downgrade():
    op.alter_column('orders', 'status_v2', nullable=True)
"""

步骤五：在线 DDL 安全实践

# =============================================
# 在线 DDL 安全清单
# =============================================

# 安全的列添加（PostgreSQL）：
"""
# 1. 加 nullable 列 —— 轻量操作，无锁表
op.add_column('orders', sa.Column('notes', sa.Text(), nullable=True))
# ✅ 安全，瞬时完成

# 2. 加 NOT NULL 列 + default —— 全表重写！
op.add_column('orders', sa.Column('version', sa.Integer(), nullable=False, server_default='1'))
# ⚠️ 会锁表重写！PostgreSQL 11+ 加 volatile default 较安全

# 安全顺序：
# ① 先加 nullable 列
# ② 数据回填
# ③ 再加 NOT NULL 约束（或 CHECK 约束）
"""

# 不安全的更改类型：
UNSAFE_DDL = """
-- !!! 危险操作（会全表重写） !!!
ALTER TABLE orders ALTER COLUMN total_amount TYPE numeric(15,2);
ALTER TABLE orders ALTER COLUMN order_no TYPE varchar(50);

-- 安全替代方案（PostgreSQL）：
-- ① 加新列（nullable）
-- ② 分批回填（带 LIMIT + 事务分批）
-- ③ 切换业务代码使用新列
-- ④ 再删旧列

-- 或使用 USING 子句原地转换（仅当 USING 不阻塞时）：
-- ALTER TABLE orders ALTER COLUMN total_amount TYPE numeric(15,2) USING total_amount::numeric(15,2);
"""

# 分批数据迁移模板：
BATCH_MIGRATION_TEMPLATE = """
from alembic import op
import time

def upgrade():
    conn = op.get_bind()
    batch_size = 10000
    offset = 0
    while True:
        result = conn.execute(
            orders_table.update()
            .where(orders_table.c.status == '0')
            .where(orders_table.c.status_v2 == None)
            .limit(batch_size)
            .values(status_v2='pending')
        )
        if result.rowcount == 0:
            break
        print(f"  回填 {result.rowcount} 行，offset={offset}")
        offset += batch_size
        time.sleep(0.1)  # 给其他查询让路
"""

print("在线 DDL 安全原则：")
print("  1. 加 nullable 列 → 回填数据 → 加约束（分三步，不一步到位）")
print("  2. 避免 ALTER COLUMN TYPE（原地改类型 = 全表重写）")
print("  3. 大表操作分批，每批后 sleep 或等待 replication lag 小于阈值")
print("  4. 生产变更安排在低峰期窗口，有回滚预案")

步骤六：自动化迁移校验

# =============================================
# pytest 中的 Alembic 迁移校验
# =============================================

"""
# tests/test_migrations.py
import pytest
from alembic.config import Config
from alembic import command
from sqlalchemy import create_engine, inspect, text

ALEMBIC_CFG_PATH = "alembic.ini"

@pytest.fixture
def alembic_config():
    return Config(ALEMBIC_CFG_PATH)

def test_migration_upgrade_downgrade_cycle(test_db_url):
    '''验证迁移链可完整 upgrade → downgrade → upgrade'''
    engine = create_engine(test_db_url)

    # 1. Upgrade to head
    command.upgrade(alembic_config, "head")

    # 2. Downgrade to base
    command.downgrade(alembic_config, "base")

    # 3. Upgrade again（验证 downgrade 清理干净了）
    command.upgrade(alembic_config, "head")

    # 4. 验证表结构完整性
    insp = inspect(engine)
    tables = insp.get_table_names()
    assert "orders" in tables, "orders 表应存在"

def test_data_migration_status_values(test_db_url):
    '''验证数据迁移：status_v2 全部已回填'''
    engine = create_engine(test_db_url)
    with engine.connect() as conn:
        null_count = conn.execute(
            text("SELECT COUNT(*) FROM orders WHERE status_v2 IS NULL")
        ).scalar()
        assert null_count == 0, f"有 {null_count} 行未回填 status_v2"

def test_unique_index_after_migration(test_db_url):
    '''验证迁移后唯一索引仍然生效'''
    engine = create_engine(test_db_url)
    # 确保部分唯一索引正确创建
    insp = inspect(engine)
    indexes = insp.get_indexes("orders")
    index_names = [idx["name"] for idx in indexes]
    assert "idx_orders_coupon" in index_names
"""

# =============================================
# CI 中的自动迁移检测
# =============================================

CI_MIGRATION_CHECK = """
# .github/workflows/alembic-check.yml
steps:
  - name: Check for missing migrations
    run: |
      alembic check
      # 如果模型与最新 migration 不一致，返回非零退出码

  - name: Verify migration chain
    run: |
      alembic history --verbose
      # 检查是否有多个 head

  - name: Test upgrade → downgrade cycle
    run: |
      pytest tests/test_migrations.py -v
"""

可能遇到的坑及解决方法

autogenerate 不能检测到所有变更

不检测的变更：表重命名、列重命名、ENUM 值增删、约束/索引的修改（部分）。
解决：autogenerate 生成的迁移检查一遍，手动补充遗漏项。表重命名用 op.rename_table()，列重命名用 op.alter_column(..., new_column_name=...)。

alembic upgrade 在已有数据的环境失败

现象：加了 nullable=False 的列，已有行没有默认值，ALTER TABLE 失败。
解决：先 nullable=True + 回填默认值 + 再 nullable=False。三步走。

alembic downgrade 不可靠（罕见）

现象：downgrade 脚本写错了（如删列时写错了列名），无法回滚。
解决：每次写升级脚本的同时写降级脚本，并在 CI 中测试 upgrade → downgrade → upgrade 循环。

迁移脚本在生产超长耗时

现象：ALTER TABLE 加索引（PostgreSQL CREATE INDEX CONCURRENTLY 除外）会锁表，导致 API 超时。
解决：对 PostgreSQL 使用 CREATE INDEX CONCURRENTLY（op.create_index 默认不支持），需手动 op.execute("CREATE INDEX CONCURRENTLY ...")。

完整代码清单

完整迁移脚本示例参见 alembic/versions/ 目录（本章已逐段展示）

测试验证

# tests/test_ch26_migrations.py
import pytest
from sqlalchemy import create_engine, text, inspect

def test_upgrade_creates_all_tables(test_db_url):
    engine = create_engine(test_db_url)
    insp = inspect(engine)
    tables = insp.get_table_names()
    assert "orders" in tables
    assert "alembic_version" in tables

def test_column_exists_after_migration(test_db_url):
    engine = create_engine(test_db_url)
    insp = inspect(engine)
    cols = [c["name"] for c in insp.get_columns("orders")]
    assert "discount_amount" in cols
    assert "coupon_code" in cols
    assert "status_v2" in cols

def test_data_backfill_complete(test_db_url):
    engine = create_engine(test_db_url)
    with engine.connect() as conn:
        count = conn.execute(
            text("SELECT COUNT(*) FROM orders WHERE status_v2 IS NULL")
        ).scalar()
    assert count == 0

四、项目总结

Alembic 迁移模式对比

模式	使用场景	复杂度	风险等级
单人开发	个人项目，`autogenerate` 直接生成	低	低
多人协作	团队开发，需 merge revision + code review	中	中
在线 DDL	生产环境变更，分步执行 + 反向兼容	高	高
数据迁移	枚举回填、大表拆分、历史归档	极高	极高

适用场景

自动生成 + 人工审查：alembic revision --autogenerate 后检查——90% 场景适用。
手动编写：复杂的数据回填、CONCURRENTLY 索引创建、表分区操作——手动控制。
merge revision：多人并行开发后合并迁移分支。
分步迁移：大表结构变更——先加列、后回填、再删旧列。
CI 自动校验：在 pipeline 中跑 alembic check + upgrade → downgrade 循环测试。

不适用场景：

极高频变更（每天数次改表）——应重新评估 Schema 设计是否合理。
数据量极大的回填（数十亿行）——不应在 Alembic 迁移中做，应使用外部批处理作业。

注意事项

永远在测试环境先跑 alembic upgrade head——不要直接在预发/生产跑。
autogenerate 不能检测：表/列重命名、唯一约束变更、ENUM 枚举值变更。
Alembic 的 env.py 中的 target_metadata 必须与最新的模型同步。
alembic_version 表是轻量级的——只存一条记录指向最新 revision。

常见踩坑经验

案例 1：autogenerate 把已删除的字段生成了一次 DROP + 一次 ADD

现象：将字段 email 改名为 user_email，autogenerate 生成 op.drop_column('email') 和 op.add_column('user_email')——数据会丢失！
修复：手动改为 op.alter_column('users', 'email', new_column_name='user_email')。autogenerate 无法检测列重命名——它只对比"现在有哪些列"和"模型声明了哪些列"。

案例 2：Alembic downgrade 中调用 drop_table 时漏掉了依赖的 FK

现象：降级时 op.drop_table('order_items') 报错 cannot drop table because other objects depend on it。
根因：orders 表通过 FK 引用了 order_items——但没有在降级脚本中先处理外键。
修复：降级脚本按依赖顺序操作——先解 FK、再删表。

案例 3：迁移中的 server_default 与实际数据库的默认值格式不匹配

现象：server_default=sa.text("'active'") 生成的 DDL 依赖于数据库引号习惯，跨方言可能报错。
修复：使用 sa.text("'active'") 且只在 Alembic 中测试目标数据库（不要用 SQLite 测试 PostgreSQL 的迁移）。

思考题

生产环境的一张 5000 万行的表中需加一个 NOT NULL DEFAULT 0 的列。如果直接在高峰期执行 ALTER TABLE 会锁表几分钟——你如何设计一个零停机时间的变更方案？涉及哪些步骤和 Alembic 迁移脚本？
Alembic 的 branch_labels 机制（不同于 merge）用于支持不同数据库环境（如 PostgreSQL 生产 vs SQLite 测试）的迁移分支。如果你的团队本地用 SQLite、CI 用 PostgreSQL，应该如何配置 branch_labels 和 depends_on 来管理两种方言的迁移链？