Alembic执行upgrade()出错仍建表，如何实现事务化执行？

核心问题原因 #

你的情况是因为迁移脚本执行时，表a的创建成功提交，但表b的创建失败，而Alembic未将整个迁移过程纳入单个事务，导致部分变更持久化，同时版本记录未更新。这通常和数据库对事务DDL的支持以及Alembic的配置有关。

解决方案 #

1. 针对支持事务DDL的数据库（PostgreSQL、SQL Server等） #

这类数据库允许DDL操作在事务中执行，可通过修改Alembic的env.py文件，强制将整个迁移过程包裹在单个事务中：

修改run_migrations_online函数如下：

def run_migrations_online():
    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
        )

        # 启动全局事务包裹整个迁移
        trans = connection.begin()
        try:
            with context.begin_transaction():
                context.run_migrations()
            trans.commit()
        except Exception:
            trans.rollback()
            raise

这样，只要迁移过程中任何一步出错，整个事务都会回滚，不会留下部分创建的表（比如a），alembic_version表也不会更新。

2. 针对MySQL（InnoDB/MyISAM） #

MySQL的DDL操作（如CREATE TABLE）是隐式自动提交的，无法通过事务回滚。这种情况下需要从以下方面规避：

• 提前验证迁移脚本：执行迁移前，先用alembic upgrade head --sql命令生成对应的SQL语句，手动检查语法和逻辑错误（比如重复列名），确认无误后再执行实际迁移。
• 拆分迁移脚本：将多个表的创建拆分为独立的迁移文件，每次只执行一个DDL操作，避免单个迁移中包含多个可能失败的步骤。
• 修复当前遗留问题：手动删除数据库中已创建的表a，然后执行修复后的迁移脚本。

3. 迁移脚本本身的修复 #

先修正表b的重复列问题，比如将第二个id列改为其他名称：

op.create_table(
    'b',
    sa.Column('id', sa.BIGINT, primary_key=True),
    sa.Column('other_id', sa.BIGINT, nullable=False),
)

当前问题的临时修复 #

现在数据库中存在表a但alembic_version无记录，可通过以下步骤解决：

1. 手动删除数据库中的表a；
2. 执行修复后的迁移脚本，此时整个迁移会正常执行，alembic_version也会正确记录版本。

内容的提问来源于stack exchange，提问作者Stefan Falk