alembic

alembic — SKILL.md

变体： 标准 · 使用时机： 调用此技能时，它会生成配置好的 Alembic 配置和经过审核的迁移脚本（或相关部分），随后控制权返回调用方。

概述

此技能指导 Agent 基于现有的 SQLAlchemy 2.x 数据层，在 SQLite、PostgreSQL 和 MySQL 上使用同一份迁移历史运行 Alembic 架构迁移。它假设已有可正常使用的 Base.metadata（这是同级 sqlalchemy 技能负责的内容），并将 Alembic 接入其中。主要路径为同步——使用同步连接的标准 run_migrations_online() 结构——另附简短的异步说明。两点至关重要：（1）自动生成的草稿需要你审核——其已知盲点若不经审核直接应用，可能导致静默数据丢失；（2）SQLite 批量操作／移动并复制的注意事项——SQLite 无法对大多数对象执行 ALTER，因此迁移采用 batch_alter_table 风格：在 SQLite 上重建表，在 PG/MySQL 上则生成普通的 ALTER。此技能还负责处理 create_all() 与 Alembic 之间的选择，以及在已创建的数据库上采用 Alembic 时使用的基线标记流程。

何时启用

• ✅ 将 Alembic（env.py + alembic.ini）接入现有的 SQLAlchemy Base.metadata，以便自动生成迁移时对比实时架构与模型。
• ✅ 编写迁移——空迁移（revision）或自动生成的草稿（revision --autogenerate）——并在应用前审核和编辑。
• ✅ 运行 upgrade / downgrade，检查 history / current / heads，或使用 merge 解决多个头。
• ✅ 编写一个必须同时适用于 SQLite、PostgreSQL 和 MySQL 的迁移——使用批量操作（移动并复制）。
• ✅ 为项目决定使用 Base.metadata.create_all() 还是 Alembic，或为通过 create_all 创建的数据库标记基线，以采用 Alembic。

请勿在以下情况启用：

• 定义 SQLAlchemy 模型 / Engine / 连接 URL / 会话 → sqlalchemy。此技能仅将 Alembic 接入已有的 Base.metadata。
• 在迁移后的架构上构建作业队列 / 调度器 / 租约循环 → sql-job-queue。
• 手动编写没有版本历史的原始 SQL DDL——Alembic 的用途正是提供这类历史；如果确实不需要，请参阅第 6 步中关于 create_all() 的选择。

工作流

第 1 步：初始化迁移环境

alembic init 会在项目旁生成目录结构。

alembic init alembic        # creates alembic.ini + alembic/ (env.py, script.py.mako, versions/)

这会生成 alembic.ini（配置文件）、alembic/env.py（运行时钩子）、alembic/script.py.mako（迁移版本模板）和 alembic/versions/（迁移版本文件存放位置）。alembic.ini 中的 script_location 指向 alembic/ 目录；version_locations（可选）允许将迁移版本拆分到多个目录中。

第 2 步：从环境中读取数据库 URL（不要硬编码）

生成的 alembic.ini 中包含字面值 sqlalchemy.url = driver://...。不要在其中提交真实 URL。可以将其留空，并在 env.py 中从环境读取，或者使用插值。在 env.py 中设置可以让应用和迁移共用同一个真实来源：

# alembic/env.py
import os
from alembic import context

config = context.config
config.set_main_option("sqlalchemy.url", os.environ["DB_URL"])

set_main_option 会在运行时覆盖 alembic.ini 中的任何值，因此只需更改 DB_URL，同一份迁移历史就能在本地针对 SQLite 运行，在其他环境中针对 PostgreSQL/MySQL 运行。

第 3 步：将 target_metadata 指向模型的 Base.metadata

这是让 --autogenerate 生效的关键——它会将实时数据库与传入的 MetaData 进行对比。导入模型的 Base（以便所有表都注册到其 metadata 上），并赋值：

# alembic/env.py (continued)
from myapp.models import Base   # your DeclarativeBase subclass — sqlalchemy sibling owns this

target_metadata = Base.metadata

必须导入定义模型的模块——从未导入的表对 target_metadata 不可见，因此自动生成迁移时也不会发现它（这是一个盲点——见第 5 步）。

步骤 4：env.py 在线运行器（优先同步），支持批处理和命名约定

脚手架提供离线和在线运行器。在线运行器使用实时连接（也是最常见的路径）；离线运行器不会连接数据库，而是将 SQL 输出到标准输出或文件（--sql 模式），用于手动应用或需要审核后才能部署的场景。编辑在线运行器以启用批处理模式，让 SQLite 迁移能够正常工作（步骤 7），并定义 naming_convention，以便批处理表重建时能按名称引用约束：

# alembic/env.py (continued)
from sqlalchemy import engine_from_config, pool, MetaData

NAMING_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",
}
# Best applied on Base.metadata itself in the models module so live + autogen agree;
# shown here for visibility. If set in models: MetaData(naming_convention=NAMING_CONVENTION)

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,
            render_as_batch=True,          # makes SQLite migrations move-and-copy (Step 7)
            compare_type=True,             # also diff column TYPE changes (still review them)
            compare_server_default=True,   # attempt server_default diffs (imperfect — Step 5)
        )
        with context.begin_transaction():
            context.run_migrations()

run_migrations_online()   # the scaffold guards this with context.is_offline_mode()

render_as_batch=True 是全局开关，会将生成的变更语句包装在批处理模式中。对于短期运行的迁移进程，NullPool 是合适的连接池。（脚手架的离线运行器——run_migrations_offline()——会调用 context.configure(url=..., literal_binds=True) 并生成 SQL；请保留它，以便用于 --sql 部署。）

步骤 5：编写修订——自动生成草稿，由你审核

创建修订有两种方式：

alembic revision -m "add jobs table"              # EMPTY — you write upgrade()/downgrade() by hand
alembic revision --autogenerate -m "add jobs table"   # DRAFTS ops by diffing target_metadata vs DB

自动生成是草稿生成器，而不是事实来源。它会将 target_metadata 与实时数据库进行比较，并写入它认为最合适的结果，但它有一些已知盲区。**始终打开生成的文件，在应用前进行审核和编辑。**自动生成无法可靠检测的内容包括：

• 表名或列名重命名——它会识别为删除旧项并添加新项。如果不加审核地直接应用，会丢失数据。请手动将其改写为 op.rename_table(...) 或批处理列重命名。
• 大多数 server_default 变更——即使设置了 compare_server_default=True，许多变更仍会被漏检或错误生成。
• CHECK 约束和某些命名约束——经常无法检测到。
• 某些索引变更和列类型变更——compare_type=True 会有所帮助，但并不全面；请核对生成的类型。
• target_metadata 无法访问的任何内容——例如忘记导入的模型，或 Alembic 未被告知要比较的架构对象（其他架构、某些对象类型）。看不到，就不会出现在差异中。

审核后的自动生成文件如下所示——请注意显式且可逆的 downgrade()：

"""add jobs table"""
from alembic import op
import sqlalchemy as sa

revision = "a1b2c3d4e5f6"
down_revision = None        # first migration; otherwise the prior revision id
branch_labels = None
depends_on = None

def upgrade() -> None:
    op.create_table(
        "jobs",
        sa.Column("id", sa.Integer(), primary_key=True),
        sa.Column("name", sa.String(length=200), nullable=False),
        sa.Column("data", sa.JSON(), nullable=True),
    )
    op.create_index("ix_jobs_name", "jobs", ["name"])

def downgrade() -> None:
    op.drop_index("ix_jobs_name", table_name="jobs")
    op.drop_table("jobs")

将 downgrade() 写成 upgrade() 的真正逆操作——顺序相反，每个操作也都反向执行。错误的 downgrade() 或只写 pass 的迁移无法回滚。有关安全重命名和数据迁移，请参阅 references/alembic-extras.md。

步骤 6：应用、回滚和检查

alembic upgrade head        # apply every unapplied revision up to the latest
alembic upgrade +1          # apply the next one only
alembic downgrade -1        # roll back the most recent revision
alembic downgrade base      # roll back everything (empty schema)

alembic current             # which revision the DB is stamped at
alembic history             # the revision graph, newest first
alembic heads               # the current head(s) — more than one means a branch to merge

步骤 7：多方言——采用批处理风格编写迁移

这是多方言支持的关键规则。SQLite 无法 ALTER 大多数架构元素——不能删除列、更改列的类型或可空性，也不能通过普通的 ALTER TABLE 添加大多数约束。Alembic 的批处理操作通过迁移并复制来解决这个问题：创建符合目标架构的新表，将行跨表复制（INSERT ... SELECT），删除旧表，然后将新表重命名到位。在 PostgreSQL 和 MySQL 上，同样的批处理代码会生成普通的 ALTER 语句。因此，所有变更都采用批处理风格，就能编写一份可在三种方言上应用的迁移脚本。

def upgrade() -> None:
    # Same code path on all three; SQLite does move-and-copy, PG/MySQL do ALTER.
    with op.batch_alter_table("jobs") as batch_op:
        batch_op.add_column(sa.Column("priority", sa.Integer(), nullable=False, server_default="0"))
        batch_op.alter_column("name", existing_type=sa.String(200), nullable=True)
        batch_op.create_unique_constraint("uq_jobs_name", ["name"])

def downgrade() -> None:
    with op.batch_alter_table("jobs") as batch_op:
        batch_op.drop_constraint("uq_jobs_name", type_="unique")
        batch_op.alter_column("name", existing_type=sa.String(200), nullable=False)
        batch_op.drop_column("priority")

render_as_batch=True（步骤 4）会让自动生成器为你生成 batch_alter_table 代码块。naming_convention（步骤 4）能让 SQLite 在迁移并复制时保留表的约束；如果没有命名约束，重建过程可能会删除未命名的约束。审核生成的 SQL 时，预期 SQLite 会出现 CREATE TABLE _alembic_tmp_... / INSERT ... SELECT / DROP / ALTER ... RENAME 序列，而 PG/MySQL 上则会出现单条 ALTER TABLE。有关反射信息不足时使用的 batch_alter_table(..., copy_from=...)，以及重建模式控制选项，请参阅 references/alembic-extras.md。

步骤 8：分支——通过 merge 解决多个头

当两个人（或两个功能分支）从同一个父修订分别添加修订时，alembic heads 会显示两个头，而 upgrade head 会因目标不明确而报错。使用合并修订将它们连接起来——该修订以两个头为父修订，自身不应用任何架构变更：

alembic heads                              # shows e.g. two head ids: rev_a, rev_b
alembic merge -m "merge a and b" rev_a rev_b
alembic upgrade head                       # now unambiguous

第 9 步：异步说明（简短）

对于异步 SQLAlchemy 技术栈，迁移通常仍使用同步方式运行——最简单的做法是让 Alembic 使用同步驱动 URL，并原样使用第 4 步的运行器。如果必须使用异步引擎，env.py 的在线运行器将变为：

import asyncio
from sqlalchemy.ext.asyncio import async_engine_from_config

def do_run_migrations(connection):
    context.configure(connection=connection, target_metadata=target_metadata, render_as_batch=True)
    with context.begin_transaction():
        context.run_migrations()

async def run_async_migrations():
    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section, {}), prefix="sqlalchemy.", poolclass=pool.NullPool
    )
    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)   # greenlet bridge to Alembic's sync context
    await connectable.dispose()

asyncio.run(run_async_migrations())

run_sync(do_run_migrations) 通过 greenlet 桥接，从异步连接切换到 Alembic 的同步迁移上下文。这是唯一与异步相关的部分；其他所有内容（修订、批处理、审查）都相同。

规则

硬性规则（绝不违反）：

• 自动生成的只是草稿；应用前必须审查并编辑每个生成的脚本。 绝不要升级未经审查的自动生成修订——重命名和约束变更会悄无声息地导致数据丢失。
• 如果目标包含 SQLite，请以批处理方式编写迁移。 对于 SQLite 无法通过普通 ALTER 完成的任何变更，使用 render_as_batch=True + op.batch_alter_table(...)。相同代码在 PG/MySQL 上也能正常运行。
• 设置 target_metadata = Base.metadata 并导入模型，让自动生成能够看到每张表。未导入的表不可见。
• 从环境变量读取数据库 URL，不要将其写成已提交的 alembic.ini 字面值——一份迁移历史，多种方言／环境。
• 每个 upgrade() 都必须有真正可逆的 downgrade()。 对可逆操作，不要让 downgrade() 只执行 pass。
• 使用 naming_convention，这样 SQLite 上的批量迁移和复制操作才能引用并保留具名约束。

建议（可覆盖）：

• 优先使用同步运行器；只有在确实无法使用同步驱动 URL 时，才采用异步的 run_async_migrations 结构。
• 优先使用 --autogenerate 创建修订初稿，然后手动编辑，而不是从头编写操作；纯数据迁移除外（应手动编写）。
• 保持启用 compare_type=True / compare_server_default=True，以获得更好的差异，但应将其输出视为草稿，而非保证准确。
• 迁移进程使用 NullPool——它的生命周期很短，不应持有连接池。

注意事项

• 自动生成会将重命名识别为删除并新增 → 导致数据丢失。 这是最重要的审查项。如果列／表被重命名，生成的 drop_column+add_column（或 drop_table+create_table）会销毁数据。应改写为 op.rename_table(...) 或批处理 alter_column(..., new_column_name=...)。
• 未导入的新模型会导致自动生成结果为空。 由于表从未注册到 target_metadata，自动生成会认为“无需执行任何操作”。应在 env.py 中或之前导入模型模块。
• 未使用批处理时，SQLite 上的 ALTER 会失败。 如果未用 batch_alter_table 包装，带有某些约束的 add_column、任何 drop_column／alter_column 以及大多数 add_constraint 操作都会在 SQLite 上报错。通常只有在迁移实际运行于 SQLite 上时才会出现该错误，而不是在你开发时使用的 PG 上。应从一开始就启用 render_as_batch=True。
• SQLite 批量重建时，未命名约束会消失。 移动并复制的重建过程会重新创建表；Alembic 无法命名的约束（未设置 naming_convention）可能无法保留。应在数据可能丢失之前定义命名约定。
• alembic upgrade head 报错“multiple heads”。 存在两个尚未合并的分支。运行 alembic heads 查看它们，再运行 alembic merge 将其合并（第 8 步）。这不是损坏，只是分支尚未解决。
• server_default 差异并不可靠。 即使启用了 compare_server_default=True，也经常会漏掉默认值变更或生成错误结果。应手动验证默认值；不要在这方面相信自动生成。
• downgrade base 会清空架构。 它会运行所有 downgrade()，直到数据库变空。在开发环境中没问题，但在生产环境中具有破坏性——运行前确认 DB_URL 指向哪个数据库。

反模式

• 不要不读自动生成的迁移就直接应用，理由是“自动生成通常没错”——它的盲点（重命名、服务器默认值、CHECK／命名约束）恰恰会导致数据被破坏。
• 不要写普通的 op.alter_column／op.drop_column，并以为它“在 Postgres 上能用”就能在所有地方运行——它在 SQLite 上会报错。请使用 batch_alter_table。
• 不要把 Base.metadata.create_all() 当作迁移方案交付，理由是“这样更简单”——它没有版本历史、没有增量变更，也无法降级。请参阅下方的决策；一旦架构已交付，就应使用 Alembic。
• 不要在 alembic.ini 中硬编码生产环境 URL，理由是“这只是配置”——这会泄露凭据，并固定一种方言。请从环境中读取。
• 不要让 downgrade() 保持为 pass，理由是“我们永远不会回滚”——不可逆的迁移会阻止所有后续迁移回滚。
• 不要通过编辑 down_revision 手动合并两个分支头——请使用 alembic merge，它会正确记录两个父版本。

create_all() 与 Alembic 的取舍（由本技能负责）

Base.metadata.create_all(engine) 会一次性为当前模型状态生成 CREATE TABLE。它仅适用于：全新／尚未交付的架构、临时数据库，以及测试／开发环境的初始化（sqlalchemy 同级技能正是在这些场景中使用它）。对于任何已交付的架构，它的局限都不可接受：没有版本历史、没有增量变更，也无法降级——它只能从头创建，永远无法演进。

当架构已交付、在多个环境中运行，或会随时间演进时，就应使用 Alembic（团队按顺序应用的变更有向无环图）。这正是迁移历史的意义所在。

在使用 create_all() 构建的数据库上采用 Alembic——不要重新运行建表操作，而应：

1. 初始化 Alembic（步骤 1–4），并编写一个基线迁移，让其 upgrade() 创建当前架构（针对空数据库运行自动生成，或手动编写，使其与 create_all 生成的内容一致）。
2. 在现有的、已通过 create_all() 创建的数据库上运行 alembic stamp head——这会记录数据库已处于基线修订版本，但不会运行该迁移（不会重新执行 CREATE TABLE）。
3. 从此以后，每项变更都作为新的修订版本通过 upgrade 应用。

alembic stamp <rev> 会写入 alembic_version 行，将某个修订版本标记为已应用，但不会执行其内容——这是将未托管的 create_all() 数据库接入托管迁移历史的桥梁。

输出

配置好的 Alembic（alembic.ini + 一个 target_metadata = Base.metadata 的 env.py，数据库 URL 从环境中读取，设置 render_as_batch=True，并配置 naming_convention），以及经过审查、可逆的 versions/ 修订文件——采用批处理风格编写，使同一套迁移历史可应用于 SQLite、PostgreSQL 和 MySQL。下一层会使用这一产物：部署步骤运行 alembic upgrade head，或更高层的模式（例如 sql-job-queue）基于已迁移的架构构建。

相关

• sqlalchemy——本技能所依托的数据层：DeclarativeBase／Mapped 模型、Engine／URL、会话，以及跨方言 ORM 的注意事项。本技能假定其 Base.metadata 已存在，只负责将 Alembic 接入其中；sqlalchemy 使用 create_all() 初始化开发／测试环境，并将托管变更的取舍留给本技能。
• sql-job-queue——基于已迁移架构构建的作业存储／调度器模式（就绪集查询、原子租约、公平共享）。本技能负责生成架构历史，不讲解队列。
• pydantic-v2——应用边界验证（而非数据库架构或迁移）。

渐进式披露

• references/alembic-extras.md — 更完整的迁移示例：安全重命名（保留数据）、通过 op.bulk_insert / op.execute 进行数据迁移、在反射不足以满足需求时使用 batch_alter_table(..., copy_from=..., recreate=...)、离线 --sql 模式，以及使用 include_object/include_name 限定自动生成的范围。**在以下情况下加载：**正文中的代码片段不足以应对特定迁移场景时。
• references/sources.md — 此技能的研究来源。**在以下情况下加载：**核实某项声明的来源，或进行新一轮审查时。

正文预算

• description ≤ 1,024 个字符。
• 正文 ≤ 约 500 行 / 5,000 个令牌。
• 每个参考文件：超过 10k 个令牌时发出警告，超过 25k 个令牌时报错。参考文件总计：超过 25k 个令牌时发出警告，超过 50k 个令牌时报错。