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 并激活对应环境。这里固定大版本范围，完成验收后生成锁定文件。

文件：终端 / 当前课时

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

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

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

.env
.venv/
__pycache__/
.pytest_cache/

5. 验证目标库与现有表

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

文件：终端 / 当前课时

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 应触发校验失败。

技术文档

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

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

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

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 才表示事务已开启但没有继续工作。请求数不应使连接数持续增长。

文件：终端 / 当前课时

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

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。

技术文档

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

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 纳入版本迁移。

文件：终端 / 当前课时

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

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

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

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 校验。

技术文档

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 数值而不设外键，使笔记删除后审计仍保留。

文件：终端 / 当前课时

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

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

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

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 的路径。

文件：终端 / 当前课时

输入校验 → 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 查询找不到本次标题。移除故障注入后再次请求成功；不要保留公开故障接口。

技术文档

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 执行。

文件：终端 / 当前课时

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

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

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

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 更易排错。

文件：终端 / 当前课时

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。

技术文档

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。

文件：终端 / 当前课时

数据库上限 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

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

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

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；不要对业务库执行这些命令。

文件：终端 / 当前课时

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 约减半，同时验证排队时间与吞吐。连接配额只是上限，不能保证响应时间；需要限流和负载测量，而非只修改参数。

技术文档

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。

文件：终端 / 当前课时

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

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

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 课继续使用新库，避免把教学迁移应用到已有表。

文件：终端 / 当前课时

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/_expand_note_version.py

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；验证与锁预算在预发布副本测量。

文件：终端 / 当前课时

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。

文件：终端 / 当前课时

# 在当前连接的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 客户端。启用版本校验前仍需完成所有写路径升级，否则旧应用不会递增版本。

技术文档

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/_notes_cursor_index.py

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

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

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/_notes_title_check.py

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

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 请求有自己的事务。

参考答案：

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

技术文档

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

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

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

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

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

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。

技术文档

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 误判为多余表。

文件：终端 / 当前课时

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

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

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

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

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 前后两个时点，不能从客户端断开推断数据库一定回滚。

技术文档

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 误判删除。

文件：终端 / 当前课时

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

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

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

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

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

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 的事件静默删除。

文件：终端 / 当前课时

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；否则仍存在发送成功但记录失败的窗口。要按业务允许的重复程度设计去重、状态查询或人工补偿，不能称为恰好一次。

技术文档

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 核对数据库名称。不要对共享库执行测试清理。

文件：终端 / 当前课时

$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

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

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

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

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 课演练做真实提交验收，并记录错误码、数据库最终状态和恢复探针。以下矩阵是需要实际执行的验收任务，不代表课件已运行并通过；不要在未执行时把“预期”写成“结果”。

文件：终端 / 当前课时

场景 | 注入方式 | 预期状态 | 恢复验证
提交约束失败 | 重复 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 供核对。数据库名检查只是第一道防线，不能替代网络与权限隔离。

api-database-lessons.md 97 KB تاريخچه خام

API + 数据库

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

本节目标

必要概念

项目边界

配置与代码

生产账号分工

动手实现

1. 新建项目并安装依赖

2. 新建环境模板

3. 新建配置模块

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

5. 验证目标库与现有表

验收结果

生产约束与故障演练

实现决策

隔离环境演练

恢复验收

常见错误

导入配置提示 Field required

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

扩展练习

技术文档

02 异步 Session 的生命周期

本节目标

必要概念

选择一条完整的执行路径

对象的生命周期

并发安全

动手实现

1. 新建数据库模块

2. 新建应用入口

3. 启动并检查资源释放

4. 观察借还连接

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

验收结果

生产约束与故障演练

实现决策

隔离环境演练

恢复验收

常见错误

出现 MissingGreenlet 或异步驱动不兼容

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

扩展练习

技术文档

03 ORM CRUD 与输入输出模型

本节目标

必要概念

三种模型回答不同问题

flush 与 commit

PATCH 的省略与空值

动手实现

1. 新建基础 ORM

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

3. 新建输入输出模型

4. 新建 CRUD 路由

5. 通过 HTTP 与 SQL 双向验收

验收结果

生产约束与故障演练

实现决策

隔离环境演练

恢复验收

常见错误

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

PATCH 把未传字段改成 None

扩展练习

技术文档

04 事务边界与失败提交

本节目标

必要概念

事务由业务入口负责

成功返回的顺序

失败之后的 Session

动手实现

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

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

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

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

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

api-database-lessons.md 97 KB

تاريخچه خام