# 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 再激活。

文件：终端 / 当前课时

````bash
# 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

````bash
fastapi>=0.115,<1
uvicorn[standard]>=0.30,<1
pydantic>=2,<3
````

#### 3. 编写健康检查接口

在 api-lab 根目录创建 main.py。FastAPI 实例只创建一次。路由装饰器紧接在函数定义上方，返回值使用 Python 字典，JSON 编码由框架完成。

文件：main.py

````python
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。

文件：终端 / 当前课时

````bash
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。

文件：终端 / 当前课时

````http
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。

### 技术文档

- [Python：venv](https://docs.python.org/3/library/venv.html)
- [FastAPI：First Steps](https://fastapi.tiangolo.com/tutorial/first-steps/)
- [FastAPI：手动运行服务器](https://fastapi.tiangolo.com/deployment/manually/)

---

## 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 不需要请求体。

文件：终端 / 当前课时

````http
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

````python
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

````json
{
  "title": "first note",
  "content": "learn HTTP",
  "pinned": false
}
````

#### 4. 对照原始请求与服务器结果

命令行应返回 200 和 received 对象。改动 note.json 中的 title 后再次发送，确认响应随之变化。在浏览器开发者工具 Network 中展开 /docs 发出的请求，也能查看 Method、Request Headers、Payload 和 Response。

文件：终端 / 当前课时

````http
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 描述客户端希望接收的格式。

### 技术文档

- [RFC 9110：HTTP 消息与内容语义](https://www.rfc-editor.org/rfc/rfc9110.html#section-6)
- [RFC 8259：JSON 语法](https://www.rfc-editor.org/rfc/rfc8259)
- [FastAPI：直接使用 Request](https://fastapi.tiangolo.com/advanced/using-request-directly/)

---

## 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

````python
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

````python
@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

````python
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

````python
@app.post("/notes")
def inspect_note_body(note: NoteCreate):
    return note.model_dump()
````

#### 5. 建立有效值和无效值对照

在 /docs 连续完成以下四次调用，记录状态码与响应中的字段位置。缺少 title 的 POST 应在路由运行前失败。下一课会继续收紧 title 的长度与多余字段规则。

文件：终端 / 当前课时

````http
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。

### 技术文档

- [FastAPI：路径参数](https://fastapi.tiangolo.com/tutorial/path-params/)
- [FastAPI：查询参数](https://fastapi.tiangolo.com/tutorial/query-params/)
- [FastAPI：请求体](https://fastapi.tiangolo.com/tutorial/body/)

---

## 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

````python
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

````python
class NoteOut(BaseModel):
    id: int
    title: str
    content: str
    pinned: bool
````

#### 3. 用 response_model 约束 POST 响应

替换 POST /notes 的装饰器和函数。本节的 id=1 仍是演示值，下一阶段才生成真实编号。返回字典故意增加 internal_source，查看 HTTP 响应时该字段应被过滤。

文件：main.py

````python
@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。

文件：终端 / 当前课时

````json
{
  "title": "",
  "content": "title must not be empty",
  "pinned": false
}
````

#### 5. 检查文档中的契约

刷新 /docs。展开 POST /notes 的 Schema，检查 title 的 minLength、maxLength，以及 NoteOut 的 id。发送 {"title":"schema check"} 后应返回下面四个字段。保存一份响应作为后续 CRUD 的验收样本。

文件：终端 / 当前课时

````json
{
  "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'。

### 技术文档

- [Pydantic：Field 与字段约束](https://docs.pydantic.dev/latest/concepts/fields/)
- [Pydantic：验证器](https://docs.pydantic.dev/latest/concepts/validators/)
- [FastAPI：响应模型](https://fastapi.tiangolo.com/tutorial/response-model/)

---

## 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

````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 字段校验无关。

文件：终端 / 当前课时

````bash
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

````python
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

````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 空体，因此需同时调整状态码和响应内容。

### 技术文档

- [RFC 9110：请求方法](https://www.rfc-editor.org/rfc/rfc9110.html#section-9)
- [RFC 9110：204 No Content](https://www.rfc-editor.org/rfc/rfc9110.html#section-15.3.5)
- [FastAPI：更新请求体](https://fastapi.tiangolo.com/tutorial/body-updates/)

---

## 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

````python
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

````python
@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

````python
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

````python
@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。操作过程中不要修改代码触发重载。最后重新创建一条，检查编号没有因删除而回退。

文件：终端 / 当前课时

````json
{
  "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。

### 技术文档

- [FastAPI：响应状态码](https://fastapi.tiangolo.com/tutorial/response-status-code/)
- [FastAPI：直接返回 Response](https://fastapi.tiangolo.com/advanced/response-directly/)
- [FastAPI：处理错误](https://fastapi.tiangolo.com/tutorial/handling-errors/)

---

## 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

````python
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

````python
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

````python
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 可能随依赖版本变化，客户端不要解析这段文本作业务判断。

文件：终端 / 当前课时

````json
{
  "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 用于显示，不作为稳定的判断条件。

### 技术文档

- [FastAPI：覆盖异常处理器](https://fastapi.tiangolo.com/tutorial/handling-errors/)
- [Starlette：Exceptions](https://github.com/Kludex/starlette/blob/main/docs/exceptions.md)
- [Pydantic：错误处理](https://docs.pydantic.dev/latest/errors/errors/)

---

## 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

````python
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

````python
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

````python
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 是公开教学校验值。

文件：终端 / 当前课时

````bash
# 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，确认该本地练习接口未要求密钥。

文件：终端 / 当前课时

````bash
# 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 依赖。所有人使用同一公开密钥，服务只能判断是否匹配，无法识别具体用户。

### 技术文档

- [FastAPI：依赖注入](https://fastapi.tiangolo.com/tutorial/dependencies/)
- [FastAPI：Header 参数](https://fastapi.tiangolo.com/tutorial/header-params/)
- [FastAPI：多文件应用](https://fastapi.tiangolo.com/tutorial/bigger-applications/)
- [Python：compare_digest](https://docs.python.org/3/library/secrets.html#secrets.compare_digest)

---

## 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

````python
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

````python
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，观察阻塞时它也要等待。结果会受机器和调度开销影响，不以毫秒误差判断对错。

文件：终端 / 当前课时

````bash
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 计算加速方案。

文件：终端 / 当前课时

````python
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。

### 技术文档

- [FastAPI：并发与 async / await](https://fastapi.tiangolo.com/async/)
- [Python：协程、gather 与 to_thread](https://docs.python.org/3/library/asyncio-task.html)

---

## 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

````python
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

````python
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

````python
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

````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 加回白名单。

文件：终端 / 当前课时

````bash
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。

### 技术文档

- [HTTPX：异步客户端](https://www.python-httpx.org/async/)
- [HTTPX：超时阶段](https://www.python-httpx.org/advanced/timeouts/)
- [HTTPX：状态码与异常](https://www.python-httpx.org/quickstart/)
- [FastAPI：lifespan](https://fastapi.tiangolo.com/advanced/events/)
- [FastAPI：CORS](https://fastapi.tiangolo.com/tutorial/cors/)

---

## 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

````python
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。结果约在所有步骤完成后出现。区分响应头到达与可用业务内容到达；观察何时收到第一段内容。

文件：终端 / 当前课时

````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。

### 技术文档

- [FastAPI：StreamingResponse](https://fastapi.tiangolo.com/advanced/custom-response/)
- [MDN：SSE 事件流](https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events)
- [FastAPI：WebSocket](https://fastapi.tiangolo.com/advanced/websockets/)

---

## 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

````python
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

````python
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

````python
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；断开通常通过取消路径或断开检查被观察到。

文件：终端 / 当前课时

````bash
curl.exe -N -i "http://127.0.0.1:8000/stream?prompt=hello"
````

#### 5. 核对线上实际字节格式

以下是事件体格式展示，其中空行是协议的一部分。data 内的 JSON 字符串若有换行，应由 JSON 编码为转义序列，不能破坏事件边界。

文件：终端 / 当前课时

````http
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；新增监听器后可以显示统计值。

### 技术文档

- [FastAPI：自定义与流式响应](https://fastapi.tiangolo.com/advanced/custom-response/)
- [MDN：SSE 消息格式](https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events)
- [FastAPI：Request 类与断开检查](https://fastapi.tiangolo.com/reference/request/)

---

## 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

````javascript
{
  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

````javascript
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

````javascript
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

````python
@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。

### 技术文档

- [MDN：EventSource](https://developer.mozilla.org/en-US/docs/Web/API/EventSource)
- [MDN：读取 ReadableStream](https://developer.mozilla.org/en-US/docs/Web/API/Streams_API/Using_readable_streams)
- [MDN：TextDecoder.decode](https://developer.mozilla.org/en-US/docs/Web/API/TextDecoder/decode)
- [MDN：AbortController](https://developer.mozilla.org/en-US/docs/Web/API/AbortController)

---

## 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

````python
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

````javascript
{
  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

````javascript
if (echoSocket.readyState === WebSocket.OPEN) {
  echoSocket.send("hello");
  echoSocket.send("ping");
}
````

#### 4. 主动关闭并确认生命周期结束

调用 close 后，浏览器输出关闭事件，服务端输出清理日志。关闭后不能复用旧 socket 继续发送；需要新建 WebSocket。重新连接并再发送一次 hello，确认服务仍可用。

文件：浏览器 Console

````javascript
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。

### 技术文档

- [FastAPI：WebSocket 收发与断开](https://fastapi.tiangolo.com/advanced/websockets/)
- [MDN：编写 WebSocket 客户端](https://developer.mozilla.org/en-US/docs/Web/API/WebSockets_API/Writing_WebSocket_client_applications)

---

## 15 连接断开后：重连、心跳与取消

给 WebSocket 加上有限重试和定时器清理，验证停止后不会自行恢复。

### 本节目标

- 在异常关闭后用指数退避重新连接
- 通过应用层 ping / pong 检测连接是否仍可用
- 清理重连和心跳定时器，不自动重发未确认消息

### 必要概念

#### 重连创建一条新连接

新连接不会自动恢复旧连接内的执行状态，也不保证断开期间消息送达。本课不缓存、不重发业务消息；需要可靠恢复时，必须另行设计消息编号、确认和服务端去重。

#### 退避限制重试频率

连续失败时逐渐延长等待，并加少量随机抖动，避免同时快速请求。设置最大次数；用户主动停止后取消待执行重连。

#### 心跳与页面状态

这里的 ping 是普通文本消息，pong 是业务 JSON，属于应用层心跳。浏览器 WebSocket API 不提供主动发送协议级 Ping 帧的方法。计时器会受后台页面节流影响，超时阈值需要按场景调整。

### 动手实现

#### 1. 先定义连接状态与定时器清理

在本地 /docs 的 Console 按步骤运行。先关闭上一课 echoSocket，避免同时观察两条连接。下面状态只属于本次练习；刷新页面可重新开始。

文件：浏览器 Console

````javascript
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

````javascript
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

````javascript
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

````javascript
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() 重置次数并重新连接，若服务已恢复即可继续回显。

### 技术文档

- [MDN：WebSocket 客户端生命周期](https://developer.mozilla.org/en-US/docs/Web/API/WebSockets_API/Writing_WebSocket_client_applications)
- [MDN：WebSocket.close](https://developer.mozilla.org/en-US/docs/Web/API/WebSocket/close)
- [RFC 6455：异常关闭后的恢复建议](https://www.rfc-editor.org/rfc/rfc6455#section-7.2.3)

---

## 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

````python
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

````python
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

````python
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

````python
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

````python
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 清理。这个测试同时检查完整替换约定和校验失败不写入的行为。

### 技术文档

- [FastAPI：TestClient 与 pytest](https://fastapi.tiangolo.com/tutorial/testing/)
- [FastAPI：测试 lifespan](https://fastapi.tiangolo.com/advanced/testing-events/)
- [FastAPI：测试 WebSocket](https://fastapi.tiangolo.com/advanced/testing-websockets/)
- [pytest：fixtures](https://docs.pytest.org/en/stable/how-to/fixtures.html)
- [pytest：临时环境变量](https://docs.pytest.org/en/stable/how-to/monkeypatch.html)
- [HTTPX：流式响应](https://www.python-httpx.org/quickstart/)