异步数据库

前面几章的数据库操作都是「同步」的：执行查询时，Python 会原地等待数据库返回结果。如果数据库在远程、网络慢，这个等待就白白占着工作线程。异步数据库让等待期间能去处理别的请求，吞吐更高。

36-1 同步为什么会被拖慢

想象餐厅只有一个服务员。他给一桌点完单，必须站在厨房门口等到菜做好才端走，期间别的桌只能干等。这就是同步：一个慢操作卡住所有人。

异步像多了个叫号系统：服务员把单子递给厨房（发出请求），立刻去招呼别的桌；菜好了再回来端。他「等」的时候在做别的事，整体效率更高。

数据库查询正是典型的「等网络」操作，所以异步收益明显。但需要驱动和库都支持 await，普通 sqlite3 不行。

36-2 选对异步驱动 asyncpg

异步数据库必须用支持 asyncio 的驱动。PostgreSQL 用 asyncpg，MySQL 用 aiomysql，SQLite 可用 aiosqlite。驱动名要写进连接字符串的「协议前缀」：

pip install asyncpg

# 同步：postgresql://...  异步：postgresql+asyncpg://...
URL = "postgresql+asyncpg://user:password@localhost:5432/mydb"

注意前缀多了 +asyncpg。写错驱动，连库时就会报错。

36-3 create_async_engine 建异步引擎

和同步的 create_engine 对应，异步用 create_async_engine，它来自 sqlalchemy.ext.asyncio：

from sqlalchemy.ext.asyncio import create_async_engine

async_engine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost:5432/mydb",
    echo=True,          # 打印执行的 SQL，调试方便
    pool_pre_ping=True, # 每次取连接前先测一下是否还活着
)

echo=True 会把每条 SQL 打到控制台，初学很有用；生产可关掉。pool_pre_ping 能避免拿到失效连接。

36-4 async_sessionmaker 与 AsyncSession

会话工厂也得换成异步版 async_sessionmaker，它产出 AsyncSession：

from sqlalchemy.ext.asyncio import async_sessionmaker, AsyncSession

AsyncSessionLocal = async_sessionmaker(
    bind=async_engine,
    class_=AsyncSession,
    expire_on_commit=False,  # 提交后不让对象过期，省一次查库
)

expire_on_commit=False 是异步场景的常用设置：提交后对象属性仍然可读，不会因「过期」而触发额外查询。

36-5 模型定义完全不变

重点来了：模型类（表结构）和同步版一字不差，不用为异步重写。

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

class Base(DeclarativeBase):
    pass

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

异步只改变「连库方式」和「调用时要 await」，表怎么定义是数据库层面的事，和异步无关。

36-6 异步查询：await session.execute(select(…))

异步会话的操作都要 await。查询用 await session.execute(stmt)，再从结果取对象：

from sqlalchemy import select

async def get_users_async():
    async with AsyncSessionLocal() as session:
        stmt = select(User)
        result = await session.execute(stmt)
        users = result.scalars().all()   # scalars() 提取 ORM 对象
        return users

几个关键点：

• async with AsyncSessionLocal() as session：进入即开会话，退出自动关闭。
• await session.execute(stmt)：真正发查询到数据库，必须 await。
• result.scalars().all()：拿到所有用户对象的列表；.first() 拿第一个。

36-7 异步增删改

写操作和同步几乎一样，只是 commit、refresh 前加 await：

async def create_user_async(name: str, email: str):
    async with AsyncSessionLocal() as session:
        user = User(name=name, email=email)
        session.add(user)
        await session.commit()
        await session.refresh(user)
        return user

async def delete_user_async(user_id: int):
    async with AsyncSessionLocal() as session:
        user = await session.get(User, user_id)
        await session.delete(user)
        await session.commit()

即使出错想回滚，也是 await session.rollback()。

36-8 接到 FastAPI 的 async def 接口

把异步会话做成带 yield 的依赖，路径操作用 async def，整条链路都是异步的：

from typing import Annotated
from fastapi import Depends, FastAPI
from sqlalchemy.ext.asyncio import AsyncSession
from pydantic import BaseModel, ConfigDict

app = FastAPI()

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

DbDep = Annotated[AsyncSession, Depends(get_db)]

class UserOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    name: str

@app.get("/users/", response_model=list[UserOut])
async def read_users(db: DbDep):
    stmt = select(User)
    result = await db.execute(stmt)
    return result.scalars().all()

@app.post("/users/", response_model=UserOut)
async def create_user(name: str, db: DbDep):
    user = User(name=name)
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

36-9 启动与小结

异步项目启动方式不变，依然用 FastAPI CLI：

fastapi dev main.py
# 等价命令
uvicorn main:app --reload

回顾这套异步组合拳：

• 驱动 asyncpg 写进连接字符串前缀。
• create_async_engine 建异步引擎。
• async_sessionmaker 产 AsyncSession。
• 所有操作 await，查询用 await session.execute(select(...))。
• 路径操作与依赖都用 async def，全程不阻塞。

36-10 并行查询与常见坑

异步最大的红利，是能「同时」发起多个互不依赖的查询。用 asyncio.gather 即可让它们并发执行：

import asyncio

users, articles = await asyncio.gather(
    get_all_users(),
    get_all_articles(),
)

两个查询同时发出，总耗时约等于较慢的那个，而不是两者相加。接口要聚合多个数据源时，这招提速明显。

不过这里有个前提容易被忽略：并发执行的这几个查询必须各自使用独立的会话。若它们共用同一个 AsyncSession，反而会因为争抢同一条连接而报错，因为一个会话在同一时刻只能处理一次操作。正确做法是在每个待并发的函数内部各开一个会话，或者干脆按顺序 await，别为了并发牺牲正确性。

但异步也有几个坑要避开。第一，别在 async def 里调用同步阻塞库（如普通 requests、pymysql），那会卡住整个事件循环，必须换对应的异步版本。第二，异步 AsyncSession 不能跨线程、也不能被多个协程同时共享，每个任务用自己的会话。第三，查询前一定记得 await，漏写拿到的是「协程对象」而不是结果，接口会直接报错。

学完异步数据库，你已经能应对大多数高性能场景。但关系型数据库不是唯一选择——下一章我们看另一种思路：用 MongoDB 做文档型存储。