Featured image of post Async SQLAlchemy之旅：从困惑到明朗
Python FastAPI SQLAlchemy

Async SQLAlchemy之旅：从困惑到明朗

探索从同步到异步SQLAlchemy与FastAPI迁移中的陷阱与启示

English

当我第一次听说要在FastAPI中使用异步SQLAlchemy时，我的第一反应是：“又一个流行词，能有多复杂？”结果发现，这可不是简单地加个async就能搞定的。以下就是我在困惑、误区和最终豁然开朗之间跌跌撞撞的心路历程。

初印象：Async、Await与SQLAlchemy

我一直写同步代码，对async总觉得像是给代码撒了点魔法粉，让它“更好”。而且我用TypeScript的async模式很熟，觉得Python的async应该也差不多。很简单，对吧？

错了。

我最初的想法是：只要在FastAPI路由上加个async def，对SQLAlchemy查询加点await，API就能飞快了。

剧透： 根本不是这么回事。

心智模型的问题

我一开始把async当成简单的性能增强器——就像给汽车加了氮气加速。但这个心智模型是错的。Python里的async更像是把单车道公路换成了多车道高速公路，还配有智能交通管理。

在TypeScript/JavaScript里，事件循环是语言底层自带的。而在Python里，async是一种架构选择，不同场景有不同模式。

深入异步SQLAlchemy：第一道困惑

最初，我常用的同步SQLAlchemy查询是这样的：

def get_user(db: Session, email: str):
    return db.query(User).filter(User.email == email).first()

很直观。但当我试图加上async def，并随意加点await：

async def get_user(db: Session, email: str):
    return await db.query(User).filter(User.email == email).first()  # 这是错的！

我很快发现哪里不对劲。Python立刻报错：

AttributeError: 'Query' object has no attribute '__await__'

啊，经典错误。原来我犯了第一个致命错误：

错误假设： SQLAlchemy经典的session.query()模式会自动支持async。

并不会。这就像试图把普通电器插到USB口——系统根本不兼容！

“阻塞”带来的顿悟

这让我更深入地理解了“阻塞”到底是什么意思。想象一下，餐厅里只有一个服务员（事件循环）。当服务员去厨房（数据库）取餐时，其他客人只能等着。

同步系统里，服务员进了厨房就出不来，直到餐做好。异步系统里，服务员可以一边让厨房做菜，一边继续接新单。

问题在于，我的SQLAlchemy session就是那个死守厨房、拒绝多任务的服务员：

@app.get("/users/{user_id}")
async def read_user(user_id: int, db: Session = Depends(get_db)):
    # 即使在async def函数里，这依然是阻塞调用！
    user = db.query(User).filter(User.id == user_id).first()
    return user

这段代码是最糟糕的混合体——用了async语法，但数据库调用还是阻塞的！

神秘的.execute()和select()登场

疯狂翻阅Stack Overflow后（大家都懂的），我发现异步SQLAlchemy用的是完全不同的语法：

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

async def get_user(db: AsyncSession, email: str):
    result = await db.execute(select(User).where(User.email == email))
    return result.scalars().first()

“等等，”我盯着.scalars().first()发呆，“查一条记录怎么突然变成两步了？”

感觉太复杂了。老的.query().first()多好啊！

数据库依赖：被遗忘的一环

我后来意识到，FastAPI的依赖也要变。同步的session依赖：

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

要变成：

async def get_async_db():
    async with async_session() as session:
        yield session

这不仅是语法变化，更是session管理方式的根本转变！

拆解我的误区：.scalars()到底是什么？

我一开始以为.scalars()是个异步收集器，等流结束后自动吐出对象。后来才发现，.execute()返回的是一个Result对象，本质上是个表结构，就算你只查一列也是如此。

让我用一个比喻来解释，终于让我恍然大悟：

把SQLAlchemy查询比作点餐：

同步SQLAlchemy（老方式）：

pizza = restaurant.order(Pizza).with_topping('pepperoni').make()

很简单，直接拿到披萨。

异步SQLAlchemy（新方式）：

order_slip = await restaurant.execute(order(Pizza).with_topping('pepperoni'))
pizza = order_slip.items().first()

为啥多了几步？想象服务员给你端来一个托盘（Result），上面有多个盒子（Row），每个盒子里装着你点的东西（如User对象）。哪怕你只点了一份披萨，也会被装在托盘的盒子里：

  • execute() = 服务员端来托盘
  • .scalars() = 打开所有盒子，把食物直接放到托盘上
  • .first() = 从托盘上拿第一个食物

也就是：

# execute() 得到: [Row(User(id=1),), Row(User(id=2),)]
result = await db.execute(select(User).where(...))

# scalars() 得到: [User(id=1), User(id=2)]
users = result.scalars()

# first() 得到: User(id=1)
first_user = users.first()

CRUD语法对照表

理解了这个模式后，我需要把常用的CRUD操作翻译成新语法：

旧同步语法新异步语法
db.query(User).first()(await db.execute(select(User))).scalars().first()
db.query(User).all()(await db.execute(select(User))).scalars().all()
db.query(User).filter(User.name == name).one()(await db.execute(select(User).where(User.name == name))).scalars().one()
db.add(user); db.commit()db.add(user); await db.commit()
db.query(User).filter(User.id == id).update({User.name: new_name})await db.execute(update(User).where(User.id == id).values(name=new_name))

这张表成了我迁移代码的“罗塞塔石碑”。

性能焦虑：我真的需要异步吗？

下一个问题：折腾这么多，真的有用吗？

我做了个简单基准测试，用典型API接口做数据库调用。1000个并发请求下结果如下：

  • 同步FastAPI + 同步SQLAlchemy：约600请求/秒
  • 异步FastAPI + 同步SQLAlchemy：约550请求/秒（反而更慢！）
  • 异步FastAPI + 异步SQLAlchemy：约1400请求/秒

中间那个最差，因为有async的开销却没有async的好处！就像雇了个多任务服务员，却只让他一次只服务一桌。

如果你要做高并发API（每秒上千请求），async的优势才明显。同步DB调用会阻塞整个API，异步则能让Python在等数据库时处理更多请求。

Python 3.13的剧情反转

后来我又听说Python 3.13要解锁GIL（全局解释器锁），实现真正多线程。那是不是可以用线程代替async？我一开始也这么想——但又错了。

我以为Python 3.13的GIL改进能让多线程通吃一切，但其实我混淆了两类问题：

  • CPU密集型任务：解锁GIL后多线程才有用
  • I/O密集型任务（如数据库查询）：无论GIL如何，async依然更优

解锁GIL对CPU密集型有帮助，但对于网络I/O密集型，async依然是王道——这正是大多数API的场景。所以，即使多线程更强，async依然很重要。

混合方案：务实的过渡

我的项目有几百个路由，没法一次性全部重写。我找到了一个渐进迁移的办法：

from fastapi.concurrency import run_in_threadpool

@app.get("/legacy-but-important")
async def read_complex_report(db: Session = Depends(get_db)):
    # 把阻塞代码放到线程池，避免阻塞事件循环
    result = await run_in_threadpool(
        lambda: db.query(ComplexReport).all()
    )
    return result

这样可以先迁移最常用的接口，逐步推进。

特殊数据库扩展，比如pgvector怎么办？

我最后的顾虑是：像pgvector这样的特殊扩展能用async吗？查了下资料，async和pgvector配合得很好。用asyncpg作为PostgreSQL驱动，异步集成这些扩展毫无障碍。

下面是用pgvector异步查询的例子：

stmt = select(Document).order_by(
    l2_distance(Document.embedding, query_vector)
).limit(5)

result = await db.execute(stmt)
docs = result.scalars().all()

我很欣慰地发现，未来要加向量检索也不会被async迁移卡住。

辅助函数：让生活更简单

迁移了几个路由后，我发现新语法太啰嗦了，于是写了辅助函数：

async def db_get(db: AsyncSession, model, **kwargs):
    stmt = select(model).filter_by(**kwargs)
    result = await db.execute(stmt)
    return result.scalars().first()

# 用法
user = await db_get(db, User, email="[email protected]")

这些辅助函数让新语法恢复了同步时代的简洁，同时保留了async的优势。

总结我的Async之旅

从一开始觉得async只是个噱头，到被语法折磨、误解.scalars()，再到最终欣然接受它的优点——这一路既痛苦又收获满满。异步SQLAlchemy其实没那么可怕，但确实需要转变思维和拥抱新模式。

如果你要从我的经历里记住三点：

  1. 心智模型很重要：async关注的是并发，不是并行
  2. 要么全异步，要么全同步：async路由配同步DB调用是最差组合
  3. 抽象很有用：写辅助函数简化冗长语法

所以，如果你正盯着db.query()发愁要不要迁移到异步SQLAlchemy——别怕！刚开始确实有点懵，但坚持下去，明朗和高性能就在前方。

© 2022 - 2025 张欣耕

保留所有权利