如何将已创建的SQLAlchemy引擎连接传入Alembic执行迁移？

我之前也碰到过一模一样的场景，核心痛点在于SQLite内存数据库每个新连接都是完全独立的实例——默认Alembic通过engine_from_config创建的新连接，根本看不到你单元测试里已经建好的表。要搞定这个问题，关键就是让Alembic直接复用你已经创建好的SQLAlchemy引擎实例，而不是重新创建连接。

下面是具体的实现步骤：

1. 修改Alembic的env.py，支持外部传入引擎 #

打开你的Alembic配置目录下的env.py，调整run_migrations_online函数，添加一个接收外部引擎的入口：

# 在env.py顶部添加全局变量，用于接收外部传入的引擎
target_engine = None

def run_migrations_online():
    global target_engine
    # 优先使用外部传入的引擎，没有的话再走默认的engine_from_config逻辑
    connectable = target_engine or 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
        )

        with context.begin_transaction():
            context.run_migrations()

2. 在单元测试中传递现有引擎给Alembic #

在你的单元测试代码里，创建好引擎和表之后，把这个引擎实例传给Alembic的env.py，然后执行迁移命令：

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from alembic.config import Config
from alembic import command
import alembic.env  # 导入你的alembic.env模块

# 你原有的创建引擎和表的代码
Base = declarative_base()
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

# 将已创建的引擎传给Alembic的env.py
alembic.env.target_engine = engine

# 加载Alembic配置并执行迁移
alembic_cfg = Config("alembic.ini")
# 执行到最新版本的迁移（包含你的数据填充逻辑）
command.upgrade(alembic_cfg, "head")

# 之后你就可以用这个engine进行单元测试，数据已经是迁移后的状态了

关键注意点 #

• 别再通过命令行传递dbPath参数了，那样会触发engine_from_config创建新的内存连接，和你单元测试的库完全独立。
• 确保你的迁移脚本逻辑和Base.metadata.create_all的表结构兼容——比如如果迁移是初始化表，可能需要调整，但你提到迁移包含数据填充，只要表结构一致就没问题。
• SQLite内存库的生命周期和引擎实例绑定，只要你复用这个engine，所有操作都在同一个内存库中进行。

内容的提问来源于stack exchange，提问作者Javier Novoa C.