Pytest中使用Alembic异步迁移初始化测试数据库的问题咨询

所有报错本质都是事件循环冲突：

• 直接在异步fixture里调用command.upgrade不生效，是因为Alembic默认自行创建异步数据库引擎，会生成独立的事件循环，和pytest-asyncio管理的测试事件循环完全隔离，迁移操作不会作用到测试用的数据库连接上下文。
• pytest-alembic的两个报错（跨loop绑定Future、运行中loop内不能调用asyncio.run()）是因为其内部硬编码了asyncio.run()执行迁移逻辑，强制新建事件循环，和pytest运行时已存在的事件循环直接冲突，自定义event_loopfixture也无法修复——它根本不会复用你注入的loop。
• nest-asyncio挂起是因为asyncpg驱动和嵌套事件循环兼容性极差，很容易触发连接死锁，不适合这个场景。
• Base.metadata.create_all直接建表的方案完全跳过Alembic迁移脚本执行流程，无法覆盖数据迁移、约束变更、默认值调整等迁移逻辑，满足不了验证迁移正确性的需求。

不需要依赖pytest-alembic，手动实现fixture即可，完全兼容异步测试流程，同时不影响生产环境正常的Alembic命令使用。

第一步：调整Alembic的env.py配置 #

修改两个位置，避免导入时自动执行迁移，同时支持外部传入数据库连接：

1. 把文件末尾直接调用的asyncio.run(run_migrations_online())放到主程序判断里，只有命令行直接执行alembic时才触发：

# 删掉原来直接写的 asyncio.run(run_migrations_online())
if __name__ == "__main__":
    asyncio.run(run_migrations_online())

2. 修改run_migrations_online逻辑，优先读取外部传入的连接，没有传入时才自行创建引擎（兼容生产命令行场景）：

async def run_migrations_online() -> None:
    """以'online'模式运行迁移"""
    # 优先从配置属性读取外部传入的连接（测试场景使用）
    connectable = config.attributes.get("connection", None)
    
    if connectable is None:
        # 生产命令行场景：自行创建引擎
        connectable = engine
        async with connectable.connect() as connection:
            await connection.run_sync(do_run_migrations)
        await connectable.dispose()
    else:
        # 测试场景：直接复用外部传入的连接，不新建引擎和事件循环
        await connectable.run_sync(do_run_migrations)

第二步：编写pytest fixture #

三个fixture配合，整个测试会话共用同一个事件循环和数据库引擎，从根源避免跨loop问题：

import asyncio
from typing import Generator
import pytest
from alembic import config, command
from sqlalchemy.ext.asyncio import create_async_engine
# 替换成你项目自身的测试数据库地址导入路径
from your_project.config import TEST_DATABASE_URL

@pytest.fixture(scope="session")
def event_loop() -> Generator:
    """会话级事件循环，整个测试流程共用同一个loop"""
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

@pytest.fixture(scope="session")
async def db_engine(event_loop):
    """会话级异步数据库引擎，所有数据库操作复用该实例"""
    engine = create_async_engine(TEST_DATABASE_URL)
    yield engine
    await engine.dispose()

@pytest.fixture(scope="session")
async def migrated_tables(db_engine):
    """测试前执行Alembic升级到最新版本，测试结束后降级清空表结构"""
    alembic_cfg = config.Config("alembic.ini")
    # 核心：把当前测试用引擎的同步连接包装传入Alembic配置，复用同一个连接上下文
    alembic_cfg.attributes["connection"] = db_engine.sync_engine
    # 执行升级
    command.upgrade(alembic_cfg, "head")
    yield
    # 所有测试结束后降级到初始版本
    command.downgrade(alembic_cfg, "base")

使用说明 #

• 所有需要访问数据库的异步测试、fixture都依赖migrated_tablesfixture，确保迁移完成后再执行数据库操作。
• 如果需要测试间数据隔离，可以在migrated_tables基础上新增函数级fixture，通过事务回滚、表清空等方式重置数据，不需要重复执行迁移。
• 生产环境下alembic upgrade head等命令的使用完全不受影响，和原有流程一致。

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