api-database-lessons.md 97 KB
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 供核对。数据库名检查只是第一道防线，不能替代网络与权限隔离。