Python 基础体系 · 第 84/112 篇。示例统一以 Python 3.14 为语言基线；第三方库使用与其兼容的现代稳定版本，版本敏感行为会单独说明。

Alembic 数据库迁移：Revision、自动生成、分支、上线和回滚

数据库表结构会随着业务代码一起变化：

  • 用户表增加 email；
  • 订单表增加状态字段；
  • 某个字符串字段改成更严格的类型；
  • 新增索引、外键或唯一约束；
  • 拆分一张表，或者把列重命名；
  • 在不停止服务的情况下逐步发布新结构。

如果只修改 SQLAlchemy 模型，而不修改数据库，代码与实际数据库就会产生偏差。如果直接在生产数据库上手工执行 SQL，又很难回答以下问题：

  1. 生产数据库当前到底执行过哪些结构变更？
  2. 某个变更是谁、何时、以什么顺序执行的？
  3. 测试环境和生产环境是否处于同一迁移版本？
  4. 发布失败后，能否安全恢复？
  5. 两个开发分支分别修改了数据库结构，合并代码后如何处理？

Alembic 解决的不是“自动修改数据库”这么简单，而是把数据库结构变化记录为一组有依赖关系的版本脚本，再根据数据库当前状态计算需要执行的路径。Alembic 是面向 SQLAlchemy 的数据库迁移工具；它的迁移脚本使用 Python 编写，并通过 Operations 对象发出数据库结构操作。(alembic.sqlalchemy.org)

本文示例使用 Python 3.14 语法，ORM 部分采用 SQLAlchemy 2.0 风格。Python 版本与 Alembic、SQLAlchemy 版本应分别固定在项目依赖中，不能因为 Python 版本升级就隐含升级数据库工具。Python 3.14 的官方文档目前对应 3.14.7。(docs.python.org)

一、先区分三个对象：模型、数据库和迁移脚本

Alembic 的核心比较过程可以抽象成：

迁移候选=diff⁡(数据库当前结构,代码中的目标 Metadata)\text{迁移候选} = \operatorname{diff}(\text{数据库当前结构}, \text{代码中的目标 Metadata})

其中：

  • 数据库当前结构：通过数据库连接和数据库方言反射得到；
  • 目标 Metadata：通常是 SQLAlchemy 声明式模型注册到的 Base.metadata；
  • 迁移候选：Alembic 根据差异生成的 op.add_column()、op.create_table() 等 Python 操作。

这三个对象的职责不同：

对象 含义 是否直接用于生产变更
SQLAlchemy 模型 应用程序希望使用的结构 否，模型本身不会自动改变数据库
实际数据库结构 当前数据库真实状态 是，迁移最终作用于它
Alembic revision 描述一次结构变更的版本脚本 是，生产环境执行它

例如，模型从：

class User(Base):
    __tablename__ = "user_account"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))

变成：

class User(Base):
    __tablename__ = "user_account"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    email: Mapped[str | None] = mapped_column(String(320), nullable=True)

这里仅仅表示：

目标模型声明了一个新的可空列。

数据库不会因为 Python 类多了一个属性，就自动执行：

ALTER TABLE user_account ADD COLUMN email VARCHAR(320);

Alembic 的职责是发现这个差异，生成一份候选 revision，再由工程师审查并执行。

二、Revision 是什么：数据库迁移图中的一个节点

2.1 Revision 不是简单的“序号”

一个 Alembic revision 通常包含以下元数据：

"""add email to user account."""

from collections.abc import Sequence

from alembic import op
import sqlalchemy as sa

revision: str = "a1b2c3d4e5f6"
down_revision: str | None = "001_create_user_account"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None

其中：

  • revision：当前迁移节点的唯一标识；
  • down_revision：当前节点的父节点；
  • branch_labels：可选的分支名称；
  • depends_on：可选的额外依赖；
  • upgrade()：从父节点升级到当前节点；
  • downgrade()：从当前节点退回父节点。

完整脚本通常是：

"""add email to user account."""

from alembic import op
import sqlalchemy as sa


revision = "a1b2c3d4e5f6"
down_revision = "001_create_user_account"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.add_column(
        "user_account",
        sa.Column("email", sa.String(length=320), nullable=True),
    )


def downgrade() -> None:
    op.drop_column("user_account", "email")

revision 看起来像一串随机字符串，但它不是数据库中的自增 ID。它的主要作用是让迁移文件能够被稳定引用。Alembic 根据 revision 和 down_revision 解析迁移关系，而不是根据文件名中的时间戳或字典序判断顺序。(alembic.sqlalchemy.org)

因此，下列文件名中的时间先后不决定执行顺序：

versions/
├── 20240101_create_user.py
└── 20231231_add_email.py

真正决定依赖关系的是：

# add_email.py
revision = "b"
down_revision = "a"

只要 b 的 down_revision 是 a，Alembic 就会把 a -> b 视为迁移路径。

2.2 Revision 组成一张有向无环图

如果每个 revision 只有一个父节点，图看起来像链表：

flowchart LR
    B[base] --> A[001_create_user]
    A --> C[002_add_email]
    C --> D[003_add_index]

但真实项目中经常出现分支：

flowchart LR
    B[base] --> A[001_create_user]
    A --> C[002_add_email]
    A --> E[003_add_avatar]
    C --> F[004_add_email_index]
    E --> G[005_add_avatar_index]

这里：

  • C 和 E 的父节点都是 A；
  • F 和 G 分别是两条分支上的后续 revision；
  • 当前存在两个 head：F 和 G。

因此，Alembic 的迁移文件不是一条简单链，而是一张有向无环图。升级过程需要在这张图上进行拓扑排序，找到从数据库当前版本到目标版本的可执行路径。Alembic 官方文档也明确说明，迁移过程将版本文件视为有向无环图，并通过拓扑排序处理分支和合并。(alembic.sqlalchemy.org)

2.3 alembic_version 表记录的是数据库状态

Alembic 通常会创建一张版本表：

CREATE TABLE alembic_version (
    version_num VARCHAR(32) NOT NULL,
    CONSTRAINT alembic_version_pkc PRIMARY KEY (version_num)
);

在单一 head 的普通项目中，它通常只有一行：

version_num
-----------
002_add_email

这表示：

该数据库已执行到 002_add_email，并且当前数据库状态对应这个 revision。

它不是迁移历史日志。历史信息仍然来自代码仓库中的 revision 文件；alembic_version 主要保存当前已应用的版本节点。

当迁移图存在多个 head 时，版本表可以有多行：

version_num
-----------
004_add_email_index
005_add_avatar_index

这两行分别表示两条分支都已执行到各自的 head。Alembic 在存在多个分支 head 时，会在版本表中保存多个版本值。(alembic.sqlalchemy.org)

这也解释了为什么以下命令含义不同：

alembic current
alembic heads
alembic history
  • current：查询目标数据库版本表中的当前版本；
  • heads：查询代码仓库迁移图中的所有末端节点；
  • history：查询迁移文件描述的历史关系。

如果 current 显示的版本不是 heads，不一定是错误，也可能表示数据库尚未升级到最新版本。

三、建立 Alembic 环境

3.1 安装依赖

一个最小项目可以固定如下依赖：

SQLAlchemy>=2.0,<2.1
alembic>=1.19,<1.20

安装：

python3.14 -m venv .venv
source .venv/bin/activate

python -m pip install -U pip
python -m pip install "SQLAlchemy>=2.0,<2.1" "alembic>=1.19,<1.20"

验证：

python --version
alembic --version

预期输出类似：

Python 3.14.x
alembic 1.19.1

实际版本以项目锁文件和运行环境为准。版本范围应与数据库驱动、SQLAlchemy 版本一起测试，而不是只看 Alembic 命令是否能启动。

3.2 初始化目录

在项目根目录执行：

alembic init migrations

典型结果：

Creating directory .../migrations ... done
Creating directory .../migrations/versions ... done
Generating .../alembic.ini ... done
Generating .../migrations/env.py ... done
Generating .../migrations/script.py.mako ... done
Please edit configuration/connection/logging settings in ...

生成目录：

.
├── alembic.ini
├── migrations
│   ├── env.py
│   ├── script.py.mako
│   └── versions
└── app
    ├── db.py
    └── models.py

各文件作用如下：

  • alembic.ini：命令行配置；
  • migrations/env.py：迁移运行环境；
  • migrations/script.py.mako：revision 文件模板；
  • migrations/versions/：实际迁移脚本；
  • alembic_version：运行迁移后在数据库中创建的版本表。

3.3 配置 SQLAlchemy URL

开发环境可以在 alembic.ini 中配置：

[alembic]
script_location = %(here)s/migrations
prepend_sys_path = .
sqlalchemy.url = sqlite:///./app.db

生产环境不应把密码直接提交到仓库。常见做法是在 env.py 中从环境变量读取：

import os

from alembic import context
from sqlalchemy import engine_from_config, pool


def get_database_url() -> str:
    url = os.environ["DATABASE_URL"]
    if not url:
        raise RuntimeError("DATABASE_URL is required")
    return url

然后配置 Engine：

def run_migrations_online() -> None:
    configuration = context.config.get_section(
        context.config.config_ini_section,
    )

    if configuration is None:
        raise RuntimeError("Alembic configuration section is missing")

    configuration["sqlalchemy.url"] = get_database_url()

    connectable = engine_from_config(
        configuration,
        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()

迁移进程使用 NullPool 是常见实现，因为迁移是一次性管理任务，不需要复用应用服务的连接池。但这不是 Alembic 的强制规范；重点是迁移任务要使用明确、独立、可审计的数据库连接配置。

四、env.py 为什么决定自动生成是否有效

4.1 target_metadata 必须完整

SQLAlchemy 模型通常定义一个声明式基类：

# app/db.py
from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass

模型：

# app/models.py
from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column

from app.db import Base


class User(Base):
    __tablename__ = "user_account"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100), nullable=False)

env.py 必须导入模型模块，使模型类注册到 Base.metadata：

from app.db import Base
from app import models  # noqa: F401

target_metadata = Base.metadata

只导入 Base，但没有导入 models，可能导致 Base.metadata.tables 为空。此时 Alembic 会把数据库中的业务表误认为“代码中不存在的表”，自动生成大量：

op.drop_table("user_account")

这不是数据库真的应该删除，而是目标 Metadata 不完整。

可以在命令行或测试中检查：

from app.db import Base
from app import models  # noqa: F401

print(sorted(Base.metadata.tables))

预期至少包含：

['user_account']

4.2 典型 env.py 结构

简化后的在线和离线配置如下：

from __future__ import annotations

import os
from logging.config import fileConfig

from alembic import context
from sqlalchemy import engine_from_config, pool
from sqlalchemy.engine import Connection

from app.db import Base
from app import models  # noqa: F401


config = context.config

if config.config_file_name is not None:
    fileConfig(config.config_file_name)

target_metadata = Base.metadata


def database_url() -> str:
    try:
        return os.environ["DATABASE_URL"]
    except KeyError as exc:
        raise RuntimeError("DATABASE_URL is required") from exc


def run_migrations_offline() -> None:
    context.configure(
        url=database_url(),
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()


def run_migrations_online() -> None:
    configuration = config.get_section(config.config_ini_section)
    if configuration is None:
        raise RuntimeError("Alembic config section is missing")

    configuration["sqlalchemy.url"] = database_url()

    connectable = engine_from_config(
        configuration,
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        run_migrations_with_connection(connection)


def run_migrations_with_connection(connection: Connection) -> None:
    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()

这里有两个不同的运行模式：

  • 在线模式：连接真实数据库并执行迁移；
  • 离线模式：不连接数据库，只生成 SQL 文本。

Alembic 的离线模式会把迁移操作输出到文件或标准输出，而不是直接操作数据库。(alembic.sqlalchemy.org)

五、创建 Revision：空白脚本与自动生成

5.1 创建空白 Revision

执行：

alembic revision -m "create user account"

预期输出：

Generating .../migrations/versions/8f2c_create_user_account.py ... done

生成的文件大致是：

"""create user account."""

from alembic import op
import sqlalchemy as sa


revision = "8f2c"
down_revision = None
branch_labels = None
depends_on = None


def upgrade() -> None:
    pass


def downgrade() -> None:
    pass

空白 revision 适合以下场景：

  • 数据迁移；
  • 创建视图、触发器、存储过程；
  • 执行数据库方言特有的 SQL；
  • 自动生成无法识别的结构变更；
  • 需要严格控制执行顺序的复杂变更。

例如初始化表：

def upgrade() -> None:
    op.create_table(
        "user_account",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("name", sa.String(length=100), nullable=False),
        sa.PrimaryKeyConstraint("id"),
    )


def downgrade() -> None:
    op.drop_table("user_account")

upgrade() 是前进操作，downgrade() 是逆向操作。两者不一定能在数学意义上完全互逆，因为数据删除、数据合并和不可逆类型转换可能无法恢复原状。

5.2 使用 --autogenerate

假设当前数据库已经执行到：

8f2c_create_user_account

然后模型增加：

email: Mapped[str | None] = mapped_column(
    String(320),
    nullable=True,
)

执行：

alembic revision --autogenerate -m "add email to user account"

Alembic 会：

  1. 连接配置中的数据库；
  2. 读取目标数据库的实际结构；
  3. 读取 target_metadata；
  4. 调用比较器生成差异；
  5. 把差异渲染为 Python 操作；
  6. 创建一个新的 revision 文件。

生成结果可能类似：

def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column(
        "user_account",
        sa.Column("email", sa.String(length=320), nullable=True),
    )
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_column("user_account", "email")
    # ### end Alembic commands ###

--autogenerate 生成的是候选迁移，不是经过验证的最终迁移。官方文档明确要求人工检查和修正自动生成结果。(alembic.sqlalchemy.org)

5.3 自动生成实际比较了什么

设：

  • DD 表示数据库当前结构；
  • MM 表示模型 Metadata；
  • CC 表示 Alembic 配置；
  • PP 表示数据库方言能力。

自动生成可以近似表示为：

R=Render⁡(Compare⁡(D,M,C,P))R = \operatorname{Render}(\operatorname{Compare}(D, M, C, P))

它不是把 Python 类直接翻译成 SQL，而是先比较两个结构描述，再把差异转成迁移操作。

通常可以检测：

  • 新增或删除表；
  • 新增或删除列；
  • 可空性变化；
  • 基本索引变化；
  • 显式命名的唯一约束；
  • 基本外键变化；
  • 某些命名 CHECK 约束变化；
  • 默认情况下的主要类型变化。

类型比较默认启用，但可以通过 compare_type 调整。服务器端默认值比较默认关闭，需要设置 compare_server_default=True 或提供自定义比较函数。(alembic.sqlalchemy.org)

例如：

context.configure(
    connection=connection,
    target_metadata=target_metadata,
    compare_type=True,
    compare_server_default=True,
)

开启默认值比较后，仍需检查结果。不同数据库对默认表达式的格式化和等价判断可能不同，某些后端还会实际执行表达式进行比较，因此自动判断不应替代人工审查。(alembic.sqlalchemy.org)

5.4 自动生成无法可靠理解重命名

假设模型从：

display_name = mapped_column(String(100))

改成：

full_name = mapped_column(String(100))

从业务角度看，这是重命名。但从结构差异角度看，Alembic 看到的是：

删除 display_name
新增 full_name

于是可能生成：

op.add_column(
    "user_account",
    sa.Column("full_name", sa.String(length=100), nullable=True),
)
op.drop_column("user_account", "display_name")

这会导致原列数据丢失。

正确做法是手工改为：

def upgrade() -> None:
    op.alter_column(
        "user_account",
        "display_name",
        new_column_name="full_name",
    )


def downgrade() -> None:
    op.alter_column(
        "user_account",
        "full_name",
        new_column_name="display_name",
    )

表重命名同理。Alembic 官方文档明确指出，表名和列名变化通常会被识别为“删除旧对象、创建新对象”，必须人工修改为重命名操作。匿名约束也无法稳定匹配，因此约束应使用明确名称。(alembic.sqlalchemy.org)

5.5 自动生成无法替代数据迁移

模型增加一个非空列：

nickname: Mapped[str] = mapped_column(
    String(100),
    nullable=False,
)

如果已有数据，直接生成：

op.add_column(
    "user_account",
    sa.Column("nickname", sa.String(length=100), nullable=False),
)

通常会失败，因为已有行没有 nickname 值。

安全的迁移需要分阶段：

def upgrade() -> None:
    op.add_column(
        "user_account",
        sa.Column("nickname", sa.String(length=100), nullable=True),
    )

    op.execute(
        """
        UPDATE user_account
        SET nickname = name
        WHERE nickname IS NULL
        """
    )

    op.alter_column(
        "user_account",
        "nickname",
        nullable=False,
    )

这里的中间状态是：

不存在 nickname→可空 nickname→填充历史数据→非空 nickname\text{不存在 nickname} \rightarrow \text{可空 nickname} \rightarrow \text{填充历史数据} \rightarrow \text{非空 nickname}

如果数据量很大，直接在一次事务中更新全表可能产生长事务、锁等待或复制延迟。此时可以拆成多个 revision：

  1. 新增可空列；
  2. 应用双写；
  3. 分批回填；
  4. 校验无空值；
  5. 增加非空约束；
  6. 删除旧字段或停止旧写入。

自动生成只能发现“列属性发生变化”，不能知道历史数据应如何转换。

六、Migration Operation：迁移文件真正执行的内容

Alembic 迁移脚本中的 op 是 Operations 对象。常见操作包括：

op.create_table(...)
op.drop_table(...)
op.add_column(...)
op.drop_column(...)
op.alter_column(...)
op.create_index(...)
op.drop_index(...)
op.create_unique_constraint(...)
op.create_foreign_key(...)
op.execute(...)

6.1 创建表

def upgrade() -> None:
    op.create_table(
        "article",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("title", sa.String(200), nullable=False),
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("CURRENT_TIMESTAMP"),
            nullable=False,
        ),
        sa.PrimaryKeyConstraint("id"),
    )

server_default 表示由数据库服务器提供默认值，和 Python 层的 default 不同：

# Python/SQLAlchemy 侧默认值
default=datetime.now

# 数据库服务器侧默认值
server_default=sa.text("CURRENT_TIMESTAMP")

迁移脚本描述的是数据库结构，因此需要明确区分这两类默认值。

6.2 修改列属性

def upgrade() -> None:
    op.alter_column(
        "article",
        "title",
        existing_type=sa.String(length=200),
        type_=sa.String(length=500),
        existing_nullable=False,
        nullable=False,
    )

在某些数据库方言中，existing_type、existing_nullable、existing_server_default 等信息有助于 Alembic 生成完整且正确的 DDL。它们不是装饰性参数，而是告诉 Alembic 当前列的原始状态。

6.3 创建索引

def upgrade() -> None:
    op.create_index(
        "ix_article_created_at",
        "article",
        ["created_at"],
    )


def downgrade() -> None:
    op.drop_index(
        "ix_article_created_at",
        table_name="article",
    )

索引名应稳定、明确，并在回滚中使用同一个名字。不要依赖数据库自动生成的匿名名称，否则不同数据库、不同环境中的名称可能不一致。

6.4 执行数据迁移 SQL

from sqlalchemy import text


def upgrade() -> None:
    op.execute(
        text(
            """
            UPDATE article
            SET title = TRIM(title)
            WHERE title IS NOT NULL
            """
        )
    )

如果 SQL 使用参数，应避免字符串拼接：

op.execute(
    text(
        """
        UPDATE article
        SET title = :title
        WHERE id = :id
        """
    ).bindparams(title="untitled", id=1)
)

迁移脚本中的 SQL 应考虑：

  • 数据库方言；
  • 锁范围；
  • 执行时间；
  • 是否可重复执行；
  • 是否需要分批；
  • 是否需要先验证数据；
  • 失败后是否能恢复。

七、分支：为什么会出现多个 Head

7.1 分支的形成

假设主干当前是：

A -> B

开发者甲基于 B 创建：

B -> C

开发者乙也基于 B 创建：

B -> D

合并代码后，迁移图变成：

flowchart LR
    A[001 A] --> B[002 B]
    B --> C[003 C]
    B --> D[004 D]

此时 C 和 D 都没有子节点，因此都是 head：

alembic heads

可能输出：

003_c
004_d

head 的定义是：

当前迁移图中没有其他 revision 将它作为父节点的末端 revision。

它不是“最新时间生成的文件”，也不是“版本号最大”的节点。

7.2 多个 Head 是否一定错误

不一定。

如果多个分支代表独立且长期存在的迁移线，可以保留多个 head。例如不同数据库模块、不同版本树或独立的数据库环境可能需要不同分支。

但在普通单体应用中，多个 head 往往表示两个开发分支分别修改了数据库结构。此时通常应创建 merge revision，把它们合并为一个共同后继。

7.3 head、heads 与分支标签

当只有一个 head 时：

alembic upgrade head

表示升级到唯一 head。

当有多个 head 时：

alembic upgrade head

无法唯一确定目标，通常需要：

alembic upgrade heads

这表示升级到所有 head。

也可以指定某个具体分支：

alembic upgrade 003_c

如果使用了分支标签：

branch_labels = ("billing",)

就可以写：

alembic upgrade billing@head

分支标签可以作用于该分支的后代，并允许通过 branch@head 形式引用某个分支的 head。(alembic.sqlalchemy.org)

八、Merge Revision：合并迁移分支，而不是合并表结构

8.1 创建合并节点

假设当前有两个 head：

003_c
004_d

执行：

alembic merge -m "merge feature migrations" 003_c 004_d

生成：

"""merge feature migrations."""

from collections.abc import Sequence

from alembic import op


revision = "005_merge"
down_revision: tuple[str, str] = ("003_c", "004_d")
branch_labels = None
depends_on = None


def upgrade() -> None:
    pass


def downgrade() -> None:
    pass

迁移图从：

      C
     / 
B --<
     \
      D

变成：

      C
     / \
B --<   M
     \ /
      D

合并 revision 的关键不是执行某个新的表结构操作，而是声明：

只有当 C 和 D 都完成后，才认为迁移图到达 M。

Alembic 官方示例中的 merge 文件通过 down_revision 指向多个 revision，通常 upgrade() 和 downgrade() 可以为空；如果两个分支需要额外协调，也可以把协调逻辑放入合并节点。(alembic.sqlalchemy.org)

8.2 Merge 不能自动解决结构冲突

假设两个分支分别执行：

分支 C：把 user.name 改成 String(200)
分支 D：把 user.name 改成 Integer

创建 merge revision 并不会判断哪个类型正确，也不会自动合并这两个操作。因为两个分支都可能已经在不同数据库中独立执行过，迁移历史的合并与业务结构冲突的解决是两个问题。

处理步骤应是：

  1. 查看两个分支的 upgrade()；
  2. 确认两条路径分别对数据库做了什么；
  3. 设计最终目标结构；
  4. 必要时修改其中一个 revision，或增加新的修正 revision；
  5. 在干净数据库上从 base 执行完整迁移；
  6. 在分别应用过两条分支的数据库上测试升级到 merge 点。

8.3 --splice 的含义

通常新 revision 会接在某个 head 后面：

alembic revision -m "add payment table" --head 005_merge

如果指定的父版本不是当前 head，却希望从它创建一条新的独立分支，需要：

alembic revision \
  -m "create historical branch" \
  --head 002_b \
  --splice

--splice 的意义是：

明确告诉 Alembic：即使指定的父节点不是当前 head，也要从这里创建新的 head。

这不是普通开发流程中的常用选项。误用它会让迁移图出现额外分支，之后可能需要 merge。

九、迁移命令的状态变化

9.1 初始状态

代码仓库：

A -> B -> C

数据库：

alembic_version = A

执行：

alembic upgrade head

Alembic 计算路径：

A -> B -> C

依次执行：

upgrade A -> B
upgrade B -> C

最终：

alembic_version = C

9.2 指定版本

alembic upgrade B

表示：

将数据库升级到 B，只执行到该节点所需的祖先路径。

如果数据库当前是 A，只执行 A -> B。如果数据库当前已经是 C，则这个命令会执行降级路径，具体行为取决于目标版本相对于当前版本的位置。

查看当前状态：

alembic current

查看完整历史：

alembic history --verbose

查看所有 head：

alembic heads --verbose

查看某个 revision：

alembic show C

相对版本也可以使用：

alembic upgrade +2
alembic downgrade -1

其中 +2 表示从当前状态向前两个迁移步骤，-1 表示向后一个步骤。(alembic.sqlalchemy.org)

9.3 stamp 不执行迁移

alembic stamp C

它只修改版本表，不执行 A -> B -> C 中的任何 DDL 或数据迁移。

因此：

真实数据库结构：仍然是 A
alembic_version：被标记为 C

这会制造严重的不一致。

stamp 适合的场景是：

  • 已经通过外部工具完成了结构初始化；
  • 需要把一个已人工确认等价于某版本的数据库纳入 Alembic 管理；
  • 修复经过审计的版本表状态。

它不适合代替正式迁移。Alembic 官方对 stamp 的定义就是“更新版本表但不运行迁移”。(alembic.sqlalchemy.org)

十、上线：迁移不是“服务启动时顺便执行”

10.1 推荐的数据流

生产发布至少包含以下对象：

sequenceDiagram
    participant CI as CI/CD
    participant M as Migration Job
    participant DB as Database
    participant APP as Application Instances
    participant U as Users

    CI->>M: 发布迁移版本
    M->>DB: alembic current
    M->>DB: alembic upgrade head
    DB-->>M: 迁移成功或失败
    M->>DB: 结构与数据校验
    M-->>CI: 迁移结果
    CI->>APP: 发布兼容版本
    APP->>DB: 读写新旧兼容结构
    U->>APP: 正常请求

核心顺序是：

  1. 确认连接的是正确数据库；
  2. 查询当前 revision；
  3. 执行目标 revision；
  4. 验证结构、索引和数据；
  5. 再发布依赖新结构的应用代码。

把迁移放在每个 Web 进程启动时执行，会产生竞争：

进程 P1 启动 -> 尝试迁移
进程 P2 启动 -> 同时尝试迁移
进程 P3 启动 -> 同时尝试迁移

不同数据库对 DDL、锁和版本表更新的行为不同。即使某些情况下第二个进程最终会等待，也不应把迁移竞争交给应用进程处理。更可控的做法是使用一个独立 migration job，并让应用进程只启动已经确认结构兼容的版本。

10.2 迁移必须在正确代码版本下执行

迁移脚本通常会导入：

from app import models

因此迁移 job 使用的应用代码必须包含完整的模型和迁移文件。常见错误是：

代码仓库包含新 revision
容器镜像却是旧代码
migration job 看不到新脚本

或者：

应用模型已经改变
migration job 使用的镜像没有对应模型
--autogenerate 产生错误差异

生产迁移应绑定到不可变的构建产物，例如镜像 digest 或明确 commit，而不是直接读取工作目录中的未提交文件。

10.3 在线迁移中的锁风险

以下操作可能造成明显锁等待或长时间 DDL：

  • 重建大表；
  • 为大表创建非并发索引；
  • 修改大字段类型；
  • 增加需要扫描全表的约束；
  • 全表数据回填；
  • 删除列或重写表结构。

迁移成功不等于业务无感。应在测试环境观察：

  • DDL 执行时间；
  • 锁等待；
  • 事务日志增长；
  • 复制延迟；
  • 连接池耗尽；
  • 应用请求错误率。

某些数据库支持特定的在线索引创建方式，但这属于数据库方言能力，不能把某一个数据库的 SQL 当成 Alembic 的通用保证。

10.4 Offline SQL

如果上线流程要求 DBA 审核 SQL，可以生成离线脚本：

alembic upgrade head --sql > migration.sql

检查：

sed -n '1,240p' migration.sql

离线模式的输入仍然是迁移脚本和 Alembic 配置，而不是数据库当前状态。它不会像在线模式那样连接目标数据库判断真实版本，因此执行前必须明确：

  • SQL 对应的起始 revision；
  • DBA 将在哪个数据库执行；
  • 数据库方言是否正确；
  • 是否需要手工分段；
  • 是否包含不可在事务中执行的语句。

--sql 生成的是待审阅脚本，不代表已经成功应用。Alembic 官方将其定义为输出 SQL 而不连接数据库执行迁移。(alembic.sqlalchemy.org)

十一、Expand/Contract：零停机变更的基本模型

当新旧应用版本会并存时，数据库结构必须同时兼容它们。一个典型流程是：

flowchart LR
    A[旧结构] --> B[Expand: 增加兼容结构]
    B --> C[发布可同时读写旧新字段的应用]
    C --> D[回填历史数据]
    D --> E[校验数据一致性]
    E --> F[发布只依赖新结构的应用]
    F --> G[Contract: 删除旧结构]

例如把：

user.name

迁移为：

user.first_name
user.last_name

不应直接执行：

op.drop_column("user", "name")
op.add_column("user", sa.Column("first_name", ...))
op.add_column("user", sa.Column("last_name", ...))

因为旧版本应用仍然读取 name。

更安全的过程是：

第一步：Expand

def upgrade() -> None:
    op.add_column(
        "user",
        sa.Column("first_name", sa.String(100), nullable=True),
    )
    op.add_column(
        "user",
        sa.Column("last_name", sa.String(100), nullable=True),
    )

第二步：应用双写

新应用写入：

name = "Ada Lovelace"
first_name = "Ada"
last_name = "Lovelace"

旧应用仍然可以继续读写 name。

第三步：回填

def upgrade() -> None:
    op.execute(
        """
        UPDATE user
        SET first_name = split_part(name, ' ', 1),
            last_name = substring(name from position(' ' in name) + 1)
        WHERE first_name IS NULL
          AND last_name IS NULL
        """
    )

这段 SQL 只适用于支持相应字符串函数的数据库。跨数据库项目应使用目标数据库方言，或在 Python 任务中分批处理。

第四步：Contract

当所有应用实例都不再依赖 name，并且确认数据已完成迁移后，才删除旧列：

def upgrade() -> None:
    op.drop_column("user", "name")

删除旧列通常是不可逆的，因此应放在单独发布阶段，而不是和新增列放在同一个不可观察的大迁移中。

十二、回滚：代码回滚、迁移回滚和数据恢复不是一回事

12.1 Alembic downgrade 的含义

如果当前版本是：

C

执行：

alembic downgrade -1

Alembic 会调用 C 的：

def downgrade() -> None:
    ...

然后把版本表更新到 C 的父节点。

例如：

alembic downgrade 002_add_email

表示回退到指定 revision，而不是“回退最近一个文件名”。

12.2 可逆 DDL 与不可逆数据变化

以下操作通常较容易逆转：

新增索引 -> 删除索引
新增可空列 -> 删除列
创建表 -> 删除表

但以下操作可能不可逆：

删除列并丢弃数据
合并两列且无法拆分
截断表
改变值域并丢失原始信息
删除枚举值
压缩或聚合历史记录

例如：

def upgrade() -> None:
    op.drop_column("user_account", "legacy_status")

对应的 downgrade() 即使写成：

def downgrade() -> None:
    op.add_column(
        "user_account",
        sa.Column("legacy_status", sa.String(20), nullable=True),
    )

也只恢复了列，不可能恢复已经删除的业务数据。

因此，downgrade() 的正确性至少有两个层次：

  1. 结构可逆：数据库表结构能回到原状态；
  2. 数据可逆：原有业务数据也能恢复。

第二个条件往往无法仅靠 Alembic 保证，需要备份、审计表或专门的数据回滚脚本。

12.3 生产回滚的判断顺序

发布失败后，不应机械地执行：

alembic downgrade -1

更合理的判断顺序是：

  1. 失败发生在迁移前、迁移中还是迁移后；
  2. 数据库事务是否已经回滚；
  3. 当前 alembic current 是什么；
  4. 新旧应用是否都能读当前结构；
  5. 是否已经发生不可逆数据变更；
  6. 是否应该回滚应用代码，还是只修复应用；
  7. 是否需要从备份恢复。

如果迁移失败且数据库支持事务性 DDL，可能整个 migration transaction 已回滚；但不是所有数据库和所有 DDL 都有相同保证。迁移日志和数据库实际结构必须同时验证，不能仅凭命令退出码推断最终状态。

十三、事务边界与失败路径

典型的在线迁移代码是：

with connectable.connect() as connection:
    context.configure(connection=connection)

    with context.begin_transaction():
        context.run_migrations()

迁移流程可以抽象为：

开始事务
  ↓
执行 revision 1
  ↓
更新版本表
  ↓
执行 revision 2
  ↓
更新版本表
  ↓
提交事务

如果某一步抛出异常：

开始事务
  ↓
执行 revision 1
  ↓
执行 revision 2
  ↓
异常
  ↓
回滚事务

但这只是迁移框架层面的事务边界。实际行为取决于：

  • 数据库是否支持事务性 DDL；
  • 某条 DDL 是否隐式提交；
  • 是否使用在线索引、并发索引等特殊语句；
  • 迁移脚本是否显式开启了独立事务；
  • 数据库驱动如何处理异常。

因此，迁移脚本不能假设所有 DDL 都像普通 INSERT 一样可回滚。

十四、SQLite 的特殊边界：Batch Migration

SQLite 对部分 ALTER TABLE 能力有限，修改列、删除列、修改约束时，常见做法是：

  1. 创建新表；
  2. 复制旧表数据；
  3. 删除旧表；
  4. 重命名新表；
  5. 重建索引和约束。

Alembic 提供 batch operations：

def upgrade() -> None:
    with op.batch_alter_table("user_account") as batch_op:
        batch_op.alter_column(
            "name",
            existing_type=sa.String(length=100),
            type_=sa.String(length=200),
        )

SQLite 是否采用重建表流程取决于具体操作和配置；不能把 SQLite 的行为推断到 PostgreSQL、MySQL 或其他数据库。生产项目若使用 SQLite，还应专门测试：

  • 外键约束；
  • 索引恢复；
  • 默认值；
  • 触发器；
  • 大表复制时间；
  • 中断后的恢复。

Alembic 官方文档将 SQLite 的 batch migration 作为单独主题处理，原因正是不同数据库对 DDL 的能力和实现方式不同。(alembic.sqlalchemy.org)

十五、约束命名：自动生成能否识别对象的前提

不稳定或匿名的约束会使自动生成难以判断“这是同一个约束被修改了，还是新建了另一个约束”。

可以在 SQLAlchemy Metadata 中统一配置命名规则：

from sqlalchemy import MetaData
from sqlalchemy.orm import DeclarativeBase


convention = {
    "ix": "ix_%(column_0_label)s",
    "uq": "uq_%(table_name)s_%(column_0_name)s",
    "ck": "ck_%(table_name)s_%(constraint_name)s",
    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
    "pk": "pk_%(table_name)s",
}


class Base(DeclarativeBase):
    metadata = MetaData(naming_convention=convention)

显式命名示例：

from sqlalchemy import ForeignKey, UniqueConstraint


class Article(Base):
    __tablename__ = "article"

    id: Mapped[int] = mapped_column(primary_key=True)
    slug: Mapped[str] = mapped_column(String(200), nullable=False)

    __table_args__ = (
        UniqueConstraint("slug", name="uq_article_slug"),
    )

命名的价值不仅是自动生成：

  • 回滚时可以准确删除；
  • 线上诊断时可以定位；
  • 不同环境的名称更稳定；
  • 数据库错误信息更容易理解；
  • 迁移脚本不依赖数据库随机生成名称。

十六、校验：让 CI 阻止结构漂移

16.1 在干净数据库上执行完整迁移

CI 应至少运行：

alembic upgrade head

然后运行应用测试或结构检查。

这能发现：

  • revision 依赖错误；
  • 迁移顺序错误；
  • 遗漏导入；
  • 不兼容的 DDL；
  • 新增非空列没有默认值；
  • 数据迁移 SQL 无法执行。

16.2 测试 downgrade

如果项目要求支持回滚，可以运行：

alembic downgrade base
alembic upgrade head

但这个测试需要明确其含义：

  • downgrade base 会执行全部逆向迁移；
  • 可能删除业务数据；
  • 不应对真实生产数据库执行；
  • 对不可逆迁移必须单独设计测试和恢复方案。

更细粒度的测试：

alembic upgrade 002_add_email
alembic downgrade 001_create_user_account

这样可以验证相邻 revision 的结构往返。

16.3 使用 alembic check

Alembic 提供：

alembic check

它使用与 revision --autogenerate 相同的模型比较过程，检查当前数据库与目标 Metadata 之间是否存在新的升级操作。没有差异时，输出类似：

No new upgrade operations detected.

有差异时，命令失败并显示候选操作。(alembic.sqlalchemy.org)

需要注意，alembic check 继承自动生成的局限：

  • 它不能可靠识别重命名；
  • 它依赖 target_metadata 完整；
  • 它可能无法发现某些数据库对象；
  • 它发现的是结构差异，不是业务数据差异。

因此 CI 可以把它作为结构漂移检查，但不能把它当成迁移正确性的完整证明。

十七、常见失败表现与诊断路径

17.1 Target database is not up to date

常见原因是：

数据库当前版本：A
代码迁移 head：C

但你直接执行：

alembic revision --autogenerate -m "new change"

先查询：

alembic current
alembic heads

如果数据库不是最新 head，应先：

alembic upgrade head

然后再生成新的 revision。否则自动生成比较的是一个过旧的数据库结构，候选结果可能混合历史差异和新差异。

17.2 自动生成大量 drop_table

首先检查：

print(sorted(target_metadata.tables))

如果模型表没有出现在 Metadata 中，说明模型模块未导入或导入路径不正确。

其次检查：

  • 连接的是否是目标数据库；
  • DATABASE_URL 是否指向测试库；
  • 数据库是否包含由其他系统管理的表；
  • include_object 是否需要过滤外部表。

把错误的 drop_table() 提交到生产是高风险操作。自动生成文件必须经过人工审查。

17.3 Multiple head revisions are present

说明代码仓库有多个 head。诊断：

alembic heads --verbose
alembic branches --verbose
alembic history --verbose

如果这些 head 是同一应用的并行开发分支，创建 merge revision：

alembic merge -m "merge migration heads" heads

如果它们是设计上的独立分支，则使用：

alembic upgrade heads

或：

alembic upgrade branch_name@head

不要为了消除错误而随意删除其中一个 revision。删除已经被环境执行过的 revision，会破坏数据库状态与代码历史的对应关系。

17.4 Can't locate revision identified by ...

这表示数据库版本表引用了一个当前代码仓库找不到的 revision。常见原因：

  • 迁移文件没有提交；
  • 部署漏了 migrations/versions；
  • 错误删除或重命名了 revision；
  • 数据库使用了另一套迁移仓库；
  • 多个服务共用数据库但迁移代码不一致。

处理时应先保留现场：

SELECT version_num FROM alembic_version;

再检查代码仓库是否存在相应 revision。不要直接执行：

alembic stamp base

因为这会掩盖真实结构状态，除非已经通过备份、结构检查和人工审计确认数据库实际结构。

十八、生产迁移的最小闭环

一个可审计的迁移流程可以是：

修改 SQLAlchemy 模型
        ↓
确认 target_metadata 包含全部模型
        ↓
本地生成 revision --autogenerate
        ↓
人工检查重命名、默认值、约束和数据迁移
        ↓
在干净数据库 upgrade head
        ↓
在带历史数据的数据库 upgrade head
        ↓
测试应用新旧版本兼容性
        ↓
CI 执行 alembic check
        ↓
合并代码
        ↓
生产 migration job
        ↓
验证 current、结构、数据和应用指标
        ↓
发布应用代码

发布前至少应回答：

  • 迁移从哪个 down_revision 开始？
  • 当前生产数据库是什么 revision？
  • 是否存在多个 head？
  • 是否有非空列、唯一约束或外键会因历史数据失败？
  • 是否存在重命名但自动生成误判为删除加新增？
  • DDL 是否会锁大表？
  • 迁移失败后数据库处于什么状态？
  • 应用回滚时是否仍能兼容已经扩展的结构？
  • 哪些数据变化不可逆？
  • 是否有备份或补偿脚本？

十九、核心理解：Alembic 管理的是“状态转换路径”

Alembic 的关键抽象不是“保存几份 SQL 文件”，而是：

S0→r1S1→r2S2→r3S3S_0 \xrightarrow{r_1} S_1 \xrightarrow{r_2} S_2 \xrightarrow{r_3} S_3

其中：

  • SiS_i 是数据库结构和数据状态；
  • rir_i 是一个 revision；
  • upgrade() 描述 Si→Si+1S_i \to S_{i+1}；
  • downgrade() 尝试描述 Si+1→SiS_{i+1} \to S_i；
  • alembic_version 记录当前状态节点；
  • revision 文件通过 down_revision 组成迁移图。

自动生成解决的是：

从 Metadata 和数据库结构之间找出候选差异

它不解决：

业务数据如何转换
发布期间如何兼容
多个分支如何治理
大表 DDL 如何控制风险
不可逆操作如何恢复

分支解决的是：

多个 revision 路径如何共存

merge 解决的是：

多个 head 如何形成共同后继

上线解决的是：

如何在真实流量和并发环境中安全执行状态转换

回滚解决的是：

失败后是否能恢复应用、结构和数据

当这几个层次被区分后，Alembic 的使用就不再是机械执行 revision --autogenerate 和 upgrade head，而是对数据库状态、迁移图、应用兼容性和恢复路径进行明确建模。

系列导航与关联阅读

官方资料

本文依据 Python 官方文档、相关 PEP 与生态项目官方文档重新梳理；正文、示例与工程清单由 WR BLOG 编写。