第 17 章 SQLAlchemy 2.0 在 FastAPI 中的集成与最佳实践

从一个生活场景说起

前面 16 章学的都是"单机演练"。现在要上一线了：把 SQLAlchemy 装进一个 FastAPI 网店后台，让它同时接待成百上千的 HTTP 请求。

继续用第 16 章的"银行"比喻。一个 Web 服务就是一家银行网点：

• 网点开门时，只建一套叫号系统（Engine 全局唯一，连接池共享）；
• 每位顾客（HTTP 请求）进门取号，分配一位专属柜员（每请求一个 Session）；
• 业务办完柜员立刻复位接待下一位（请求结束关闭 Session）；
• 顾客填的业务单（Pydantic 模型，校验入参）和柜台内部的档案卡（ORM 模型）是两套东西——顾客永远摸不到档案卡（模型分层）。

这一章就是把这四个原则落成可运行的代码。

核心概念

FastAPI 与 SQLAlchemy 集成的三个关键问题：

1. Engine 全局唯一（连接池共享），Session 按请求创建；
2. 用依赖注入（Depends）管理 Session 生命周期——请求开始创建、请求结束关闭；
3. 模型（SQLAlchemy）与数据结构（Pydantic）分离——ORM 模型管数据库，Pydantic 模型管接口入参/出参。

术语解释 —— 依赖注入（Dependency Injection）：FastAPI 的一种机制。你在路由函数参数里声明"我需要一个 db"，FastAPI 就会自动调用你写的 get_db() 函数造一个出来、用完再帮你收尾。你不用管"怎么造、怎么收"，只管"用"。

术语解释 —— Pydantic 模型：FastAPI 用来做数据校验和序列化的类。它定义"接口接收/返回的数据长什么样"，与数据库表结构无关。

环境准备：

pip install fastapi uvicorn sqlalchemy pymysql

典型项目结构

myapp/
├── main.py            # FastAPI 入口
├── database.py        # Engine + Session 工厂 + Base
├── models.py          # SQLAlchemy ORM 模型
├── schemas.py         # Pydantic 模型（请求/响应）
└── routers/
    └── user.py        # 路由，通过 Depends 获取 Session

为什么这样分文件？职责清晰：database.py 是"网点的基础设施"，models.py 是"档案卡格式"，schemas.py 是"对外业务单格式"，routers/ 是"各个业务窗口"。文件各司其职，改表结构不用翻路由代码。

完整示例

database.py —— 基础设施层

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

DATABASE_URL = "mysql+pymysql://root:123456@localhost:3306/demo_db?charset=utf8mb4"

engine = create_engine(
    DATABASE_URL,
    pool_size=10,          # 连接池常驻连接数
    max_overflow=20,       # 高峰期允许额外创建的连接
    pool_recycle=3600,     # 小于 MySQL 默认的 wait_timeout(28800s)，防止断连
    pool_pre_ping=True,    # 取连接前先探测有效性
)

SessionLocal = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

# FastAPI 依赖：每个请求一个 Session
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

get_db() 是整个集成的枢纽，逐行拆解：

• db = SessionLocal()：顾客进门，分配柜员；
• yield db：把柜员交给路由函数使用（yield 是"带暂存的返回"：先把 db 交出去，等路由执行完，再回来继续执行 finally）；
• finally: db.close()：无论请求成功还是抛异常，柜员都一定复位。这就是"请求结束自动关 Session"的全部秘密。

models.py —— ORM 模型

from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column
from database import Base

class User(Base):
    __tablename__ = "user"
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(50))
    age: Mapped[int | None]

schemas.py —— Pydantic 模型

from pydantic import BaseModel, ConfigDict

class UserCreate(BaseModel):
    """创建用户时客户端要提交的数据（没有 id，id 由数据库生成）"""
    name: str
    age: int | None = None

class UserOut(BaseModel):
    """返回给客户端的数据（包含 id）"""
    model_config = ConfigDict(from_attributes=True)  # 允许从 ORM 对象转换
    id: int
    name: str
    age: int | None

为什么需要两套模型？

• 职责不同：UserCreate 描述"客户端该提交什么"（不该有 id）；UserOut 描述"我们返回什么"（必须有 id，但绝不会有密码字段——如果 User 表里有的话）；
• from_attributes=True：告诉 Pydantic"允许从对象的属性读数据"，这样路由里可以直接 return user（ORM 对象），FastAPI 自动转换成 UserOut 再序列化成 JSON。

routers/user.py —— 路由层

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import select
from sqlalchemy.orm import Session

from database import get_db
from models import User
from schemas import UserCreate, UserOut

router = APIRouter(prefix="/users", tags=["用户"])

@router.post("", response_model=UserOut)
def create_user(data: UserCreate, db: Session = Depends(get_db)):
    user = User(name=data.name, age=data.age)
    db.add(user)
    db.commit()
    db.refresh(user)        # 重新加载，拿到自增 id
    return user

@router.get("/{user_id}", response_model=UserOut)
def get_user(user_id: int, db: Session = Depends(get_db)):
    user = db.get(User, user_id)
    if user is None:
        raise HTTPException(status_code=404, detail="用户不存在")
    return user

@router.get("", response_model=list[UserOut])
def list_users(page: int = 1, size: int = 10, db: Session = Depends(get_db)):
    stmt = select(User).offset((page - 1) * size).limit(size)
    return db.scalars(stmt).all()

@router.put("/{user_id}", response_model=UserOut)
def update_user(user_id: int, data: UserCreate, db: Session = Depends(get_db)):
    user = db.get(User, user_id)
    if user is None:
        raise HTTPException(status_code=404, detail="用户不存在")
    user.name = data.name
    user.age = data.age
    db.commit()
    return user

@router.delete("/{user_id}")
def delete_user(user_id: int, db: Session = Depends(get_db)):
    user = db.get(User, user_id)
    if user is None:
        raise HTTPException(status_code=404, detail="用户不存在")
    db.delete(user)
    db.commit()
    return {"message": "删除成功"}

五个接口正好是一套完整的 CRUD（Create 增 / Read 查 / Update 改 / Delete 删），每个都遵循同一套路：

1. db: Session = Depends(get_db) —— 声明"我要一个柜员"；
2. 业务操作（add / get / 改属性 / delete）；
3. 成功 commit，找不到资源抛 404；
4. return ORM 对象，由 response_model 过滤成对外格式。

main.py —— 入口

from fastapi import FastAPI
from database import Base, engine
from routers import user

Base.metadata.create_all(bind=engine)   # 开发阶段建表；生产环境用 Alembic 迁移

app = FastAPI(title="SQLAlchemy 2.0 + FastAPI 示例")
app.include_router(user.router)

启动：

uvicorn main:app --reload

打开浏览器访问 http://127.0.0.1:8000/docs，FastAPI 自动生成交互式 API 文档，可以直接在网页上点按钮测试每个接口。

最佳实践清单

实践	说明
Engine 单例	全局只创建一次，复用连接池
Session 按请求	用 Depends(get_db)，请求结束自动 close
事务边界明确	业务成功 commit，失败 rollback；get_db 的 finally 兜底 close
模型分层	ORM 模型 ≠ Pydantic 模型，不直接把 ORM 类暴露给接口
response_model	用 Pydantic 控制输出字段，避免泄露敏感信息
表结构变更用 Alembic	生产环境不用 create_all，用迁移管理版本
防 N+1	列表接口返回关联数据时用 selectinload 预加载
连接池参数	MySQL 场景显式设置 pool_size、max_overflow、pool_recycle=3600、pool_pre_ping=True
字符集	连接串带 charset=utf8mb4，库表也统一 utf8mb4

深入理解：两个容易困惑的设计

1. 为什么 get_db 里不 commit？ 因为"是否提交"是业务决定，不是基础设施决定。同一个 get_db，查询接口根本不需要 commit，创建接口需要 commit，复杂接口可能要先 flush 再决定。依赖只负责"借"和"还"，业务代码自己决定"何时盖章"（回顾第 16 章）。

2. 为什么查询接口路由用 def 而不是 async def？ 同步 SQLAlchemy 的每次数据库调用都是"阻塞"的——在 async def 里执行会把整个事件循环卡住，所有请求一起排队。用普通 def 定义路由时，FastAPI 会自动把它放到线程池里执行，互不阻塞。想真正拥抱异步，就要全链路换异步 SQLAlchemy（见下文）。

进阶：异步 SQLAlchemy

高并发场景可用异步驱动（MySQL 用 aiomysql），FastAPI 路由写成 async def：

pip install aiomysql

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy import select

async_engine = create_async_engine(
    "mysql+aiomysql://root:123456@localhost:3306/demo_db?charset=utf8mb4"
)
AsyncSessionLocal = async_sessionmaker(async_engine, expire_on_commit=False)

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

@router.get("/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).where(User.id == user_id))
    return result.scalar_one_or_none()

与同步版的差异：引擎换成 create_async_engine、驱动换成 aiomysql、所有数据库操作前加 await。注意：异步代码里不能混用任何同步 SQLAlchemy 调用（包括对象的懒加载属性访问），要预加载请用 selectinload。

进阶：Alembic 数据库迁移

create_all() 只会建表、不会改表（第 2 章）。生产环境表结构变更多次后，需要 Alembic 管理：

pip install alembic
alembic init alembic                 # 初始化迁移目录
alembic revision --autogenerate -m "add email column"  # 对比模型与数据库，生成迁移脚本
alembic upgrade head                 # 执行迁移

它像 Git 一样记录表结构的每次变更，可以升级也可以回滚，是团队协作和生产部署的标准做法。

注意事项

• 同步 SQLAlchemy 在 async def 路由中会阻塞事件循环——要么用 def 路由（FastAPI 会放到线程池执行），要么全链路用异步 SQLAlchemy。
• db.refresh(obj) 用于 commit 后重新加载对象，拿到数据库生成的字段（如自增 id、默认时间戳）。
• 不要在 get_db 里做 commit——事务的决定权交给业务代码，依赖只负责创建和关闭。
• 数据库连接串不要硬编码在代码里提交到 Git，生产环境用环境变量或配置中心管理：os.getenv("DATABASE_URL")。
• MySQL 的 wait_timeout 默认 28800 秒，务必配置 pool_recycle 小于该值，否则高峰后第一批请求会拿到已被服务端断开的连接而报错。
• 列表接口若返回关联数据（如用户及其文章数），务必 selectinload 预加载，否则 N 个用户就是 N+1 条 SQL（第 14 章）。
• autoflush=False 在 Web 场景是常见选择：避免查询中途意外触发 flush 带来的副作用，需要时显式 flush()。

---

结语

至此你已经掌握了 SQLAlchemy 2.0 的完整学习路径：

1. 底层（Core）：Engine → Table → insert/select/update/delete，理解 SQL 如何被构建与执行；
2. 上层（ORM）：映射类 → relationship（一对多/多对多/一对一）→ 面向对象的查询与批量操作；
3. 会话与事务：Session 生命周期、commit/rollback 的正确姿势；
4. 工程集成：在 FastAPI 中以依赖注入管理 Session、模型分层、Alembic 迁移、异步支持。

建议动手做一个完整的 FastAPI + MySQL 小项目（如简易博客 API）来巩固全教程内容：定义 User / Article / Comment 三个模型，配好一对多关系，写出五个 CRUD 接口——做完它，本教程的每一个知识点你都会真正用上。