Python异步数据库测试套件执行异常问题排查求助

一、先解决 AttributeError: 'async_generator' object has no attribute 'get_async_session' #

问题根源 #

Database.get_async_session 方法标注返回异步生成器，但实际用了 return 直接返回会话对象，导致 async with 无法正确识别上下文管理器，同时可能在测试注入时出现对象类型混淆。

修正代码 #

修改 Database 类的 get_async_session 方法，用 yield 实现正确的异步生成器上下文管理：

async def get_async_session(self) -> AsyncGenerator[AsyncSession, None]:
    session = self.AsyncSourceSessionLocal()
    try:
        yield session
    finally:
        await session.close()

二、解决批量执行测试时的 another operation is in progress 错误 #

问题根源 #

多个测试共享数据库会话/连接，且未正确隔离数据，导致异步操作冲突；同时原fixture每次创建/删除表的操作效率低且易引发冲突。

解决方案 #

通过会话隔离+事务回滚实现测试间数据隔离，优化fixture作用域：

1. 优化测试Fixture

import pytest
from sqlalchemy.ext.asyncio import AsyncSession
# 替换为实际的模型/类导入路径
from your_module import Database, InventoryRepository, Base, Inventory

@pytest.fixture(scope="session")
async def test_database() -> AsyncGenerator[Database, None]:
    """测试会话级fixture：仅创建/删除一次表"""
    db = Database()
    
    # 创建所有表
    async with db.engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    
    yield db
    
    # 测试结束后删除表并释放引擎
    async with db.engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    await db.engine.dispose()

@pytest.fixture(scope="function")
async def db_session(test_database: Database) -> AsyncGenerator[AsyncSession, None]:
    """测试用例级fixture：每个测试用独立会话，执行后回滚"""
    session = test_database.AsyncSourceSessionLocal()
    try:
        await session.begin()
        yield session
        await session.rollback()
    except Exception:
        await session.rollback()
        raise
    finally:
        await session.close()

2. 改造Repository支持会话注入

让Repository可以接受外部传入的会话，方便测试时使用隔离会话：

from typing import Optional, AsyncGenerator
from sqlalchemy import insert, select
from sqlalchemy.ext.asyncio import AsyncSession
from your_module import Inventory

class InventoryRepository:
    def __init__(self, database: Database, session: Optional[AsyncSession] = None) -> None:
        self.database = database
        self._session = session

    async def get_session(self) -> AsyncGenerator[AsyncSession, None]:
        if self._session:
            yield self._session
        else:
            async with self.database.get_async_session() as session:
                yield session

    async def create_inventory(self, inventory_data: dict) -> None:
        async with self.get_session() as session:
            stmt = insert(Inventory).values(**inventory_data)
            await session.execute(stmt)

    async def get_inventory(self, flight: str, flight_booking_class: str) -> Optional[Inventory]:
        async with self.get_session() as session:
            stmt = select(Inventory).where(
                Inventory.flight == flight,
                Inventory.flight_booking_class == flight_booking_class
            )
            result = await session.execute(stmt)
            return result.scalar_one_or_none()

3. 更新测试用例

使用隔离的会话fixture执行测试：

@pytest.mark.asyncio
async def test_create_inventory(test_database: Database, db_session: AsyncSession):
    repo = InventoryRepository(test_database, db_session)
    inventory_data = {
        "time": 1234567890,
        "flight": "FL456",
        "departure": 9876543211,
        "flight_booking_class": "Economy",
        "idle_seats_count": 15
    }
    
    await repo.create_inventory(inventory_data)

    new_inventory = await repo.get_inventory(
        flight=inventory_data["flight"], 
        flight_booking_class=inventory_data["flight_booking_class"]
    )
    
    assert new_inventory.flight == inventory_data["flight"]
    assert new_inventory.idle_seats_count == inventory_data["idle_seats_count"]

关键修正总结 #

• 修复异步会话生成器的实现，解决上下文管理器类型错误。
• 用会话级fixture管理表的创建/删除，用例级fixture提供隔离会话并自动回滚，彻底避免测试间数据干扰。
• Repository支持会话注入，兼顾生产环境和测试场景的不同需求。
• 确保async_scoped_session的scopefunc=current_task正确导入，保证异步任务间会话隔离。

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