FastAPI异步数据库会话配置异常：查询时应用崩溃无响应

问题根源 #

你遇到的冻结崩溃问题，核心是异步会话配置错误，误用了同步的Session类，导致异步上下文里执行阻塞操作，卡死FastAPI的事件循环。

错误点拆解 #

• 异步Session类选错：异步引擎必须搭配AsyncSession（SQLAlchemy提供的异步会话类），而不是同步的DBSession。
• 生成器返回类型错误：异步依赖生成器的返回类型应该标注为AsyncIterator[AsyncSession]，而非普通的DBSession。
• 会话创建方式：异步sessionmaker创建的会话，需要确保是异步实例，否则会在异步上下文里执行同步阻塞代码。

修正后的异步配置代码 #

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from fastapi import FastAPI, Depends
from sqlalchemy import select
from typing import AsyncIterator

# 假设User是你的模型类
from app.models import User

def setup_database(app: FastAPI):
    # 初始化异步引擎
    engine = create_async_engine(SQLALCHEMY_DATABASE_URI, future=True, echo=True)
    # 绑定异步会话类AsyncSession
    app.state.database = sessionmaker(
        engine, 
        expire_on_commit=False,
        class_=AsyncSession  # 关键：使用异步会话类
    )

async def request_db_session() -> AsyncIterator[AsyncSession]:
    from app import app
    db_session: AsyncSession = app.state.database()

    try:
        yield db_session
    finally:
        await db_session.close()

async def request_user(email: str, db_session: AsyncSession = Depends(request_db_session)):
    # 异步查询必须用await，且会话是AsyncSession实例
    user_query = await db_session.scalars(select(User).filter(User.email == email))
    return user_query.first()  # 示例：返回第一个匹配用户

为什么同步模式能正常运行？ #

同步模式用的是create_engine（同步引擎）+ 普通Session类，所有操作都是阻塞式的，FastAPI在同步路由里会自动用线程池处理，不会卡主事件循环，所以能正常工作。但异步模式必须严格遵循异步会话的使用规范，否则就会出现冻结崩溃。

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