第 13 章 · 数据库集成（SQLModel 与异步 SQLAlchemy）

本章目标：用官方推荐的 SQLModel 完成建表、Session 依赖注入与完整 CRUD，理解"表模型 vs 数据模型"的拆分策略，并掌握异步引擎版本。

13.1 SQLModel：Pydantic 与 SQLAlchemy 的合体

FastAPI 不绑定任何数据库，但作者 tiangolo 专门写了 SQLModel——它建立在 SQLAlchemy 和 Pydantic 之上，一个类既是 Pydantic 模型（可做请求/响应校验）又是 ORM 表模型。SQLAlchemy 支持的数据库（PostgreSQL、MySQL、SQLite 等）它都支持。本章用 SQLite 演示，零配置可运行。

bash

uv add sqlmodel   # 或 pip install sqlmodel

13.2 单模型版：从建表到 CRUD

先看最简可运行的全量代码：

python

# main.py
from sqlmodel import Field, Session, SQLModel, create_engine, select
from fastapi import Depends, FastAPI, HTTPException
from contextlib import asynccontextmanager

class Hero(SQLModel, table=True):
    # table=True 声明这是"表模型"；主键用 int | None，
    # 表示 Python 里创建对象时可以没有 id，由数据库生成
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)      # index=True 为该列建索引，加速按名字查询
    age: int | None = None
    secret_name: str

engine = create_engine(
    "sqlite:///database.db",
    connect_args={"check_same_thread": False},  # SQLite 专用：允许跨线程使用连接
)

def create_db_and_tables():
    SQLModel.metadata.create_all(engine)   # 按所有表模型建表（已存在则跳过）

@asynccontextmanager
async def lifespan(app: FastAPI):
    create_db_and_tables()   # 生产环境建议改用迁移工具（如 Alembic）
    yield

app = FastAPI(lifespan=lifespan)

def get_session():
    with Session(engine) as session:
        yield session

# Annotated 依赖别名：全项目复用同一个类型声明
SessionDep = Annotated[Session, Depends(get_session)]

@app.post("/heroes/")
def create_hero(hero: Hero, session: SessionDep):
    session.add(hero)
    session.commit()
    session.refresh(hero)    # 刷新以取回数据库生成的 id
    return hero

@app.get("/heroes/")
def read_heroes(session: SessionDep, offset: int = 0, limit: int = 100):
    heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
    return heroes

@app.get("/heroes/{hero_id}")
def read_hero(hero_id: int, session: SessionDep):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    return hero

@app.delete("/heroes/{hero_id}")
def delete_hero(hero_id: int, session: SessionDep):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    session.delete(hero)
    session.commit()
    return {"ok": True}

要点解读：

check_same_thread=False 只针对 SQLite。FastAPI 可能在同一次请求里用多个线程（例如同步依赖），放开这个限制后，靠"每请求一个 Session"保证安全；
get_session 用 yield：请求开始时创建会话，响应完成后自动关闭——这正是第 11 章 yield 依赖的标准应用；
session.exec() 是 SQLModel 提供的封装（返回带类型的行），等价于 SQLAlchemy 的 session.execute(select(...))。

更新操作怎么做

SQLModel/SQLAlchemy 没有内置 update 端点模式：先 hero = session.get(Hero, id) 取出对象，用 Pydantic 的 hero_data.model_dump(exclude_unset=True) 过滤出客户端传来的字段，再逐个 setattr 后 commit()。exclude_unset=True 是关键，它让"未传字段不覆盖旧值"成为可能。

13.3 多模型拆分：别把表模型直接暴露给 API

单模型版有两个安全问题：客户端可以在创建时指定 id（可能覆盖已有数据）；secret_name 被原样返回给所有人。正确做法是拆成多个模型，用继承消除重复字段：

python

from sqlmodel import SQLModel, Field

class HeroBase(SQLModel):
    """所有模型共享的字段"""
    name: str = Field(index=True)
    age: int | None = None

class Hero(HeroBase, table=True):
    """表模型：额外拥有 id 和 secret_name，对应数据库里的真实表"""
    id: int | None = Field(default=None, primary_key=True)
    secret_name: str

class HeroPublic(HeroBase):
    """返回给客户端的模型：不含 secret_name；id 必为 int 而非 None"""
    id: int

class HeroCreate(HeroBase):
    """创建时校验客户端输入：允许传 secret_name，但永远不会回显"""
    secret_name: str

路由随之变化：

python

@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate, session: SessionDep):
    db_hero = Hero.model_validate(hero)     # 数据模型 -> 表模型
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero                          # response_model 自动过滤为 HeroPublic 形状

@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate, session: SessionDep):
    db_hero = session.get(Hero, hero_id)
    if not db_hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    data = hero.model_dump(exclude_unset=True)   # 只取客户端实际传了的字段
    db_hero.sqlmodel_update(data)                # 批量更新
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero

这套 XxxBase / Xxx(table=True) / XxxPublic / XxxCreate / XxxUpdate 五件套是官方文档的推荐结构：密码类敏感字段"收得进、出不来"，接口契约也更稳定。

13.4 异步版本

如果路径操作全是 async def，同步的数据库驱动会阻塞事件循环。换用异步引擎：

python

import asyncio
from contextlib import asynccontextmanager

from fastapi import Depends, FastAPI
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlmodel import SQLModel, select

class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)

# aiosqlite 驱动：pip install aiosqlite；PostgreSQL 则用 postgresql+asyncpg://...
engine = create_async_engine("sqlite+aiosqlite:///database.db")

# async_sessionmaker 预先绑定 engine 和 expire_on_commit 配置
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

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

AsyncSessionDep = Annotated[AsyncSession, Depends(get_async_session)]

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)
    yield

app = FastAPI(lifespan=lifespan)

@app.get("/heroes/")
async def read_heroes(session: AsyncSessionDep):
    result = await session.execute(select(Hero))
    return result.scalars().all()

@app.post("/heroes/")
async def create_hero(hero: Hero, session: AsyncSessionDep):
    session.add(hero)
    await session.commit()
    await session.refresh(hero)
    return hero

注意 expire_on_commit=False：异步场景下 commit 之后对象属性会被标记过期，若保持默认值，随后访问属性会触发隐式同步 IO 报错，所以官方示例都显式关闭它。另外 run_sync 用于在异步引擎上执行同步风格的 DDL（如 create_all）。

13.5 本章小结

SQLModel 一个类同时是 Pydantic 模型和 SQLAlchemy 表模型，table=True 区分二者；
全局只建一个 engine；每个请求通过 yield 依赖获得独立 Session；
生产级做法是 Base/Public/Create/Update 多模型拆分，防止客户端指定 id、泄露敏感字段；
异步栈：create_async_engine + async_sessionmaker(expire_on_commit=False)，SQLite 用 sqlite+aiosqlite；
建表放 lifespan 仅适合开发，生产用 Alembic 迁移。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. SQLModel 中 table=True 的作用是什么？

2. 为什么 SQLite 引擎要设置 check_same_thread=False？

3. PATCH 更新时 hero.model_dump(exclude_unset=True) 的目的是？

4. 异步版中 async_sessionmaker(..., expire_on_commit=False) 的原因是？

🛠️ 动手实践

把 13.2 的单模型应用补全 PUT 全量更新和 PATCH 部分更新端点，并用 /docs 页面验证"只传 name 时 age 不变"。
按 13.3 的五件套模式，给 User 资源建模：要求 password 能创建但任何响应模型都不包含它。
将 13.2 的应用改造成异步版（aiosqlite），用 ab 或 wrk 对比同步版和异步版的吞吐差异，记录结论。

数据库就位后，下一步保护它——请进入下一章：安全基础 OAuth2 与 JWT。

第 13 章 · 数据库集成（SQLModel 与异步 SQLAlchemy） ​

13.1 SQLModel：Pydantic 与 SQLAlchemy 的合体 ​

13.2 单模型版：从建表到 CRUD ​

13.3 多模型拆分：别把表模型直接暴露给 API ​

13.4 异步版本 ​

13.5 本章小结 ​

🧪 随堂测验

🛠️ 动手实践 ​

第 13 章 · 数据库集成（SQLModel 与异步 SQLAlchemy）

13.1 SQLModel：Pydantic 与 SQLAlchemy 的合体

13.2 单模型版：从建表到 CRUD

13.3 多模型拆分：别把表模型直接暴露给 API

13.4 异步版本

13.5 本章小结

🛠️ 动手实践