基于Alembic与SQLAlchemy的数据迁移方案选型咨询

我接触SQLAlchemy + Alembic已有约两天时间，了解Alembic用于schema迁移，自动生成功能很实用，但对数据迁移的实现方式有疑问。

需求举例：我创建了一张表，希望以**自动化方式（支持升级与回滚）**向其中插入数据，理想状态是开发者更新至最新迁移版本时自动完成操作。

调研的三种可行方案 #

方案1：独立编写ORM脚本 #

编写常规代码更新表、插入或删除数据，ORM写法比直接操作SQL简洁清晰，但缺乏自动化能力，开发者需手动运行脚本，且无法实现回滚等操作。

示例代码：

def create_users():
    with session_maker() as session:
        for user in users:
            session.add(user)
        session.commit()

方案2：使用Alembic原生操作方法 #

使用Alembic提供的bulk_insert、batch_alter_table等方法，开发者只需运行最新迁移即可完成操作，但写法繁琐，不像ORM那样便捷。

示例代码：

op.bulk_insert(accounts_table,
[
    {'id':1, 'name':'John Smith', 'create_date':date(2010, 10, 5)},
    {'id':2, 'name':'Ed Williams', 'create_date':date(2007, 5, 27)},
    {'id':3, 'name':'Wendy Jones', 'create_date':date(2008, 8, 15)},
]
)

方案3：迁移中混合使用ORM代码 #

在Alembic的迁移脚本里直接调用ORM相关代码，兼顾自动化和ORM的便捷性。

示例代码：

def create_business_units():
    session_maker = sessionmaker(bind=create_engine(db_url))

    business_units = [
        BusinessUnits(name="Marketing"),
        BusinessUnits(name="Sales"),
        BusinessUnits(name="Software"),
    ]

    with session_maker() as session:
        for unit in business_units:
            session.add(unit)
        session.commit()


def upgrade() -> None:
    create_business_units()
    # ### commands auto generated by Alembic - please adjust! ###
    pass
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    pass
    # ### end Alembic commands ###

我的疑问 #

1. 第三种方案是否存在问题？若没有，为何会有第二种方案？第三种方案似乎兼顾了两者的优势。
2. 哪种方案是最优选择？还是说没有绝对最优，需根据场景而定？我知道这带有主观性，但肯定存在选型的判断标准？

解答 #

关于方案3的潜在问题及方案2存在的意义 #

方案3确实看起来兼顾了自动化和ORM的便捷性，但存在几个容易忽略的问题：

• 依赖模型定义的稳定性：迁移脚本是长期存在的，若后续BusinessUnits模型的字段、约束被修改，旧的迁移脚本可能因为模型结构变化而无法正常运行（比如新增了必填字段但旧迁移里没设置）。而方案2直接操作表结构（accounts_table是Alembic迁移里的表对象），不依赖应用层的ORM模型，更稳定。
• 事务一致性风险：Alembic的op操作默认是在迁移的事务中执行的，但手动创建的ORM Session如果没有绑定到Alembic的事务上下文，可能会出现部分操作成功、部分失败的情况，导致数据不一致。
• 兼容性问题：不同数据库的SQL语法差异，Alembic的op方法会自动做适配，但ORM操作虽然也适配，但若模型里用了数据库特定的特性，旧迁移脚本在切换数据库时可能出问题。

方案2存在的核心意义就是迁移脚本的独立性和稳定性——它不依赖应用代码的变化，只针对数据库表结构本身操作，这在长期维护、多环境部署的场景下非常重要。

选型判断标准 #

没有绝对最优的方案，核心看你的场景需求，判断标准如下：

1. 是否需要长期维护迁移脚本：如果项目会持续迭代，需要保证几年前的迁移脚本还能正常运行，优先选方案2。方案3依赖ORM模型，后续模型变更会导致旧迁移失效。
2. 数据操作的复杂度：如果只是简单的批量插入、删除，方案2足够；如果数据操作涉及复杂的业务逻辑（比如关联查询、数据转换、依赖其他业务模型），方案3的ORM写法更简洁不易出错，但要注意绑定Alembic的事务（可以通过op.get_bind()获取迁移的连接来创建Session）。
3. 团队协作与规范：如果团队统一要求迁移脚本不依赖应用层代码，那方案2是标准选择；如果追求开发效率，且能接受后续维护时对旧迁移脚本做适配，方案3可以用。
4. 回滚需求：方案1无法自动回滚，直接排除；方案2和3都能在downgrade里编写回滚逻辑，但方案2的回滚（比如op.execute("DELETE FROM accounts WHERE id IN (1,2,3)")）更直接，方案3需要在downgrade里写对应的ORM删除逻辑。

另外补充一个优化后的方案3写法，解决事务一致性问题：

def upgrade() -> None:
    # 获取Alembic的迁移连接，绑定到Session
    bind = op.get_bind()
    session = sessionmaker(bind=bind)()
    
    business_units = [
        BusinessUnits(name="Marketing"),
        BusinessUnits(name="Sales"),
        BusinessUnits(name="Software"),
    ]
    
    session.add_all(business_units)
    session.commit()

def downgrade() -> None:
    bind = op.get_bind()
    session = sessionmaker(bind=bind)()
    
    session.query(BusinessUnits).filter(BusinessUnits.name.in_(["Marketing", "Sales", "Software"])).delete()
    session.commit()

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