SQLAlchemy 2.0+ 才原生支持 FastAPI 异步连接，需搭配 asyncpg 等异步驱动；旧版仅伪异步，易阻塞事件循环。推荐用 create_async_engine、AsyncSession 和 async with 依赖管理会话生命周期。

FastAPI 异步连接 SQLAlchemy 的核心要点

FastAPI 本身支持异步（async/await），但 SQLAlchemy 2.0+ 才原生支持真正的异步操作，且必须搭配异步数据库驱动（如 asyncpg（PostgreSQL）或 aiomysql（MySQL））。旧版 SQLAlchemy（1.x）的 create_engine 不支持 async，强行用 async def 包裹同步 ORM 操作只会阻塞事件循环，起不到异步效果。

使用 SQLAlchemy 2.0+ 异步配置（推荐方式）

以 PostgreSQL 为例，需安装 sqlalchemy>=2.0 和 asyncpg：

• 安装依赖：pip install "sqlalchemy[asyncio]" asyncpg
• 创建异步引擎：
  from sqlalchemy.ext.asyncio import create_async_engine
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db", echo=True)
• 定义异步 Session：
  from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import sessionmaker
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
• 在路由中使用：
  @app.get("/users")
   async def get_users(db: AsyncSession = Depends(get_db)):
     result = await db.execute(select(User))
     users = result.scalars().all()
     return users

注意事务与会话生命周期管理

异步 Session 不能像同步那样直接 session.close()，要 await session.close()。建议用依赖项封装会话获取与清理：

• 定义依赖函数：
  async def get_db():
     async with AsyncSessionLocal() as session:
       yield session
       # 自动 await session.close()
• 避免跨 await 复用同一 Session 实例 —— 异步 Session 不是线程安全的，也不支持并发执行多个查询
• 复杂事务需显式 await session.begin() 和 await session.commit()，回滚用 await session.rollback()

不推荐的“伪异步”方案（慎用）

有人尝试用 run_in_executor 把同步 SQLAlchemy 操作扔进线程池，这虽能避免阻塞主协程，但：

• 丧失数据库连接池的异步复用能力，开销大
• 无法利用 asyncpg 等驱动的底层异步 I/O 优势
• ORM 对象可能在不同线程间传递失败，引发状态错误
• 事务隔离、会话一致性难以保障

除非你暂时无法升级到 SQLAlchemy 2.0+ 或数据库驱动不支持异步，否则应避免该方案。