FastAPI异步项目pytest测试报错：'async_generator'对象无'add'属性

错误信息 #

session.add(user)
AttributeError: 'async_generator' object has no attribute 'add'

相关代码 #

conftest.py #

SQLALCHEMY_DATABASE_URL = f'postgresql+asyncpg://{USER}:{PASSWORD}@{HOST}/{DB}_test'

engine = asyncio.create_async_engine(SQLALCHEMY_DATABASE_URL)
SessionTesting = sessionmaker(
    bind=engine,
    autocommit=False,
    autoflush=False,
    expire_on_commit=False,
    class_=asyncio.AsyncSession
)


@pytest.fixture(name='session')
async def session_fixture() -> Generator[SessionTesting, None, None]:
    Base.metadata.create_all(engine)
    async with SessionTesting(bind=engine.connect()) as session:
        yield session
    Base.metadata.drop_all(engine)

test_auth.py #

@pytest.mark.asyncio
async def test_get_token(session: SessionTesting, create_user):
    # async with AsyncClient(app=app, base_url='http://test') as ac:
    user = User(
        first_name='fixture',
        last_name='fixture',
        username='fixture',
        password=await get_hash('fixture')
    )
    session.add(user)
    await session.commit()
    async with AsyncClient(app=app, base_url='http://localhost:8000') as ac:
        credentials = {
            'username': 'fixture',
            'password': 'fixture'
        }
        response = await ac.post('/token/get_token', data=credentials, headers={"content-type": "application/x-www-form-urlencoded"})
        assert response.status_code == 200

解决思路 #

• 问题根源：异步 fixture 返回的是async_generator，但测试函数中直接将其当作AsyncSession实例使用；同时同步的数据库表操作方法在异步环境下执行会引发异常，且会话创建方式冗余。

• 具体修改步骤：

  1. 修正类型标注：将session_fixture的返回类型从Generator改为AsyncGenerator（需从typing或typing_extensions导入，取决于Python版本），匹配异步生成器的类型。
  2. 异步处理表创建/删除：Base.metadata.create_all和drop_all是同步方法，需用engine.begin()获取连接后，通过run_sync异步执行，避免阻塞事件循环。
  3. 简化会话创建：SessionTesting已绑定engine，无需重复传入bind=engine.connect()，直接通过SessionTesting()创建会话即可。

修改后的conftest.py代码 #

from typing import AsyncGenerator
import asyncio
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = f'postgresql+asyncpg://{USER}:{PASSWORD}@{HOST}/{DB}_test'

engine = create_async_engine(SQLALCHEMY_DATABASE_URL)
SessionTesting = sessionmaker(
    bind=engine,
    autocommit=False,
    autoflush=False,
    expire_on_commit=False,
    class_=AsyncSession
)


@pytest.fixture(name='session')
async def session_fixture() -> AsyncGenerator[SessionTesting, None]:
    # 异步创建数据库表
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    
    async with SessionTesting() as session:
        yield session
    
    # 异步删除数据库表
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

修改后，测试函数中的session会被正确注入AsyncSession实例，即可正常调用add、commit等方法。

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