api-lessons.md 95 KB
API 服务开发
术核 AI · 后端工程课件
01 启动第一个 FastAPI 服务
建立 api-lab 项目，让 /health 返回 JSON，并通过交互文档发出第一次请求。
本节目标
- 为项目创建独立的 Python 环境
- 解释 main:app、端口与 --reload 的作用
- 分别在浏览器和 /docs 验证同一个接口
必要概念
服务进程与路由函数
普通脚本运行结束后会退出。API 服务由 Uvicorn 持续监听本机端口，收到 HTTP 请求后调用 FastAPI 中匹配的方法与路径。@app.get('/health') 注册的是 GET /health；函数的返回值会被序列化为 JSON 响应。
虚拟环境决定从哪里导入依赖
同一台电脑可以有多个 Python。虚拟环境把当前项目的解释器入口和第三方包放在 .venv 中。用 python -m pip 安装，能使 pip 与当前 python 指向同一环境；编辑器也应选择 .venv 中的解释器。
开发服务器的几个参数
main:app 表示从当前目录的 main.py 导入 app 对象。127.0.0.1 是本机地址，8000 是默认端口。--reload 会在代码变化后重启开发进程，重启也会清空之后课程中的内存数据。
动手实现
1. 建立目录和虚拟环境
准备 Python 3.11 或更新版本。以下两组命令按系统选择一组。激活只影响当前终端；以后重新打开终端时，要先切换到 api-lab 再激活。
文件：终端 / 当前课时
# Windows PowerShell
mkdir api-lab
cd api-lab
py -3 -m venv .venv
.\.venv\Scripts\Activate.ps1
# macOS / Linux
mkdir api-lab
cd api-lab
python3 -m venv .venv
source .venv/bin/activate
# 激活后检查解释器路径，结果应包含 api-lab/.venv
python -c "import sys; print(sys.executable)"
2. 安装依赖并记录环境
创建 requirements.txt，写入下面三行。然后在激活后的终端运行 python -m pip install -r requirements.txt。本模块使用 Pydantic v2 的 model_dump() 写法。安装成功后运行 python -m pip freeze > requirements-lock.txt，保留当前环境实际使用的版本；另一个环境可从这个锁定文件安装。 如果直接运行下载包，请使用包内已经锁定版本的 requirements.txt。
文件：requirements.txt
fastapi>=0.115,<1
uvicorn[standard]>=0.30,<1
pydantic>=2,<3
3. 编写健康检查接口
在 api-lab 根目录创建 main.py。FastAPI 实例只创建一次。路由装饰器紧接在函数定义上方，返回值使用 Python 字典，JSON 编码由框架完成。
文件：main.py
from fastapi import FastAPI
app = FastAPI(title="API Lab", version="0.1.0")
@app.get("/health", tags=["system"])
def health():
return {"status": "ok"}
4. 启动服务并观察终端
在 main.py 所在目录运行以下命令。终端保持运行状态是正常现象；看到 Application startup complete 后再访问浏览器。服务运行期间用另一个终端执行请求，停止服务使用 Ctrl+C。
文件：终端 / 当前课时
python -m uvicorn main:app --reload
# 默认访问地址：http://127.0.0.1:8000
# 8000 被占用时，可加 --port 8001
5. 通过浏览器和交互文档验收
先打开 http://127.0.0.1:8000/health，确认显示 {"status":"ok"}。再打开 /docs，展开 GET /health → Try it out → Execute，查看状态码、响应头和响应体。最后把 main.py 中的 ok 改为 ready 并保存，观察终端重载日志和新响应，完成后改回 ok。
文件：终端 / 当前课时
GET /health HTTP/1.1
Host: 127.0.0.1:8000
HTTP/1.1 200 OK
Content-Type: application/json
{"status":"ok"}
验收结果
- 终端的 sys.executable 路径包含 api-lab/.venv。
- GET /health 返回 200 和 {"status":"ok"}；访问未定义的 / 返回 404，这不表示服务未启动。
- /docs 中出现 GET /health，点击 Execute 后能看到真实响应。
常见错误
ModuleNotFoundError: No module named 'fastapi'
原因：安装依赖和启动服务使用了不同解释器。
处理：激活 .venv，检查 sys.executable，再执行 python -m pip install -r requirements.txt。
Could not import module 'main'
原因：当前目录没有 main.py，或者文件中存在语法错误。
处理：切换到 api-lab 根目录，确认文件名为 main.py，阅读终端中最早出现的 Python 错误。
PowerShell 阻止 Activate.ps1 运行
原因：当前执行策略限制脚本运行。
处理：无需修改全局策略，可直接使用 ..venv\Scripts\python.exe -m pip install -r requirements.txt，以及 ..venv\Scripts\python.exe -m uvicorn main:app --reload。
扩展练习
添加 GET /version，返回 {"service":"api-lab","version":"0.1.0"}，并在 /docs 中执行一次。
提示：保留已有 app，在另一个函数上方注册新的 GET 路由。
参考答案：
在 main.py 末尾添加： @app.get('/version') def version():
return {'service': 'api-lab', 'version': '0.1.0'}
保存后访问 /version，应返回 200；刷新 /docs 后应同时看到 /health 与 /version。
技术文档
02 读懂一次 HTTP 请求与响应
拆开方法、路径、查询参数、请求头和请求体，亲手发送 GET 与 POST。
本节目标
- 在请求中指出方法、URL、请求头和请求体的位置
- 区分 Content-Type、Accept 与 HTTP 状态码
- 用 /docs 和命令行复现同一条请求
必要概念
请求由几个独立部分组成
以 POST /notes?source=lesson 为例：POST 是方法，/notes 是路径，source=lesson 是查询参数。请求头携带内容类型等元信息，请求体携带笔记数据。下面的 HTTP/1.1 文本用于辨认字段；浏览器也可能使用其他 HTTP 版本，业务含义一致。
Content-Type 描述实际内容
请求头 Content-Type: application/json 表示请求体按 JSON 编码；Accept: application/json 表示客户端期望的响应格式。服务器仍按接口定义产生响应，并非只改 Accept 就会自动得到另一种格式。JSON 键和字符串使用双引号，布尔值使用 true / false。
响应状态与业务字段分开读取
响应先给出 HTTP 状态码，再给出响应头和响应体。200 表示本次 HTTP 操作成功，422 表示请求数据未通过当前接口校验。把 {"code":404} 放在一个 200 响应里，不会把 HTTP 状态变成 404。
动手实现
1. 观察最小 GET 请求
保持第 01 课服务运行。在 /docs 执行 /health，同时查看本页的请求拆解演示。把方法、路径、Host 和响应体分别定位；GET /health 不需要请求体。
文件：终端 / 当前课时
GET /health HTTP/1.1
Host: 127.0.0.1:8000
Accept: application/json
HTTP/1.1 200 OK
Content-Type: application/json
{"status":"ok"}
2. 添加临时的 POST 回显接口
把以下代码追加到 main.py。这个临时接口只读取请求体并返回，不保存笔记，也不宣称创建成功。第 03 课会用带类型的模型替换它。这里先借助 Request 观察 HTTP 数据，异步语法在后续课程集中练习。
文件：main.py
from fastapi import Request
@app.post("/notes")
async def inspect_note(request: Request):
data = await request.json()
return {"received": data}
3. 准备 JSON 文件并发送 POST
在 api-lab 中创建 note.json，使用 UTF-8 保存。macOS / Linux 执行 curl -i -X POST http://127.0.0.1:8000/notes -H "Content-Type: application/json" --data-binary @note.json；Windows PowerShell 使用 curl.exe，并把 @note.json 写成引号包围的 "@note.json"。-i 同时显示响应头。
文件：note.json
{
"title": "first note",
"content": "learn HTTP",
"pinned": false
}
4. 对照原始请求与服务器结果
命令行应返回 200 和 received 对象。改动 note.json 中的 title 后再次发送，确认响应随之变化。在浏览器开发者工具 Network 中展开 /docs 发出的请求，也能查看 Method、Request Headers、Payload 和 Response。
文件：终端 / 当前课时
POST /notes HTTP/1.1
Host: 127.0.0.1:8000
Content-Type: application/json
{"title":"first note","content":"learn HTTP","pinned":false}
HTTP/1.1 200 OK
Content-Type: application/json
{"received":{"title":"first note","content":"learn HTTP","pinned":false}}
验收结果
- 能说明查询参数属于 URL，请求体位于请求头之后，两者不是同一块数据。
- 命令行 POST 返回 200，并在 received 中保留 note.json 的三个字段。
- 浏览器直接打开 /notes 会发出 GET；当前只注册了 POST，因此应返回 405 Method Not Allowed。
常见错误
PowerShell 提示无法识别 -H 参数
原因：使用了旧版 PowerShell 的 curl 别名。
处理：明确执行 curl.exe；或者在 /docs 中操作。
请求体解析失败
原因：把 Python 字典写法当成 JSON，例如使用单引号或 False。
处理：使用 note.json 文件，字符串和字段名用双引号，布尔值写 true / false。
浏览器地址栏打不开 POST 接口
原因：地址栏导航通常发送 GET，没有发送预期的 POST 和 JSON 请求体。
处理：使用 curl、/docs 或能够配置 HTTP 方法的客户端。
扩展练习
把请求体的 title 改成 second note，再发送一次。记录响应的 HTTP 状态码、Content-Type 和 received.title；最后说明 Accept 和 Content-Type 的区别。
提示：使用 -i 查看响应头，不要只看最后一行 JSON。
参考答案：
状态码为 200；响应 Content-Type 为 application/json；received.title 为 second note。请求 Content-Type 描述这次发送的数据格式，Accept 描述客户端希望接收的格式。
技术文档
03 从路径、查询参数和 JSON 中接收数据
为笔记接口声明参数类型，让 FastAPI 完成解析、默认值填充和基础校验。
本节目标
- 将 note_id 放在路径中，将筛选条件放在查询参数中
- 用 Pydantic 模型接收 JSON 请求体
- 通过 422 响应定位错误参数
必要概念
路径标识对象，查询参数控制读取方式
/notes/12 中的 12 表示要操作的笔记；/notes?limit=10&pinned=true 中的参数表示读取条件。参数名出现在路由路径中时，FastAPI 将其识别为路径参数。未出现在路径中的简单类型参数通常来自查询字符串。
类型声明参与运行时解析
HTTP 路径和查询字符串首先都是文本。note_id: int 要求它能解析为整数；limit: int = 10 提供默认值。无法转换的值会在进入路由函数前产生 422 响应，因此函数里不用重复写 int() 和大量类型判断。
请求模型决定必填与默认值
继承 BaseModel 的 NoteCreate 声明 JSON 对象的结构。title 没有默认值，所以必须发送；content 默认空字符串，pinned 默认 false。str | None 表示允许 null，不自动表示可以省略；若要允许省略，还需写默认值，例如 = None。
动手实现
1. 声明创建笔记的输入模型
在 main.py 的路由定义之前添加模型。删除第 02 课的 inspect_note 路由函数及其装饰器，随后用第 04 步的新 POST /notes 替换，避免同一方法和路径出现多个处理函数。
文件：main.py
from pydantic import BaseModel
class NoteCreate(BaseModel):
title: str
content: str = ""
pinned: bool = False
2. 添加路径参数
添加 GET /notes/{note_id}。本节先回显解析后的编号，暂不查询存储。依次请求 /notes/12 与 /notes/abc，比较状态码；整数解析失败时，错误位置会包含 path 和 note_id。
文件：main.py
@app.get("/notes/{note_id}")
def inspect_note_id(note_id: int):
return {"note_id": note_id}
3. 添加查询参数和范围
添加 GET /notes。Annotated 把 Python 类型和 FastAPI 的参数规则写在一起。limit 默认为 10，只允许 1 到 100；pinned 未传时为 None，传 false 时是明确的 False，后续筛选中要区分这两种情况。
文件：main.py
from typing import Annotated
from fastapi import Query
@app.get("/notes")
def inspect_filters(
limit: Annotated[int, Query(ge=1, le=100)] = 10,
pinned: bool | None = None,
):
return {"limit": limit, "pinned": pinned}
4. 用类型模型替换原始请求体
定义新的 POST /notes。在 /docs 中仅发送 {"title":"first note"}，观察缺省字段被补齐。本节仍是参数练习，不写入存储。model_dump() 把 Pydantic 模型转换为 Python 字典。
文件：main.py
@app.post("/notes")
def inspect_note_body(note: NoteCreate):
return note.model_dump()
5. 建立有效值和无效值对照
在 /docs 连续完成以下四次调用，记录状态码与响应中的字段位置。缺少 title 的 POST 应在路由运行前失败。下一课会继续收紧 title 的长度与多余字段规则。
文件：终端 / 当前课时
GET /notes/12
# 200: {"note_id":12}
GET /notes?limit=5&pinned=false
# 200: {"limit":5,"pinned":false}
GET /notes?limit=0
# 422，错误位置包含 ["query","limit"]
POST /notes
Content-Type: application/json
{}
# 422，错误位置包含 ["body","title"]
验收结果
- GET /notes/12 返回数字 12，/notes/abc 返回 422。
- GET /notes 返回 {"limit":10,"pinned":null}；?pinned=false 返回 false。
- POST /notes 只传 title 时补齐 content 和 pinned；缺少 title 返回 422。
常见错误
把 note_id 放进查询字符串，得到列表而不是单条笔记
原因：GET /notes?note_id=12 匹配的是 GET /notes。本课列表路由没有声明 note_id，FastAPI 默认忽略这个额外查询参数，因此返回 200 和列表；它不会匹配 /notes/{note_id}。
处理：读取单条笔记时改为 GET /notes/12。先检查方法和路径，再检查查询参数；只有确实没有匹配路由或资源不存在时才会得到相应的 404。
写了 title: str | None 后，省略 title 仍然报错
原因：允许 null 和允许省略是两个条件。
处理：如果业务允许省略，显式写默认值 = None；本例 title 应保持必填字符串。
修改 POST /notes 后响应还像上一课
原因：旧路由没有移除，重复注册了相同方法和路径。
处理：搜索 @app.post('/notes')，只保留当前一个定义，保存并确认服务完成重载。
扩展练习
为 GET /notes 添加 offset 查询参数，默认 0，只允许非负整数，并把解析结果加进响应。验证 offset=3 与 offset=-1。
提示：参考 limit 的 Annotated[int, Query(...)]，使用 ge=0。
参考答案：
在函数参数中加入 offset: Annotated[int, Query(ge=0)] = 0，并在返回字典中加入 'offset': offset。?offset=3 返回 200 且 offset 为 3；?offset=-1 返回 422，loc 包含 query、offset。
技术文档
04 用 Pydantic 定义输入和输出契约
限制标题长度、拒绝拼错的字段，并通过 response_model 控制客户端能看到的数据。
本节目标
- 用 Field 声明范围，用 extra='forbid' 发现字段拼写错误
- 分开定义创建输入与接口输出模型
- 区分请求校验失败与响应校验失败
必要概念
模型把接口规则放到一起
类型描述数据形状，Field 描述长度或数值约束。本例 title 的长度为 1 到 120。extra='forbid' 会拒绝模型中未声明的字段，能及时发现 pinnd 这样的拼写错误；Pydantic 默认会忽略多余字段，因此需要显式配置。
输入模型与输出模型承担不同职责
创建时客户端只提交 title、content、pinned；id 由服务端产生。NoteOut 在输入字段之外增加 id。response_model 会校验并筛选普通返回数据，未声明的内部字段不会进入响应；这也会出现在 /docs 的响应结构中。
校验成功不代表全部业务规则成立
长度为 1 的字符串可能只是一个空格，类型与长度规则不会自动决定标题是否有意义。输入校验失败通常返回 422；如果路由返回的数据不符合输出模型，例如漏掉 id，则属于服务端代码错误，默认返回 500，并在终端记录具体原因。
动手实现
1. 收紧 NoteCreate 的输入规则
用下面定义替换原来的 NoteCreate，放在路由之前。保留其余路由。随后在 /docs 中分别发送空标题和拼错的 pinnd 字段，确认都得到 422。
文件：main.py
from pydantic import BaseModel, ConfigDict, Field
class NoteCreate(BaseModel):
model_config = ConfigDict(extra="forbid")
title: str = Field(min_length=1, max_length=120)
content: str = ""
pinned: bool = False
2. 增加响应模型
在 NoteCreate 后添加 NoteOut。输出字段单独声明，避免把输入的 extra='forbid' 规则带入本节的输出过滤实验。id 在输出里必填；客户端创建时不能自行指定 id。
文件：main.py
class NoteOut(BaseModel):
id: int
title: str
content: str
pinned: bool
3. 用 response_model 约束 POST 响应
替换 POST /notes 的装饰器和函数。本节的 id=1 仍是演示值，下一阶段才生成真实编号。返回字典故意增加 internal_source，查看 HTTP 响应时该字段应被过滤。
文件：main.py
@app.post("/notes", response_model=NoteOut)
def inspect_note_body(note: NoteCreate):
return {
"id": 1,
**note.model_dump(),
"internal_source": "validation-lesson",
}
4. 分别制造输入与输出错误
先发送下面的空标题请求，确认 422 的错误位置包含 body、title。再临时删除返回字典中的 id，发送正常请求，观察 500 和终端中的响应校验异常。恢复 id 后再继续；不要把服务端输出错误改成客户端的 422。
文件：终端 / 当前课时
{
"title": "",
"content": "title must not be empty",
"pinned": false
}
5. 检查文档中的契约
刷新 /docs。展开 POST /notes 的 Schema，检查 title 的 minLength、maxLength，以及 NoteOut 的 id。发送 {"title":"schema check"} 后应返回下面四个字段。保存一份响应作为后续 CRUD 的验收样本。
文件：终端 / 当前课时
{
"id": 1,
"title": "schema check",
"content": "",
"pinned": false
}
验收结果
- 空 title 或额外 pinnd 字段均返回 422。
- 正常创建演示返回 id、title、content、pinned 四个字段，不包含 internal_source。
- 故意漏掉输出 id 时出现 500；恢复后为 200，当前尚未设置创建成功的 201。
常见错误
空格标题通过了 min_length=1
原因：空格也是字符串字符，长度校验不执行去空格。
处理：根据业务要求加入 strip_whitespace 或字段校验器，去空格后再检查是否为空。
response_model 没有过滤某个自定义响应
原因：直接返回 JSONResponse 等 Response 对象时，框架不会按普通返回值继续执行响应模型转换。
处理：本节返回字典或模型对象，让 FastAPI 使用 response_model 处理。
把模型直接拼到字典时提示类型不匹配
原因：Pydantic 模型对象不等于普通字典。
处理：需要字典时调用 note.model_dump()；不要使用旧教材中的 .dict() 作为本模块标准写法。
扩展练习
拒绝只有空白的标题，并将 ' first note ' 保存为 'first note'。保持 120 字符上限。
提示：为 title 添加 mode='before' 的 field_validator，先去除字符串两端空白，再交给 Field 长度校验。
参考答案：
从 pydantic 导入 field_validator，在 NoteCreate 内添加： @field_validator('title', mode='before') @classmethod def strip_title(cls, value):
return value.strip() if isinstance(value, str) else value
只含空格会变为空字符串并返回 422；' first note ' 经解析后为 'first note'。
技术文档
05 设计笔记资源与 HTTP 行为
确定集合、单条资源、全量替换和删除的接口约定，再用请求验证方法与状态码。
本节目标
- 用资源路径表达笔记集合和单条笔记
- 区分 POST、PUT、PATCH、DELETE 的语义
- 解释 200、201、204、404、405、422 与幂等性
必要概念
先确定资源，再确定操作
本项目使用 /notes 表示笔记集合，/notes/{note_id} 表示一条笔记。读取用 GET，创建用 POST，完整替换用 PUT，删除用 DELETE。方法和路径共同确定接口，因此 GET /notes 与 POST /notes 可以共用路径。
PUT 的完整替换需要明确输入规则
本课程的 PUT 要求 title、content、pinned 全部出现，替换指定笔记可编辑字段，id 保持不变。PATCH 通常表达局部更新，本模块暂不实现。若把创建模型直接用于 PUT，省略 content 会触发默认空字符串，容易被误认为保留旧值。
幂等描述重复请求后的目标状态
重复相同 PUT，目标笔记应保持相同内容；重复 DELETE，目标笔记都处于不存在状态，第一次可以返回 204，第二次可以返回 404。幂等不要求每次响应完全相同。POST 创建通常每次产生新资源，不能假设自动重复发送不会多建一条。
状态码也是接口契约
读取与更新成功使用 200，创建成功使用 201，删除成功且无响应体使用 204。找不到资源使用 404，路径存在但方法不支持使用 405，输入校验失败使用 422。4xx、5xx 响应不能只靠 JSON 业务字段表达，HTTP 状态也必须正确。
动手实现
1. 写下接口清单
以下是第 06 课要实现的目标契约。当前服务还处于参数练习阶段，不应把回显结果视为已保存。先在纸上或请求文件中列出方法、路径、输入和成功状态，后续逐项实现。
文件：notes-contract.http
GET /notes # 查询集合，200，响应为数组
POST /notes # 创建，201，响应包含新 id
GET /notes/{note_id} # 查询单条，200 或 404
PUT /notes/{note_id} # 完整替换，200 或 404
DELETE /notes/{note_id} # 删除，204 或 404
2. 用已有服务区分 404 与 405
在 /docs 执行 GET /health，应返回 200。然后命令行发送 POST /health，应返回 405，并检查 Allow 响应头；再 GET /missing-path，应返回 404。Windows 将下面 curl 改为 curl.exe。这两类错误都与 JSON 字段校验无关。
文件：终端 / 当前课时
curl -i http://127.0.0.1:8000/health
curl -i -X POST http://127.0.0.1:8000/health
curl -i http://127.0.0.1:8000/missing-path
3. 为完整替换定义独立输入
在 main.py 的模型区添加 NoteReplace。三个字段均无默认值，确保调用方明确提交完整内容。在 Python 终端或单独脚本中使用下面示例验证；此时只定义模型，尚未注册 PUT 路由。
文件：main.py
class NoteReplace(BaseModel):
model_config = ConfigDict(extra="forbid")
title: str = Field(min_length=1, max_length=120)
content: str
pinned: bool
# 在独立 Python 终端中测试：
# from main import NoteReplace
# NoteReplace(title="updated", content="", pinned=False) # 通过
# NoteReplace(title="updated") # ValidationError：缺少两项
4. 准备替换与删除的验收请求
把完整请求体保存为 replace.json。下一课先创建一条笔记，记录真实 id，再执行两次同样 PUT 和两次 DELETE。不要写死假定 id 永远为 1；服务重载会影响当前内存。204 响应后不要继续解析 JSON。
文件：replace.json
{
"title": "updated note",
"content": "replace all editable fields",
"pinned": true
}
验收结果
- GET /health 是 200；POST /health 是 405；不存在的路径是 404。
- NoteReplace 缺少 content 或 pinned 时验证失败，空 content 和 false 本身都合法。
- 能解释 DELETE 第二次返回 404 仍然满足幂等，以及 204 为什么不能带 {"ok":true}。
常见错误
PUT 请求只传 title，意外把 content 清空
原因：复用了带默认值的创建模型，未明确全量替换规则。
处理：使用所有可编辑字段均必填的 NoteReplace；如果需要局部更新，应另行设计 PATCH。
客户端对成功 DELETE 调用 response.json() 后报错
原因：204 响应没有响应体。
处理：先判断状态码；遇到 204 直接按成功处理，不执行 JSON 解码。
网络超时后重复 POST，出现两条笔记
原因：客户端没收到响应，并不等于服务器没创建资源。
处理：本地实验先查询列表核对；生产创建接口需根据业务设计幂等键等重复提交控制，不能直接假定 POST 可安全重试。
扩展练习
评审三个接口：GET /delete-note?id=1、PUT /notes/1 只接收 pinned、DELETE /notes/1 返回 200 和 {"deleted":true}。给出本课程约定下的修改方案，并说明第三种设计是否违反 HTTP。
提示：分清 HTTP 协议允许的行为与课程选择的统一接口约定。
参考答案：
第一种改为 DELETE /notes/1，GET 不用于用户请求的数据删除。第二种若只做局部修改应定义 PATCH，或者要求 PUT 传齐 title/content/pinned。第三种返回 200 和结果体在 HTTP 中可以成立；本课程统一选择 204 空体，因此需同时调整状态码和响应内容。
技术文档
06 完成内存版笔记 CRUD
把接口契约变成可运行程序，按创建、查询、替换、删除的顺序完成一次完整操作。
本节目标
- 用字典保存笔记并生成不重复的进程内编号
- 实现带校验与响应模型的笔记 REST 接口
- 按真实创建结果串联后续请求并验证失败路径
必要概念
请求之间需要共享状态
局部变量在函数调用结束后不用于保存下一次请求的数据。这里用模块级 notes 字典，键为整数 id，值为笔记字段。它只存在于当前 Python 进程内；重启、重载都会清空，多个 worker 也不会共享这份字典。
编号与列表位置不是同一个概念
使用单独递增的 next_id 分配编号，删除笔记后不回退。不能用 len(notes)+1 生成 id，否则删掉中间一条后可能复用仍存在的编号。客户端应从创建响应读取 id，再构造单条资源路径。
演示存储的运行条件
本节路由使用 async def，内部没有 await，字典读写在一次事件循环执行段中完成。以单个 Uvicorn worker 运行本地实验；不要在更新共享状态的中间加入等待操作。异步函数与等待的细节将在后面的并发课程中单独演示。
动手实现
1. 重建 main.py 的公共部分
先备份前几课的实验文件，再将 main.py 替换为本步骤代码，并按顺序追加第 2、3、4 步。不要同时保留临时回显路由。全部代码完成后再启动；本节不含 /version 等附加练习接口。
文件：main.py
from typing import Annotated
from fastapi import FastAPI, HTTPException, Query, Response
from pydantic import BaseModel, ConfigDict, Field
app = FastAPI(title="API Lab")
class NoteCreate(BaseModel):
model_config = ConfigDict(extra="forbid")
title: str = Field(min_length=1, max_length=120)
content: str = ""
pinned: bool = False
class NoteReplace(BaseModel):
model_config = ConfigDict(extra="forbid")
title: str = Field(min_length=1, max_length=120)
content: str
pinned: bool
class NoteOut(BaseModel):
id: int
title: str
content: str
pinned: bool
notes: dict[int, dict] = {}
next_id = 1
@app.get("/health", tags=["system"])
async def health():
return {"status": "ok"}
2. 实现列表与创建
追加两个集合路由。pinned is not None 保证 false 也会触发筛选。创建成功后返回 201，并通过 Location 响应头给出新资源路径。next_id 在当前进程递增，保存时写入所有输入字段。
文件：main.py
@app.get("/notes", response_model=list[NoteOut], tags=["notes"])
async def list_notes(
limit: Annotated[int, Query(ge=1, le=100)] = 10,
pinned: bool | None = None,
):
rows = list(notes.values())
if pinned is not None:
rows = [row for row in rows if row["pinned"] == pinned]
return rows[:limit]
@app.post("/notes", response_model=NoteOut, status_code=201, tags=["notes"])
async def create_note(note: NoteCreate, response: Response):
global next_id
note_id = next_id
next_id += 1
row = {"id": note_id, **note.model_dump()}
notes[note_id] = row
response.headers["Location"] = f"/notes/{note_id}"
return row
3. 实现单条读取与全量替换
先定义统一查找函数。不存在时 raise HTTPException 会立即中断本次请求，下一课再统一其错误格式。PUT 创建全新的字段字典并保留路径 id，不把旧 content 或 pinned 合并进去。
文件：main.py
def require_note(note_id: int) -> dict:
if note_id not in notes:
raise HTTPException(status_code=404, detail="笔记不存在")
return notes[note_id]
@app.get("/notes/{note_id}", response_model=NoteOut, tags=["notes"])
async def get_note(note_id: int):
return require_note(note_id)
@app.put("/notes/{note_id}", response_model=NoteOut, tags=["notes"])
async def replace_note(note_id: int, note: NoteReplace):
require_note(note_id)
row = {"id": note_id, **note.model_dump()}
notes[note_id] = row
return row
4. 实现无响应体的删除
删除前先检查是否存在；返回 Response(status_code=204) 明确表示空响应。全部代码写好后运行 python -m uvicorn main:app --reload，打开 /docs，确认一个健康检查加五个笔记操作，共六个路由方法。
文件：main.py
@app.delete("/notes/{note_id}", status_code=204, tags=["notes"])
async def delete_note(note_id: int):
require_note(note_id)
del notes[note_id]
return Response(status_code=204)
5. 按顺序完成一轮 CRUD
使用 /docs：先 GET /notes 确認 []；POST 下列对象并记下实际 id；GET /notes/{id}；PUT 提交 title='updated'、content=''、pinned=true；再 GET 确认内容已替换；DELETE 后再次 GET，确认 404。操作过程中不要修改代码触发重载。最后重新创建一条，检查编号没有因删除而回退。
文件：终端 / 当前课时
{
"title": "first CRUD",
"content": "create, read, replace, delete",
"pinned": false
}
验收结果
- 空列表返回 200 和 []；创建返回 201，响应含整数 id，Location 指向 /notes/{id}。
- PUT 传齐字段时返回 200，id 保持不变；仅传 title 时返回 422。
- 删除返回 204 且响应体为空；再 GET 或 DELETE 相同 id 均返回 404。
- 创建两条并删除其中一条后，再创建不会覆盖另一条；服务重启后列表重新为空。
常见错误
保存代码后刚创建的笔记找不到了
原因：--reload 重启了服务进程，内存字典被重新初始化。
处理：完成代码编辑后重新准备数据；验证一轮 CRUD 时不要中途保存服务端文件。
筛选 pinned=false 返回全部数据
原因：使用 if pinned 判断是否筛选，False 被当成未提供条件。
处理：使用 if pinned is not None，再比较字段的布尔值。
删除一条后，新笔记覆盖了旧笔记
原因：用 len(notes)+1 计算编号导致重复。
处理：独立维护递增 next_id，不因删除回退；本地练习保持单进程运行。
扩展练习
在列表接口加入 offset>=0，先按 pinned 筛选，再返回 rows[offset:offset+limit]。创建三条笔记，第二条和第三条设为 pinned=true，验证 pinned=true&offset=1&limit=1 只返回第三条。
提示：分页应该作用在筛选后的结果上；把切片放在筛选后，limit 和 offset 均添加参数约束。
参考答案：
添加 offset: Annotated[int, Query(ge=0)] = 0；保持 pinned 筛选顺序；把 return rows[:limit] 改为 return rows[offset:offset + limit]。按题目创建顺序，返回数组长度为 1，元素为第三条笔记；offset=-1 返回 422。
技术文档
07 返回可定位、格式一致的错误
用 HTTPException 表达已知失败，统一 404、405 与 422 的响应结构，并保留字段位置。
本节目标
- 在业务失败时抛出 HTTPException
- 为 HTTPException 和 RequestValidationError 注册处理器
- 保持 HTTP 状态码，同时返回稳定的错误结构
必要概念
已知失败应在对应位置结束请求
查不到笔记是可以预期的业务结果，用 raise HTTPException(status_code=404, detail='笔记不存在') 结束当前处理。return HTTPException(...) 只是在返回一个对象，不能表达相同控制流程。未知编程错误需要修复，不能一律包装成 200。
统一外层结构，保留不同错误的细节
本项目约定 {error:{code,message,details}}。HTTP 错误使用 code='http_error'；请求校验使用 code='validation_error'。HTTP 状态码仍保留 404、405、422 等差异。details 中只返回定位所需的字段路径、错误类型和说明，不回显整个请求体。
校验失败可能发生在路由执行之前
参数解析失败会抛出 RequestValidationError，不会进入路由里的 try/except。因此在应用级处理器统一处理。注册 Starlette 的 HTTPException 基类还能覆盖路由不存在和方法不支持等框架错误；处理时保留 Allow 等原有响应头。
动手实现
1. 确认已有业务错误入口
保留第 06 课代码，找到 require_note。确认它使用 raise 而非 return。请求一个不存在的 id，此时默认响应通常为 {"detail":"笔记不存在"}，接下来统一其外层结构。
文件：main.py
def require_note(note_id: int) -> dict:
if note_id not in notes:
raise HTTPException(status_code=404, detail="笔记不存在")
return notes[note_id]
2. 统一 HTTP 错误处理
在 main.py 中添加导入与处理器。这里演示业务 detail 为字符串；若后续传入字典，应单独约定字段映射，不把任意内部对象直接返回。保留 exc.headers，确保例如 405 的 Allow 头仍然存在。
文件：main.py
from fastapi import Request
from fastapi.responses import JSONResponse
from starlette.exceptions import HTTPException as StarletteHTTPException
@app.exception_handler(StarletteHTTPException)
async def handle_http_error(request: Request, exc: StarletteHTTPException):
return JSONResponse(
status_code=exc.status_code,
headers=exc.headers,
content={
"error": {
"code": "http_error",
"message": str(exc.detail),
"details": [],
}
},
)
3. 统一 422 并保留字段位置
添加请求校验处理器。不要直接把 exc.errors() 中的所有对象交给 JSON 序列化；自定义校验器的 ctx 可能包含异常对象。这里只提取 JSON 可序列化的 loc、type、msg，也避免在响应中回显原始输入。
文件：main.py
from fastapi.exceptions import RequestValidationError
@app.exception_handler(RequestValidationError)
async def handle_validation_error(request: Request, exc: RequestValidationError):
details = [
{
"loc": list(error["loc"]),
"type": error["type"],
"msg": error["msg"],
}
for error in exc.errors()
]
return JSONResponse(
status_code=422,
content={
"error": {
"code": "validation_error",
"message": "请求参数不符合接口要求",
"details": details,
}
},
)
4. 用三类失败验证处理器
先 GET /notes/999999，确认 404 且 error.code='http_error'。再 GET /notes/abc 和 POST /notes 的空对象，确认 422；前者定位 path.note_id，后者定位 body.title。最后 POST /health，检查 405、http_error 以及 Allow 头。具体英文 msg 可能随依赖版本变化，客户端不要解析这段文本作业务判断。
文件：终端 / 当前课时
{
"error": {
"code": "validation_error",
"message": "请求参数不符合接口要求",
"details": [
{
"loc": ["body", "title"],
"type": "missing",
"msg": "Field required"
}
]
}
}
验收结果
- 不存在的笔记返回 404、http_error 和“笔记不存在”。
- 非法 note_id 与缺失 title 都返回 422、validation_error，details.loc 能指出不同位置。
- POST /health 仍返回 405，Allow 响应头被保留。
- 成功的笔记响应不被包在 error 字段中；204 删除仍然没有响应体。
常见错误
错误响应有 message，但 HTTP 状态一直是 200
原因：只返回了错误字典，没有设置 JSONResponse 的 status_code。
处理：在异常处理器中显式设置对应状态，HTTP 错误使用 exc.status_code。
注册 HTTPException 处理器后，未知路径仍是默认格式
原因：只处理 FastAPI 的异常子类，没有覆盖框架抛出的 Starlette 基类异常。
处理：导入并为 StarletteHTTPException 注册处理器，路由仍可抛出 FastAPI 的 HTTPException。
自定义标题校验失败后出现 JSON 序列化异常
原因：把 exc.errors() 的 ctx 异常对象直接放进 JSONResponse。
处理：只提取 loc、type、msg 等必要字段，不直接返回任意异常对象或原始请求内容。
扩展练习
对 GET /notes?limit=0、GET /notes/abc、POST /notes 的 {} 逐一记录 status、error.code 和第一条 details.loc，写出客户端应如何定位表单或请求参数错误。
提示：三次都是 422，但错误位置的来源不同：query、path、body。
参考答案：
三次 status 都是 422，code 都是 validation_error。loc 分别为 ['query','limit']、['path','note_id']、['body','title']。客户端先按 status/code 判断错误类别，再用 loc 映射对应输入位置；msg 用于显示，不作为稳定的判断条件。
技术文档
08 复用依赖与拆分路由文件
用 X-API-Key 练习依赖执行顺序，把 /private 放入独立 APIRouter，并通过环境变量配置课堂密钥。
本节目标
- 用 Depends 在路由执行前完成可复用校验
- 从请求头读取 X-API-Key，从环境变量读取预期值
- 用 APIRouter 拆分文件并注册到同一个应用
必要概念
Depends 声明请求处理的前置步骤
FastAPI 收到请求后先解析依赖需要的参数，执行依赖函数，再进入路由。依赖抛出 HTTPException 时，路由不会执行。同一套校验可以挂在多个接口或整个 APIRouter 上，避免每个路由重复读取请求头。
Header 与查询参数的来源不同
x_api_key: str 这种普通参数默认会作为查询参数。使用 Header(alias='X-API-Key') 才从请求头读取。这里给缺失值默认 None，再由依赖统一返回课堂约定的 401；缺少请求头不会先变成默认必填参数的 422。401 响应携带 WWW-Authenticate: APIKey，说明本实验使用的凭据类型。
配置值与身份系统分开理解
DEMO_API_KEY 从启动终端的环境变量读取，网页教程中的 classroom-key 仅用于本地实验。这个共享字符串只演示请求校验，不能区分用户，也没有密钥轮换或权限管理。本节只保护 /private，公开的笔记接口用于本地练习。
APIRouter 负责组织，不会自动启动服务
独立文件创建 router 并注册其路由，主入口再 app.include_router(router)。依赖模块不导入 main，避免循环导入。多个文件最终仍组成同一个 FastAPI 应用，共享第 07 课注册的异常处理器。
动手实现
1. 建立依赖函数文件
在 api-lab 根目录新增 dependencies.py，不修改前面的笔记接口。用 secrets.compare_digest 比较字符串。未配置环境变量返回 503，缺失或错误请求头返回 401，这两种失败帮助区分服务配置和调用参数。
文件：dependencies.py
import os
import secrets
from typing import Annotated
from fastapi import Header, HTTPException
def require_demo_key(
x_api_key: Annotated[str | None, Header(alias="X-API-Key")] = None,
) -> None:
expected = os.getenv("DEMO_API_KEY")
if not expected:
raise HTTPException(status_code=503, detail="服务未配置 DEMO_API_KEY")
if x_api_key is None or not secrets.compare_digest(
x_api_key.encode("utf-8"), expected.encode("utf-8")
):
raise HTTPException(
status_code=401,
detail="X-API-Key 无效",
headers={"WWW-Authenticate": "APIKey"},
)
2. 创建独立的私有路由
新增 routers 目录，其中创建空的 init.py 和下面的 private.py。把依赖挂在 APIRouter 上，当前文件中的接口都会执行相同校验。装饰器使用 @router.get，路径保持 /private。
文件：routers/private.py
from fastapi import APIRouter, Depends
from dependencies import require_demo_key
router = APIRouter(
tags=["private"],
dependencies=[Depends(require_demo_key)],
)
@router.get("/private")
def read_private():
return {"message": "API key accepted"}
3. 在主入口注册 router
在 main.py 导入路由模块，并在 app 创建之后调用 include_router。保留原有 /health、笔记路由和两个异常处理器。刷新 /docs，确认出现 private 分组及 X-API-Key 参数；如果没有出现，先检查是否真正注册。
文件：main.py
from routers import private
# 放在 app = FastAPI(...) 之后
app.include_router(private.router)
# 目录结构
# api-lab/
# main.py
# dependencies.py
# routers/
# __init__.py
# private.py
4. 设置当前终端环境并重新启动
先 Ctrl+C 停止旧服务，在同一个终端按系统选择一组命令。新启动的进程继承该终端环境；在另一窗口设置变量不会修改正在运行的服务。classroom-key 是公开教学校验值。
文件：终端 / 当前课时
# Windows PowerShell
$env:DEMO_API_KEY = "classroom-key"
python -m uvicorn main:app --reload
# macOS / Linux
export DEMO_API_KEY="classroom-key"
python -m uvicorn main:app --reload
5. 验证缺失、错误与正确密钥
在 /docs 连续执行 GET /private：不填写头部、填写 wrong-key、填写 classroom-key。前两次返回 401，第三次返回 200。第 07 课错误处理器会把失败转换为统一 error 对象。再访问 /notes，确认该本地练习接口未要求密钥。
文件：终端 / 当前课时
# macOS / Linux；Windows PowerShell 使用 curl.exe
curl -i http://127.0.0.1:8000/private
curl -i http://127.0.0.1:8000/private -H "X-API-Key: wrong-key"
curl -i http://127.0.0.1:8000/private -H "X-API-Key: classroom-key"
验收结果
- /docs 显示 /private 和 X-API-Key 请求头输入项。
- 配置环境变量后，缺失或错误密钥返回 401 和 http_error；正确密钥返回 200 和 {"message":"API key accepted"}。
- 未设置 DEMO_API_KEY 的新服务进程返回 503，而不会因为请求头与配置都为空而通过。
- 拆分文件后 /health、原有笔记 CRUD 与错误结构仍然可用。
常见错误
已在客户端写 X-API-Key，却显示密钥无效
原因：服务读取的头部来源声明不正确，或者客户端把值放到了查询字符串。
处理：依赖明确使用 Header(alias='X-API-Key')；在 Network 或 curl 输出中核对 Request Headers。
环境变量已设置，接口仍返回 503
原因：变量设置在另一个终端，或者旧服务进程没有重启。
处理：在设置变量的同一个终端停止并重启 Uvicorn。
新增 private.py 后 /private 返回 404
原因：只创建了 router，没有在主 app 上 include_router，或启动时目录不正确。
处理：确认 main.py 注册 private.router，并从 api-lab 根目录启动 main:app。
扩展练习
在 routers/private.py 中新增 GET /private/status，返回 {"enabled":true}。复用 router 上的依赖，分别验证无头部与正确头部。再说明为什么这仍不能判断是哪位学生发出的请求。
提示：只添加 @router.get('/private/status') 的函数，不必复制校验代码。
参考答案：
添加： @router.get('/private/status') def private_status():
return {'enabled': True}
无 X-API-Key 返回 401；classroom-key 返回 200 和 {'enabled':true}。两个接口自动复用 router 依赖。所有人使用同一公开密钥，服务只能判断是否匹配，无法识别具体用户。
技术文档
09 使用 async / await 处理 I/O 等待
用两个并发请求复现阻塞，掌握 async / await 在接口中的实际作用。
本节目标
- 写出不阻塞事件循环的 /slow 接口
- 测量串行与并发请求的耗时差异
- 判断同步 I/O 应改用异步客户端，还是移交线程
必要概念
async def 定义协程函数
调用后得到可等待对象；await 等待结果，并在操作尚未完成时让出执行机会。仅给函数添加 async 不会把内部的阻塞代码变为异步。
等待与计算分开判断
网络响应、磁盘访问属于 I/O；大量 Python 计算属于 CPU 工作。异步可以交错等待，本课不通过 async 加速计算，也不在接口里引入后台任务队列。
FastAPI 的执行位置
FastAPI 调用普通 def 路由时使用线程池；async def 路由在事件循环中执行。你在 async def 内直接调用普通辅助函数时，FastAPI 不会自动把辅助函数移到线程。
动手实现
1. 增加一个可测量的等待接口
开始前，在已激活的虚拟环境中运行 python -m pip install "httpx>=0.27,<1"，第 09 节的客户端脚本会导入它。下载包的 requirements.txt 已包含此依赖。
在 api-lab 的应用中加入路由。delay 限制在 0–5 秒，避免误输很大数值；运行单个服务进程进行本课实验。
文件：main.py
import asyncio
from fastapi import Query
@app.get("/slow")
async def slow(delay: float = Query(default=1, ge=0, le=5)):
await asyncio.sleep(delay)
return {"delay": delay}
2. 同时发出两次请求
新建脚本，保持服务已启动后执行 python concurrent_check.py。先建立一个客户端，再让两次请求同时等待；打印状态码和总耗时。页面演示是浏览器模拟，真实耗时以此脚本为准。
文件：concurrent_check.py
import asyncio
from time import perf_counter
import httpx
async def main():
async with httpx.AsyncClient(base_url="http://127.0.0.1:8000") as client:
started = perf_counter()
results = await asyncio.gather(
client.get("/slow", params={"delay": 1}),
client.get("/slow", params={"delay": 1}),
)
print([r.status_code for r in results])
print(f"elapsed={perf_counter() - started:.2f}s")
asyncio.run(main())
3. 复现阻塞，再恢复
暂时把 await asyncio.sleep(delay) 换成 time.sleep(delay)，并 import time。保存后等服务完成重载，再运行相同脚本。单进程时两次等待通常接近 2 秒；恢复 async sleep 后通常接近 1 秒。另开终端访问 /health，观察阻塞时它也要等待。结果会受机器和调度开销影响，不以毫秒误差判断对错。
文件：终端 / 当前课时
curl.exe "http://127.0.0.1:8000/health"
4. 为无法替换的同步 I/O 选择线程
优先使用库提供的异步方法；若必须在 async def 路由中调用同步 I/O，用 asyncio.to_thread。此处 slow_sync 用 sleep 模拟同步 I/O，实际替换为该库的函数；线程不是本课 CPU 计算加速方案。
文件：终端 / 当前课时
import time
def slow_sync(delay: float):
time.sleep(delay)
return {"delay": delay}
# 放在 async def 路由内：
# result = await asyncio.to_thread(slow_sync, delay)
# return result
验收结果
- 异步版本两个 delay=1 请求均返回 200，总耗时通常接近 1 秒。
- 阻塞版本在单进程下耗时明显增加，恢复后 /health 不再等待 slow 的 sleep 结束。
- /slow?delay=-1 与 /slow?delay=6 都返回 422。
常见错误
写了 async def，其他请求仍卡住
原因：路由内使用 time.sleep、同步 HTTP 客户端或长时间计算，占住事件循环。
处理：定位耗时调用；I/O 改用 await 支持的 API，必要时把同步 I/O 放到线程。
两次 await 请求仍耗时约 2 秒
原因：先 await 第一次，再创建第二次，代码仍是串行等待。
处理：用 asyncio.gather 同时调度两个请求，再比较结果。
出现 coroutine was never awaited
原因：调用异步函数但没有 await，也没有交给任务调度。
处理：在 async 函数内 await 结果；脚本最外层用 asyncio.run(main())。
扩展练习
把脚本改成同时请求 3 个 delay=1 的 /slow，再增加一次 /health，记录总耗时及各状态码。
提示：把第三个 /slow 和 /health 一并放进 gather；不要依次 await。
参考答案：
四个请求应全部返回 200。使用非阻塞等待时，总耗时通常仍接近最长的一次等待，即约 1 秒；若接近 3 秒，先检查服务是否恢复了 asyncio.sleep。
技术文档
10 使用 HTTPX 调用服务与调试 CORS
复用 HTTPX 客户端，处理超时与上游错误，并定位浏览器跨域问题。
本节目标
- 在应用生命周期中创建和关闭 AsyncClient
- 区分超时、连接失败和非成功状态码
- 说明为什么 curl 正常而浏览器请求失败
必要概念
一次请求有两个方向
浏览器到 FastAPI 是入站请求；FastAPI 到另一个服务是出站请求。调用失败时，先确定错误发生在哪一段。
客户端与连接池
AsyncClient 可以复用连接。应用启动时创建，退出时关闭；不要在每次高频请求内重新创建客户端。connect、read、write、pool 是不同阶段的超时，read 通常是等待下一块数据的上限，不是请求总时限。
CORS 由浏览器检查
Origin 由协议、主机和端口组成。http://localhost:5500 与 http://127.0.0.1:5500 是两个来源；curl 和服务端 HTTPX 不执行浏览器的 CORS 检查。CORS 也不替代 API 认证。
动手实现
1. 把客户端交给 lifespan 管理
安装 httpx。把现有 FastAPI() 初始化改为使用 lifespan；若应用已有 lifespan，把下面的客户端管理合并进去，不要再创建第二个 app。上游使用本机已完成的 /slow，让超时实验不依赖公网。
文件：main.py
from contextlib import asynccontextmanager
import httpx
from fastapi import FastAPI
@asynccontextmanager
async def lifespan(app: FastAPI):
timeout = httpx.Timeout(3.0, connect=1.0)
async with httpx.AsyncClient(
base_url="http://127.0.0.1:8000", timeout=timeout
) as client:
app.state.http = client
yield
app = FastAPI(lifespan=lifespan)
2. 把出站失败转换为可判断的响应
增加 /upstream-demo。默认 delay=1 会成功，delay=4 超过 read 超时。这里调用本机接口只是本地实验；正式业务中使用服务方的固定地址，避免让用户任意指定出站 URL。raise_for_status 会把非 2xx 响应转为 HTTPStatusError。
文件：main.py
from fastapi import HTTPException, Query, Request
@app.get("/upstream-demo")
async def upstream_demo(
request: Request,
delay: float = Query(default=1, ge=0, le=5),
):
try:
response = await request.app.state.http.get(
"/slow", params={"delay": delay}
)
response.raise_for_status()
return {"upstream": response.json()}
except httpx.TimeoutException as exc:
raise HTTPException(504, "上游响应超时") from exc
except httpx.HTTPStatusError as exc:
raise HTTPException(502, "上游返回非成功状态") from exc
except httpx.RequestError as exc:
raise HTTPException(502, "无法连接上游服务") from exc
3. 在两个页面来源之间复现 CORS
API 保持运行在 8000 端口。把下面中间件加入现有 app；已有 CORSMiddleware 时修改原配置，不要重复添加。本实验只允许 http://127.0.0.1:5500。如果使用下载的参考项目，先暂时从 allow_origins 中移除 http://localhost:5500，才能复现后面来源不匹配的情况。
文件：main.py
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=["http://127.0.0.1:5500"],
allow_credentials=False,
allow_methods=["GET", "POST", "PUT", "DELETE"],
allow_headers=["Content-Type", "X-API-Key"],
)
4. 建立前端页面并从 5500 访问 API
在 api-lab 中新建 cors-demo.html，粘贴下面内容。另开一个终端，在 api-lab 目录执行 python -m http.server 5500，保留 API 的 8000 终端。打开 http://127.0.0.1:5500/cors-demo.html 并点击按钮，在 Network 中观察 OPTIONS 预检和 POST；JSON 的 Content-Type 会触发本次跨域预检。再通过 http://localhost:5500/cors-demo.html 打开同一文件，对照来源不匹配时的浏览器错误。此时简单的 GET /health 不一定产生 OPTIONS。
文件：cors-demo.html
<!doctype html>
<html lang="zh-CN">
<meta charset="utf-8">
<title>CORS 实验</title>
<button id="send">向 API 创建笔记</button>
<pre id="result"></pre>
<script>
document.querySelector("#send").onclick = async () => {
const output = document.querySelector("#result");
try {
const response = await fetch("http://127.0.0.1:8000/notes", {
method: "POST",
headers: {"Content-Type": "application/json"},
body: JSON.stringify({title: "CORS 实验"}),
});
output.textContent = response.status + "\n" + await response.text();
} catch (error) {
output.textContent = "浏览器未能读取响应；检查 Network 和 Console：" + error.message;
}
};
</script>
</html>
5. 读预检请求和响应头
在浏览器 Network 面板查看 Origin、OPTIONS 和 Access-Control-Allow-Origin。curl.exe 可手动查看预检响应，但它不会替浏览器决定是否允许页面读取响应。把来源改为 localhost:5500 后复现不匹配，再恢复准确来源。 macOS / Linux 把 curl.exe 改为 curl。实验完成后，如需同时使用两个前端域名，再把 localhost:5500 加回白名单。
文件：终端 / 当前课时
curl.exe -i -X OPTIONS "http://127.0.0.1:8000/notes" -H "Origin: http://127.0.0.1:5500" -H "Access-Control-Request-Method: POST" -H "Access-Control-Request-Headers: content-type"
curl.exe -i "http://127.0.0.1:8000/upstream-demo?delay=4"
验收结果
- /upstream-demo?delay=1 返回 200，包含上游的 delay 字段。
- /upstream-demo?delay=4 返回 504；把上游地址临时改为未监听端口可观察 502，随后恢复。
- 允许来源的预检响应包含 Access-Control-Allow-Origin: http://127.0.0.1:5500，浏览器可以读取响应。
常见错误
上游返回 404，接口却继续执行正常处理
原因：HTTP 请求收到响应不等于业务成功，代码没有检查状态。
处理：调用 raise_for_status 并单独处理 HTTPStatusError。
curl 成功，网页报 CORS 错误
原因：浏览器页面来源与 allow_origins 不匹配，或预检方法、请求头未被允许。
处理：逐项核对 Origin、方法和自定义请求头，再看浏览器 Network 中的 OPTIONS 响应。
用 timeout=3 就以为整个请求最多 3 秒
原因：把阶段性超时误认为总请求截止时间，流式响应每次及时发送数据时可能持续更久。
处理：先明确要限制连接、块间等待还是总耗时；本课实验只验证读超时。
扩展练习
让 /upstream-demo 在等待超过 0.2 秒时返回 504，同时保持直接访问 /slow?delay=1 能正常返回。
提示：只覆盖 client.get 的 timeout 参数，保留异步 /slow。
参考答案：
给该次请求设置 timeout=httpx.Timeout(0.2, connect=1.0)。请求 /upstream-demo?delay=1 会在等待上游数据时触发超时；直接请求 /slow?delay=1 仍约 1 秒后返回 200。
技术文档
11 完整响应、流式响应、SSE 与 WebSocket
从通信方向与消息格式选择接口形态，用同一份文本比较接收过程。
本节目标
- 分清 REST 风格、HTTP 响应方式和 WebSocket 协议
- 根据发送方向与数据频率选择实现
- 观察首段到达时间与全部完成时间
必要概念
REST 不规定必须一次返回全部数据
REST 主要讨论资源及 HTTP 操作方式。普通 JSON 响应先得到完整结果再返回；HTTP 流式响应逐段发送同一个响应体。流式发送不自动定义每段的业务含义。
SSE 为文本流定义事件边界
SSE 在 HTTP 响应体中使用 text/event-stream，以空行分隔事件。连接内数据由服务端推向客户端；浏览器可用 EventSource 解析。SSE 是 HTTP 流式响应的一种约定。
WebSocket 保持双向通信
连接建立后，双方可以独立发送消息，不必每条消息新建 HTTP 请求。业务仍需定义消息类型、结束、错误与连接关闭后的行为。
动手实现
1. 先跑一次完整结果接口
用延时代表逐步产生内容，POST /generate 等待全部内容完成后返回一个 JSON 对象。PromptIn 限制长度，本课使用固定文本拼接，便于观察协议而不引入模型服务。
文件：main.py
import asyncio
from pydantic import BaseModel, Field
class PromptIn(BaseModel):
prompt: str = Field(min_length=1, max_length=200)
@app.post("/generate")
async def generate(payload: PromptIn):
parts = []
for text in ["收到：", payload.prompt, "。处理完成。"]:
await asyncio.sleep(0.4)
parts.append(text)
return {"text": "".join(parts)}
2. 记录完整响应的等待
在 /docs 对 POST /generate 输入下面 JSON。结果约在所有步骤完成后出现。区分响应头到达与可用业务内容到达；观察何时收到第一段内容。
文件：终端 / 当前课时
{"prompt":"hello"}
3. 对照三条连接过程
在本页协议演示中依次选择完整 HTTP、SSE、WebSocket：完整响应在完成后显示文本；SSE 逐条显示 token 事件；WebSocket 可在连接保持期间发送新消息。此交互是浏览器模拟；第 12–15 课会用 api-lab 进行真实收发验证。
4. 把场景写成协议选择题
保存笔记通常使用 HTTP 的 POST /notes；一段处理结果要边产生边显示，可使用 HTTP 流式响应，文本事件采用 SSE；前后端要频繁互发消息，可使用 WebSocket。若只需偶尔查询状态，也可以定期 HTTP 查询，不需要先维护长连接。
验收结果
- 能说清 SSE 使用 HTTP，且不是 WebSocket 的别名。
- /generate 返回单个 {text:...} 对象；不会在每个步骤结束时返回一个新 HTTP 响应。
- 能为保存笔记、逐步显示文本、双向实时消息分别选择方案并说明原因。
常见错误
为了逐字显示文字，直接把所有接口都改成 WebSocket
原因：把流式接收和双向消息通信混为一谈。
处理：先确定客户端是否需要在同一条连接上持续主动发送；单向文本结果通常先考虑 HTTP 流式响应。
把服务器 yield 的次数当成客户端 read 的次数
原因：网络、服务器和浏览器可能合并或拆开数据块。
处理：依赖协议的事件分隔符解析消息，不依赖网络块边界。
扩展练习
用户提交一次短文本后只接收生成结果，偶尔点击“停止”。是否必须使用 WebSocket？
提示：思考取消 HTTP 请求是否已经满足停止读取的需要。
参考答案：
不必须。可以提交 HTTP 请求并读取流式响应，通过取消请求停止读取；服务端同时处理断开并释放资源。如果还需要在同一会话持续双向发送独立消息，再评估 WebSocket。
技术文档
12 用 StreamingResponse 输出 SSE 事件
实现有限长度文本流，定义 token / done 事件，并处理客户端断开。
本节目标
- 用异步生成器逐段发送 text/event-stream
- 使用 JSON 编码 data 字段并保留空行边界
- 区分正常完成、连接断开和流内错误
必要概念
yield 交出一段响应体
异步生成器可在产生下一段数据前 await。StreamingResponse 消费生成器并发送响应体；生成器必须有明确的正常结束条件。
消息格式属于双方约定
本课每个 token 事件的 data 是 {text:...}，最后发送 done 事件且 data 为 {}。JSON 编码处理引号和换行，空行负责分隔事件。不能直接把任意用户输入拼接成 SSE 行。
响应开始后不能改写 HTTP 状态
参数验证应在开始流式响应前完成。发送响应头后若处理中断，应通过约定的流内错误事件或连接结束通知客户端；不能再把已经开始的 200 改成 500。
动手实现
1. 定义一处统一的事件编码
把事件名限制为代码中定义的固定值；用 json.dumps 编码数据。返回字符串末尾是两个换行字符，即一个空行。ensure_ascii=False 保留中文可读性。
文件：main.py
import json
def encode_sse(event: str, data: dict) -> str:
payload = json.dumps(data, ensure_ascii=False)
return f"event: {event}\ndata: {payload}\n\n"
2. 产生有限长度 token 流
每隔约 0.4 秒发送一个 token，完成后发 done。使用 finally 观察连接关闭后的清理路径；真实上游迭代器、文件等资源应在这里或上下文管理器内关闭。不要吞掉取消异常。
文件：main.py
import asyncio
from fastapi import Request
async def token_events(request: Request, prompt: str):
try:
for text in ["收到：", prompt, "。处理完成。"]:
await asyncio.sleep(0.4)
if await request.is_disconnected():
return
yield encode_sse("token", {"text": text})
yield encode_sse("done", {})
finally:
print("stream closed")
3. 在 /stream 返回 StreamingResponse
GET 查询参数适合 EventSource 演示；不要在 URL 中传密钥或敏感正文。prompt 先经过长度验证，再进入生成器。已有相同路由时替换定义，避免重复注册。
文件：main.py
from fastapi import Query
from fastapi.responses import StreamingResponse
@app.get("/stream")
async def stream(
request: Request,
prompt: str = Query(default="hello", min_length=1, max_length=200),
):
return StreamingResponse(
token_events(request, prompt),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache"},
)
4. 从终端逐段观察，再主动取消
使用 curl.exe 的 -N 禁用它自己的输出缓冲。应先后看到三条 token 与一条 done，每条后有空行。第二次请求在生成中按 Ctrl+C，服务端最终应打印 stream closed；断开通常通过取消路径或断开检查被观察到。
文件：终端 / 当前课时
curl.exe -N -i "http://127.0.0.1:8000/stream?prompt=hello"
5. 核对线上实际字节格式
以下是事件体格式展示，其中空行是协议的一部分。data 内的 JSON 字符串若有换行，应由 JSON 编码为转义序列，不能破坏事件边界。
文件：终端 / 当前课时
event: token
data: {"text": "hello"}
event: done
data: {}
验收结果
- 响应 Content-Type 以 text/event-stream 开头；终端能在流结束前看到 token。
- 正常结束只有一个 done，三个 token 的 text 拼接后与 /generate 的 text 相同。
- prompt 为空返回 422；请求中途取消后清理日志出现，服务仍能处理 /health。
常见错误
客户端一直等不到事件回调
原因：发送了 data 行却漏了事件末尾的空行。
处理：每条事件统一通过 encode_sse 编码，并确认末尾为 \n\n。
输入含换行后事件解析失败
原因：手工拼接 JSON 或把原始输入直接放进 data 行。
处理：先 json.dumps，再写入 SSE data 字段。
断开后仍持续执行，或者没有清理资源
原因：无限循环未处理断开、没有 await 让取消生效，或遗漏 finally 清理。
处理：在迭代中检查断开，使用可等待的 I/O，并在 finally 中释放资源；保持本课生成器有限长度。
扩展练习
在三个 token 后、done 前增加一个 stats 事件，返回 token_count:3；保持已有客户端仍能完成。
提示：stats 是独立事件类型，不需要把它伪装成 token。
参考答案：
在 done 前增加 yield encode_sse("stats", {"token_count": 3})。原客户端只监听 token 和 done 时会忽略 stats；新增监听器后可以显示统计值。
技术文档
13 在浏览器中读取流，并在中途停止
分别使用 EventSource 与 fetch，处理 UTF-8 分块、事件边界和取消请求。
本节目标
- 用 EventSource 监听命名事件并主动关闭
- 用 ReadableStream 解析可能被拆开的 SSE 数据
- 通过 AbortController 停止请求并区分取消与错误
必要概念
EventSource 简单，但请求形态有限
原生 EventSource 发起 GET 请求，自动解析 SSE 并具备重连行为；构造参数不能任意设置请求体和自定义请求头。收到本课有限流的 done 后应主动 close，防止结束后重新请求。
fetch 能发送 POST，再读取响应流
POST 请求不妨碍响应逐段返回。需要 JSON 请求体或自定义头时可用 fetch，但必须自己解析响应体。调用 response.json() 或 response.text() 会等到响应体读完。
网络块不等于事件
一个中文字符的 UTF-8 字节、一个 JSON 或空行都可能跨越两次 read；一次 read 也可能包含多条事件。先连续解码字节，再缓存文本并按事件边界提取。
动手实现
1. 先用 EventSource 接收命名事件
打开本地 http://127.0.0.1:8000/docs，在开发者工具 Console 运行。URL 与页面同源，先排除 CORS 干扰。控制台应逐条输出文本，done 后主动关闭；本课把网络失败也关闭，避免自动重连重复执行。
文件：浏览器 Console
{
const source = new EventSource("/stream?prompt=hello");
source.addEventListener("token", (event) => {
console.log(JSON.parse(event.data).text);
});
source.addEventListener("done", () => {
console.log("完成");
source.close();
});
source.onerror = () => {
console.error("流连接失败，已停止");
source.close();
};
window.stopSse = () => source.close();
}
2. 把事件解析与网络读取分开
先在 Console 定义解析函数。它为本课 LF / CRLF 事件格式处理跨块与多条 data 行，且要求正常收到 done；不实现完整 EventSource 的 id、retry 与自动重连功能。TextDecoder 的 stream:true 会暂存未完整的 UTF-8 字节，最后再刷新。
文件：浏览器 Console
async function consumeSse(response, onEvent) {
if (!response.ok) throw new Error(`HTTP ${response.status}`);
if (!response.headers.get("content-type")?.includes("text/event-stream")) {
throw new Error("响应不是 SSE");
}
if (!response.body) throw new Error("没有响应体");
const reader = response.body.getReader();
const decoder = new TextDecoder("utf-8");
let buffer = "";
let completed = false;
try {
while (!completed) {
const { value, done } = await reader.read();
buffer += done ? decoder.decode() : decoder.decode(value, { stream: true });
let boundary;
while ((boundary = /\r?\n\r?\n/.exec(buffer)) !== null) {
const block = buffer.slice(0, boundary.index);
buffer = buffer.slice(boundary.index + boundary[0].length);
const lines = block.split(/\r?\n/);
const name = lines.find(line => line.startsWith("event:"))
?.slice(6).trim() || "message";
const data = lines.filter(line => line.startsWith("data:"))
.map(line => line.slice(5).replace(/^ /, "")).join("\n");
if (data) onEvent(name, JSON.parse(data));
if (name === "done") { completed = true; break; }
}
if (done && !completed) throw new Error("连接结束，但没有收到 done");
}
await reader.cancel();
} catch (error) {
await reader.cancel().catch(() => {});
throw error;
} finally {
reader.releaseLock();
}
}
3. 用 fetch 读取同一个 GET 接口
在同一 Console 定义 startStream，然后执行 startStream()。紧接着执行 stopStream() 可中途取消；AbortError 代表用户主动取消，不应显示为服务故障。新的请求使用新的 AbortController。
文件：浏览器 Console
async function startStream() {
const controller = new AbortController();
window.stopStream = () => controller.abort();
try {
const response = await fetch("/stream?prompt=hello", {
signal: controller.signal,
});
await consumeSse(response, (name, data) => {
if (name === "token") console.log(data.text);
if (name === "done") console.log("完成");
});
} catch (error) {
if (error.name === "AbortError") console.log("已取消");
else console.error(error);
}
}
startStream();
4. 扩展：需要 POST 请求体时
先在 main.py 额外加入以下 POST /stream 路由，复用第 11 课 PromptIn 与第 12 课 token_events。然后把上一步 fetch 的 URL 改为 /stream，并在配置对象中加入 method:"POST"、headers:{"Content-Type":"application/json"}、body:JSON.stringify({prompt:"hello"})；保留 signal。POST /generate 仍返回完整 JSON，不可拿它套用 SSE 解析函数。
文件：main.py
@app.post("/stream")
async def stream_post(request: Request, payload: PromptIn):
return StreamingResponse(
token_events(request, payload.prompt),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache"},
)
5. 核对流为什么没有逐段显示
先用 curl.exe -N 直连本地 API。若本地逐段到达但页面一次出现，检查是否调用了 response.text/json，或渲染是否缓存了结果。若只有经过中间服务才一次出现，再检查中间服务的响应缓冲；本课只定位差异，具体代理配置留到部署模块。
验收结果
- EventSource 收到三次 token 和一次 done 后关闭，不会周期性重新发起 /stream。
- fetch 版本能正常显示中文；主动 stopStream 后输出“已取消”。
- 扩展 POST /stream 返回 SSE；POST /generate 仍返回 application/json。
常见错误
偶发 JSON.parse 错误，或中文出现替换字符
原因：直接对每次 read 的内容解码和解析，忽略了字符、事件可能跨块。
处理：保持同一 TextDecoder 并设置 stream:true，再缓存到完整空行边界后解析。
一次成功结束后又出现同样文本
原因：有限 SSE 结束后 EventSource 自动重连，客户端没有处理 done。
处理：监听 done 调用 close；真实恢复能力需要单独定义，不默认重复执行。
点击停止后仍把状态改成失败
原因：把所有 fetch 异常都当成网络故障。
处理：单独识别 AbortError，并同步把页面状态改为已取消。
扩展练习
把一条包含“你好”的 SSE 消息拆成逐个字节的 ReadableStream，验证解析函数仍得到完整中文和 done。
提示：用 TextEncoder 生成字节，再通过 Response 包装本地 ReadableStream；这只是解析器实验，不是 API 网络测试。
参考答案：
构造文本 'event: token\ndata: {"text":"你好"}\n\nevent: done\ndata: {}\n\n'；对 TextEncoder().encode(text) 的每个字节调用 controller.enqueue(new Uint8Array([byte]))，最后 close。用 new Response(stream,{headers:{"Content-Type":"text/event-stream"}}) 传入 consumeSse，回调应只收到完整 token 和 done。
技术文档
14 建立第一条 WebSocket 双向连接
实现 /ws 回显，观察建立、发送、接收和关闭的完整过程。
本节目标
- 使用 accept、receive_text 和 send_json 完成回显
- 在浏览器连接打开后发送消息
- 捕获断开并释放连接相关资源
必要概念
连接先建立，消息再发送
服务端 accept 后进入收发阶段；浏览器 WebSocket 从 CONNECTING 转为 OPEN 后才能发送。send 只表示数据进入发送流程，不证明对方已经处理。
一条连接，多条消息
本课客户端发送文本，服务端回复 JSON。普通文本对应 type:echo；文本 ping 对应 type:pong。双方必须约定数据格式，不能把 HTTP 路由的返回值当作 WebSocket 消息发送。
断开是正常生命周期
用户关页、主动关闭和网络中断都可能结束连接。receive_text 在断开时会抛出 WebSocketDisconnect；在 finally 中清理该连接持有的资源。
动手实现
1. 在 main.py 插入 /ws 路由
接受连接后循环读取消息。回复必须调用 send_json，而不是 return 一个字典。这个本地练习只接收文本消息，不实现用户身份或消息保存。
文件：main.py
from fastapi import WebSocket, WebSocketDisconnect
@app.websocket("/ws")
async def websocket_echo(websocket: WebSocket):
await websocket.accept()
try:
while True:
text = await websocket.receive_text()
if text == "ping":
await websocket.send_json({"type": "pong"})
else:
await websocket.send_json({"type": "echo", "text": text})
except WebSocketDisconnect:
pass
finally:
print("websocket closed")
2. 从浏览器建立连接
打开本地 http://127.0.0.1:8000/docs，在 Console 运行。代码把连接保留到 window.echoSocket，方便随后发送和关闭。open、message、close 的日志对应三个不同阶段。
文件：浏览器 Console
{
const socket = new WebSocket("ws://127.0.0.1:8000/ws");
window.echoSocket = socket;
socket.onopen = () => console.log("连接已打开");
socket.onmessage = event => console.log("收到", JSON.parse(event.data));
socket.onerror = () => console.error("连接发生错误");
socket.onclose = event => console.log("已关闭", event.code, event.reason);
}
3. 在 OPEN 状态发送两种消息
等“连接已打开”出现后，执行下列代码。在 Network 面板选择 WS 请求并查看消息列表，区分客户端发出的原始文本与服务端返回的 JSON。WebSocket 请求不使用 /docs 的 HTTP Execute 按钮测试。
文件：浏览器 Console
if (echoSocket.readyState === WebSocket.OPEN) {
echoSocket.send("hello");
echoSocket.send("ping");
}
4. 主动关闭并确认生命周期结束
调用 close 后，浏览器输出关闭事件，服务端输出清理日志。关闭后不能复用旧 socket 继续发送；需要新建 WebSocket。重新连接并再发送一次 hello，确认服务仍可用。
文件：浏览器 Console
echoSocket.close(1000, "lesson finished");
验收结果
- 浏览器先收到 open，发送 hello 后收到 {type:'echo',text:'hello'}。
- 发送 ping 后收到 {type:'pong'}；同一连接可重复收发。
- 主动关闭后收到 close，服务端清理日志出现，重新连接仍正常。
常见错误
调用 send 报连接尚未建立
原因：new WebSocket 返回时握手仍在进行。
处理：在 open 回调中发送，或先检查 readyState === WebSocket.OPEN。
浏览器无任何回复，路由却 return 了字典
原因：WebSocket 消息不会像 HTTP 响应一样自动序列化返回。
处理：在连接内 await websocket.send_json(...)。
把 WebSocket 地址粘到普通地址栏后得到 404
原因：使用 HTTP GET 测试了 WebSocket 端点，没有进行 WebSocket 握手。
处理：通过浏览器 WebSocket API 或支持 WS 的客户端测试，使用 ws:// 地址。
扩展练习
增加一个文本命令 upper:hello，让服务端回复 {type:'upper',text:'HELLO'}，其他消息保持回显。
提示：把 startswith('upper:') 分支放在默认 echo 分支之前。
参考答案：
在 echo 的 else 之前加入 elif text.startswith("upper:"): await websocket.send_json({"type":"upper","text":text.removeprefix("upper:").upper()})。浏览器收到的 JSON type 应为 upper。
技术文档
15 连接断开后：重连、心跳与取消
给 WebSocket 加上有限重试和定时器清理，验证停止后不会自行恢复。
本节目标
- 在异常关闭后用指数退避重新连接
- 通过应用层 ping / pong 检测连接是否仍可用
- 清理重连和心跳定时器，不自动重发未确认消息
必要概念
重连创建一条新连接
新连接不会自动恢复旧连接内的执行状态，也不保证断开期间消息送达。本课不缓存、不重发业务消息；需要可靠恢复时，必须另行设计消息编号、确认和服务端去重。
退避限制重试频率
连续失败时逐渐延长等待，并加少量随机抖动，避免同时快速请求。设置最大次数；用户主动停止后取消待执行重连。
心跳与页面状态
这里的 ping 是普通文本消息，pong 是业务 JSON，属于应用层心跳。浏览器 WebSocket API 不提供主动发送协议级 Ping 帧的方法。计时器会受后台页面节流影响，超时阈值需要按场景调整。
动手实现
1. 先定义连接状态与定时器清理
在本地 /docs 的 Console 按步骤运行。先关闭上一课 echoSocket，避免同时观察两条连接。下面状态只属于本次练习；刷新页面可重新开始。
文件：浏览器 Console
let wsSocket = null;
let retryTimer = null;
let heartbeatTimer = null;
let pongTimer = null;
let wsStopped = true;
let retryCount = 0;
function clearHeartbeat() {
clearInterval(heartbeatTimer);
clearTimeout(pongTimer);
heartbeatTimer = null;
pongTimer = null;
}
2. 加入应用层心跳
每 5 秒发送一次 ping，3 秒内没有 pong 就关闭连接，让同一条关闭处理路径决定是否重连。收到 pong 时取消对应超时。这里只允许一个等待中的心跳。
文件：浏览器 Console
function startHeartbeat(socket) {
clearHeartbeat();
heartbeatTimer = setInterval(() => {
if (socket.readyState !== WebSocket.OPEN) return;
socket.send("ping");
pongTimer = setTimeout(() => {
socket.close(4000, "heartbeat timeout");
}, 3000);
}, 5000);
}
3. 把重连集中在 close 事件中
首次失败约 0.5 秒后重连，随后约 1、2、4、8 秒，另加少量随机抖动；最多连续重试 5 次。error 只记录，避免 error 和 close 各重试一次。收到服务端有效消息后清零连续失败计数。
文件：浏览器 Console
function connectWs() {
if (wsStopped) return;
const socket = new WebSocket("ws://127.0.0.1:8000/ws");
wsSocket = socket;
socket.onopen = () => {
if (wsStopped || wsSocket !== socket) { socket.close(); return; }
console.log("已连接");
startHeartbeat(socket);
};
socket.onmessage = event => {
if (wsSocket !== socket) return;
const message = JSON.parse(event.data);
retryCount = 0;
if (message.type === "pong") {
clearTimeout(pongTimer);
pongTimer = null;
} else {
console.log("收到", message);
}
};
socket.onerror = () => console.log("连接错误，等待 close");
socket.onclose = () => {
if (wsSocket !== socket) return;
clearHeartbeat();
if (wsStopped) return;
if (retryCount >= 5) {
wsStopped = true;
console.log("连续重试已达上限，请手动开始");
return;
}
const delay = Math.min(8000, 500 * 2 ** retryCount) + Math.random() * 200;
retryCount += 1;
console.log(`第 ${retryCount} 次重试，等待 ${Math.round(delay)}ms`);
retryTimer = setTimeout(connectWs, delay);
};
}
4. 实现开始、发送与停止
开始函数防止重复创建连接；发送函数在未连接时明确拒绝；停止函数先记录主动停止，再清理所有计时器和连接。stopWs 同时可供页面关闭时调用。
文件：浏览器 Console
function startWs() {
if (!wsStopped) return;
wsStopped = false;
retryCount = 0;
connectWs();
}
function sendWs(text) {
if (wsSocket?.readyState !== WebSocket.OPEN) {
console.log("尚未连接，本条消息未发送");
return;
}
wsSocket.send(text);
}
function stopWs() {
wsStopped = true;
clearTimeout(retryTimer);
clearHeartbeat();
wsSocket?.close(1000, "user stopped");
wsSocket = null;
console.log("已停止");
}
window.addEventListener("pagehide", stopWs);
startWs();
5. 执行三组失败实验
正常连接后用 sendWs('hello') 验证回显。接着停止本地服务，观察逐渐延长的重试间隔，重启服务后确认新连接可收发。最后在下一次重试前调用 stopWs()，等待超过 9 秒，确认没有新建 WS 请求。断线期间尝试发送的消息不能在重连后自动出现。
验收结果
- 服务停止时重连间隔逐步增加，连续失败达到 5 次重试后停止。
- 服务恢复后能建立新连接并再次收到 echo，心跳 ping 对应 pong。
- stopWs 后超过 9 秒没有新连接；未连接时调用 sendWs 不会排队重发。
常见错误
每次断线都会产生两条新连接
原因：在 error 和 close 中分别启动重试，或开始按钮允许重复创建。
处理：只在 close 调度重试；开始前检查当前是否已经运行。
点停止后几秒钟又自动连上
原因：只关闭 socket，没有取消已经安排的 setTimeout。
处理：先设置主动停止标记，再清理重试、心跳和 pong 超时定时器。
重连后重复执行了一条操作
原因：客户端自动重发消息，但无法确定服务端在断开前是否已经执行。
处理：本课不自动重发；需要重试业务操作时另行加入确认、操作编号与幂等处理。
扩展练习
把连续重试上限改为 3 次，并验证用户手动 startWs() 后可以开始新一轮连接。
提示：更改重试边界，保留达到上限时 wsStopped=true，以及开始时 retryCount=0。
参考答案：
把 if (retryCount >= 5) 改为 >= 3。连续失败后控制台打印到达上限；随后调用 startWs() 重置次数并重新连接，若服务已恢复即可继续回显。
技术文档
16 把能运行的接口变成可重复验收的服务
用 pytest 检查 CRUD、错误、认证、SSE 和 WebSocket，再完成一次真实浏览器联调。
本节目标
- 使用 TestClient 在测试中运行 FastAPI 应用
- 为正常路径和 401 / 404 / 422 写出明确断言
- 区分接口逻辑验收与真实网络流式验证
必要概念
断言检查可观察结果
测试应检查状态码、响应字段与操作后结果，而不是只确认请求没报异常。每个测试创建自己的数据，通过返回 id 继续操作，避免假设某条笔记固定为 id=1。
TestClient 不等同于浏览器
TestClient 基于 HTTPX，可在进程内调用应用，不必先启动 Uvicorn。用 with 管理客户端才会执行 lifespan；它不验证浏览器 CORS，也不能证明真实网络按预期逐段到达。
错误与正常结束分别验收
422 验证输入约束；401 验证 /private 的 API Key；404 验证资源不存在。SSE 既检查事件数据，也检查 done；WebSocket 检查来回消息并关闭会话。
动手实现
1. 建立测试客户端
在 api-lab 新建 tests/test_acceptance.py；以下第 1–4 步代码依次放入同一文件。先运行 python -m pip install pytest httpx。monkeypatch 只在测试期间设置演示 Key，测试结束自动恢复；导入入口为根目录 main.py。
文件：tests/test_acceptance.py
import json
from uuid import uuid4
import pytest
from fastapi.testclient import TestClient
from main import app
@pytest.fixture
def client(monkeypatch):
monkeypatch.setenv("DEMO_API_KEY", "classroom-key")
with TestClient(app) as test_client:
yield test_client
def test_health(client):
response = client.get("/health")
assert response.status_code == 200
2. 验证创建、查询、更新与删除
从 POST 返回值提取 id，再验证默认值、列表、PUT 完整替换和 DELETE 后 404。PUT 必须包含 title、content、pinned 三个可写字段。DELETE 204 响应体应为空，不能再对它调用 json()。
文件：tests/test_acceptance.py
def test_note_lifecycle(client):
title = f"acceptance-{uuid4()}"
created = client.post("/notes", json={"title": title})
assert created.status_code == 201
note = created.json()
note_id = note["id"]
assert note["content"] == ""
assert note["pinned"] is False
assert client.get(f"/notes/{note_id}").json() == note
listed = client.get("/notes", params={"q": title, "limit": 100})
assert listed.status_code == 200
assert any(item["id"] == note_id for item in listed.json())
replacement = {"title": "更新后", "content": "测试正文", "pinned": True}
updated = client.put(f"/notes/{note_id}", json=replacement)
assert updated.status_code == 200
assert updated.json() == {"id": note_id, **replacement}
assert client.get(f"/notes/{note_id}").json() == updated.json()
deleted = client.delete(f"/notes/{note_id}")
assert deleted.status_code == 204
assert deleted.content == b""
missing = client.get(f"/notes/{note_id}")
assert missing.status_code == 404
assert missing.json()["error"]["code"] == "http_error"
3. 验证输入错误与受保护接口
空标题应被输入校验拒绝。只有 /private 在本地样例中需要 X-API-Key；不要把笔记的公开练习路由与认证演示混在一起。错误体沿用前面课程的统一结构 error.code / message / details。
文件：tests/test_acceptance.py
def test_validation(client):
invalid = client.post("/notes", json={"title": ""})
assert invalid.status_code == 422
assert invalid.json()["error"]["code"] == "validation_error"
assert client.get("/slow", params={"delay": -1}).status_code == 422
def test_api_key(client):
assert client.get("/private").status_code == 401
wrong = client.get("/private", headers={"X-API-Key": "wrong-key"})
assert wrong.status_code == 401
assert wrong.json()["error"]["code"] == "http_error"
valid = client.get("/private", headers={"X-API-Key": "classroom-key"})
assert valid.status_code == 200
4. 验证 SSE 内容和 WebSocket 来回消息
TestClient 读取有限 SSE 的完整响应后检查协议内容；该测试不用于断言首段到达时间。WebSocket 使用上下文管理器，离开时关闭连接。 如果完成了第 12 课的 stats 扩展，测试允许保留该事件，只对 token/done 的主线顺序和内容断言；done 仍必须是最后一个事件。
文件：tests/test_acceptance.py
def test_stream_matches_complete_result(client):
complete = client.post("/generate", json={"prompt": "hello"})
assert complete.status_code == 200
response = client.get("/stream", params={"prompt": "hello"})
assert response.status_code == 200
assert response.headers["content-type"].startswith("text/event-stream")
events = []
for block in response.text.replace("\r\n", "\n").strip().split("\n\n"):
fields = {}
for line in block.split("\n"):
name, value = line.split(":", 1)
fields[name] = value.lstrip(" ")
events.append((fields["event"], json.loads(fields["data"])))
assert [name for name, _ in events if name != "stats"] == ["token", "token", "token", "done"]
text = "".join(data["text"] for name, data in events if name == "token")
assert text == complete.json()["text"]
assert events[-1] == ("done", {})
def test_websocket(client):
with client.websocket_connect("/ws") as socket:
socket.send_text("hello")
assert socket.receive_json() == {"type": "echo", "text": "hello"}
socket.send_text("ping")
assert socket.receive_json() == {"type": "pong"}
5. 运行自动验收，再测真实网络
在 api-lab 执行 python -m pytest -q tests/test_acceptance.py。通过后另开终端运行 python -m uvicorn main:app --reload，再运行下面脚本。SSE 的 data 行应分批出现，时间通常约为 0.4、0.8、1.2 秒，而非都等到最后。最后在浏览器完成 CORS、主动停止流、WebSocket 断开重连三个场景。
文件：verify_stream_network.py
from time import perf_counter
import httpx
with httpx.Client(base_url="http://127.0.0.1:8000", timeout=5.0) as client:
started = perf_counter()
with client.stream("GET", "/stream", params={"prompt": "hello"}) as response:
response.raise_for_status()
for line in response.iter_lines():
if line.startswith("data:"):
print(f"{perf_counter() - started:.2f}s {line}", flush=True)
验收结果
- 自动测试覆盖健康检查、完整 CRUD、422 校验、401 认证、删除后 404、SSE 与 WebSocket，运行结果全部通过。
- 真实 HTTPX / curl 客户端在响应结束前接收到 token，浏览器停止读取后服务端执行清理。
- 允许来源的浏览器能够调用 API；未允许来源不能读取响应；WebSocket 主动停止后不再重连。
常见错误
单个测试能过，整套测试偶尔失败
原因：多个测试假定相同 id 或依赖前一个测试创建的数据。
处理：每个测试自行创建记录，用返回 id 操作，并在生命周期测试中删除。
测试里 app.state.http 不存在
原因：没有通过 with TestClient(app) 启动 lifespan，或代码未正确注册 lifespan。
处理：在 fixture 的上下文中创建客户端，并核对 FastAPI(lifespan=lifespan)。
SSE 测试通过，网页仍一次显示全部内容
原因：进程内测试只检查完整内容，没有经过浏览器、真实网络和中间服务。
处理：用真实 Uvicorn 配合 httpx.stream / curl -N 观察时间，再检查浏览器读取与渲染。
扩展练习
增加一个测试：创建笔记后，PUT 只发送 title 应返回 422，随后 GET 仍返回原来的完整内容。
提示：输入不合法时不能发生部分更新；先保存创建响应，再比较读取结果。
参考答案：
创建笔记并保存 note；使用 client.put(f'/notes/{note["id"]}', json={"title":"修改"})，断言 422；再次 GET 并断言等于 note，最后 DELETE 清理。这个测试同时检查完整替换约定和校验失败不写入的行为。