使用SQLAlchemy AsyncSession执行查询报错，FastAPI异步数据库操作求助

你遇到的错误核心原因是：get_async_session()是一个异步生成器函数（用yield定义），直接调用它返回的是异步生成器对象，而非实际的AsyncSession实例，所以无法调用execute方法。

修复方案一：通过async for从生成器中获取会话 #

修改check_db_connect函数，用async for迭代异步生成器来拿到会话实例：

async def check_db_connect():
    async for session in get_async_session():
        try:
            await session.execute(text("SELECT 1"))
            print("-- Database connection successful! --")
            await engine.dispose()
        except Exception as e:
            print(f"Database connection failed: {e}")

修复方案二：直接使用会话工厂创建会话 #

get_async_session主要是给FastAPI的依赖注入设计的（比如路由里用Depends(get_async_session)），在启动检查这种场景下，可以直接用async_session_maker创建会话，更简洁：

async def check_db_connect():
    async with async_session_maker() as session:
        try:
            await session.execute(text("SELECT 1"))
            print("-- Database connection successful! --")
            await engine.dispose()
        except Exception as e:
            print(f"Database connection failed: {e}")

补充说明：FastAPI依赖中正确使用生成器 #

你的get_async_session作为依赖函数是完全正确的，在路由中使用Depends时，FastAPI会自动处理异步生成器，提取出AsyncSession实例：

from fastapi import Depends

@app.get("/items")
async def get_items(session: AsyncSession = Depends(get_async_session)):
    result = await session.execute(text("SELECT * FROM items"))
    return result.scalars().all()

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