Alembic迁移测试在已有数据库场景下执行失败求助

问题根源 #

• 内存数据库初始化时直接加载了最新版本的models.py，导致user表一开始就包含column4，但Alembic的alembic_version表是空的，没有任何迁移记录。
• 执行downgrade("base")时，因为当前没有已应用的迁移版本，该操作实际上没有执行任何降级逻辑（没有可回滚的迁移步骤）。
• 后续执行upgrade("head")时，Alembic会尝试执行所有未应用的迁移脚本（即添加column4的操作），但此时表中已经存在该字段，因此触发SQLite的duplicate column name错误。

解决方案 #

方案1：基于初始模型初始化数据库 #

测试前先创建不含column4的初始版本表，再执行完整的迁移流程：

# 定义初始版本的表（或临时注释models.py中的column4）
USERS_INITIAL = Table('user', admin_meta,
    Column('column1', ...),
    Column('column2', ...),
    Column('column3', ...),
)

def test_upgrade(dbclient):
    # 用初始模型创建表
    admin_meta.create_all(dbclient.engine, tables=[USERS_INITIAL])
    with dbclient.engine.connect() as connection:
        config.attributes['connection'] = connection
        # 执行升级→降级→升级的完整流程
        alembic.command.upgrade(config, "head")
        alembic.command.downgrade(config, "base")
        alembic.command.upgrade(config, "head")

方案2：用Alembic的stamp命令标记当前版本 #

利用stamp命令将内存数据库的版本表标记为head（即最新版本），让Alembic识别当前状态已完成所有迁移，再执行降级和升级：

def test_upgrade(dbclient):
    with dbclient.engine.connect() as connection:
        config.attributes['connection'] = connection
        # 标记数据库为head版本（不执行迁移，仅更新版本表）
        alembic.command.stamp(config, "head")
        # 降级到初始版本（此时会执行删除column4的逻辑）
        alembic.command.downgrade(config, "base")
        # 升级回最新版本（重新添加column4）
        alembic.command.upgrade(config, "head")

方案3：手动同步版本表（仅作参考，不推荐） #

手动创建alembic_version表并插入最新版本号，模拟已完成迁移的状态：

from sqlalchemy import text

def test_upgrade(dbclient):
    with dbclient.engine.connect() as connection:
        # 创建Alembic版本表
        connection.execute(text("CREATE TABLE IF NOT EXISTS alembic_version (version_num VARCHAR(32) PRIMARY KEY)"))
        # 插入当前head版本号（替换为你的迁移脚本revision）
        connection.execute(text("REPLACE INTO alembic_version (version_num) VALUES ('2aa4282617d4')"))
        connection.commit()
        
        config.attributes['connection'] = connection
        alembic.command.downgrade(config, "base")
        alembic.command.upgrade(config, "head")

额外注意点 #

SQLite默认不支持直接DROP COLUMN操作，需要在迁移脚本中使用Alembic的批处理模式来兼容：

def downgrade():
    with op.batch_alter_table('users') as batch_op:
        batch_op.drop_column('column4')

同时确保alembic.ini中的SQLite连接配置添加foreign_keys=on，避免其他兼容性问题：

sqlalchemy.url = sqlite:///./test.db?foreign_keys=on

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