# API Lab · FastAPI 笔记服务

这份代码配合「API 服务开发」模块。用同一个小服务练习请求参数、REST CRUD、错误响应、依赖、异步等待、SSE 与 WebSocket。代码直接运行，不需要模型账户或模型密钥。

这是各课核心功能的参考实现。课后扩展由学生按课件添加，例如第 06 课的 offset 分页、第 13 课的 POST /stream，以及 /private/status。`PromptIn`、`encode_sse()` 与 `token_events()` 已提供，第 13 课可直接复用。代码不预置全部练习答案。

推荐 Python 3.11 或更新版本；本项目在 Windows、Python 3.12.6 下实际验证。依赖版本固定在 `requirements.txt`。数据库、Redis、任务队列、Docker 与服务器部署留到后续模块。

## 1. 启动服务

先解压并进入 `api-lab` 文件夹。后续命令都在这个文件夹执行。

Windows PowerShell：

```powershell
python -m venv .venv
.\.venv\Scripts\Activate.ps1
python -m pip install -r requirements.txt
$env:DEMO_API_KEY = 'classroom-key'
python -m uvicorn main:app --reload
```

如果 PowerShell 不允许执行激活脚本，不需要修改系统策略，改为直接调用虚拟环境中的 Python：

```powershell
.\.venv\Scripts\python.exe -m pip install -r requirements.txt
$env:DEMO_API_KEY = 'classroom-key'
.\.venv\Scripts\python.exe -m uvicorn main:app --reload
```

macOS / Linux：

```bash
python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt
export DEMO_API_KEY='classroom-key'
python -m uvicorn main:app --reload
```

终端保留运行，打开：

- `http://127.0.0.1:8000/health`：应返回 `{"status":"ok"}`。
- `http://127.0.0.1:8000/docs`：展开端点，点击 **Try it out → Execute**，观察 Request URL、Response code 和 Response body。
- `http://127.0.0.1:8000/openapi.json`：查看接口描述。

`main:app` 表示导入 `main.py` 中的 `app` 对象。`--reload` 在代码改动后重启进程，因此笔记会清空。按 `Ctrl+C` 停止服务。

## 2. 跑一遍 CRUD

在 `/docs` 按顺序执行：

1. `POST /notes` 提交 `{"title":"第一篇笔记"}`，得到 **201**，记录返回的 `id` 与响应头 `Location: /notes/{id}`。`content` 默认空字符串，`pinned` 默认 `false`。
2. `GET /notes` 得到 JSON 数组，默认最多 10 条；填写 `q` 过滤标题或正文，填写 `pinned=true` 或 `pinned=false` 筛选置顶状态，填写 `limit` 限制返回条数。先筛选再限制条数，不传 `pinned` 时保留两种状态。
3. `GET /notes/{note_id}` 用刚才的 ID 读取笔记。
4. `PUT /notes/{note_id}` 提交 `{"title":"修改后的笔记","content":"完整替换","pinned":true}`。三个可修改字段都必须出现，少一个得到 **422**。
5. `DELETE /notes/{note_id}` 得到 **204** 和空响应体。此时不要调用前端的 `response.json()`。
6. 再读同一个 ID，得到 **404**；标题提交空字符串、纯空白或超过 120 个字符，得到 **422**。

标题、正文会去掉首尾空白。`pinned` 是布尔类型，示例使用 JSON 的 `true` / `false`；Pydantic 默认还允许部分类型转换，不等同于严格模式。未声明的请求体字段会返回 422。

## 3. 浏览器联调

打开第二个终端，同样进入 `api-lab`，激活虚拟环境后执行：

```bash
python -m http.server 5500
```

打开 `http://localhost:5500/web-client.html`。不要双击 HTML 使用 `file://`。页面有三个可以直接操作的实验：

- 创建笔记：`fetch()` 发送 POST JSON，输出状态码与响应体。
- SSE：通过 `EventSource` 收到三个 `token` 事件和一个 `done` 事件；完整文本逐段显示。
- WebSocket：连接后发送文本并收到 echo，发送文本 `ping` 收到 `pong`，然后断开并重连。

打开浏览器开发者工具的 **Network** 面板。POST JSON 的跨域请求会触发 OPTIONS 预检；SSE 在连接内连续收到事件；WebSocket 在 Messages / Frames 中显示双向消息。不同浏览器的面板名称略有差异。

服务仅允许 `http://localhost:5500` 与 `http://127.0.0.1:5500` 这两个浏览器来源。域名、协议或端口不同就是不同来源。CORS 是浏览器的跨域读取规则，不能替代身份验证；WebSocket 也不由这段 CORS 中间件提供鉴权。

要复现第 10 课的来源不匹配，先临时把 `allow_origins` 收窄为只有 `http://127.0.0.1:5500`，再从 `http://localhost:5500` 打开前端；实验后恢复两个来源。

## 4. SSE 的实际内容

先启动 API。在另一终端执行以下命令，观察每个事件约隔 0.4 秒到达：

Windows PowerShell：

```powershell
curl.exe -N 'http://127.0.0.1:8000/stream?prompt=hello'
```

macOS / Linux：

```bash
curl -N 'http://127.0.0.1:8000/stream?prompt=hello'
```

协议内容如下。每个事件末尾的空行负责划分事件边界。

```text
event: token
data: {"text": "收到："}

event: token
data: {"text": "hello"}

event: token
data: {"text": "。处理完成。"}

event: done
data: {}

```

`token` 和 `done` 是本例约定的事件名；每个 `token` 的内容是演示片段，不代表模型分词器中的 token。`data` 用 `json.dumps()` 编码，所以文本里的引号和换行不会破坏 SSE 帧。

对照 `POST /generate`：它也模拟三个 0.4 秒处理步骤，但等约 1.2 秒后才一次返回完整 JSON。SSE 每完成一个步骤就发送一次。流正常完成或中途关闭时，`token_events()` 的 `finally` 会在服务端终端打印 `stream closed`。

`EventSource` 默认使用 GET，不能直接提交 JSON 请求体或任意自定义请求头。本例使用查询参数；需要 POST 请求体时可按课程扩展为 `fetch()` 与 `ReadableStream`。页面在收到 `done`、发生错误、手动停止、再次开始或离开页面时关闭旧连接，避免默认重连再次生成内容。

服务检查客户端是否断开。`Cache-Control: no-cache` 减少缓存干扰；`X-Accel-Buffering: no` 用于支持它的代理。真实线上流式到达仍取决于代理与客户端配置，不能只靠这个响应头保证。

## 5. 依赖与异步等待

`GET /private` 演示 `Depends()` 与 `X-API-Key` 请求头：

- 服务进程未设置 `DEMO_API_KEY`：返回 **503**，其余端点仍可使用。
- 设置后，漏传或传错 `X-API-Key`：返回 **401**。
- 设置为 `classroom-key` 后传入同样的请求头：返回 **200**。

`classroom-key` 只是课堂示例值。只有 `/private` 受这个演示依赖保护，`/notes`、`/stream` 和 `/ws` 没有身份验证。

`GET /slow?delay=1` 使用 `await asyncio.sleep()` 模拟可等待的 I/O。合法等待时间是 0～5 秒。服务启动后运行：

```bash
python concurrency_client.py
```

脚本同时发出四个 1 秒请求，打印每个响应与总耗时。在本机空闲状态下，总耗时通常接近 1 秒加网络与调度开销。`async def` 不会自动把同步文件读写、同步网络调用或 CPU 密集计算变为不阻塞的工作。

`GET /upstream-demo?delay=1` 通过共享的 HTTPX AsyncClient 调用本机 `/slow`，成功返回 `{"upstream":{"delay":1,"status":"done"}}`。客户端在 lifespan 中创建，并在应用关闭时释放。连接超时为 1 秒，其余默认超时为 3 秒；`delay=4` 会触发读取超时，接口返回 **504**。上游非成功状态或连接失败返回 **502**。

这个出站实验固定调用 `http://127.0.0.1:8000`，不访问公网，也不接受用户输入任意 URL。请按启动步骤使用 8000 端口；若修改服务端口，要同时修改 lifespan 中的 `base_url`。

## 6. 端点契约

| 方法与路径 | 输入 | 成功响应 | 练习目标 |
| --- | --- | --- | --- |
| GET `/health` | 无 | 200 `{"status":"ok"}` | 路由与启动 |
| GET `/notes` | 可选 `q`、`pinned`；`limit` 默认 10，范围 1～100 | 200 笔记数组 | 查询参数 |
| POST `/notes` | `title`；可选 `content`、`pinned` | 201 笔记对象；`Location` 指向新资源 | 请求体与默认值 |
| GET `/notes/{note_id}` | 正整数 ID | 200 笔记对象 | 路径参数、404 |
| PUT `/notes/{note_id}` | `title`、`content`、`pinned` 都必填 | 200 笔记对象 | 完整替换 |
| DELETE `/notes/{note_id}` | 正整数 ID | 204 空响应体 | 删除语义 |
| GET `/private` | 请求头 `X-API-Key` | 200 `{"message":"API key accepted"}` | 依赖与环境变量 |
| GET `/slow` | `delay` 默认 1，范围 0～5 | 200 `{"delay":1,"status":"done"}` | 异步等待 |
| GET `/upstream-demo` | `delay` 默认 1，范围 0～5 | 200 `{"upstream":{"delay":1,"status":"done"}}` | HTTPX 生命周期、502/504 |
| POST `/generate` | `{"prompt":"hello"}`，1～200 字符 | 200 `{"text":"收到：hello。处理完成。"}` | 一次性响应 |
| GET `/stream` | `prompt` 默认 hello，1～200 字符 | 200 `text/event-stream` | 有限 SSE 流 |
| WS `/ws` | 文本；文本 `ping` 是特殊约定 | echo 或 pong JSON 消息 | 同一连接上的双向通信 |

HTTP 异常与请求校验错误统一为：

```json
{"error":{"code":"http_error","message":"Note not found","details":[]}}
```

校验失败时，`code` 为 `validation_error`，`details` 包含 `loc`、`msg`、`type`。这个教学处理器覆盖 HTTPException 与 RequestValidationError；未处理的程序异常、CORS 预检拒绝和 WebSocket 消息不属于该 JSON 契约。生产环境的日志、鉴权与错误追踪在后续模块补充。

## 7. 验收

```bash
python -m pytest -q
python -m pip check
```

测试覆盖 CRUD 全流程与 Location、PUT 必填字段、404/422、标题限制、q/pinned 组合过滤与默认 limit、环境变量与请求头、SSE 帧边界、中文与换行内容、WebSocket echo/ping/重连、CORS 预检和健康端点。HTTPX 上游成功、超时、HTTP 错误、连接失败使用 MockTransport 验证，不访问公网；同时检查 lifespan 关闭客户端。`TestClient` 的 SSE 测试检查协议格式与完整内容，不用于判断真实网络的首段到达时间。

实际验证结果记录在 `VERIFICATION.md`。示例的测试会重置内存笔记，适合独立测试进程，不要将测试代码作为线上操作入口。

## 8. 文件与使用边界

- `main.py`：核心功能参考实现，保留在一个文件中便于逐节查找。课程可进一步练习拆分 models、routers 和 dependencies。
- `web-client.html`：不需要构建工具的本地浏览器客户端。
- `concurrency_client.py`：四个并发请求的实测脚本。
- `test_main.py`：可执行验收。
- `requirements.txt`：已验证的固定依赖。

所有笔记保存在当前 Python 进程的内存中，重启清空、多个 worker 彼此不共享。请按示例运行单个开发进程。这里没有用户隔离、持久化、线上密钥管理、限流或任务队列；课程后续再接入数据库及部署能力。

## 官方参考

- FastAPI 请求体与模型：https://fastapi.tiangolo.com/tutorial/body/
- FastAPI 错误处理：https://fastapi.tiangolo.com/tutorial/handling-errors/
- FastAPI 自定义与流式响应：https://fastapi.tiangolo.com/advanced/custom-response/
- FastAPI WebSocket：https://fastapi.tiangolo.com/advanced/websockets/
- FastAPI CORS：https://fastapi.tiangolo.com/tutorial/cors/
- FastAPI 测试：https://fastapi.tiangolo.com/tutorial/testing/