# API + 数据库

术核 AI · 后端工程课件

## 01 建立 notes-service 的配置与目录

把 API 路由、数据库连接和环境配置分开，为后续事务与迁移建立可重复的练习环境。

### 本节目标

- 建立可独立启动的 notes-service 项目
- 从环境读取数据库地址并在启动时校验
- 区分练习账号、迁移权限和线上运行权限

### 必要概念

#### 项目边界

本模块在新的 notes-service 目录逐课演进。数据库沿用数据库开发模块的 PostgreSQL 16、notes_lab 与 app_user。它与已有 api-lab 下载包相互独立；下面的端点需要按课件新建和整合，不能在旧下载包直接调用。没有做过数据库模块的同学先准备独立的本机 PostgreSQL 16 与空 notes_lab 数据库。

#### 配置与代码

数据库地址随机器变化，表结构与业务规则随代码版本变化。配置放在环境变量或本机 .env，代码只保存字段声明和 .env.example。URL 中的密码若含 @、/、: 等字符要百分号编码；日志只报告配置项缺失，不输出完整连接串。

#### 生产账号分工

练习 app_user 可创建表以便完成迁移；上线通常由独立迁移账号拥有 DDL 权限，运行账号仅获得当前服务所需表的读写及 identity 序列权限。健康检查不证明账号能完成所有业务写操作，权限仍需在部署验收中覆盖。

### 动手实现

#### 1. 新建项目并安装依赖

在新的 notes-service 根目录执行。以下为 PowerShell 命令；Linux 用 python3 -m venv .venv 并激活对应环境。这里固定大版本范围，完成验收后生成锁定文件。

文件：终端 / 当前课时

````powershell
mkdir notes-service
cd notes-service
py -3 -m venv .venv
.\.venv\Scripts\python.exe -m pip install "fastapi>=0.121,<1" "uvicorn[standard]>=0.30,<1" "pydantic-settings>=2,<3" "SQLAlchemy>=2.0,<2.1" "psycopg[binary]>=3,<4" "alembic>=1.13,<2" "httpx>=0.27,<1" "pytest>=8,<10" "pytest-asyncio>=1,<2"
mkdir app
New-Item app/__init__.py -ItemType File
.\.venv\Scripts\python.exe -m pip freeze > requirements-lock.txt
````

#### 2. 新建环境模板

新建 .env.example，再复制为 .env，填写你在独立练习库设置的密码。不要把此示例密码用在其他环境。DATABASE_URL 的 psycopg 方言由下一课 create_async_engine 选择异步驱动。

文件：.env.example

````dotenv
DATABASE_URL=postgresql+psycopg://app_user:CHANGE_ME@127.0.0.1:5433/notes_lab
DB_POOL_SIZE=5
DB_MAX_OVERFLOW=0
DB_POOL_TIMEOUT=2
````

#### 3. 新建配置模块

Settings() 会读取当前项目根目录下的 .env；启动命令始终在 notes-service 根目录运行。env 中同名变量优先于文件。约束使拼错的负数连接数在启动时失败。

文件：app/config.py

````python
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")
    database_url: str = Field(min_length=1)
    db_pool_size: int = Field(default=5, ge=1, le=50)
    db_max_overflow: int = Field(default=0, ge=0, le=50)
    db_pool_timeout: float = Field(default=2, gt=0, le=30)

settings = Settings()
````

#### 4. 隔离本机配置并验证导入

新建 .gitignore，内容如下。在虚拟环境运行 python -c "from app.config import settings; print(settings.db_pool_size)"，只应输出 5。临时把 DB_POOL_SIZE 改成 -1，导入应出现校验失败；恢复后继续。

文件：.gitignore

````text
.env
.venv/
__pycache__/
.pytest_cache/
````

#### 5. 验证目标库与现有表

使用你安装的 psql 连接本机练习库；-W 交互输入密码。检查数据库和账号，禁止对共享数据库练习建表和失败注入。已有 notes 表在第 3 课核对结构。 沿用数据库模块的 Docker 本机端口映射 5433:5432；若安装原生 PostgreSQL，按实际监听端口同步修改 .env 与 psql 参数。

文件：终端 / 当前课时

````sql
psql -h 127.0.0.1 -p 5433 -U app_user -d notes_lab -W
-- 以下在 psql 中执行：
SELECT current_database(), current_user, version();
SELECT to_regclass('public.notes');
````

### 验收结果

- 缺少 DATABASE_URL 时启动配置校验失败；日志未打印密码。
- requirements-lock.txt 记录了实际安装版本，app/__init__.py 已存在。
- psql 显示数据库为 notes_lab、账号为 app_user，且是专用练习环境。

### 生产约束与故障演练

同一份代码在开发、测试和线上读取不同数据库配置。

#### 实现决策

- 配置缺失立即失败，不回退连接开发数据库。
- 运行账号与迁移账号分开管理；练习账号权限不直接复制到线上。

#### 隔离环境演练

只在本机把端口改为未监听端口，记录配置校验成功而连接失败的区别，恢复配置。

#### 恢复验收

- 配置不含默认真实密码。
- 能用脱敏信息判断地址、认证、权限三个故障层。

### 常见错误

#### 导入配置提示 Field required

原因：未创建 .env，或从错误工作目录运行。

处理：在 notes-service 根目录复制 .env.example 为 .env 并填写数据库连接；不要在 Python 中硬编码密码绕过。

#### 数据库地址正确却认证失败

原因：URL 特殊字符未编码，或账号密码与本机实例不一致。

处理：先用 psql -W 验证凭据，再修正 URL 的密码编码；不要把连接串贴进排错截图。

### 扩展练习

新增 app_env 配置，只接受 local、test、production。

提示：用 typing.Literal 声明允许值。

参考答案：

在 config.py 导入 Literal，然后在 Settings 中加入 app_env: Literal["local", "test", "production"] = "local"。在 .env 设置 APP_ENV=staging 应触发校验失败。

### 技术文档

- [Pydantic Settings：环境配置](https://docs.pydantic.dev/latest/concepts/pydantic_settings/)
- [SQLAlchemy：数据库 URL](https://docs.sqlalchemy.org/en/20/core/engines.html#database-urls)

---

## 02 异步 Session 的生命周期

建立一个进程级连接池，每个请求创建自己的 AsyncSession，并明确同步与异步驱动的使用位置。

### 本节目标

- 解释 Engine、连接池、连接与 Session 的关系
- 通过依赖管理请求 Session 并释放资源
- 避免在 asyncio 并发任务或线程之间共享 Session

### 必要概念

#### 选择一条完整的执行路径

本模块使用 async def 路由、SQLAlchemy AsyncSession 与 psycopg 3 异步连接。同步方案也可用于阻塞型后端：def 路由搭配同步 Session，让框架在线程池执行；不能在 async def 中直接执行同步数据库 I/O。异步提升等待期间的并发利用率，不使数据库查询本身更快。

#### 对象的生命周期

Engine 与 sessionmaker 在每个服务进程内复用；Session 属于一次请求或一个任务。Session 通常到执行 SQL 时才借连接。事务结束后连接归还池，Session 关闭清理剩余状态；进程退出还需 dispose Engine。

#### 并发安全

Session 跟踪待写对象、事务与连接，不能传给 asyncio.gather 的多个并发分支，也不能传进工作线程。独立事务的任务各自通过 SessionFactory() 获取 Session；要求原子的多步操作应在同一任务中顺序使用同一事务。

### 动手实现

#### 1. 新建数据库模块

依赖上一课 app/config.py。create_async_engine 使用 postgresql+psycopg URL 自动选择异步连接。连接与 SQL 超时是本课程起始练习值，后续用预算与测量调整。

文件：app/db.py

````python
from typing import Annotated
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from app.config import settings

engine = create_async_engine(
    settings.database_url,
    pool_size=settings.db_pool_size,
    max_overflow=settings.db_max_overflow,
    pool_timeout=settings.db_pool_timeout,
    pool_pre_ping=True,
    connect_args={"connect_timeout": 3, "options": "-c statement_timeout=3000 -c lock_timeout=1000"},
)
SessionFactory = async_sessionmaker(engine, expire_on_commit=False)

async def get_session():
    async with SessionFactory() as session:
        yield session

Db = Annotated[AsyncSession, Depends(get_session, scope="function")]
````

#### 2. 新建应用入口

依赖上一步 db.py。/live 仅证明进程能响应；/ready 尝试查询数据库。生产探针需结合平台超时设置，readiness 失败使流量停止进入该实例，不应用于不断重启所有实例。

文件：app/main.py

````python
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from sqlalchemy import text
from sqlalchemy.exc import SQLAlchemyError
from app.db import Db, engine

@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    await engine.dispose()

app = FastAPI(title="notes-service", lifespan=lifespan)

@app.get("/live")
async def live():
    return {"status": "ok"}

@app.get("/ready")
async def ready(db: Db):
    try:
        async with db.begin():
            await db.execute(text("SELECT 1"))
    except SQLAlchemyError as exc:
        raise HTTPException(503, "database unavailable") from exc
    return {"status": "ready"}
````

#### 3. 启动并检查资源释放

Linux 可执行 python -m uvicorn app.main:app --reload。Windows 的 psycopg 异步连接需要 SelectorEventLoop；新建下面的本机启动器后执行 python run_server.py。此入口只用于本机单进程练习，修改代码后手工重启。

文件：run_server.py

````python
import asyncio
import sys
import uvicorn

if __name__ == "__main__":
    if sys.platform == "win32":
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    uvicorn.run("app.main:app", host="127.0.0.1", port=8000, loop="asyncio")
````

#### 4. 观察借还连接

访问 /ready 十次，再在 psql 查询下面 SQL。看到少量 idle 池连接是正常的；idle in transaction 才表示事务已开启但没有继续工作。请求数不应使连接数持续增长。

文件：终端 / 当前课时

````sql
SELECT application_name, state, count(*)
FROM pg_stat_activity
WHERE datname = 'notes_lab' AND usename = 'app_user'
GROUP BY application_name, state;
````

#### 5. 单独任务使用自己的会话

新建 scripts 目录及 __init__.py，再建立 scripts/probe.py，执行 python -m scripts.probe。不要把路由 db 传入 worker；每个 probe 仅持有自己的事务。Windows 在此脚本同样使用 Selector 策略。

文件：scripts/probe.py

````python
import asyncio
import sys
from sqlalchemy import text
from app.db import SessionFactory, engine

async def probe():
    async with SessionFactory.begin() as db:
        return await db.scalar(text("SELECT 1"))

async def main():
    try:
        print(await asyncio.gather(probe(), probe()))
    finally:
        await engine.dispose()

if __name__ == "__main__":
    if sys.platform == "win32":
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    asyncio.run(main())
````

### 验收结果

- /live 和 /ready 都返回 200；停止本机数据库后 /live 仍成功而 /ready 为 503。
- 连续请求结束后没有属于 API 的 idle in transaction 连接。
- 两个 probe 返回 [1, 1]，各自创建会话，退出时 Engine 已 dispose。

### 生产约束与故障演练

一个实例同时接收普通 API、慢客户端与后台任务。

#### 实现决策

- 请求依赖只负责会话关闭，业务函数在返回前完成事务。
- 慢流式响应先查询有限数据并转 DTO，再关闭事务；不把连接绑定整个流。

#### 隔离环境演练

在隔离环境连续调用 readiness 并中断客户端，查询 pg_stat_activity 确认没有遗留事务。

#### 恢复验收

- Session 不跨并发任务共享。
- 数据库不可用时 readiness 为 503，响应不包含数据库密码或原始异常。

### 常见错误

#### 出现 MissingGreenlet 或异步驱动不兼容

原因：异步 ORM 触发隐式 I/O，或 Windows 使用了 Proactor 事件循环。

处理：先确认错误堆栈：循环问题使用本课启动器；属性加载问题按第 5 课显式加载，不要混用同步驱动。

#### 并发时提示 transaction already begun / operation in progress

原因：同一个 Session 被多个协程共享。

处理：把 SessionFactory 传给任务或在每个任务内部创建 Session；同一原子事务保持顺序执行。

### 扩展练习

让两个并发任务分别返回当前数据库连接的 PID。

提示：SELECT pg_backend_pid() 代替 SELECT 1；不要对 PID 值做相等或不等的断言。

参考答案：

probe 中返回 await db.scalar(text("SELECT pg_backend_pid()"))。由于连接池会复用连接，PID 可能相同也可能不同；判断会话隔离应检查每个任务创建了自己的 Session，而非只看 PID。

### 技术文档

- [SQLAlchemy：异步 ORM](https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html)
- [Psycopg：异步与 Windows](https://www.psycopg.org/psycopg3/docs/advanced/async.html)
- [FastAPI：依赖作用域](https://fastapi.tiangolo.com/advanced/advanced-dependencies/)

---

## 03 ORM CRUD 与输入输出模型

把 notes 表映射为 ORM，对输入约束、数据库实体和 API 响应分别建模。

### 本节目标

- 对齐 PostgreSQL 字段类型与 Python 映射
- 实现创建、读取、修改和删除笔记
- 将 ORM 对象在事务内转换为稳定的 DTO

### 必要概念

#### 三种模型回答不同问题

数据库表定义存储约束，ORM 定义 Python 对象与表的映射，Pydantic DTO 定义允许客户端提交和读取的字段。客户端不能提交 id 或 created_at。即使 Pydantic 校验通过，数据库约束仍是其他写入路径的最后防线。

#### flush 与 commit

add 只把对象加入会话；flush 执行 INSERT 并取得 id，但事务尚未提交。离开 db.begin() 的成功路径才 commit。先生成 DTO 可以发现类型或字段缺失问题，成功离开事务后再返回 HTTP 响应。

#### PATCH 的省略与空值

PATCH 只更新本次提供的字段。exclude_unset=True 保留显式传入的 null，因此必须决定 null 的业务含义。本服务的三个可修改字段均 NOT NULL，显式 null 返回 422；没有传入的字段保持原值。

### 动手实现

#### 1. 新建基础 ORM

依赖 app/db.py。数据库模块已有 public.notes 时先用 \d+ notes 核对：本课不改变已有表。没有表时执行下一步初始化 SQL。updated_at 由本模块所有写路径主动刷新，不声称 server_default 会在 UPDATE 自动执行。

文件：app/models.py

````python
from datetime import datetime
from sqlalchemy import BigInteger, Boolean, DateTime, Identity, String, Text, func, text
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class Note(Base):
    __tablename__ = "notes"
    id: Mapped[int] = mapped_column(BigInteger, Identity(), primary_key=True)
    title: Mapped[str] = mapped_column(String(120), nullable=False)
    content: Mapped[str] = mapped_column(Text, nullable=False, server_default=text("''"))
    pinned: Mapped[bool] = mapped_column(Boolean, nullable=False, server_default=text("false"))
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now())
````

#### 2. 仅对没有 notes 表的练习库初始化

在 notes_lab 的 psql 执行。已有表不执行 CREATE；表缺字段时回到数据库模块对齐，不使用 drop/recreate 破坏数据。第 7 课将这些 DDL 纳入版本迁移。

文件：终端 / 当前课时

````sql
CREATE TABLE notes (
    id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    title varchar(120) NOT NULL,
    content text NOT NULL DEFAULT '',
    pinned boolean NOT NULL DEFAULT false,
    created_at timestamptz NOT NULL DEFAULT now(),
    updated_at timestamptz NOT NULL DEFAULT now()
);
````

#### 3. 新建输入输出模型

创建、响应和局部更新分开。内容限制为 100000 字符是练习的业务上限，代理请求体字节上限还需在部署模块设置；字符数不是字节数。

文件：app/schemas.py

````python
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field, model_validator

class NoteCreate(BaseModel):
    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
    title: str = Field(min_length=1, max_length=120)
    content: str = Field(default="", max_length=100000)
    pinned: bool = False

class NotePatch(BaseModel):
    model_config = ConfigDict(extra="forbid")
    title: str | None = Field(default=None, min_length=1, max_length=120)
    content: str | None = Field(default=None, max_length=100000)
    pinned: bool | None = None

    @model_validator(mode="after")
    def reject_null(self):
        if any(getattr(self, name) is None for name in self.model_fields_set):
            raise ValueError("provided fields must not be null")
        if self.title is not None:
            self.title = self.title.strip()
            if not self.title:
                raise ValueError("title must not be blank")
        return self

class NoteOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    title: str
    content: str
    pinned: bool
    created_at: datetime
    updated_at: datetime
````

#### 4. 新建 CRUD 路由

新建 app/routes.py，然后在 app/main.py 的 app 创建后添加 from app.routes import router 与 app.include_router(router)。这些写路径每次请求使用一个明确事务；生产鉴权与归属隔离需要在项目接入身份系统后增加，本课程接口只绑定本机练习。

文件：app/routes.py

````python
from datetime import datetime, timezone
from fastapi import APIRouter, HTTPException, Response
from app.db import Db
from app.models import Note
from app.schemas import NoteCreate, NoteOut, NotePatch

router = APIRouter(prefix="/notes", tags=["notes"])

@router.post("", response_model=NoteOut, status_code=201)
async def create_note(body: NoteCreate, db: Db):
    async with db.begin():
        note = Note(**body.model_dump())
        db.add(note)
        await db.flush()
        await db.refresh(note)
        result = NoteOut.model_validate(note)
    return result

@router.get("/{note_id}", response_model=NoteOut)
async def get_note(note_id: int, db: Db):
    async with db.begin():
        note = await db.get(Note, note_id)
        if note is None:
            raise HTTPException(404, "note not found")
        result = NoteOut.model_validate(note)
    return result

@router.patch("/{note_id}", response_model=NoteOut)
async def patch_note(note_id: int, body: NotePatch, db: Db):
    async with db.begin():
        note = await db.get(Note, note_id)
        if note is None:
            raise HTTPException(404, "note not found")
        for name, value in body.model_dump(exclude_unset=True).items():
            setattr(note, name, value)
        note.updated_at = datetime.now(timezone.utc)
        await db.flush()
        result = NoteOut.model_validate(note)
    return result

@router.delete("/{note_id}", status_code=204)
async def delete_note(note_id: int, db: Db):
    async with db.begin():
        note = await db.get(Note, note_id)
        if note is None:
            raise HTTPException(404, "note not found")
        await db.delete(note)
    return Response(status_code=204)
````

#### 5. 通过 HTTP 与 SQL 双向验收

启动服务，在 /docs 创建笔记，记录响应 id，随后读取、PATCH pinned=true 并删除。下面 Python 片段保存为 scripts/crud_probe.py 后运行 python -m scripts.crud_probe。

文件：scripts/crud_probe.py

````python
import httpx

with httpx.Client(base_url="http://127.0.0.1:8000", timeout=5) as client:
    created = client.post("/notes", json={"title": "持久化练习"})
    assert created.status_code == 201, created.text
    note_id = created.json()["id"]
    assert client.get(f"/notes/{note_id}").status_code == 200
    patched = client.patch(f"/notes/{note_id}", json={"pinned": True})
    assert patched.json()["pinned"] is True
    assert client.patch(f"/notes/{note_id}", json={"title": None}).status_code == 422
    assert client.delete(f"/notes/{note_id}").status_code == 204
    assert client.get(f"/notes/{note_id}").status_code == 404
````

### 验收结果

- POST 返回 201、数据库生成的 id 和带时区时间，重启 API 后仍能读取未删除的记录。
- PATCH 省略 content 不改变原文；显式 null 与空白标题返回 422。
- DELETE 成功返回空响应体 204，后续 GET 返回 404。

### 生产约束与故障演练

API 从内存存储迁移到真实数据库，需要保持响应与持久化一致。

#### 实现决策

- 输入模型不暴露数据库生成字段。
- 业务 DTO 在事务内完成校验，提交成功后才返回 2xx。

#### 隔离环境演练

在本机记录创建 id，重启 API 再读取；另开 psql 核对同一行，排除误用内存列表。

#### 恢复验收

- 数据跨进程重启保留。
- 请求、响应、数据库约束各有明确职责。

### 常见错误

#### 返回对象时报字段缺失或隐式查询错误

原因：在服务器默认值尚未加载、Session 已关闭后才序列化 ORM。

处理：flush 后 refresh，并在事务内部 NoteOut.model_validate；不依赖响应序列化期间访问数据库。

#### PATCH 把未传字段改成 None

原因：直接 model_dump() 将默认值也写入对象。

处理：使用 exclude_unset=True，并对显式 null 规定并验证行为。

### 扩展练习

禁止创建请求自定义 id，并验证超长标题不会到达数据库。

提示：extra="forbid" 与 max_length 分别解决两种输入问题。

参考答案：

向 POST /notes 发送 {"id":1,"title":"x"} 应为 422；发送 121 字符标题也应为 422。直接 SQL 写入超长标题仍由 varchar(120) 拒绝，不能只依赖 HTTP 校验。

### 技术文档

- [SQLAlchemy：声明式映射](https://docs.sqlalchemy.org/en/20/orm/declarative_tables.html)
- [Pydantic：模型与属性读取](https://docs.pydantic.dev/latest/concepts/models/)
- [FastAPI：局部更新](https://fastapi.tiangolo.com/tutorial/body-updates/)

---

## 04 事务边界与失败提交

把一项业务操作的多次写入放进同一事务，复现 flush 成功但 commit 失败的情况。

### 本节目标

- 识别业务事务的开始、提交和回滚位置
- 避免在 yield 依赖结束时隐藏 commit
- 让数据库提交失败发生在 HTTP 成功响应之前

### 必要概念

#### 事务由业务入口负责

get_session 负责提供和关闭资源，不知道一次业务操作包含哪些写入。路由或服务函数用 db.begin() 包住相关操作；被调用的辅助函数只执行 SQL 或 flush，不自行 commit。HTTP 请求不是数据库事务的天然边界，流式和外部网络调用尤其需要拆开。

#### 成功返回的顺序

顺序应是校验输入 → begin → SQL → 生成 DTO → commit → 返回响应。若把 commit 放在依赖 yield 后，提交时机可能晚于响应发送，客户端收到成功却没有落库。本模块只在显式事务块内提交，与依赖清理时机无关。

#### 失败之后的 Session

数据库错误会让当前事务不可继续使用。退出 db.begin() 的异常路径自动回滚，之后才映射业务错误或开始新事务；不要在已失败事务内部捕获 IntegrityError 后继续查询。提交连接中断还可能造成结果未知，重试需幂等协议，不能简单认定一定失败。

### 动手实现

#### 1. 为审计记录准备练习表

在本机 notes_lab 执行一次。DEFERRABLE 用来演示提交时检查；第 7 课会将该表纳入基线迁移。审计保存 note_id 数值而不设外键，使笔记删除后审计仍保留。

文件：终端 / 当前课时

````sql
CREATE TABLE note_audits (
    id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    note_id bigint NOT NULL,
    action text NOT NULL,
    request_key text NOT NULL,
    created_at timestamptz NOT NULL DEFAULT now(),
    CONSTRAINT uq_note_audits_request UNIQUE (request_key) DEFERRABLE INITIALLY DEFERRED
);
````

#### 2. 把创建与审计组合为事务

新建 app/services.py。辅助函数不 commit；session 的事务由调用者控制。request_key 在这里只用于审计提交冲突实验，第 10 课才实现完整幂等语义。

文件：app/services.py

````python
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession
from app.models import Note
from app.schemas import NoteCreate, NoteOut

async def create_with_audit(db: AsyncSession, body: NoteCreate, request_key: str):
    note = Note(**body.model_dump())
    db.add(note)
    await db.flush()
    await db.refresh(note)
    await db.execute(text("""
        INSERT INTO note_audits (note_id, action, request_key)
        VALUES (:note_id, 'created', :request_key)
    """), {"note_id": note.id, "request_key": request_key})
    return NoteOut.model_validate(note)
````

#### 3. 新增演示端点，在事务外捕获约束错误

追加到 app/routes.py；保留原 POST /notes。该端点路径为 POST /notes/audited，无 GET 路由冲突。异常匹配使用约束名称，未知 IntegrityError 继续抛出交给错误日志，不把所有数据库异常都说成输入错误。

文件：app/routes.py

````python
from typing import Annotated
from fastapi import Header
from sqlalchemy.exc import IntegrityError
from app.services import create_with_audit

@router.post("/audited", response_model=NoteOut, status_code=201)
async def create_audited(body: NoteCreate, db: Db, request_key: Annotated[str, Header(min_length=1, max_length=100)]):
    try:
        async with db.begin():
            result = await create_with_audit(db, body, request_key)
    except IntegrityError as exc:
        constraint = getattr(getattr(exc.orig, "diag", None), "constraint_name", None)
        if constraint == "uq_note_audits_request":
            raise HTTPException(409, "audit request key already exists") from exc
        raise
    return result
````

#### 4. 复现延迟约束在 commit 才报错

用同一个 request-key 连续调用两次。第二次 SQL INSERT 本身可以执行，离开事务块时才触发唯一约束。运行前后分别查询 notes 和 note_audits 的条数，第二次不应增加任一表。

文件：scripts/commit_probe.py

````python
import httpx

with httpx.Client(base_url="http://127.0.0.1:8000", timeout=5) as client:
    headers = {"request-key": "commit-lab-001"}
    first = client.post("/notes/audited", json={"title": "first"}, headers=headers)
    second = client.post("/notes/audited", json={"title": "must rollback"}, headers=headers)
    print(first.status_code, second.status_code)
    assert first.status_code == 201
    assert second.status_code == 409
````

#### 5. 画出并核对实际控制流

在纸上或白板沿每个箭头指出对应代码行。出现异常时从 commit 进入 rollback，再到 HTTP 409；不应有从 flush 直接跳到 HTTP 201 的路径。

文件：终端 / 当前课时

````text
输入校验 → begin → 写 notes → 写 audit → 构造 DTO → commit → HTTP 201
                                                  └─失败→ rollback → HTTP 错误
yield 依赖：创建 Session ──────────────────────────────────→ close
````

### 验收结果

- 第二次重复 request-key 返回 409，不能返回 201 后才打印提交异常。
- 重复请求失败后两张表都没有新增第二条业务数据。
- 普通 POST /notes 仍能成功，说明失败事务已回滚释放。

### 生产约束与故障演练

创建笔记必须同时写审计，客户端只能在两项都提交后看到成功。

#### 实现决策

- 业务辅助函数无隐式 commit。
- 已知约束映射成可识别错误；未知失败保留内部堆栈但响应脱敏。

#### 隔离环境演练

使用可延迟唯一约束让第二次请求在 commit 失败，核对 HTTP 与两张表的最终状态。

#### 恢复验收

- commit 失败没有 2xx。
- 回滚范围覆盖全部相关写入。

### 常见错误

#### 客户端得到成功，日志随后才报告 commit 失败

原因：依赖清理或后台任务中执行关键业务 commit。

处理：把提交放回业务事务块，离开事务后才 return；后台任务不能承诺当前请求已持久化。

#### 捕获异常后执行 SELECT 又报 InFailedSqlTransaction

原因：仍在失败事务内部复用会话。

处理：在 async with db.begin() 外捕获；需要局部容错时明确设计 SAVEPOINT，而非吞异常。

### 扩展练习

验证服务函数中间抛异常会回滚笔记。

提示：只在隔离练习分支，在审计插入前临时 raise RuntimeError。

参考答案：

在 create_with_audit 的 flush 后加入 raise RuntimeError("drill")，调用一次端点应返回服务错误，psql 查询找不到本次标题。移除故障注入后再次请求成功；不要保留公开故障接口。

### 技术文档

- [SQLAlchemy：事务管理](https://docs.sqlalchemy.org/en/20/orm/session_transaction.html)
- [PostgreSQL 16：SET CONSTRAINTS](https://www.postgresql.org/docs/16/sql-set-constraints.html)
- [FastAPI：yield 依赖](https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-with-yield/)

---

## 05 关系加载与 N+1 查询

为笔记增加评论，比较逐条懒加载、selectinload 和有界查询的 SQL 次数。

### 本节目标

- 建立一对多关系并约束删除行为
- 用显式加载避免异步隐式 I/O
- 限制主列表和关联集合的读取规模

### 必要概念

#### N+1 如何发生

先查 20 条笔记，再为每条笔记查询评论，会形成 1+20 次数据库往返。网络延迟叠加后，即使每条 SQL 都很快，接口仍慢。异步 Session 中直接读取尚未加载的关系还可能触发 MissingGreenlet。

#### 显式加载的选择

selectinload 将关联查询批量合并，适合规模受控的一对多集合。joinedload 用 JOIN 合并结果，集合关联会放大结果行并通常需要 unique()。分页笔记数不等于限制评论总数，不能把 selectinload 当作无限集合的分页方案。

#### 生产接口的边界

笔记列表只返回笔记摘要或聚合计数，评论列表单独分页。本课演示 eager loading 只使用每条两条评论的固定测试数据；真实产品默认使用 /notes/{id}/comments 有界接口，删除策略明确为数据库 CASCADE。

### 动手实现

#### 1. 新增评论表

在本机练习库执行。外键建立归属；索引支持按笔记读取并按 id 翻页。生产变更在第 7 课以后均通过 Alembic 执行。

文件：终端 / 当前课时

````sql
CREATE TABLE comments (
    id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    note_id bigint NOT NULL REFERENCES notes(id) ON DELETE CASCADE,
    body varchar(2000) NOT NULL,
    created_at timestamptz NOT NULL DEFAULT now()
);
CREATE INDEX ix_comments_note_id_id ON comments (note_id, id);
````

#### 2. 扩展 ORM 映射

将下列 import 合并到 app/models.py。把 comments 属性放入已有 Note 类体，Comment 类放在文件末尾。lazy="raise" 让未显式加载成为确定错误，避免悄悄执行 N+1。

文件：app/models.py

````python
from sqlalchemy import ForeignKey, Index
from sqlalchemy.orm import relationship

# 插入已有 Note 类内部：
# comments: Mapped[list["Comment"]] = relationship(
#     back_populates="note", lazy="raise", passive_deletes="all"
# )

class Comment(Base):
    __tablename__ = "comments"
    __table_args__ = (Index("ix_comments_note_id_id", "note_id", "id"),)
    id: Mapped[int] = mapped_column(BigInteger, Identity(), primary_key=True)
    note_id: Mapped[int] = mapped_column(ForeignKey("notes.id", ondelete="CASCADE"), nullable=False)
    body: Mapped[str] = mapped_column(String(2000), nullable=False)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    note: Mapped["Note"] = relationship(back_populates="comments", lazy="raise")
````

#### 3. 新增生产形式的分页评论接口

追加到 app/routes.py。每次最多 100 条，以严格递增 id 翻页；无评论与笔记不存在分别返回空数组与 404。输出为明确字典，未访问 Note.comments。

文件：app/routes.py

````python
from typing import Annotated
from fastapi import Query
from sqlalchemy import select
from app.models import Comment

@router.get("/{note_id}/comments")
async def list_comments(note_id: int, db: Db,
    after_id: Annotated[int, Query(ge=0)] = 0,
    limit: Annotated[int, Query(ge=1, le=100)] = 20):
    async with db.begin():
        if await db.get(Note, note_id) is None:
            raise HTTPException(404, "note not found")
        rows = (await db.scalars(select(Comment).where(
            Comment.note_id == note_id, Comment.id > after_id
        ).order_by(Comment.id).limit(limit + 1))).all()
        visible = rows[:limit]
        result = {"items": [{"id": x.id, "body": x.body} for x in visible],
                  "next_after_id": visible[-1].id if len(rows) > limit else None}
    return result
````

#### 4. 在固定小数据集比较加载方式

先在 /docs 创建 3 条专用笔记，用 psql 为每条插入 2 条评论。把下面函数追加到 scripts/probe.py，并在其 main 内调用 await compare()；只对这组 id 运行，替换 [1,2,3] 为实际值。在本机临时把 Engine echo 设为 True 观察 SQL，完成后关闭，避免记录用户内容。

文件：scripts/probe.py

````python
from sqlalchemy import select
from sqlalchemy.orm import selectinload
from app.models import Note

async def compare():
    async with SessionFactory.begin() as db:
        notes = (await db.scalars(select(Note).where(Note.id.in_([1, 2, 3]))
            .options(selectinload(Note.comments)))).all()
        print([(note.id, len(note.comments)) for note in notes])
````

#### 5. 验收集合边界与删除

访问 /notes/{id}/comments?limit=1，沿 next_after_id 翻到结尾。删除练习笔记后在 psql 查询相应 comments，应该为 0 条。将 compare 中 options 移除，访问 comments 应被 lazy="raise" 拒绝，这比隐藏 N+1 更易排错。

文件：终端 / 当前课时

````sql
SELECT note_id, count(*) FROM comments GROUP BY note_id;
EXPLAIN (ANALYZE, BUFFERS)
SELECT id, body FROM comments
WHERE note_id = 1 AND id > 0 ORDER BY id LIMIT 21;
````

### 验收结果

- 小数据 selectinload 示例用笔记查询与评论查询完成读取，没有每条笔记再发一次 SQL。
- 评论接口 limit=101 返回 422，翻页不重复已见 id。
- 删除笔记后对应评论被数据库删除，无孤立外键记录。

### 生产约束与故障演练

笔记列表最初只有少量评论，运行一段时间后单条笔记积累大量评论。

#### 实现决策

- 主列表不嵌入无界集合。
- 通过 SQL 次数与返回行数共同判断优化效果。

#### 隔离环境演练

只在练习库为单条笔记插入较多评论，比较 eager loading 与分页接口的返回规模；限制数据量避免占满磁盘。

#### 恢复验收

- 关联读取有明确上限。
- 响应序列化不触发数据库查询。

### 常见错误

#### 分页笔记 20 条仍返回几十万条评论

原因：主表分页没有限制关联集合规模。

处理：默认分离评论分页接口，只在已知小集合使用 selectinload；聚合计数使用 GROUP BY 子查询。

#### 删除带评论的笔记出现外键非空错误

原因：ORM 尝试把外键置空，与数据库 CASCADE 策略不一致。

处理：使用本课 passive_deletes="all" 让数据库处理删除；同一会话内已加载对象可能过期，删除后不再返回其评论对象。

### 扩展练习

给评论接口增加排序反向读取。

提示：反向页需同时改变比较符与 ORDER BY。

参考答案：

使用 before_id 游标，WHERE id < before_id ORDER BY id DESC LIMIT limit+1。首次无游标时不加比较条件，下一页传上一页末尾 id；不要只改排序而保留 id > after_id。

### 技术文档

- [SQLAlchemy：关系加载策略](https://docs.sqlalchemy.org/en/20/orm/queryguide/relationships.html)
- [SQLAlchemy：数据库级删除级联](https://docs.sqlalchemy.org/en/20/orm/cascades.html#using-foreign-key-on-delete-cascade-with-orm-relationships)

---

## 06 连接池预算、超时与多 Worker

从 PostgreSQL 连接上限反推每个进程的池容量，区分借连接、建连接、锁等待与 SQL 执行超时。

### 本节目标

- 计算多实例多 Worker 下的最坏连接数
- 为数据库等待设置分层预算
- 在隔离环境复现连接池耗尽并观察恢复

### 必要概念

#### 连接数按进程相乘

每个 Uvicorn worker 都拥有自己的 Engine 与连接池。应用最坏连接数约为实例数 × worker 数 × (pool_size + max_overflow)，还要加任务消费者、迁移、运维与其他应用。连接池预算小于数据库可用配额，不能仅把 max_connections 全部分给 API。

#### 四种等待

pool_timeout 是等待池中连接的时间，connect_timeout 是建立新连接的时间，lock_timeout 是 SQL 等锁的上限，statement_timeout 是单条 SQL 执行总上限。它们有重叠，不能简单相加等同请求耗时；请求预算还包括网络、序列化与多条 SQL。

#### 拥塞时及时拒绝

扩大池会增加数据库并发和内存压力。先缩短事务、排查慢 SQL、控制入口与任务并发，再测量池容量。pool_pre_ping 能识别借出前的失效连接，不能修复事务执行中断，也不会安全重放写操作。

### 动手实现

#### 1. 计算部署预算

先用 psql SHOW max_connections，并列出其他客户端。以下为练习计算，不是通用推荐值；2 个实例、每个 3 worker、每 worker 5 连接占 30，滚动发布若新旧全部重叠可能占 60。

文件：终端 / 当前课时

````text
数据库上限 100
保留给管理与故障处理 10
其他服务与后台任务 20
API 稳态 2 × 3 × (5 + 0) = 30
发布峰值按 4 × 3 × 5 = 60 估算
峰值总需求 60 + 20 + 10 = 90 < 100
````

#### 2. 检查当前生效超时

第 2 课 Engine 已设置连接和 SQL 超时。运行下列 psql 只看到当前 psql 会话配置；要验证应用连接，给 scripts/probe.py 的 probe 函数临时替换为下一段，运行后恢复。

文件：scripts/probe.py

````python
async def probe():
    async with SessionFactory.begin() as db:
        rows = (await db.execute(text("""
            SELECT current_setting('statement_timeout') AS statement_timeout,
                   current_setting('lock_timeout') AS lock_timeout
        """))).mappings().one()
        return dict(rows)
````

#### 3. 为池耗尽提供明确响应

添加到 app/main.py。只映射 SQLAlchemy 的池等待 TimeoutError；SQL 执行取消与数据库断连需要分别记录和处理，不能都笼统重试。请求日志应包含 request_id、延迟、异常类别，排除参数与连接串。

文件：app/main.py

````python
from fastapi import Request
from fastapi.responses import JSONResponse
from sqlalchemy.exc import TimeoutError as PoolTimeout

@app.exception_handler(PoolTimeout)
async def pool_timeout_handler(request: Request, exc: PoolTimeout):
    return JSONResponse(status_code=503,
        content={"detail": "database capacity exhausted"},
        headers={"Retry-After": "1"})
````

#### 4. 本机复现池竞争

新建 scripts/pool_drill.py。它创建单独练习 Engine，不改服务的配置。SELECT pg_sleep 仅在本机 notes_lab 使用；两任务各自用连接，一个等不到连接触发池超时，finally 确保释放。Windows 使用本课代码内 Selector 策略。

文件：scripts/pool_drill.py

````python
import asyncio
import sys
from sqlalchemy import text
from sqlalchemy.exc import TimeoutError
from sqlalchemy.ext.asyncio import create_async_engine
from app.config import settings

async def main():
    engine = create_async_engine(settings.database_url, pool_size=1,
        max_overflow=0, pool_timeout=0.1)
    async def hold():
        async with engine.begin() as conn:
            await conn.execute(text("SELECT pg_sleep(0.5)"))
    try:
        results = await asyncio.gather(hold(), hold(), return_exceptions=True)
        assert sum(isinstance(x, TimeoutError) for x in results) == 1, results
        async with engine.begin() as conn:
            assert await conn.scalar(text("SELECT 1")) == 1
        print("one rejected; pool recovered")
    finally:
        await engine.dispose()

if __name__ == "__main__":
    if sys.platform == "win32":
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    asyncio.run(main())
````

#### 5. 验证慢 SQL 与锁等待分开

应用 probe 事务中分别执行 pg_sleep(4)，或在 psql 事务锁定一条专用笔记再由 API PATCH。记录哪个超时先发生。锁练习结束在 psql 执行 ROLLBACK；不要对业务库执行这些命令。

文件：终端 / 当前课时

````sql
BEGIN;
SELECT id FROM notes WHERE id = 1 FOR UPDATE;
-- 将 1 换成本机专用练习笔记 id；此时在另一个终端 PATCH 同一笔记。
ROLLBACK;
````

### 验收结果

- 预算表包含 worker、实例、发布重叠和后台任务。
- 池演练输出 one rejected; pool recovered；失败后新查询仍成功。
- 能区分池等待超时、statement_timeout 与 lock_timeout，并记录实际延迟。

### 生产约束与故障演练

服务扩容和滚动发布时，短时间内进程数显著增加。

#### 实现决策

- max_overflow=0 起步使容量上限可预测。
- 容量不足返回可恢复错误，重试使用退避与幂等；不在应用层无限排队。

#### 隔离环境演练

独立池限制为 1，两个任务争抢连接；确认一次失败不会泄露唯一连接。

#### 恢复验收

- 失败后池恢复。
- 最坏连接需求含部署峰值且留管理余量。

### 常见错误

#### 把 worker 从 2 加到 8 后数据库连接爆满

原因：每个 worker 都创建自己的池，遗漏乘数。

处理：重新预算所有实例与发布重叠；根据 CPU 与等待特征调 worker，按数据库配额分配池。

#### 设置 pool_timeout 后慢查询仍占用很久

原因：pool_timeout 只限制借连接等待，不限制已借到连接上的 SQL。

处理：分别设置 statement_timeout 与 lock_timeout，并检查事务持有连接时是否等待外部 API。

### 扩展练习

如果实例翻倍但数据库配额不变，如何保持最坏连接数不变？

提示：把进程总数和每池连接上限一起计算。

参考答案：

在 worker 不变时将每 worker 的 pool_size + max_overflow 约减半，同时验证排队时间与吞吐。连接配额只是上限，不能保证响应时间；需要限流和负载测量，而非只修改参数。

### 技术文档

- [SQLAlchemy：连接池](https://docs.sqlalchemy.org/en/20/core/pooling.html)
- [PostgreSQL 16：客户端连接默认设置](https://www.postgresql.org/docs/16/runtime-config-client.html)
- [Uvicorn：部署与 Worker](https://uvicorn.dev/deployment/)

---

## 07 Alembic 迁移与 Expand–Contract

把手工练习表纳入版本管理，以新增可空字段、分批回填、收紧约束完成兼容发布。

### 本节目标

- 为 async Engine 建立 Alembic 环境
- 区分 upgrade、autogenerate 与 stamp
- 设计新旧应用可同时运行的数据库演进

### 必要概念

#### 迁移版本是发布状态

ORM 模型描述当前目标结构，迁移文件描述从旧版如何到新版。autogenerate 是候选差异，不能理解改名或业务回填语义。生产服务启动不执行 create_all；数据库迁移由单独发布步骤执行一次并记录版本。

#### 接管已有数据库

数据库开发模块的 notes_lab 可能已包含 version、检查约束与额外索引，不能直接套用本课从零生成的迁移历史。本课固定使用新建 notes_lab_migration 数据库做迁移实验，原 notes_lab 保留不动。stamp 只登记版本，既不建表也不验证结构；本练习不对已有库执行 stamp 或重复 ADD version。接管真实已有库需要单独的结构盘点与基线计划。

#### 兼容发布

Expand 先增加兼容字段或表；新旧应用共存期间双写或保留旧字段；后台按主键分批回填并核验；最后 Contract 收紧约束或删除旧字段。每步都需锁等待预算和回退窗口，数据删除不保证可通过 downgrade 恢复。

### 动手实现

#### 1. 初始化异步迁移环境

在 notes-service 根目录执行。生成 migrations/env.py 后按下一步修改，migration URL 从配置读取，不写密码进 alembic.ini。

文件：终端 / 当前课时

````bash
python -m alembic init -t async migrations
````

#### 2. 配置 metadata 与连接地址

在生成的 migrations/env.py 中导入下面内容，替换原 target_metadata = None；使用 set_main_option 时将 % 转为 %%，因为 Alembic 配置有插值规则。保留 async 模板的 run_sync、NullPool 与上下文，不自行改成 asyncio.run 包套 asyncio.run。 Windows 在 env.py 导入 asyncio 后、执行 run_migrations_online 之前，添加 import sys，并在 sys.platform == "win32" 时设置 asyncio.WindowsSelectorEventLoopPolicy()，与应用入口保持一致。

文件：migrations/env.py

````python
from app.config import settings
from app.models import Base

target_metadata = Base.metadata
config.set_main_option("sqlalchemy.url", settings.database_url.replace("%", "%%"))

import sys
if sys.platform == "win32":
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
````

#### 3. 生成并审核基线

先把下列 note_audits Core Table 追加到 app/models.py，与现有 Note/Comment 一起进入 Base.metadata。此时 API 模型按第 3–5 课尚未声明 version，原数据库即使有该列也不影响读写。下一步只对新的空迁移实验库生成基线，不从已有 notes_lab 自动生成差异。

文件：app/models.py

````python
from sqlalchemy import Column, MetaData, Table, UniqueConstraint

note_audits = Table("note_audits", Base.metadata,
    Column("id", BigInteger, Identity(), primary_key=True),
    Column("note_id", BigInteger, nullable=False),
    Column("action", Text, nullable=False),
    Column("request_key", Text, nullable=False),
    Column("created_at", DateTime(timezone=True), nullable=False, server_default=func.now()),
    UniqueConstraint("request_key", name="uq_note_audits_request",
        deferrable=True, initially="DEFERRED"),
)
````

#### 4. 创建空迁移实验库并明确切换

在本机 PostgreSQL 管理账号可用的前提下执行前两条命令，-W 交互输入管理密码；若使用数据库模块的Docker容器，可在其postgres容器内使用管理psql执行同一CREATE DATABASE。已有同名库时先停止并核对来源，不删除重建。将 notes-service/.env 的 DATABASE_URL 数据库名从 notes_lab 改为 notes_lab_migration，保留原值在未提交的本机备份中；同时清除终端中覆盖 .env 的 DATABASE_URL。第 7–11 课继续使用新库，避免把教学迁移应用到已有表。

文件：终端 / 当前课时

````powershell
psql -h 127.0.0.1 -p 5433 -U postgres -d postgres -W -c "CREATE DATABASE notes_lab_migration OWNER app_user;"
psql -h 127.0.0.1 -p 5433 -U app_user -d notes_lab_migration -W -c "SELECT current_database(), current_user;"
# PowerShell：清除可能优先于.env的旧环境变量
Remove-Item Env:DATABASE_URL -ErrorAction SilentlyContinue
# 用编辑器将.env连接串末尾改为/notes_lab_migration，再运行：
python -m alembic revision --autogenerate -m "baseline notes comments audits"
python -m alembic upgrade head
python -m alembic check
python -m alembic current
````

#### 5. 新增 nullable version 字段

仅在刚创建的 notes_lab_migration 完成基线后执行：alembic revision -m "expand note version"，把下面函数写入生成文件并保留 revision/down_revision/import op/import sqlalchemy as sa。同步在 Note 类加入 version: Mapped[int | None] = mapped_column(nullable=True, server_default=text("1"))。原 notes_lab 已有 version 时不执行本迁移；本练习的新增与回填只发生在独立新库。

文件：migrations/versions/<revision>_expand_note_version.py

````python
def upgrade():
    op.execute("SET LOCAL lock_timeout = '1s'")
    op.add_column("notes", sa.Column("version", sa.Integer(), nullable=True, server_default=sa.text("1")))

def downgrade():
    op.drop_column("notes", "version")
````

#### 6. 分批回填后收紧非空约束

先 upgrade，再在 psql 重复下面有界 UPDATE，每次独立提交，直到 UPDATE 0。随后生成新 revision 收紧 nullable=False，并把 ORM version 注解改为 Mapped[int]。大表可先 ADD CHECK (version IS NOT NULL) NOT VALID，再 VALIDATE CONSTRAINT，最后 SET NOT NULL 并删除过渡 CHECK；验证与锁预算在预发布副本测量。

文件：终端 / 当前课时

````sql
WITH batch AS (
    SELECT id FROM notes WHERE version IS NULL ORDER BY id LIMIT 1000
)
UPDATE notes AS n SET version = 1 FROM batch WHERE n.id = batch.id;

SELECT count(*) FROM notes WHERE version IS NULL;
-- 收紧约束的下一份 Alembic revision 中：
-- op.alter_column("notes", "version", existing_type=sa.Integer(), nullable=False)
````

#### 7. 初始化练习数据并保留返回路径

迁移实验库从零开始，重启API使新.env生效，用/docs创建几条新笔记和评论再完成后续课。第7–11课操作目标始终为notes_lab_migration；不把旧库记录默默复制进新库。完成本模块后可恢复.env原notes_lab连接，但若要在那里使用新功能，需先为该库制定独立迁移接管计划，不能直接upgrade本课历史。第12课使用另一专用notes_lab_test。

文件：终端 / 当前课时

````sql
# 在当前连接的psql中确认目标及新字段
SELECT current_database();
SELECT column_name FROM information_schema.columns
WHERE table_schema='public' AND table_name='notes' ORDER BY ordinal_position;
````

### 验收结果

- notes_lab_migration 从空库 upgrade 得到三张基础表与 version；原 notes_lab 没有执行本课DDL。
- 配置确实指向 notes_lab_migration，终端环境变量不会覆盖 .env，API重启后写入新库。
- 回填仅操作新实验库，收紧前 version IS NULL 为0，alembic current 为预期版本。

### 生产约束与故障演练

服务不能停机，新旧版本会在滚动发布时共同写同一张表。

#### 实现决策

- 先兼容扩展，观测回填与应用切换，再收紧结构。
- 迁移单独运行，自动生成文件必须人工读懂每一条 DDL。

#### 隔离环境演练

在隔离验证库制造长事务阻挡 ALTER，确认 lock_timeout 使迁移快速失败，结束阻塞后可重跑。

#### 恢复验收

- 已有库不会被直接stamp或重复ADD version。
- 能够从空库复现迁移历史，清楚原库接管是独立工作。

### 常见错误

#### autogenerate 把重命名识别成删列加列

原因：工具只看结构差异，不理解业务意图。

处理：手工改为兼容新增、迁移数据与后续删除，审核生成 SQL；有数据列的 DROP 不直接上线。

#### 线上多实例启动时同时运行迁移

原因：把 migration 放入每个 worker 的启动钩子。

处理：迁移移到唯一发布作业，设置锁等待上限，成功后启动或推进应用发布。

### 扩展练习

旧应用没有 version 字段时，Expand 阶段的新插入会怎样？

提示：区分 Python default 与数据库 server_default。

参考答案：

本课 version 的 server_default=1 在数据库生效，旧应用省略该列仍得到 1。单写 Python default 无法覆盖旧应用或 SQL 客户端。启用版本校验前仍需完成所有写路径升级，否则旧应用不会递增版本。

### 技术文档

- [Alembic：异步迁移](https://alembic.sqlalchemy.org/en/latest/cookbook.html#using-asyncio-with-alembic)
- [Alembic：自动生成限制](https://alembic.sqlalchemy.org/en/latest/autogenerate.html)
- [PostgreSQL 16：ALTER TABLE](https://www.postgresql.org/docs/16/sql-altertable.html)

---

## 08 Keyset 分页与数据库错误映射

使用稳定复合游标读取笔记列表，把已知约束转成明确的 API 错误。 本课继续连接第7课的 notes_lab_migration，原 notes_lab 不执行这些教学迁移。

### 本节目标

- 实现 created_at 与 id 复合 keyset 游标
- 限制分页大小并校验游标结构
- 按约束名称映射数据库错误而不泄露 SQL

### 必要概念

#### 稳定排序

只用 created_at 排序可能出现时间相同的多条记录，必须加唯一 id 作为第二排序键。降序下一页使用 (created_at,id) < 上页末条键。游标能减少深 OFFSET 扫描，但不是一致性快照；并发删除会导致后续页变短，新插入通常落在已经翻过的前面。

#### 游标是协议

本课将时间和 id 编码为 URL 安全 Base64 JSON，并在入口限制长度与字段类型。编码不是加密或授权，客户端可伪造游标；若绑定用户、过滤器或敏感字段，需要签名与权限过滤。固定排序键 created_at 不在 API 中允许修改。

#### 约束错误分层

请求格式错误由 Pydantic 返回 422。业务唯一冲突可为 409，明确的检查约束可以映射 422；连接失效不是客户端输入问题。先用约束名称或 SQLSTATE 识别，再决定映射，不把数据库异常原文直接返回。

### 动手实现

#### 1. 创建复合索引迁移

生成一个 Alembic revision 并填入函数，使用已有 op 与 sa 导入。课程练习数据量小可普通建索引；大表生产应评估 CREATE INDEX CONCURRENTLY，注意它不能在普通事务块内执行，需独立迁移与失败索引清理预案。 同步在 Note 类的 __table_args__ 加入 Index("ix_notes_created_id", "created_at", "id")，若已有元组则追加，不能覆盖其他约束。

文件：migrations/versions/<revision>_notes_cursor_index.py

````python
def upgrade():
    op.create_index("ix_notes_created_id", "notes", ["created_at", "id"])

def downgrade():
    op.drop_index("ix_notes_created_id", table_name="notes")
````

#### 2. 新增游标编码模块

新建 app/cursors.py。aware 时间保证与 timestamptz 比较一致；对 Base64、JSON 和 Pydantic 校验错误统一返回 422。游标只用于定位，不用于证明访问权限。

文件：app/cursors.py

````python
import base64
import binascii
import json
from fastapi import HTTPException
from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, ValidationError

class Cursor(BaseModel):
    model_config = ConfigDict(extra="forbid")
    created_at: AwareDatetime
    id: int = Field(gt=0)

def encode_cursor(created_at, note_id):
    raw = Cursor(created_at=created_at, id=note_id).model_dump_json().encode()
    return base64.urlsafe_b64encode(raw).decode().rstrip("=")

def decode_cursor(value):
    try:
        raw = base64.b64decode(value + "=" * (-len(value) % 4), altchars=b"-_", validate=True)
        return Cursor.model_validate_json(raw)
    except (ValueError, ValidationError, binascii.Error) as exc:
        raise HTTPException(422, "invalid cursor") from exc
````

#### 3. 新增分页笔记接口

追加到 app/routes.py；GET /notes 与已有 GET /notes/{note_id} 不冲突。每页多读一条决定是否有下一页；在事务内转换 DTO，列表响应不会访问未加载关系。生产接入用户后，权限过滤必须加入同一 SQL 且每页一致。

文件：app/routes.py

````python
from sqlalchemy import tuple_
from app.cursors import decode_cursor, encode_cursor

@router.get("")
async def list_notes(db: Db,
    limit: Annotated[int, Query(ge=1, le=100)] = 20,
    cursor: Annotated[str | None, Query(max_length=512)] = None):
    stmt = select(Note).order_by(Note.created_at.desc(), Note.id.desc()).limit(limit + 1)
    if cursor:
        key = decode_cursor(cursor)
        stmt = stmt.where(tuple_(Note.created_at, Note.id) < tuple_(key.created_at, key.id))
    async with db.begin():
        rows = (await db.scalars(stmt)).all()
        visible = rows[:limit]
        result = {"items": [NoteOut.model_validate(x) for x in visible],
                  "next_cursor": encode_cursor(visible[-1].created_at, visible[-1].id)
                      if len(rows) > limit else None}
    return result
````

#### 4. 新增非空白标题的数据库约束

先检查存量数据，再生成迁移。下面 upgrade 中 NOT VALID 允许先建立约束而暂缓验证存量，新写入立即受限；清理练习存量空白标题后再执行 VALIDATE。同步把 CheckConstraint 加入 Note.__table_args__，与索引一同保留，防止后续 autogenerate 误报差异。

文件：migrations/versions/<revision>_notes_title_check.py

````python
def upgrade():
    op.execute("ALTER TABLE notes ADD CONSTRAINT ck_notes_title_nonblank CHECK (length(btrim(title)) > 0) NOT VALID")
    op.execute("ALTER TABLE notes VALIDATE CONSTRAINT ck_notes_title_nonblank")

def downgrade():
    op.drop_constraint("ck_notes_title_nonblank", "notes", type_="check")
````

#### 5. 统一映射明确的约束

添加到 app/main.py。现有 audited 端点会先处理自身唯一约束，其他未识别 IntegrityError 返回 500 并用不含 SQL 参数的日志记录类别与约束；真实项目接入 request_id。不要记录 str(exc)，其中可能有用户输入。

文件：app/main.py

````python
import logging
from sqlalchemy.exc import IntegrityError

logger = logging.getLogger("notes-service")

@app.exception_handler(IntegrityError)
async def integrity_handler(request: Request, exc: IntegrityError):
    diag = getattr(exc.orig, "diag", None)
    name = getattr(diag, "constraint_name", None)
    if name == "ck_notes_title_nonblank":
        return JSONResponse(status_code=422, content={"detail": "title must not be blank"})
    logger.error("database constraint failure", extra={"constraint": name, "path": request.url.path})
    return JSONResponse(status_code=500, content={"detail": "database write failed"})
````

### 验收结果

- 相同 created_at 的多条笔记分页仍无重复、无静态数据遗漏。
- limit=0、limit=101、损坏游标都为 422，最后一页 next_cursor=null。
- 直接 SQL 写空白标题由 ck_notes_title_nonblank 拒绝，API 错误响应不含 SQL 和原始参数。

### 生产约束与故障演练

笔记增长后深分页拖慢数据库，同时出现绕过 API 的导入写入。

#### 实现决策

- 固定排序、限制页大小，数据库层仍执行关键不变量。
- 游标不承担授权职责，过滤条件改变时丢弃旧游标。

#### 隔离环境演练

在本机建立时间相同的测试记录并逐页收集 id；再发送损坏游标验证不会执行数据库查询。

#### 恢复验收

- 静态数据集完整且不重复。
- 未知数据库错误不会被包装成错误的成功或业务冲突。

### 常见错误

#### 翻页时相同时间的记录重复或遗漏

原因：仅以时间作为游标，缺少唯一排序键。

处理：查询排序、比较条件和游标编码同时包含 created_at 与 id。

#### 所有 IntegrityError 都返回 409

原因：把非空、外键、检查约束和程序错误混为一类。

处理：只匹配确认过的命名约束，未知异常保持服务端错误并记录可定位信息。

### 扩展练习

分页时能否保证第一屏到最后一屏来自同一个数据库快照？

提示：每个 HTTP 请求有自己的事务。

参考答案：

本课不能保证跨请求快照。若产品需要稳定导出，可创建导出作业并在合适的一致性边界生成文件，或定义快照时间与数据版本规则；不要保持一个数据库事务等待用户慢慢翻页。

### 技术文档

- [PostgreSQL 16：行比较](https://www.postgresql.org/docs/16/functions-comparisons.html#ROW-WISE-COMPARISON)
- [PostgreSQL 16：错误码](https://www.postgresql.org/docs/16/errcodes-appendix.html)
- [SQLAlchemy：tuple_ 表达式](https://docs.sqlalchemy.org/en/20/core/sqlelement.html#sqlalchemy.sql.expression.tuple_)

---

## 09 并发更新：版本条件与行锁

复现两个客户端覆盖彼此修改，用原子条件 UPDATE 返回 409，并识别行锁适用的短事务。

### 本节目标

- 复现读后写导致的丢失更新
- 以 version 条件原子更新并对过期请求返回 409
- 区分乐观并发控制和短事务行锁

### 必要概念

#### 版本表示读到的状态

两个客户端都读到 version=1，第一个写入后版本变为 2，第二个仍提交 1 应被拒绝。检查与写入必须在同一条 SQL 的 WHERE 中完成；先 SELECT 检查再无条件 UPDATE 仍有竞争窗口。

#### 所有写路径必须遵守协议

仅新增一个安全 PATCH 不够：旧 PATCH、批量脚本和管理工具若不递增版本，仍能覆盖别人修改。第 7 课扩展字段只解决结构兼容，启用强制版本前必须完成旧写路径切换并下线不安全写入口。

#### 行锁的用途

SELECT FOR UPDATE 可以串行化同一行的短业务判断，例如检查状态后做多表更新。锁会持续到事务结束，不能在持锁时等待模型 API、发送邮件或用户输入。乐观控制减少等待但要求客户端处理冲突；二者可以按业务组合。

### 动手实现

#### 1. 收紧 version 并扩展 DTO

前提是第 7 课回填和 NOT NULL 迁移已成功。Note 类 version 改成 Mapped[int] = mapped_column(nullable=False, server_default=text("1"))；NoteOut 新增 version: int。再在 schemas.py 增加下面的更新请求，保留 NotePatch 的字段校验。

文件：app/schemas.py

````python
class NoteVersionPatch(NotePatch):
    version: int = Field(gt=0)
````

#### 2. 替换原 PATCH 端点

删除或替换 app/routes.py 原 patch_note，避免同方法同路径重复注册。原子 UPDATE 以 id 与 version 为条件；无匹配统一 409，涵盖已删除和版本冲突，客户端重新 GET 区分是否已删除。所有 SQL 更新同步刷新 updated_at 与 version。

文件：app/routes.py

````python
from sqlalchemy import update, func
from app.schemas import NoteVersionPatch

@router.patch("/{note_id}", response_model=NoteOut)
async def patch_note(note_id: int, body: NoteVersionPatch, db: Db):
    changes = body.model_dump(exclude_unset=True, exclude={"version"})
    if not changes:
        raise HTTPException(422, "provide at least one field to update")
    async with db.begin():
        stmt = update(Note).where(Note.id == note_id, Note.version == body.version).values(
            **changes, version=Note.version + 1, updated_at=func.now()).returning(Note)
        note = (await db.scalars(stmt)).one_or_none()
        if note is None:
            raise HTTPException(409, "note changed or was deleted; reload before editing")
        result = NoteOut.model_validate(note)
    return result
````

#### 3. 让删除也携带版本

替换原 delete_note。这里用必填 version 查询参数便于 /docs 练习；生产协议也可统一为 If-Match + ETag。所有客户端脚本更新为提交当前 version，不能给 version 设置默认 1 绕过检查。

文件：app/routes.py

````python
from sqlalchemy import delete

@router.delete("/{note_id}", status_code=204)
async def delete_note(note_id: int, db: Db, version: Annotated[int, Query(gt=0)]):
    async with db.begin():
        deleted = await db.scalar(delete(Note).where(
            Note.id == note_id, Note.version == version).returning(Note.id))
        if deleted is None:
            raise HTTPException(409, "note changed or was deleted")
    return Response(status_code=204)
````

#### 4. 用两个客户端快照复现冲突

在 scripts/version_probe.py 保存下面片段，运行后应得到 200 与 409。第一次写入内容应保留，不能被旧快照覆盖；创建路径可沿用 POST /notes，新字段由数据库默认值赋 1。

文件：scripts/version_probe.py

````python
import httpx

with httpx.Client(base_url="http://127.0.0.1:8000", timeout=5) as client:
    note = client.post("/notes", json={"title": "concurrency"}).json()
    path = f"/notes/{note['id']}"
    a = client.get(path).json()
    b = client.get(path).json()
    first = client.patch(path, json={"title": "client A", "version": a["version"]})
    second = client.patch(path, json={"title": "client B", "version": b["version"]})
    assert first.status_code == 200, first.text
    assert second.status_code == 409, second.text
    current = client.get(path).json()
    assert current["title"] == "client A"
    assert current["version"] == note["version"] + 1
````

#### 5. 比较短事务行锁

下面为可替换服务函数片段，加入 app/services.py；由调用者在 db.begin() 内调用。此函数只示范锁内检查与写入，不能接收旧客户端内容进行无条件覆盖。锁等待由 lock_timeout 限制。

文件：app/services.py

````python
from sqlalchemy import select, func
from fastapi import HTTPException

async def pin_once(db: AsyncSession, note_id: int):
    note = await db.scalar(select(Note).where(Note.id == note_id).with_for_update())
    if note is None:
        raise HTTPException(404, "note not found")
    if not note.pinned:
        note.pinned = True
        note.version += 1
        note.updated_at = func.now()
        await db.flush()
        await db.refresh(note)
    return NoteOut.model_validate(note)
````

### 验收结果

- 两个相同版本更新只有一个成功，另一个为 409。
- 成功更新 version 恰好增加 1，内容与 updated_at 对应本次提交。
- 原 PATCH 与 DELETE 已替换，旧客户端缺 version 会失败而非静默覆盖。

### 生产约束与故障演练

两个管理页面同时编辑同一笔记，最后保存的人不应无提示覆盖前一次修改。

#### 实现决策

- 客户端提交读到的版本，数据库原子验证。
- 脚本与管理入口遵守同一版本规则。

#### 隔离环境演练

使用两个独立 HTTP 客户端同时提交同一 version，核验一个成功、一个冲突和数据库最终内容。

#### 恢复验收

- 不出现丢失更新。
- 409 对客户端有明确恢复流程。

### 常见错误

#### 使用 ORM version_id_col 仍有批量更新绕过版本

原因：ORM 版本检测主要作用于 flush 的单对象更新，批量 SQL 要自行加条件。

处理：本课统一显式 WHERE version 与递增；若改用 mapper versioning，仍审计所有批量路径。

#### 锁等待导致接口成批超时

原因：持锁期间执行外部请求，或多行操作锁顺序不一致。

处理：缩短事务，所有路径按相同 id 顺序加锁，外部调用通过 outbox 在提交后处理。

### 扩展练习

收到 409 后客户端可以自动读取新版本并原样重放旧编辑吗？

提示：这会绕过保护覆盖别人刚写入的内容。

参考答案：

不能无条件重放。客户端重新读取并让用户合并，或只对可证明可交换的增量操作自动重试；重试必须保留业务意图，不能只换 version。

### 技术文档

- [PostgreSQL 16：显式锁](https://www.postgresql.org/docs/16/explicit-locking.html)
- [SQLAlchemy：版本计数限制](https://docs.sqlalchemy.org/en/20/orm/versioning.html)
- [SQLAlchemy：UPDATE RETURNING](https://docs.sqlalchemy.org/en/20/tutorial/data_update.html)

---

## 10 幂等键与唯一约束

让创建请求在网络重试后返回同一次结果，使用唯一约束处理并发抢占，并拒绝同键不同内容。

### 本节目标

- 区分请求重试与重复创建业务语义
- 将幂等记录与业务写入放在同一事务
- 处理同时到达的同键请求和响应丢失

### 必要概念

#### 结果未知需要可重查

客户端超时不代表数据库回滚，服务可能已经提交，只是响应丢失。为一次逻辑创建生成稳定 Idempotency-Key 并在重试时保持不变，服务保存该键对应的状态码与响应体，就能返回第一次结果。

#### 唯一约束解决竞态

先查键不存在再插入存在竞争窗口。INSERT ON CONFLICT DO NOTHING 配合唯一约束决定谁获得写入权；获胜者的业务行与幂等结果在同一事务提交。后到请求等前一事务完成，随后在 READ COMMITTED 的下一条 SELECT 中读取已提交结果。

#### 键的作用域与保留期限

键至少绑定身份主体、HTTP 操作与请求摘要。本课尚无身份系统，只允许本机单主体练习，scope 固定表示该边界；上线必须从认证结果生成主体作用域。记录删除后相同键可能再次创建，必须公布保留期限并与客户端最大重试窗口一致。

### 动手实现

#### 1. 添加幂等结果表迁移

生成 Alembic revision，把下面 SQL 放入 op.execute；downgrade 使用 op.drop_table。它是普通非延迟主键，可供 ON CONFLICT 使用。业务成功后存 status_code 与响应；失败事务不保留占位记录。后续把此 Table 加入 Base.metadata，避免 autogenerate 误判为多余表。

文件：终端 / 当前课时

````sql
CREATE TABLE idempotency_records (
    scope varchar(120) NOT NULL,
    key varchar(100) NOT NULL,
    request_hash varchar(64) NOT NULL,
    status_code integer,
    response_body jsonb,
    created_at timestamptz NOT NULL DEFAULT now(),
    PRIMARY KEY (scope, key)
);
````

#### 2. 把幂等表登记为 SQLAlchemy Core Table

追加到 app/models.py，复用已有 Table、Column、Base、String、DateTime、func。JSONB 和 Integer 使用下方导入。SQLAlchemy Core 与 ORM 可以共用一个事务。

文件：app/models.py

````python
from sqlalchemy import Integer
from sqlalchemy.dialects.postgresql import JSONB

idempotency_records = Table("idempotency_records", Base.metadata,
    Column("scope", String(120), primary_key=True),
    Column("key", String(100), primary_key=True),
    Column("request_hash", String(64), nullable=False),
    Column("status_code", Integer),
    Column("response_body", JSONB),
    Column("created_at", DateTime(timezone=True), nullable=False, server_default=func.now()),
)
````

#### 3. 新增幂等服务函数

追加到 app/services.py。它要求外层尚未完成业务写入但已经 begin；不自行提交。on_conflict_do_nothing 的返回值标识是否占位成功；同键不同请求体返回 409。创建、占位和结果保存三步都是同一事务。

文件：app/services.py

````python
import hashlib
import json
from sqlalchemy import select, update
from sqlalchemy.dialects.postgresql import insert
from app.models import idempotency_records

async def create_idempotent(db: AsyncSession, body: NoteCreate, key: str, scope: str):
    payload = json.dumps(body.model_dump(mode="json"), sort_keys=True, separators=(",", ":"), ensure_ascii=False)
    digest = hashlib.sha256(payload.encode()).hexdigest()
    claimed = await db.scalar(insert(idempotency_records).values(
        scope=scope, key=key, request_hash=digest
    ).on_conflict_do_nothing(index_elements=["scope", "key"]).returning(idempotency_records.c.key))
    if claimed is None:
        stored = (await db.execute(select(idempotency_records).where(
            idempotency_records.c.scope == scope, idempotency_records.c.key == key))).mappings().one()
        if stored["request_hash"] != digest:
            raise HTTPException(409, "idempotency key reused with different payload")
        if stored["status_code"] is None:
            raise HTTPException(503, "idempotency result not ready")
        return stored["status_code"], stored["response_body"]
    note = Note(**body.model_dump())
    db.add(note)
    await db.flush()
    await db.refresh(note)
    result = NoteOut.model_validate(note).model_dump(mode="json")
    await db.execute(update(idempotency_records).where(
        idempotency_records.c.scope == scope, idempotency_records.c.key == key
    ).values(status_code=201, response_body=result))
    return 201, result
````

#### 4. 新增 HTTP 入口

追加到 app/routes.py。本机单主体 scope 必须在接入鉴权后替换为服务器可信身份加操作名，不能接受客户端随意传 scope。外层事务提交后才返回保存的响应。

文件：app/routes.py

````python
from fastapi.responses import JSONResponse
from app.services import create_idempotent

@router.post("/idempotent", status_code=201)
async def create_idempotent_note(body: NoteCreate, db: Db,
    idempotency_key: Annotated[str, Header(min_length=1, max_length=100)]):
    async with db.begin():
        status, payload = await create_idempotent(db, body, idempotency_key,
            scope="local-student:POST:/notes/idempotent:v1")
    return JSONResponse(status_code=status, content=payload)
````

#### 5. 并发重试并核对结果

新建 scripts/idempotency_probe.py。使用随机新键避免前次练习污染；两个请求并发提交相同内容，返回同一个 id。再用相同键改内容应返回 409。大量并发测试必须控制在本机练习环境。 此时服务仍使用 notes_lab_migration，所有最终数据核对都连接同一实验库。

文件：scripts/idempotency_probe.py

````python
import asyncio
from uuid import uuid4
import httpx

async def main():
    headers = {"Idempotency-Key": str(uuid4())}
    async with httpx.AsyncClient(base_url="http://127.0.0.1:8000", timeout=5) as client:
        async def send(title):
            return await client.post("/notes/idempotent", json={"title": title}, headers=headers)
        a, b = await asyncio.gather(send("retry"), send("retry"))
        assert a.status_code == b.status_code == 201, (a.text, b.text)
        assert a.json() == b.json()
        assert (await send("different")).status_code == 409

asyncio.run(main())
````

### 验收结果

- 两个相同键相同内容请求返回相同状态码和响应体，notes 只新增一条。
- 同键不同内容返回 409，不改变最初结果。
- 业务事务失败后不会留下 status_code=null 的已提交占位记录。

### 生产约束与故障演练

移动客户端在创建接口超时后自动重试，不能重复创建笔记。

#### 实现决策

- 同一逻辑操作重用键，键绑定主体与操作。
- 幂等结果与业务数据在同一数据库事务提交。

#### 隔离环境演练

隔离环境并发发送相同键，丢弃首个响应后再重试，最后核对记录与响应一致。

#### 恢复验收

- 每个键最多创建一次本服务记录。
- 文档明确保留期限、主体作用域和不同 payload 冲突行为。

### 常见错误

#### 请求哈希把 JSON 字段顺序当成内容不同

原因：对原始 JSON 字符串直接计算摘要。

处理：先 Pydantic 规范化，再固定字段顺序和编码；版本化摘要规则，升级时维持旧键语义。

#### 清理幂等表后客户端晚到重试又创建一条

原因：保留期限短于客户端重试窗口，或没有对重复业务本身加约束。

处理：公布键保留期并记录过期行为；订单号等业务唯一性还需独立唯一约束。

### 扩展练习

如何模拟数据库已提交但客户端没收到响应？

提示：在本机先成功调用后故意丢弃响应，再以原键重发即可验证服务器语义。

参考答案：

保存 key，第一次请求完成后不使用响应，第二次用相同 key/body 发起请求。核对只新增一行且得到相同 id；若做真正断连注入，要同时覆盖 commit 前后两个时点，不能从客户端断开推断数据库一定回滚。

### 技术文档

- [PostgreSQL 16：INSERT ON CONFLICT](https://www.postgresql.org/docs/16/sql-insert.html)
- [SQLAlchemy：PostgreSQL UPSERT](https://docs.sqlalchemy.org/en/20/dialects/postgresql.html#insert-on-conflict-upsert)
- [PostgreSQL 16：READ COMMITTED](https://www.postgresql.org/docs/16/transaction-iso.html#XACT-READ-COMMITTED)

---

## 11 事务 Outbox 与跨服务一致性

将笔记变更和待发事件原子落库，再由独立发布器领取事件并在提交后发送。

### 本节目标

- 解释数据库提交和消息发送之间的双写缺口
- 实现带租约的 outbox 领取和确认
- 让接收方以 event_id 去重并承受重复投递

### 必要概念

#### 本地事务不覆盖外部服务

先提交笔记再发消息，发送失败会漏事件；先发消息再提交，事务失败会留下不存在的业务事件。Outbox 把业务记录与待发事件写在同一数据库事务内，发布器稍后重试外部发送，接受最终一致性的延迟。

#### 至少一次与租约

发布器领取一批事件，写 lease token 与 locked_until 后立即提交，再进行网络调用。崩溃后租约到期可重新领取；发送成功但标记失败会导致重复投递。系统保证的是可重试的至少一次交付流程，接收方必须去重；不是端到端恰好一次。

#### 领取、重试与顺序

SKIP LOCKED 适合队列式争抢，不能用于普通业务一致列表。按 id 领取只提供尽量有序，不保证同一笔记多个事件的严格顺序。若业务需要顺序，事件携带 aggregate_id/version，接收方按版本处理并暂存缺口。

### 动手实现

#### 1. 迁移 outbox 与本机接收表

生成 Alembic revision 并通过 op.execute 执行以下 SQL。outbox 的 payload 最小化，不复制密钥或完整敏感提示词。表定义需同步加入 Base.metadata；以下后续步骤会提供对应 Core Table 映射，避免 autogenerate 误判删除。

文件：终端 / 当前课时

````sql
CREATE TABLE outbox_events (
    id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    event_id uuid NOT NULL UNIQUE,
    aggregate_id bigint NOT NULL,
    event_type varchar(80) NOT NULL,
    payload jsonb NOT NULL,
    created_at timestamptz NOT NULL DEFAULT now(),
    available_at timestamptz NOT NULL DEFAULT now(),
    locked_until timestamptz,
    lock_token uuid,
    attempts integer NOT NULL DEFAULT 0,
    delivered_at timestamptz,
    last_error varchar(120)
);
CREATE INDEX ix_outbox_pending ON outbox_events (available_at, id) WHERE delivered_at IS NULL;
CREATE TABLE received_events (
    event_id uuid PRIMARY KEY,
    received_at timestamptz NOT NULL DEFAULT now()
);
````

#### 2. 把 Outbox 结构同步到 metadata

追加到 app/models.py，复用已有 Table、Column、Index、Base、BigInteger、Identity、Integer、String、DateTime、func、text 与 JSONB。只声明映射，不在应用启动时建表；迁移创建实际表。

文件：app/models.py

````python
from sqlalchemy.dialects.postgresql import UUID

outbox_events = Table("outbox_events", Base.metadata,
    Column("id", BigInteger, Identity(), primary_key=True),
    Column("event_id", UUID(as_uuid=True), nullable=False, unique=True),
    Column("aggregate_id", BigInteger, nullable=False),
    Column("event_type", String(80), nullable=False),
    Column("payload", JSONB, nullable=False),
    Column("created_at", DateTime(timezone=True), nullable=False, server_default=func.now()),
    Column("available_at", DateTime(timezone=True), nullable=False, server_default=func.now()),
    Column("locked_until", DateTime(timezone=True)),
    Column("lock_token", UUID(as_uuid=True)),
    Column("attempts", Integer, nullable=False, server_default=text("0")),
    Column("delivered_at", DateTime(timezone=True)),
    Column("last_error", String(120)),
)
Index("ix_outbox_pending", outbox_events.c.available_at, outbox_events.c.id,
      postgresql_where=outbox_events.c.delivered_at.is_(None))
received_events = Table("received_events", Base.metadata,
    Column("event_id", UUID(as_uuid=True), primary_key=True),
    Column("received_at", DateTime(timezone=True), nullable=False, server_default=func.now()),
)
````

#### 3. 在创建事务内追加事件

追加到 app/services.py。把 await add_created_event(db, note) 放入第 10 课 create_idempotent 的新建成功分支：在 flush/refresh 后、写入幂等结果前。重放分支不再生成事件。函数只写本地表，不做 HTTP。

文件：app/services.py

````python
from uuid import uuid4

async def add_created_event(db: AsyncSession, note: Note):
    await db.execute(text("""
        INSERT INTO outbox_events (event_id, aggregate_id, event_type, payload)
        VALUES (:event_id, :note_id, 'note.created', CAST(:payload AS jsonb))
    """), {"event_id": uuid4(), "note_id": note.id,
             "payload": json.dumps({"note_id": note.id, "version": note.version})})
````

#### 4. 新建领取与确认函数

新建 app/outbox.py。批量上限 20、租约 30 秒是练习参数；本课发布器每次只领 1 条以保证 3 秒请求预算小于租约。UPDATE 确认同时匹配 token，旧发布器无法覆盖新持有者。超过重试预算的事件停止自动领取并进入人工排查清单。

文件：app/outbox.py

````python
from uuid import uuid4
from sqlalchemy import text
from app.db import SessionFactory

async def claim(limit=1):
    if not 1 <= limit <= 20:
        raise ValueError("limit must be between 1 and 20")
    token = uuid4()
    async with SessionFactory.begin() as db:
        rows = (await db.execute(text("""
            WITH picked AS (
                SELECT id FROM outbox_events
                WHERE delivered_at IS NULL AND available_at <= now()
                  AND attempts < 8
                  AND (locked_until IS NULL OR locked_until < now())
                ORDER BY id FOR UPDATE SKIP LOCKED LIMIT :limit
            )
            UPDATE outbox_events e SET lock_token=:token,
                locked_until=now()+interval '30 seconds', attempts=attempts+1
            FROM picked WHERE e.id=picked.id
            RETURNING e.id, e.event_id, e.event_type, e.payload, e.attempts
        """), {"token": token, "limit": limit})).mappings().all()
    return token, [dict(x) for x in rows]

async def finish(event_id, token, success):
    async with SessionFactory.begin() as db:
        await db.execute(text("""
            UPDATE outbox_events
            SET delivered_at=CASE WHEN :success THEN now() ELSE NULL END,
                available_at=CASE WHEN :success THEN available_at ELSE
                    now()+make_interval(secs => least(300, power(2, attempts)::int)) END,
                locked_until=NULL, lock_token=NULL,
                last_error=CASE WHEN :success THEN NULL ELSE 'delivery_failed' END
            WHERE id=:id AND lock_token=:token AND delivered_at IS NULL
        """), {"id": event_id, "token": token, "success": success})
````

#### 5. 添加本机去重接收端点

追加到 app/main.py，仅用于本机教学模拟第二个服务。真实接收服务拥有独立数据库；去重插入与自身业务副作用必须同一事务。示例的副作用仅是记录接收，不伪称完成搜索索引等外部功能。

文件：app/main.py

````python
from uuid import UUID
from pydantic import BaseModel, Field

class IncomingEvent(BaseModel):
    event_id: UUID
    event_type: str = Field(max_length=80)
    payload: dict

@app.post("/internal/events")
async def receive_event(body: IncomingEvent, db: Db):
    async with db.begin():
        new_id = await db.scalar(text("""
            INSERT INTO received_events (event_id) VALUES (:event_id)
            ON CONFLICT (event_id) DO NOTHING RETURNING event_id
        """), {"event_id": body.event_id})
        # 真实接收方在这里执行自己的数据库副作用，和去重记录一起提交。
    return {"accepted": True, "duplicate": new_id is None}
````

#### 6. 创建单次发布脚本

新建 scripts/publish_once.py，运行 python -m scripts.publish_once。它每次处理一条，HTTP 在 claim 事务提交之后执行；没有在持锁时等待网络。成功接收但 finish 前崩溃会被再次投递，这是下一步演练目标。

文件：scripts/publish_once.py

````python
import asyncio
import sys
import httpx
from app.db import engine
from app.outbox import claim, finish

async def main():
    try:
        token, events = await claim(limit=1)
        async with httpx.AsyncClient(timeout=3) as client:
            for event in events:
                success = False
                try:
                    response = await client.post("http://127.0.0.1:8000/internal/events", json={
                        "event_id": str(event["event_id"]), "event_type": event["event_type"],
                        "payload": event["payload"]})
                    response.raise_for_status()
                    success = True
                except httpx.HTTPError:
                    pass
                await finish(event["id"], token, success)
    finally:
        await engine.dispose()

if __name__ == "__main__":
    if sys.platform == "win32":
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    asyncio.run(main())
````

#### 7. 演练发送成功、确认丢失

仅在本机将 publish_once 中 await finish 临时注释一次，创建一条新幂等笔记后发布，确认 received_events 已有记录但 outbox 未 delivered。等待租约到期后恢复 finish 再发；接收方 duplicate=true，received_events 仍只有一行。用下面查询监控积压，不将 attempts>=8 的事件静默删除。

文件：终端 / 当前课时

````sql
SELECT count(*) AS pending,
       min(created_at) AS oldest_pending,
       count(*) FILTER (WHERE attempts >= 8) AS needs_attention
FROM outbox_events WHERE delivered_at IS NULL;
````

### 验收结果

- 幂等创建只产生一条 notes 和一条对应 outbox event，重试不会再发新事件。
- claim 结束后才调用 HTTP，数据库没有等待外部服务的长事务。
- 重复投递被接收方去重，失败事件保留次数、下一次可尝试时间与排查状态。

### 生产约束与故障演练

创建笔记后要通知索引服务，但索引服务短暂不可用。

#### 实现决策

- 业务事务内只记录事件；外部传输在短租约领取后执行。
- 监控积压年龄与耗尽重试，接收方按事件 id 去重。

#### 隔离环境演练

模拟成功接收后未确认，再过期重领，核对接收记录只有一条且 outbox 最终 delivered。

#### 恢复验收

- 不丢失已提交业务的待发事件。
- 承认并验证至少一次投递与最终一致性边界。

### 常见错误

#### 消息发送成功后进程崩溃，恢复后又发送一次

原因：外部成功与本地 delivered 标记不能在一个普通数据库事务中原子提交。

处理：接受重复并按 event_id 去重，去重记录和接收方业务写入共用事务。

#### 批量领取 100 条后大量租约过期

原因：顺序发送总耗时超过租约，其他发布器重新领取。

处理：缩小批次、限制单次请求时间；扩大并行时计算整个批次最坏耗时，必要时做可验证的续租。

### 扩展练习

如果事件用于邮件发送，接收数据库去重是否保证邮件只发送一次？

提示：数据库事务不能原子覆盖外部邮件供应商。

参考答案：

不能。邮件供应商若支持幂等键，应传 event_id；否则仍存在发送成功但记录失败的窗口。要按业务允许的重复程度设计去重、状态查询或人工补偿，不能称为恰好一次。

### 技术文档

- [PostgreSQL 16：SKIP LOCKED](https://www.postgresql.org/docs/16/sql-select.html)
- [AWS：事务 Outbox 模式](https://docs.aws.amazon.com/prescriptive-guidance/latest/cloud-design-patterns/transactional-outbox.html)
- [PostgreSQL 16：唯一约束](https://www.postgresql.org/docs/16/ddl-constraints.html#DDL-CONSTRAINTS-UNIQUE-CONSTRAINTS)

---

## 12 真实 PostgreSQL 隔离测试与故障验收

用专用 PostgreSQL 测试库验证提交、并发、迁移和故障恢复，避免 SQLite 掩盖生产差异。

### 本节目标

- 区分回滚隔离测试与真实提交并发测试
- 使用依赖覆盖连接真实 PostgreSQL
- 建立错误注入与恢复验收清单

### 必要概念

#### 测试数据库必须同类

SQLite 的类型、JSON、锁、约束和并发行为与 PostgreSQL 不同。本模块针对 PostgreSQL 16 测试，测试库独立为 notes_lab_test，使用专用凭据或限定权限。测试代码启动时检查数据库名，不从默认生产 URL 继承目标。

#### 两类测试不能混用

单请求 CRUD 可以把 Session 绑定外层连接事务，通过 savepoint 让内部 commit 后最终回滚清理。并发与提交失败测试必须使用多个真实连接和真正 commit；外层回滚会隐藏可延迟约束的提交时机，也无法验证其他连接可见性。

#### 故障测试的验收对象

不只断言错误码，还要查数据库最终状态、连接是否释放、下一次正常请求是否恢复。连接中断可能造成提交结果未知，验收需用幂等键重查；不要凭一次网络错误断言写入不存在。

### 动手实现

#### 1. 准备独立测试库与迁移

先用数据库管理员在本机创建 notes_lab_test 并授予练习 app_user 对该库迁移权限。仅当前测试终端设置以下变量；把 CHANGE_ME 替换为本机测试密码。运行迁移前通过 psql 核对数据库名称。不要对共享库执行测试清理。

文件：终端 / 当前课时

````powershell
$env:TEST_DATABASE_URL="postgresql+psycopg://app_user:CHANGE_ME@127.0.0.1:5433/notes_lab_test"
$env:DATABASE_URL=$env:TEST_DATABASE_URL
python -m alembic upgrade head
python -m alembic check
````

#### 2. 新建 pytest 配置与隔离 fixture

新建 pytest.ini 内容 [pytest] 下一行 asyncio_mode = auto。再新建 tests/conftest.py。在导入 app 前覆盖 DATABASE_URL；fixture 检查测试库名并使用 NullPool，避免不同事件循环复用连接。外层事务提供回滚隔离，但不能用此 fixture 验证并发或 commit 时延迟约束。

文件：tests/conftest.py

````python
import os
import sys
import asyncio
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy.engine import make_url
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.pool import NullPool

if sys.platform == "win32":
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

test_url = os.environ["TEST_DATABASE_URL"]
if make_url(test_url).database != "notes_lab_test":
    raise RuntimeError("tests require notes_lab_test")
os.environ["DATABASE_URL"] = test_url

from app.main import app
from app.db import get_session

@pytest_asyncio.fixture
async def client():
    test_engine = create_async_engine(test_url, poolclass=NullPool)
    try:
        async with test_engine.connect() as conn:
            outer = await conn.begin()
            async def override():
                async with AsyncSession(bind=conn, expire_on_commit=False,
                    join_transaction_mode="create_savepoint") as db:
                    yield db
            app.dependency_overrides[get_session] = override
            try:
                async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
                    yield c
            finally:
                app.dependency_overrides.clear()
                await outer.rollback()
    finally:
        await test_engine.dispose()
````

#### 3. 测试最终版本 CRUD 协议

新建 tests/test_notes.py。版本字段与第 9 课保持一致，先验证成功更新再验证旧版本冲突。ASGITransport 不自动运行 lifespan；本 fixture 不依赖启动建表，迁移已在前一步完成。需要启动资源的应用应显式使用 lifespan 管理器。

文件：tests/test_notes.py

````python
async def test_create_patch_conflict_delete(client):
    response = await client.post("/notes", json={"title": "isolated"})
    assert response.status_code == 201, response.text
    original = response.json()
    path = f"/notes/{original['id']}"
    saved = await client.patch(path, json={"title": "updated", "version": original["version"]})
    assert saved.status_code == 200, saved.text
    updated = saved.json()
    assert updated["version"] == original["version"] + 1
    conflict = await client.patch(path, json={"title": "stale", "version": original["version"]})
    assert conflict.status_code == 409
    assert (await client.get(path)).json()["title"] == "updated"
    deleted = await client.delete(path, params={"version": updated["version"]})
    assert deleted.status_code == 204 and deleted.content == b""
    assert (await client.get(path)).status_code == 404

async def test_input_boundaries(client):
    assert (await client.post("/notes", json={"title": "x" * 121})).status_code == 422
    assert (await client.get("/notes", params={"limit": 101})).status_code == 422
    assert (await client.get("/notes", params={"cursor": "broken!"})).status_code == 422
````

#### 4. 增加真正提交的并发 fixture

追加到 tests/conftest.py。该 fixture 使用不同请求各自的真实连接，不使用外层回滚；只能在专用测试库运行。每个测试用随机键隔离，并在测试 finally 清理自己产生的记录。不要并发运行会修改 app.dependency_overrides 的测试。

文件：tests/conftest.py

````python
from sqlalchemy.ext.asyncio import async_sessionmaker

@pytest_asyncio.fixture
async def committed_client():
    test_engine = create_async_engine(test_url, poolclass=NullPool)
    factory = async_sessionmaker(test_engine, expire_on_commit=False)
    async def override():
        async with factory() as db:
            yield db
    app.dependency_overrides[get_session] = override
    try:
        async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
            yield c, factory
    finally:
        app.dependency_overrides.clear()
        await test_engine.dispose()
````

#### 5. 编写并发幂等测试并精确清理

新建 tests/test_concurrency.py。只按本测试随机 key 查到的 note_id 清理关联 outbox 和业务记录；不 TRUNCATE 全库。若测试进程被强制结束残留数据，应通过专用测试库重建或运行有审计的测试清理作业。

文件：tests/test_concurrency.py

````python
import asyncio
from uuid import uuid4
from sqlalchemy import text

async def test_idempotency_concurrent(committed_client):
    client, factory = committed_client
    key = str(uuid4())
    scope = "local-student:POST:/notes/idempotent:v1"
    try:
        async def send():
            return await client.post("/notes/idempotent", json={"title": "parallel"},
                headers={"Idempotency-Key": key})
        a, b = await asyncio.gather(send(), send())
        assert a.status_code == b.status_code == 201, (a.text, b.text)
        assert a.json() == b.json()
        async with factory.begin() as db:
            count = await db.scalar(text("SELECT count(*) FROM outbox_events WHERE aggregate_id=:id"), {"id": a.json()["id"]})
            assert count == 1
    finally:
        async with factory.begin() as db:
            payload = await db.scalar(text("SELECT response_body FROM idempotency_records WHERE scope=:scope AND key=:key"), {"scope": scope, "key": key})
            if payload:
                note_id = payload["id"]
                await db.execute(text("DELETE FROM outbox_events WHERE aggregate_id=:id"), {"id": note_id})
                await db.execute(text("DELETE FROM notes WHERE id=:id"), {"id": note_id})
            await db.execute(text("DELETE FROM idempotency_records WHERE scope=:scope AND key=:key"), {"scope": scope, "key": key})
````

#### 6. 运行测试与故障矩阵

执行 python -m pytest -q。再在独立本机服务对第 4、6、11 课演练做真实提交验收，并记录错误码、数据库最终状态和恢复探针。以下矩阵是需要实际执行的验收任务，不代表课件已运行并通过；不要在未执行时把“预期”写成“结果”。

文件：终端 / 当前课时

````text
场景 | 注入方式 | 预期状态 | 恢复验证
提交约束失败 | 重复 audited request-key | 409；notes/audits 同时回滚 | 新键创建成功
池耗尽 | 独立 pool_drill，池上限 1 | 一任务超时；无连接泄漏 | SELECT 1 成功
陈旧编辑 | 两客户端相同 version | 一个成功、一个 409 | GET 与成功者一致
同键重试 | 两真实连接同时创建 | 相同响应；一条笔记一条事件 | 再重试同结果
发布器崩溃 | 接收后跳过 finish | 事件到期重领；接收方去重 | 最终 delivered
迁移锁阻塞 | 专用库长事务占锁 | 超时失败；revision 未推进 | 释放锁后 upgrade 成功
````

### 验收结果

- pytest 输出真实执行结果；测试库名称防护拒绝 notes_lab 或其他库。
- 回滚 fixture 不留下 CRUD 数据，并发 fixture 使用真实提交且只清理自身记录。
- 失败演练检查了数据库最终状态和恢复能力，没有用 SQLite 替代 PostgreSQL 特性。

### 生产约束与故障演练

数据库连接、迁移和并发逻辑要进入持续集成，而不依赖开发者手工点 /docs。

#### 实现决策

- 同类数据库验证实际语义，测试环境显式隔离。
- 单请求回滚测试与真实提交集成测试分开，验收包括故障恢复。

#### 隔离环境演练

完整运行本课矩阵，仅使用本机专用测试库，保存预期、实际、证据和恢复结果四列记录。

#### 恢复验收

- CRUD、输入边界、并发与幂等有真实自动化断言。
- 迁移与失败路径有执行证据，未运行的项目明确标记未验证。

### 常见错误

#### 回滚隔离测试通过，但线上 commit 延迟约束失败

原因：测试提交的是 SAVEPOINT，不是最外层数据库 COMMIT。

处理：延迟约束、并发可见性与网络提交异常使用 committed_client 或独立进程专用测试库。

#### 测试发出请求后后台发布器读不到数据

原因：测试业务事务仍被外层回滚事务包裹，其他连接看不到未提交数据。

处理：跨连接集成测试使用真正提交，并按测试生成的唯一业务键清理。

### 扩展练习

如何防止测试运行时意外连接到线上？

提示：不要只依赖“我记得设置了环境变量”。

参考答案：

要求 TEST_DATABASE_URL 必填并校验数据库名，使用专用网络与最小权限测试凭据，CI 不注入生产凭据；日志输出脱敏 host/database 供核对。数据库名检查只是第一道防线，不能替代网络与权限隔离。

### 技术文档

- [SQLAlchemy：外部事务与测试](https://docs.sqlalchemy.org/en/20/orm/session_transaction.html#joining-a-session-into-an-external-transaction-such-as-for-test-suites)
- [FastAPI：异步测试](https://fastapi.tiangolo.com/advanced/async-tests/)
- [HTTPX：ASGI Transport](https://www.python-httpx.org/advanced/transports/#asgi-transport)
- [pytest-asyncio：配置](https://pytest-asyncio.readthedocs.io/en/stable/reference/configuration.html)