进阶 3 · 持久化与认证¶

本章把"数据怎么存"和"谁能访问"接起来。数据库用两套（你选的"两者都讲"）：SQLModel 快速上手（模型即 pydantic，简洁）、SQLAlchemy 2.0 async 生产级（成熟、async、对照 JPA/Hibernate）。认证用 API Key + 依赖注入——这是 FastAPI 灵魂 Depends 的最佳练兵场。

贯穿项目 bookmarks-api 用 SQLAlchemy 2.0 async 作为生产级主线；SQLModel 本章用独立片段对照。

3.1 两个 ORM：怎么选¶

	SQLModel	SQLAlchemy 2.0
设计	模型 = pydantic + SQLAlchemy（二合一）	独立的 ORM 模型（与 pydantic 分离）
心智	一个类同时是 API 契约和表	领域模型 ≠ API 契约（用 pydantic schema 隔离）
async	支持但较新	成熟
复杂查询	一般	强大
定位	快速原型 / 中小项目	生产级主力

经验：中小项目、追求极简用 SQLModel；复杂查询、长期维护、要分离领域与契约用 SQLAlchemy。本教程的贯穿项目选 SQLAlchemy（生产级示范）。

3.2 SQLModel 快速上手¶

uv add sqlmodel

模型同时是 pydantic 模型和表定义（FastAPI 同作者出品）：

from sqlmodel import SQLModel, Field, Session, create_engine, select

class User(SQLModel, table=True):           # table=True → 也是数据库表
    id: int | None = Field(default=None, primary_key=True)
    username: str
    api_key: str

engine = create_engine("sqlite:///app.db")  # 同步 engine
SQLModel.metadata.create_all(engine)        # 建表

with Session(engine) as session:
    # 查询：session.exec（SQLModel 风格）
    user = session.exec(select(User).where(User.username == "alice")).first()

优点：User 既是表又是 API 模型，路由里直接 user: User 当请求体——零转换。

代价：API 契约和表结构绑死（想给 API 多/少字段要拆模型）；复杂查询能力不如纯 SQLAlchemy。

3.3 SQLAlchemy 2.0 async（生产级主线）¶

这是贯穿项目的选型，对照 JPA/Hibernate 最贴切。

engine 与 session¶

# bookmarks-api/src/bookmarks_api/db.py
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase): ...                       # 所有模型基类

engine = create_async_engine("sqlite+aiosqlite:///./bookmarks.db")
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

sqlite+aiosqlite:// —— async 驱动（aiosqlite 是 sqlite3 的 async 包装）。Postgres 用 postgresql+asyncpg://。
expire_on_commit=False —— commit 后对象仍可用（async 默认会失效，需重新查询，设 False 避免麻烦）。

模型（2.0 typed mapping）¶

JPA Entity

@Entity
class User {
    @Id Long id;
    @Column(unique=true) String username;
    @OneToMany List<Bookmark> bookmarks;
}

SQLAlchemy 2.0

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50), unique=True)
    bookmarks: Mapped[list["Bookmark"]] = relationship(
        back_populates="user", cascade="all, delete-orphan")

Mapped[T] + mapped_column(...) 是 2.0 的声明式风格（比老 Column 更类型友好）。完整模型见 bookmarks-api/src/bookmarks_api/models.py。

查询（`select` 语句）¶

from sqlalchemy import select

# 查询 + 过滤
result = await db.execute(select(User).where(User.username == "alice"))
user = result.scalar_one_or_none()          # 唯一或 None

# 多结果
result = await db.execute(
    select(Bookmark).where(Bookmark.user_id == user.id).order_by(Bookmark.created_at.desc())
)
bookmarks = list(result.scalars().all())

result.scalars() 把行拆成对象（async execute 返回 Row，要 .scalars() 取实体）。

关系（一对多）¶

class User(Base):
    bookmarks: Mapped[list["Bookmark"]] = relationship(back_populates="user", ...)

class Bookmark(Base):
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    user: Mapped["User"] = relationship(back_populates="bookmarks")

back_populates 双向绑定；cascade="all, delete-orphan" 让删用户时级联删其书签（对照 JPA cascade = CascadeType.ALL）。

3.4 models 与 schemas 分离（关键工程实践）¶

这是 SQLAlchemy 方案相比 SQLModel 的核心理念：数据库模型和 API 契约分离。

models.py   (SQLAlchemy ORM)   → 数据库领域模型（含关系、约束）
schemas.py  (pydantic)         → API 契约（Create / Out，对外形状）

贯穿项目正是如此——BookmarkCreate（请求：url+title）和 BookmarkOut（响应：id+url+title+created_at）是 pydantic schema，Bookmark（含 user_id 关系）是 ORM model。转换靠 from_attributes=True（第 1 章）自动完成。

为什么分离

对照 Java：JPA @Entity（表结构）和你 Controller 的 DTO 本就该分开。把表结构直接暴露成 API 会绑死两者、泄露内部字段（如 password_hash）。SQLModel 的"二合一"在小项目方便，规模一上来就会痛。

3.5 把 DB 接进路由（依赖注入）¶

第 2 章预告的 Depends(get_db) 现在落地：

# 请求级 session：自动开关
async def get_db():
    async with async_session() as session:
        yield session

@app.post("/bookmarks", response_model=BookmarkOut)
async def create_bookmark(payload: BookmarkCreate, db: AsyncSession = Depends(get_db)):
    bookmark = Bookmark(url=str(payload.url), title=payload.title, ...)
    db.add(bookmark)
    await db.commit()
    await db.refresh(bookmark)
    return bookmark

Depends(get_db) 让每个请求拿到独立的 session，请求结束自动关闭（yield 之后）。对照 Spring 的 @Transactional + OpenSessionInView，但显式得多。

3.6 API Key 认证（依赖注入实战）¶

现在用 Depends 做认证——从请求头解析 API key，查出当前用户：

# bookmarks-api/src/bookmarks_api/deps.py
from fastapi import Depends, Header, HTTPException, status

async def get_current_user(
    x_api_key: str | None = Header(None, alias="X-API-Key"),   # 读请求头
    db: AsyncSession = Depends(get_db),                         # 依赖可以依赖！
) -> User:
    if not x_api_key:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "missing api key")
    result = await db.execute(select(User).where(User.api_key == x_api_key))
    user = result.scalar_one_or_none()
    if user is None:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid api key")
    return user

路由里声明依赖即可——user 自动注入：

@app.get("/bookmarks")
async def list_bookmarks(user: User = Depends(get_current_user), db = Depends(get_db)):
    ...   # user 已是认证过的当前用户

精妙之处：get_current_user 自己也 Depends(get_db)——依赖可以链式组合。每个受保护路由只需声明 Depends(get_current_user)，认证逻辑全局复用。对照 Spring Security 的过滤器链，但用普通函数 + 类型表达，更直观。

Header(None, alias="X-API-Key")：把 HTTP 头 X-API-Key 映射成参数（用 None 默认值，缺失时返回 401 而非 422）。

3.7 权限隔离（"只能操作自己的"）¶

认证只解决"你是谁"，授权解决"你能干嘛"。书签 API 要让用户只能 CRUD 自己的书签——靠查询条件天然实现：

async def _get_owned(bookmark_id: int, user: User, db: AsyncSession):
    # 关键：where 同时限定 id 和 user_id
    result = await db.execute(
        select(Bookmark).where(Bookmark.id == bookmark_id, Bookmark.user_id == user.id)
    )
    return result.scalar_one_or_none()

@app.delete("/bookmarks/{bookmark_id}", status_code=204)
async def delete_bookmark(bookmark_id: int, user=Depends(get_current_user), db=Depends(get_db)):
    bm = await _get_owned(bookmark_id, user, db)
    if bm is None:
        raise HTTPException(404, "bookmark not found")   # 别人的或不存在都 404
    await db.delete(bm); await db.commit()

安全细节

查询时 where id=? AND user_id=current —— 别人的书签根本查不到，返回 404（不泄露"存在但无权"）。这比"查出后判断 owner"更安全、更省事。测试 test_permission_isolation 验证了这一点。

3.8 API Key vs JWT：取舍¶

	API Key	JWT
凭证形态	固定字符串，服务端存 key→user	自包含 token（base64 负载 + 签名）
验证	查 DB	验签名（无需查 DB）
状态	有状态（服务端记录）	无状态
有效期	通常长期	短期 + refresh token
撤销	删记录即生效	难（需黑名单）
适合	内部 API、服务间、简单场景	分布式、第三方、多服务无状态验证

本教程选 API Key 是为了聚焦 FastAPI 的依赖注入（Depends 链），省去 JWT 签名/过期/refresh 的协议复杂度。

若要 JWT

FastAPI 官方文档有完整的 OAuth2 Password Flow + JWT 示例（OAuth2PasswordBearer + python-jose/PyJWT + passlib[bcrypt]）。把本章的 get_current_user 换成"验 JWT → 解出 user_id → 查 user"，其余路由结构不变。对照 Spring Security 的 JWT 过滤器链。

本章练习¶

练习 3.1

用 SQLModel 定义 Bookmark（id 主键、url、title、user_id 外键），并写一段同步查询某 user 的所有书签。

参考答案

from sqlmodel import SQLModel, Field, Session, create_engine, select, ForeignKey
class Bookmark(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    url: str
    title: str
    user_id: int = Field(foreign_key="user.id")
engine = create_engine("sqlite:///app.db")
with Session(engine) as s:
    bms = s.exec(select(Bookmark).where(Bookmark.user_id == 1)).all()

练习 3.2

解释 models.py（ORM）和 schemas.py（pydantic）为什么要分开，以及 from_attributes=True 在转换中起什么作用。

参考答案

分离避免把表结构（含关系、敏感字段）直接暴露成 API 契约，降低耦合、便于独立演进（如加字段不破坏 API）。from_attributes=True 让 pydantic 模型能从 ORM 对象的属性读取值（BookmarkOut.model_validate(orm_obj)），无需手写字段拷贝，相当于 Java 的 MapStruct。

练习 3.3

把 get_current_user 改造成"支持两种凭证"：优先用 X-API-Key，缺失时尝试 Authorization: Bearer <key>。

参考答案提示

async def get_current_user(
    x_api_key: str | None = Header(None),
    authorization: str | None = Header(None),
    db = Depends(get_db),
):
    token = x_api_key or (authorization.removeprefix("Bearer ").strip()
                          if authorization else None)
    if not token: raise HTTPException(401, "missing credentials")
    ...

练习 3.4

为什么权限隔离用"查询时 where user_id=current"比"先查后判断 owner"更好？

参考答案

一条查询既过滤又鉴权，别人的记录查不到（404），不泄露存在性；且无需先加载再判断，少一次潜在的对象暴露。先查后判断若忘记加判断，容易把别人的对象返回（越权漏洞）。

上一章：进阶 2 · FastAPI ｜ ← 回首页｜下一章：进阶 4 · 测试与部署

进阶 3 · 持久化与认证¶

3.1 两个 ORM：怎么选¶

3.2 SQLModel 快速上手¶

3.3 SQLAlchemy 2.0 async（生产级主线）¶

engine 与 session¶

模型（2.0 typed mapping）¶

查询（select 语句）¶

关系（一对多）¶

3.4 models 与 schemas 分离（关键工程实践）¶

3.5 把 DB 接进路由（依赖注入）¶

3.6 API Key 认证（依赖注入实战）¶

3.7 权限隔离（"只能操作自己的"）¶

3.8 API Key vs JWT：取舍¶

本章练习¶

查询（`select` 语句）¶