FastAPI怎么使用SQLAlchemy进行异步数据库操作-Python教程-PHP中文网

create_async_engine 是起点，不是终点；必须严格匹配异步驱动与连接串协议、用 async with 管理 AsyncSession 生命周期、继承 DeclarativeBase 并使用 Mapped 定义模型、显式调用 commit/rollback 控制事务。

fastapi怎么使用sqlalchemy进行异步数据库操作

create_async_engine

是起点，不是终点。用错驱动、漏关会话、混用同步写法，接口照样卡死。

确认数据库驱动是否真正异步

常见错误是装了 aiomysql 却用 mysql+pymysql:// 连接串，或反过来装 asyncmy 却配 mysql+aiomysql://。两者不兼容，启动时可能无报错，但首次查询直接抛 RuntimeError: This event loop is already running 或静默阻塞。

必须严格匹配：

mysql+aiomysql:// → 依赖 aiomysql
mysql+asyncmy:// → 依赖 asyncmy
postgresql+asyncpg:// → 依赖 asyncpg

检查命令：pip list | grep -i "aiomysql\|asyncmy\|asyncpg"；连接串中协议前缀必须与已安装驱动一致。

AsyncSession 必须用 async with 管理生命周期

这是最常被忽略的坑。直接写 session = AsyncSessionLocal() 然后 await session.execute(...)，会导致连接永不归还池，跑几轮压测就 TimeoutError: QueuePool limit of size 10 overflow 20 reached。

正确姿势只有一种：

async def get_user(user_id: int, session: AsyncSession = Depends(get_db)):
    result = await session.execute(select(User).where(User.id == user_id))
    return result.scalar_one_or_none()

其中 get_db 必须是：

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

注意：yield 不可省略，async with 不可替换成 sessionmaker 直接调用。

模型定义必须用 DeclarativeBase + Mapped

SQLAlchemy 2.0 异步模式下，老式 Column() 配合 declarative_base() 虽然能跑，但 await session.merge() 或 await session.refresh() 会报 AttributeError: 'AsyncSession' object has no attribute 'merge' —— 实际是模型没继承对基类。

FastAPI Flask Proxy

FastAPI + Flask 混合部署最佳实践，解决路由定义、API 代理等常见问题，适用于同时运行 FastAPI API 与 Flask 前端的场景。

下载

务必使用：

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))

不要用 from sqlalchemy.ext.declarative import declarative_base，那是 1.x 时代的同步写法，和异步会话不完全兼容。