异步数据库与 SQLAlchemy 2.0：FastAPI 怎么和存储打交道

前面三篇的模型都停在「内存里」。真实接口要把数据落库、再查出来。FastAPI 自己不带 ORM，官方推荐 SQLAlchemy，且从 2.0 起原生支持 async。这篇把「异步 SQLAlchemy + FastAPI」这条最常用链路讲清楚。

为什么是异步驱动

FastAPI 跑在事件循环上。如果用传统的同步 psycopg2，一次 cursor.execute() 会阻塞整个线程，事件循环卡住，并发垮掉。换成异步驱动（asyncpg 对应 PostgreSQL、aiosqlite 对应 SQLite），await conn.execute() 在等数据库时把控制权交回循环，别的请求照常处理。

一句话：要享受 FastAPI 的并发，数据库驱动也得是 async 的。

引擎与 session 工厂

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost:5432/app"

engine = create_async_engine(DATABASE_URL, pool_size=10, max_overflow=20)
AsyncSessionLocal = async_sessionmaker(
    bind=engine,
    expire_on_commit=False,   # 提交后对象不失效，方便直接转响应模型
    autoflush=False,
)

create_async_engine 的 URL 前缀带 +asyncpg，这是和同步版唯一的写法差异。
pool_size / max_overflow 控制连接池容量。FastAPI 的并发上限实际受连接池约束——池太小，多请求会排队等连接。
expire_on_commit=False 让提交后的对象还能读属性，否则紧接着 model_dump() 会触发懒加载报错。

模型定义

用 SQLAlchemy 的声明式，和普通 ORM 写法一致：

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column()
    content: Mapped[str] = mapped_column()

Mapped[...] 是 2.0 的新写法，比老的 Column(Integer) 更贴近类型注解风格。

依赖注入 DB session

沿用第 2 篇的依赖模式，把 session 挂到每个请求：

from contextlib import asynccontextmanager
from sqlalchemy.ext.asyncio import AsyncSession

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

@app.get("/posts/")
async def list_posts(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Post))
    return result.scalars().all()

get_db 用 async with + yield：请求进来拿到 session，请求结束自动关闭（归还连接池）。handler 里 db 就是可用的 session。不需要在每个接口里手写 try/finally 关连接。

2.0 风格的 CRUD

SQLAlchemy 2.0 用 select() 构造查询，不再推荐老的 query()：

from sqlalchemy import select, insert, update, delete

# 查列表
result = await db.execute(select(Post).where(Post.id > 0).order_by(Post.id.desc()).limit(10))
posts = result.scalars().all()

# 查单条
result = await db.execute(select(Post).where(Post.id == post_id))
post = result.scalar_one_or_none()

# 插入
stmt = insert(Post).values(title="hello", content="world").returning(Post.id)
new_id = (await db.execute(stmt)).scalar_one()
await db.commit()

# 更新
await db.execute(update(Post).where(Post.id == post_id).values(title="new"))
await db.commit()

# 删除
await db.execute(delete(Post).where(Post.id == post_id))
await db.commit()

关键点：async 下所有 IO 操作都要 await，且改数据后必须 await db.commit() 才落库。漏掉 await 是最常见的 bug——代码不报错，但数据库里什么都没变。

事务与回滚

依赖注入的 session 天然是一个事务边界。遇到错误回滚：

async def create_post(db: AsyncSession = Depends(get_db)):
    try:
        db.add(Post(title="x", content="y"))
        await db.commit()
    except Exception:
        await db.rollback()
        raise
    return {"ok": True}

或者让异常自然冒泡——FastAPI 返回 500，session 在 async with 退出时自动回滚未提交的事务，不会污染连接池。

接回响应模型

把第 3 篇的 from_attributes=True 用上，ORM 行直接变 API 输出：

from pydantic import BaseModel, ConfigDict

class PostOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    title: str
    content: str

@app.get("/posts/{post_id}", response_model=PostOut)
async def get_post(post_id: int, db: AsyncSession = Depends(get_db)):
    post = (await db.execute(select(Post).where(Post.id == post_id))).scalar_one_or_none()
    if not post:
        raise HTTPException(status_code=404, detail="not found")
    return post   # Post 对象直接返回，Pydantic 按字段取出

return post 返回的是 ORM 实例，FastAPI 用 PostOut 按属性读取并序列化。from_attributes=True 就是告诉 Pydantic「从对象属性读，不只从 dict 读」。

同步/异步的边界

这条最容易踩坑：在 async 路由里调用任何阻塞 IO，都会卡住整个事件循环。

不要直接 requests.get()（同步 HTTP 库），用 httpx.AsyncClient。
不要直接 open() 大文件做同步读写，或调同步 DB 驱动。
确实有绕不开的同步重计算，用 await run_in_threadpool(func) 丢到线程池，别在主协程里跑。

如果项目里既有同步又有异步代码，最干净的做法是：async 路由只调 async 依赖，同步库单独放到后台任务或独立服务，避免混用把并发优势抹掉。

下一篇收尾——把接口真正推上生产：OAuth2+JWT 鉴权、CORS、自定义中间件、BackgroundTasks，以及用 gunicorn+uvicorn 部署和 pytest 测试。