如何在FastAPI的Depends外从get_session获取SqlAlchemy异步会话？

你遇到的AttributeError: 'async_generator' object has no attribute 'execute'问题很常见——因为get_session()是一个异步生成器函数，直接await get_session()得到的是生成器对象，而不是真正的AsyncSession实例。FastAPI的Depends会自动帮你处理生成器的迭代和清理，但自己调用的时候就得手动处理了。

下面给你两种可行的解决方法：

方法一：手动驱动异步生成器 #

直接操作生成器对象，获取yield出来的会话，最后别忘了手动关闭生成器（确保会话被正确回收）：

async def use_session_outside_depends():
    # 获取生成器对象
    session_gen = get_session()
    try:
        # 获取yield出来的会话
        session = await session_gen.__anext__()
        # 在这里使用会话
        result = await session.execute(stmt)
        await session.commit()
    finally:
        # 关闭生成器，触发async with的退出逻辑（会话回滚/关闭）
        await session_gen.aclose()

方法二：封装一个可直接使用的上下文管理器 #

如果觉得手动操作生成器太麻烦，可以把会话获取逻辑封装成标准的异步上下文管理器，这样用起来和async with一样方便：

from contextlib import asynccontextmanager

@asynccontextmanager
async def get_session_context():
    async with async_session() as session:
        yield session

然后在任何地方都可以这样用：

async def standalone_task():
    async with get_session_context() as session:
        await session.execute(stmt)
        await session.commit()

这个封装和你原来的get_session逻辑一致，但用@asynccontextmanager标记后，它可以直接被async with使用，不需要依赖FastAPI的Depends。

需要注意的是，不管用哪种方法，都要确保会话的生命周期被正确管理——用完后要让上下文管理器完成清理，避免连接泄漏。

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