FastAPI异步SQLAlchemy测试：AsyncSession调用报错问题排查

问题 #

在FastAPI应用从同步SQLAlchemy Session迁移到异步数据库调用后，原单元测试的数据库回滚代码失效，修改后的测试代码如下：

test_engine = create_async_engine(get_db_url(), poolclass=StaticPool)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=test_engine, class_=AsyncSession)

@pytest.fixture()
async def db_session():
    connection = await test_engine.connect()
    
    transaction = await connection.begin()
    await connection.begin_nested()    
    session = TestingSessionLocal(bind=connection)
    
    @sqlalchemy.event.listens_for(session.sync_session, "after_transaction_end")
    def end_savepoint(session, transaction):
        if connection.closed:
            return
        if not connection.in_nested_transaction():
            connection.sync_connection.begin_nested()
    
    yield session
    # Rollback the overall transaction, restoring the state before the test ran.
    await session.close()
    await transaction.rollback()
    await connection.close()

@pytest.fixture()
def client(db_session):
    # import ipdb; ipdb.set_trace()
    # async def override_get_db():
    def override_get_db():
        yield db_session

    app.dependency_overrides[get_db] = override_get_db
    yield TestClient(app)
    del app.dependency_overrides[get_db]

def test_get_by_id(client):
    response = client.get("/users/1")
    assert response.status_code == 404

调用以下异步查询函数时触发错误：

async def find_by_id(cls, session: AsyncSession, pk: int):
    return await session.get(cls, pk)

错误信息：

AttributeError: 'async_generator' object has no attribute 'get'

错误原因 #

核心问题是依赖注入的返回值类型不匹配：

• 生产环境中的get_db依赖必然是异步生成器（async def定义），但测试代码里的override_get_db用普通def定义成了同步生成器。
• FastAPI处理异步依赖时，会将同步生成器的返回值包装为async_generator对象，而非直接返回db_session实例，导致业务代码拿到的不是AsyncSession，自然没有get方法。

解决方法 #

修改client fixture中的override_get_db为异步生成器，确保和生产环境的依赖类型一致：

@pytest.fixture()
async def client(db_session):
    async def override_get_db():
        yield db_session

    app.dependency_overrides[get_db] = override_get_db
    yield TestClient(app)
    del app.dependency_overrides[get_db]

额外验证点 #

确认生产环境的get_db依赖是标准异步生成器格式，示例如下：

async def get_db():
    async with TestingSessionLocal() as session:
        yield session

修改后，FastAPI会正确从override_get_db中获取AsyncSession实例，业务代码的session.get()调用即可正常执行。

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