Alembic编程式运行时表修改自动生成迁移失败

我正在为Alembic编写自动化测试，验证revision --autogenerate命令能否生成预期的迁移脚本，打算通过编程方式获取建议的迁移并执行，确认对测试数据库的影响符合预期。

我使用两个模型：initial_model_base和removed_column_base，预期第二次自动生成迁移时，Alembic能检测到第二个模型少了一列，生成移除该列的迁移脚本。但实际运行时抛出错误：

ValueError: no dispatch function for object: <alembic.operations.ops.ModifyTableOps object at 0x0000028C0FBC1ED0>

完整堆栈信息：

File "C:\Users\Ronel\Desktop\repos\test_alembic\tests\tests.py", line 152, in test_removing_column()
File "C:\Users\Ronel\Desktop\repos\test_alembic\tests\tests.py", line 109, in test_removing_column
ops.invoke(op)
File "C:\Users\Ronel\AppData\Roaming\Python\Python310\site-packages\alembic\operations\base.py", line 396, in invoke
fn = self._to_impl.dispatch(
File "C:\Users\Ronel\AppData\Roaming\Python\Python310\site-packages\alembic\util\langhelpers.py", line 258, in dispatch
raise ValueError("no dispatch function for object: %s" % obj)
ValueError: no dispatch function for object: <alembic.operations.ops.ModifyTableOps object at 0x0000028C0FBC1ED0>

测试针对表test_alembic_schema.employees，为避免影响其他表，配置了include_object函数，相关代码如下：

def include_object(object, name, type_, reflected, compare_to):
    if type_ == "table":
        if object.schema == "test_alembic_schema" and object.name == 'employees':
            return True
        else:
            return False
    else:
        return True


mc = MigrationContext.configure(
    connection=DB_DEV.connection,
    opts={
        'compare_type': True,
        'include_schemas': True,
        'include_object': include_object
    }
)

ops = Operations(mc)


def test_creation():
    diff_ops = produce_migrations(
        context=mc,
        metadata=initial_model_base.metadata
    )

    with DB_DEV.transaction(autocommit=True):
        for op in diff_ops.upgrade_ops.ops:
            ops.invoke(op)

    employees = DB_DEV.get_table(table_name='employees', schema_name='test_alembic_schema')
    assert len(employees.c) == 8
    assert employees.c.id.primary_key
    assert not employees.c.first_name.primary_key


def test_removing_column():
    diff_ops = produce_migrations(
        context=mc,
        metadata=removed_column_base.metadata
    )

    with DB_DEV.transaction(autocommit=True):
        for op in diff_ops.upgrade_ops.ops:
            ops.invoke(op)

    employees = DB_DEV.get_table(table_name='employees', schema_name='test_alembic_schema')
    col_found = False
    for col in employees.columns:
        if col.name == 'created':
            col_found = True
    assert col_found

表创建操作能成功执行，但表修改操作失败，请问如何修复才能让两种操作正常工作？

问题出在ModifyTableOps是一个组合操作对象，它内部包含了具体的子操作（比如DropColumnOp），而ops.invoke()无法直接处理这种容器类型的操作，只能处理单个原子操作。

修复步骤如下：

• 递归遍历操作树：在调用ops.invoke()前，先判断当前操作是否是包含子操作的类型（比如ModifyTableOps），如果是，则遍历其内部的ops列表，执行每个子操作。
• 修改操作执行逻辑：替换原来直接遍历diff_ops.upgrade_ops.ops的代码，改为递归处理所有层级的操作。

修改后的测试函数示例：

def execute_ops(ops, op_list):
    for op in op_list:
        # 判断是否是包含子操作的容器类型
        if hasattr(op, 'ops'):
            execute_ops(ops, op.ops)
        else:
            ops.invoke(op)

def test_removing_column():
    diff_ops = produce_migrations(
        context=mc,
        metadata=removed_column_base.metadata
    )

    with DB_DEV.transaction(autocommit=True):
        # 调用递归函数执行所有操作
        execute_ops(ops, diff_ops.upgrade_ops.ops)

    employees = DB_DEV.get_table(table_name='employees', schema_name='test_alembic_schema')
    # 简化断言逻辑
    assert 'created' not in employees.columns

另外，还可以优化断言逻辑，直接检查列是否存在，不需要循环遍历，代码更简洁。

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