实现自动执行的SQLAlchemy异步驱动Pytest数据库迁移Fixture

核心思路 #

把数据库迁移逻辑封装成异步fixture，借助Pytest对异步fixture的原生支持，让它在指定作用域（比如module）内自动执行，无需手动调用异步生成器的__anext__方法。

完整可复用代码 #

假设你已有基于declarative_base的ORM模型，以及基础的异步引擎配置，以下是直接可用的fixture实现：

import pytest
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from your_module import Base  # 替换为你的declarative_base子类导入路径

@pytest.fixture(scope="module")
async def migrated_db():
    # 初始化内存SQLite异步引擎（测试场景专用，可替换为实际数据库URL）
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    
    # 执行模型迁移：创建所有表
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    
    # 创建异步会话生成器，供测试用例使用
    async_session = sessionmaker(
        engine, 
        class_=AsyncSession, 
        expire_on_commit=False
    )
    yield async_session
    
    # 测试后清理资源（可选）
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    await engine.dispose()

# 测试用例示例
@pytest.mark.asyncio
async def test_database_operation(migrated_db):
    async with migrated_db() as session:
        # 这里直接操作数据库即可，表已自动创建完成
        result = await session.execute("SELECT 1")
        assert result.scalar() == 1

关键细节说明 #

• 作用域控制：将fixture的scope设为module，确保整个测试模块仅执行一次迁移，完美匹配“仅针对指定模块生效”的需求，不会影响其他测试文件。
• 自动执行逻辑：只要测试用例声明依赖migrated_db fixture，Pytest会自动处理异步代码的await操作，无需手动触发迁移逻辑。
• 避免手动调用的核心：不要将迁移fixture设计成需要手动迭代的异步生成器，而是让fixture自身完成迁移动作，通过yield提供后续测试需要的资源（如会话生成器）；若无需返回资源，直接执行迁移后yield即可。

之前尝试无效的原因 #

• 同步函数中使用asyncio.wait：Pytest的同步上下文与异步测试的事件循环相互隔离，这种方式无法正确触发异步迁移的执行。
• 转换async_session_maker为异步生成器：若测试用例未正确依赖该fixture，Pytest不会自动执行其逻辑；且生成器形式的fixture需由Pytest迭代，手动等待反而绕开了Pytest的自动处理机制。
• 使用pytest_sessionstart：该钩子作用域为整个测试会话，无法实现模块级别的隔离，不符合你的需求。

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