890 lines
32 KiB
Markdown
890 lines
32 KiB
Markdown
---
|
||
id: plan.week04
|
||
title: "第 4 周：Python 速成（给 JS 开发者）+ FastAPI 基础"
|
||
type: week
|
||
domain: plan
|
||
stack: [python, asyncio, fastapi]
|
||
week: [4]
|
||
est_hours: 12
|
||
order: 4
|
||
---
|
||
|
||
# 第 4 周：Python 速成（给 JS 开发者）+ FastAPI 基础
|
||
|
||
> 第二段“Python 与 AI 应用”的第 1 周。从这周起 **Python 是求职主语言**。上周 `week03-project1.md`，下周 `week05-rag-1.md`。
|
||
|
||
**本周目标**：把写 AI 应用后端需要的 Python 补齐：包管理、类型标注、异常、asyncio、pydantic、httpx、pytest、FastAPI（Uvicorn、依赖注入、中间件、鉴权、生命周期、SSE）。只学项目马上要用的。
|
||
|
||
**学法**：每个知识点先想“JS / Go 里对应什么”。本周代码示例都在 Python 3.12 上实际运行过。
|
||
|
||
**参考资料**
|
||
- Python 教程（中文）：https://docs.python.org/zh-cn/3/tutorial/
|
||
- asyncio（中文）：https://docs.python.org/zh-cn/3/library/asyncio.html
|
||
- uv：https://docs.astral.sh/uv/ ；pydantic：https://docs.pydantic.dev/ ；httpx：https://www.python-httpx.org/
|
||
- FastAPI（中文）：https://fastapi.tiangolo.com/zh/
|
||
|
||
## 本周最终目录
|
||
|
||
```text
|
||
py-learning/
|
||
├── pyproject.toml
|
||
├── day1_basics.py
|
||
├── day2_typing.py
|
||
├── day3_errors.py
|
||
├── day4_async.py
|
||
├── day5_models.py
|
||
├── day6_http.py
|
||
├── app/
|
||
│ ├── __init__.py
|
||
│ └── main.py # FastAPI 练习服务
|
||
└── tests/
|
||
├── test_day2.py
|
||
├── test_day5.py
|
||
├── test_day6.py
|
||
└── test_app.py
|
||
```
|
||
|
||
---
|
||
|
||
## 第 1 天：包管理和基础语法
|
||
|
||
### 学习内容
|
||
|
||
**1. uv / pip / poetry**
|
||
|
||
- 说明：`pip` 是官方安装工具，配合 `venv` 虚拟环境（相当于每个项目一个独立的 `node_modules`）；`poetry` 和 `uv` 是“依赖声明 + 锁文件 + 虚拟环境”一体的工具，类似 npm/pnpm。uv 速度快、命令简单，本计划用它；招聘里提到 pip、poetry 时，概念是一样的。
|
||
- 案例：
|
||
|
||
```bash
|
||
curl -LsSf https://astral.sh/uv/install.sh | sh
|
||
uv init py-learning && cd py-learning
|
||
uv python pin 3.12
|
||
uv add httpx pydantic pydantic-settings fastapi uvicorn pyjwt
|
||
uv add --dev pytest pytest-asyncio ruff pyright
|
||
uv run python day1_basics.py # 在项目虚拟环境里运行
|
||
# 对照：pip 的做法
|
||
# python -m venv .venv && source .venv/bin/activate && pip install httpx && pip freeze > requirements.txt
|
||
```
|
||
|
||
| npm / pnpm | uv | pip | poetry |
|
||
| --- | --- | --- | --- |
|
||
| `package.json` | `pyproject.toml` | `requirements.txt` | `pyproject.toml` |
|
||
| `pnpm-lock.yaml` | `uv.lock` | 无（`pip freeze` 近似） | `poetry.lock` |
|
||
| `pnpm add x` | `uv add x` | `pip install x` | `poetry add x` |
|
||
| `pnpm exec` | `uv run` | 激活 venv 后直接运行 | `poetry run` |
|
||
|
||
**2. 基础语法对照 JS**
|
||
|
||
- 说明：缩进代替大括号；`None/True/False`；列表推导式相当于 `map + filter`；解包相当于解构；f-string 相当于模板字符串。
|
||
- 案例：
|
||
|
||
```python
|
||
# 文件：py-learning/day1_basics.py
|
||
from collections import Counter
|
||
|
||
|
||
def word_count(text: str) -> dict[str, int]:
|
||
return dict(Counter(text.lower().split()))
|
||
|
||
|
||
def unique(items: list[str]) -> list[str]:
|
||
seen: set[str] = set()
|
||
return [x for x in items if not (x in seen or seen.add(x))] # 保持顺序去重
|
||
|
||
|
||
def main() -> None:
|
||
nums = [1, 2, 3, 4, 5]
|
||
squares_of_even = [n * n for n in nums if n % 2 == 0] # JS: nums.filter(...).map(...)
|
||
first, *rest = nums # JS: const [first, ...rest] = nums
|
||
user = {"name": "小王", "years": 7}
|
||
merged = {**user, "city": "上海"} # JS: {...user, city: '上海'}
|
||
print(squares_of_even, first, rest, merged)
|
||
print(f"{user['name']} 做了 {user['years']} 年前端") # JS: `${...}`
|
||
print(word_count("Go is fun and go is fast"))
|
||
print(unique(["a", "b", "a", "c"]))
|
||
|
||
|
||
def append_bad(x, items=[]): # 坑：默认参数只在定义时创建一次，所有调用共享
|
||
items.append(x)
|
||
return items
|
||
|
||
|
||
def append_good(x, items=None):
|
||
items = [] if items is None else items
|
||
items.append(x)
|
||
return items
|
||
|
||
|
||
if __name__ == "__main__": # 直接运行这个文件时才执行（被 import 时不执行）
|
||
main()
|
||
print(append_bad(1), append_bad(2)) # [1, 2] [1, 2]
|
||
print(append_good(1), append_good(2)) # [1] [2]
|
||
```
|
||
|
||
- 面试常问：`list` 和 `tuple` 的区别（可变 / 不可变）；可变默认参数的坑；`is` 和 `==` 的区别。
|
||
|
||
### 要建的文件
|
||
|
||
- `pyproject.toml`（`uv init` 生成）、`day1_basics.py`
|
||
|
||
### 运行方式
|
||
|
||
```bash
|
||
uv run python day1_basics.py
|
||
```
|
||
|
||
### 完成标准
|
||
|
||
- 能用推导式、解包、f-string 写出上面的例子；能解释可变默认参数的坑。
|
||
|
||
---
|
||
|
||
## 第 2 天：类型标注、dataclass、Protocol、pytest
|
||
|
||
### 学习内容
|
||
|
||
**1. 类型标注和检查**
|
||
|
||
- 说明：和 TS 一样，类型标注只是提示，运行时不检查，要靠 pyright / mypy 静态检查。常用：`list[str]`、`dict[str, int]`、`X | None`、`Literal["a", "b"]`、`TypedDict`。
|
||
- **dataclass**：自动生成构造函数的数据类，相当于 Go 的结构体。
|
||
- **Protocol**：结构化的“隐式接口”，和 Go 的 interface 一样，不用继承，方法对得上就算实现。后面 LLM 客户端、向量库都靠它做可替换实现。
|
||
- 案例：
|
||
|
||
```python
|
||
# 文件：py-learning/day2_typing.py
|
||
from dataclasses import dataclass, field
|
||
from typing import Literal, Protocol
|
||
|
||
TaskStatus = Literal["todo", "doing", "done"]
|
||
|
||
|
||
@dataclass
|
||
class Task:
|
||
id: int
|
||
title: str
|
||
status: TaskStatus = "todo"
|
||
tags: list[str] = field(default_factory=list) # 可变默认值要用 default_factory
|
||
|
||
|
||
class TaskStore(Protocol):
|
||
def add(self, task: Task) -> None: ...
|
||
def all(self) -> list[Task]: ...
|
||
|
||
|
||
class MemoryTaskStore: # 没有继承 TaskStore，但方法签名一致，就满足协议
|
||
def __init__(self) -> None:
|
||
self._items: list[Task] = []
|
||
|
||
def add(self, task: Task) -> None:
|
||
self._items.append(task)
|
||
|
||
def all(self) -> list[Task]:
|
||
return list(self._items)
|
||
|
||
|
||
def count_by_status(store: TaskStore) -> dict[TaskStatus, int]:
|
||
result: dict[TaskStatus, int] = {"todo": 0, "doing": 0, "done": 0}
|
||
for t in store.all():
|
||
result[t.status] += 1
|
||
return result
|
||
```
|
||
|
||
**2. pytest 和参数化测试**
|
||
|
||
- 说明：pytest 自动发现 `test_*.py` 里的 `test_*` 函数，直接用 `assert`；`@pytest.mark.parametrize` 相当于 Go 的表格驱动测试。
|
||
- 案例：
|
||
|
||
```python
|
||
# 文件：py-learning/tests/test_day2.py
|
||
import pytest
|
||
|
||
from day2_typing import MemoryTaskStore, Task, count_by_status
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"statuses, expected",
|
||
[
|
||
([], {"todo": 0, "doing": 0, "done": 0}),
|
||
(["todo", "done", "done"], {"todo": 1, "doing": 0, "done": 2}),
|
||
],
|
||
)
|
||
def test_count_by_status(statuses, expected):
|
||
store = MemoryTaskStore()
|
||
for i, s in enumerate(statuses):
|
||
store.add(Task(id=i, title=f"t{i}", status=s))
|
||
assert count_by_status(store) == expected
|
||
```
|
||
|
||
- 常见坑：故意写 `Task(id=1, title="x", status="deleted")`，Python 照样运行，只有 pyright 会报错——所以 CI 里要跑类型检查。
|
||
|
||
**易错点 / 面试常问（补充）**
|
||
|
||
- 易错点：类型标注在运行时**不检查**，写错了程序照样跑；要靠 pyright / mypy 在编辑器和 CI 里检查。
|
||
- 易错点：`@dataclass` 的可变默认值（列表、字典）要写 `field(default_factory=list)`，直接写 `= []` 会报错（普通函数参数写 `def f(x=[])` 则会让所有调用共用同一个列表）。
|
||
- 易错点：pytest 的测试文件和函数必须以 `test_` 开头才会被发现；`assert` 失败时 pytest 会显示两边的值，不要写成 `assert x == y, "失败"` 后就不看输出。
|
||
- 面试常问：`Protocol` 和抽象基类 `ABC` 的区别（前者只看方法签名对不对得上，后者要显式继承）；pytest 的 fixture 是什么（可复用的准备 / 清理代码，按参数名注入）。
|
||
|
||
### 要建的文件
|
||
|
||
- `day2_typing.py`、`tests/test_day2.py`；`pyproject.toml` 里加：
|
||
|
||
```toml
|
||
[tool.pytest.ini_options]
|
||
pythonpath = ["."]
|
||
asyncio_mode = "auto"
|
||
```
|
||
|
||
### 运行方式
|
||
|
||
```bash
|
||
uv run pytest -v && uv run pyright day2_typing.py && uv run ruff check .
|
||
```
|
||
|
||
### 完成标准
|
||
|
||
- 测试通过；故意写错 `status`，pyright 能报出来。
|
||
|
||
---
|
||
|
||
## 第 3 天：异常、上下文管理器、文件和日志
|
||
|
||
### 学习内容
|
||
|
||
- 说明：
|
||
- `raise NewError(...) from err` 保留原始异常链（相当于 Go 的 `%w`）；自定义异常建一个基类，带上错误码，方便上层统一处理。
|
||
- `with` 语句（上下文管理器）保证资源被释放，相当于 Go 的 `defer`；`@contextmanager` 可以快速写一个。
|
||
- 写文件要“先写临时文件再改名”，防止写到一半崩溃导致文件损坏（第 12 周写 Agent 状态时是同一个道理）。
|
||
- 用 `logging` 而不是 `print`，生产环境才能控制级别和格式。
|
||
- 案例：
|
||
|
||
```python
|
||
# 文件：py-learning/day3_errors.py
|
||
import json
|
||
import logging
|
||
import os
|
||
import tempfile
|
||
import time
|
||
from contextlib import contextmanager
|
||
from pathlib import Path
|
||
|
||
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
|
||
log = logging.getLogger("day3")
|
||
|
||
|
||
class AppError(Exception):
|
||
code = "APP_ERROR"
|
||
|
||
|
||
class BadDataError(AppError):
|
||
code = "BAD_DATA"
|
||
|
||
|
||
@contextmanager
|
||
def timer(name: str):
|
||
start = time.perf_counter()
|
||
try:
|
||
yield
|
||
finally: # 即使中间抛异常也会执行
|
||
log.info("%s 用时 %.1f ms", name, (time.perf_counter() - start) * 1000)
|
||
|
||
|
||
def save_json_atomic(path: Path, data: object) -> None:
|
||
fd, tmp = tempfile.mkstemp(dir=path.parent, suffix=".tmp")
|
||
with os.fdopen(fd, "w", encoding="utf-8") as f:
|
||
json.dump(data, f, ensure_ascii=False) # ensure_ascii=False 才能正常写中文
|
||
os.replace(tmp, path) # 同一文件系统内改名是原子操作
|
||
|
||
|
||
def load_json(path: Path) -> list:
|
||
if not path.exists():
|
||
return []
|
||
try:
|
||
return json.loads(path.read_text(encoding="utf-8"))
|
||
except json.JSONDecodeError as err:
|
||
raise BadDataError(f"{path} 不是合法 JSON") from err
|
||
|
||
|
||
if __name__ == "__main__":
|
||
p = Path(tempfile.gettempdir()) / "tasks_demo.json"
|
||
with timer("保存"):
|
||
save_json_atomic(p, [{"id": 1, "title": "学 Python"}])
|
||
print(load_json(p))
|
||
p.write_text("{坏数据", encoding="utf-8")
|
||
try:
|
||
load_json(p)
|
||
except AppError as e:
|
||
log.warning("错误码 %s：%s（原因：%r）", e.code, e, e.__cause__)
|
||
```
|
||
|
||
**易错点 / 面试常问（补充）**
|
||
|
||
- 易错点：不要用裸 `except:` 或 `except Exception: pass` 吞掉异常；至少记录日志，能处理的才捕获。
|
||
- 易错点：打开文件要指定 `encoding="utf-8"`，否则在 Windows 上默认编码不同，中文会乱码。
|
||
- 易错点：日志用 `logging`，不要到处 `print`；库代码里只 `getLogger(__name__)`，日志格式和级别由程序入口统一配置。
|
||
- 面试常问：`try / except / else / finally` 各自什么时候执行；为什么“先写临时文件再 `os.replace`”能避免写一半崩溃导致文件损坏。
|
||
|
||
### 要建的文件
|
||
|
||
- `day3_errors.py`
|
||
|
||
### 运行方式
|
||
|
||
```bash
|
||
uv run python day3_errors.py
|
||
```
|
||
|
||
### 完成标准
|
||
|
||
- 能说清 `raise ... from` 保留了什么，`finally` 什么时候执行。
|
||
|
||
---
|
||
|
||
## 第 4 天：asyncio（本周最重要）
|
||
|
||
### 学习内容
|
||
|
||
- 说明：
|
||
- `async def` / `await` 写法和 JS 很像，但**调用 async 函数只得到协程对象，不会自动开始执行**，必须 `await` 或 `asyncio.create_task`。
|
||
- `asyncio.gather` ≈ `Promise.all`；`asyncio.timeout` 设置超时；`asyncio.Queue` ≈ Go 的 channel；`task.cancel()` 会在任务里抛 `CancelledError`，清理工作放 `finally`，**不要吞掉这个异常**。
|
||
- 异步生成器（`async def` + `yield`）配合 `async for`，是 SSE 流式输出的基础。
|
||
- 在 async 函数里调用阻塞代码（`time.sleep`、同步数据库驱动）会卡住整个事件循环，所有请求都变慢；实在要用就 `await asyncio.to_thread(...)`。
|
||
- 案例：模拟并发调用模型、超时、生产者消费者、取消流式输出。
|
||
|
||
```python
|
||
# 文件：py-learning/day4_async.py
|
||
import asyncio
|
||
import time
|
||
|
||
|
||
async def fake_llm(name: str, delay: float) -> str:
|
||
await asyncio.sleep(delay) # 模拟一次模型调用
|
||
return f"{name} 完成"
|
||
|
||
|
||
async def demo_gather() -> None:
|
||
start = time.perf_counter()
|
||
results = await asyncio.gather(fake_llm("A", 0.3), fake_llm("B", 0.3), fake_llm("C", 0.3))
|
||
print(results, f"并发总耗时约 {time.perf_counter() - start:.1f}s") # 约 0.3s 而不是 0.9s
|
||
|
||
|
||
async def demo_timeout() -> None:
|
||
try:
|
||
async with asyncio.timeout(0.1):
|
||
await fake_llm("慢模型", 1)
|
||
except TimeoutError:
|
||
print("超时了，走降级逻辑")
|
||
|
||
|
||
async def demo_queue() -> None:
|
||
q: asyncio.Queue[int] = asyncio.Queue()
|
||
|
||
async def worker(name: str) -> None:
|
||
while True:
|
||
n = await q.get()
|
||
await asyncio.sleep(0.05)
|
||
print(f"{name} 处理 {n}")
|
||
q.task_done()
|
||
|
||
workers = [asyncio.create_task(worker(f"w{i}")) for i in range(2)]
|
||
for n in range(5):
|
||
q.put_nowait(n)
|
||
await q.join() # 等所有任务处理完
|
||
for w in workers:
|
||
w.cancel()
|
||
await asyncio.gather(*workers, return_exceptions=True)
|
||
|
||
|
||
async def ticker():
|
||
i = 0
|
||
try:
|
||
while True:
|
||
await asyncio.sleep(0.05)
|
||
i += 1
|
||
yield i
|
||
finally:
|
||
print("流被关闭，清理资源") # 客户端断开时这里会执行
|
||
|
||
|
||
async def demo_cancel_stream() -> None:
|
||
async def consume():
|
||
async for n in ticker():
|
||
print("收到", n)
|
||
|
||
task = asyncio.create_task(consume())
|
||
await asyncio.sleep(0.18)
|
||
task.cancel()
|
||
try:
|
||
await task
|
||
except asyncio.CancelledError:
|
||
print("消费任务已取消")
|
||
|
||
|
||
async def main() -> None:
|
||
await demo_gather()
|
||
await demo_timeout()
|
||
await demo_queue()
|
||
await demo_cancel_stream()
|
||
print(await asyncio.to_thread(time.sleep, 0.05) or "阻塞调用放到线程里执行")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|
||
```
|
||
|
||
| JS / Go | Python asyncio |
|
||
| --- | --- |
|
||
| `Promise.all` | `asyncio.gather` |
|
||
| `setTimeout` / `time.Sleep` | `await asyncio.sleep` |
|
||
| `go f()` | `asyncio.create_task(f())` |
|
||
| channel | `asyncio.Queue` |
|
||
| `context` 取消 | `task.cancel()` |
|
||
| `AbortController` 超时 | `asyncio.timeout()` |
|
||
|
||
- 面试常问：协程、线程、进程的区别；GIL 是什么、为什么 IO 密集型任务用 asyncio 很合适而 CPU 密集型要用多进程。
|
||
|
||
### 要建的文件
|
||
|
||
- `day4_async.py`
|
||
|
||
### 运行方式
|
||
|
||
```bash
|
||
uv run python day4_async.py
|
||
```
|
||
|
||
### 完成标准
|
||
|
||
- 能说清“调用 async 函数但没 await”会怎样，以及为什么不能在 async 函数里用 `time.sleep`。
|
||
|
||
---
|
||
|
||
## 第 5 天：pydantic 和配置
|
||
|
||
### 学习内容
|
||
|
||
- 说明：pydantic 是带运行时校验的数据模型（作用像前端的 zod）。外部进来的数据——请求体、配置、**大模型输出的 JSON**——都要先过校验。v2 常用：`BaseModel`、`Field` 约束、`model_validate` / `model_validate_json`、`model_dump`、`model_validator`。`pydantic-settings` 从环境变量和 `.env` 读配置。
|
||
- 案例：解析“协调者”模型输出的决定（模型常常会在 JSON 外包一层代码块或多说几句话）。
|
||
|
||
```python
|
||
# 文件：py-learning/day5_models.py
|
||
import re
|
||
from typing import Literal
|
||
|
||
from pydantic import BaseModel, Field, ValidationError, model_validator
|
||
from pydantic_settings import BaseSettings, SettingsConfigDict
|
||
|
||
|
||
class Decision(BaseModel):
|
||
action: Literal["dispatch", "finish", "ask_human"]
|
||
target: Literal["proposer", "critic"] | None = None
|
||
instruction: str = ""
|
||
reason: str = Field(min_length=1)
|
||
|
||
@model_validator(mode="after")
|
||
def check_target(self):
|
||
if self.action == "dispatch" and self.target is None:
|
||
raise ValueError("dispatch 必须指定 target")
|
||
return self
|
||
|
||
|
||
class BadModelOutput(Exception):
|
||
pass
|
||
|
||
|
||
def parse_decision(raw: str) -> Decision:
|
||
m = re.search(r"\{.*\}", raw, re.S) # 取出第一个 { 到最后一个 } 之间的内容
|
||
if not m:
|
||
raise BadModelOutput(f"没有找到 JSON：{raw[:200]}")
|
||
try:
|
||
return Decision.model_validate_json(m.group(0))
|
||
except ValidationError as e:
|
||
raise BadModelOutput(f"格式不对：{e.errors()[0]['msg']}") from e
|
||
|
||
|
||
class Settings(BaseSettings):
|
||
model_config = SettingsConfigDict(env_file=".env", extra="ignore")
|
||
llm_base_url: str = "https://api.example.com/v1"
|
||
llm_api_key: str = ""
|
||
llm_model: str = "demo-model"
|
||
llm_timeout_s: float = 30
|
||
|
||
|
||
if __name__ == "__main__":
|
||
print(parse_decision('好的：```json\n{"action":"dispatch","target":"critic","reason":"需要评审"}\n```'))
|
||
print(Settings().model_dump(exclude={"llm_api_key"})) # 打印配置时不要输出密钥
|
||
```
|
||
|
||
```python
|
||
# 文件：py-learning/tests/test_day5.py
|
||
import pytest
|
||
|
||
from day5_models import BadModelOutput, parse_decision
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"raw, ok",
|
||
[
|
||
('{"action":"finish","reason":"已达成共识"}', True),
|
||
('```json\n{"action":"dispatch","target":"proposer","reason":"先出方案"}\n```', True),
|
||
('我认为：{"action":"finish","reason":"够了"} 以上。', True),
|
||
('{"action":"dispatch","reason":"缺 target"}', False),
|
||
('{"action":"run","reason":"拼错了"}', False),
|
||
("完全不是 JSON", False),
|
||
],
|
||
)
|
||
def test_parse_decision(raw, ok):
|
||
if ok:
|
||
assert parse_decision(raw).reason
|
||
else:
|
||
with pytest.raises(BadModelOutput):
|
||
parse_decision(raw)
|
||
```
|
||
|
||
**易错点 / 面试常问（补充）**
|
||
|
||
- 易错点：pydantic v2 的方法名和 v1 不同：`model_validate`、`model_dump`、`model_dump_json`（v1 是 `parse_obj`、`dict`、`json`）。网上很多旧教程还是 v1 写法。
|
||
- 易错点：`BaseSettings` 在 v2 里拆到了单独的包 `pydantic-settings`，要单独安装。
|
||
- 易错点：大模型返回的 JSON 经常带解释文字或代码块标记，解析前先提取 JSON 部分，解析失败要有兜底，不要直接让接口 500。
|
||
- 面试常问：pydantic 和 dataclass 的区别（前者会校验和转换类型，适合处理外部输入）；配置为什么放环境变量（12-Factor：同一份代码在不同环境只换配置）。
|
||
|
||
### 要建的文件
|
||
|
||
- `day5_models.py`、`tests/test_day5.py`
|
||
|
||
### 运行方式
|
||
|
||
```bash
|
||
uv run pytest tests/test_day5.py -v
|
||
```
|
||
|
||
### 完成标准
|
||
|
||
- 6 个用例全部按预期；能说清“模型输出为什么必须当成不可信输入来校验”。
|
||
|
||
---
|
||
|
||
## 第 6 天：httpx 异步请求、重试、测试替身
|
||
|
||
### 学习内容
|
||
|
||
- 说明：
|
||
- `httpx.AsyncClient` 支持 `base_url`、`timeout`、连接复用，用 `async with` 管理生命周期。
|
||
- 哪些错误重试：超时、连接失败、429、5xx（指数退避：每次等待时间翻倍，加一点随机）；哪些不重试：400、401、403（重试也不会好）。
|
||
- `httpx.MockTransport` 让测试不真正发请求，可以模拟任何响应。
|
||
- 案例：
|
||
|
||
```python
|
||
# 文件：py-learning/day6_http.py
|
||
import asyncio
|
||
import random
|
||
|
||
import httpx
|
||
|
||
|
||
class HttpError(Exception):
|
||
def __init__(self, code: str, message: str, retryable: bool):
|
||
super().__init__(message)
|
||
self.code, self.retryable = code, retryable
|
||
|
||
|
||
def classify(resp: httpx.Response) -> HttpError:
|
||
s = resp.status_code
|
||
if s in (401, 403):
|
||
return HttpError("AUTH", "鉴权失败", False)
|
||
if s == 429:
|
||
return HttpError("RATE_LIMITED", "请求太频繁", True)
|
||
if s >= 500:
|
||
return HttpError("UPSTREAM", f"上游错误 {s}", True)
|
||
return HttpError("BAD_REQUEST", f"请求错误 {s}", False)
|
||
|
||
|
||
async def get_json(client: httpx.AsyncClient, url: str, retries: int = 2, base_delay: float = 0.5) -> dict:
|
||
last: HttpError | None = None
|
||
for attempt in range(retries + 1):
|
||
try:
|
||
resp = await client.get(url)
|
||
if resp.status_code == 200:
|
||
return resp.json()
|
||
last = classify(resp)
|
||
except httpx.TimeoutException:
|
||
last = HttpError("TIMEOUT", "请求超时", True)
|
||
except httpx.ConnectError:
|
||
last = HttpError("CONNECT", "连接失败", True)
|
||
if not last.retryable or attempt == retries:
|
||
break
|
||
await asyncio.sleep(base_delay * 2**attempt + random.uniform(0, base_delay))
|
||
assert last is not None
|
||
raise last
|
||
```
|
||
|
||
```python
|
||
# 文件：py-learning/tests/test_day6.py
|
||
import httpx
|
||
import pytest
|
||
|
||
from day6_http import HttpError, get_json
|
||
|
||
|
||
def client_with(handler) -> httpx.AsyncClient:
|
||
return httpx.AsyncClient(transport=httpx.MockTransport(handler), base_url="https://test")
|
||
|
||
|
||
async def test_retry_then_success():
|
||
calls = {"n": 0}
|
||
|
||
def handler(request: httpx.Request) -> httpx.Response:
|
||
calls["n"] += 1
|
||
return httpx.Response(503) if calls["n"] < 3 else httpx.Response(200, json={"ok": True})
|
||
|
||
async with client_with(handler) as c:
|
||
assert await get_json(c, "/x", base_delay=0) == {"ok": True}
|
||
assert calls["n"] == 3
|
||
|
||
|
||
async def test_auth_not_retried():
|
||
calls = {"n": 0}
|
||
|
||
def handler(request):
|
||
calls["n"] += 1
|
||
return httpx.Response(401)
|
||
|
||
async with client_with(handler) as c:
|
||
with pytest.raises(HttpError) as e:
|
||
await get_json(c, "/x", base_delay=0)
|
||
assert e.value.code == "AUTH" and calls["n"] == 1
|
||
|
||
|
||
async def test_timeout():
|
||
def handler(request):
|
||
raise httpx.ReadTimeout("slow", request=request)
|
||
|
||
async with client_with(handler) as c:
|
||
with pytest.raises(HttpError) as e:
|
||
await get_json(c, "/x", retries=1, base_delay=0)
|
||
assert e.value.code == "TIMEOUT"
|
||
```
|
||
|
||
**易错点 / 面试常问（补充）**
|
||
|
||
- 易错点：`httpx.AsyncClient` 要复用（服务启动时创建、关闭时 `aclose`），不要每次请求都新建，否则连接无法复用、性能差。
|
||
- 易错点：一定要设置超时；httpx 默认超时是 5 秒，调用大模型这类慢接口要按需调大，并区分连接超时和读取超时。
|
||
- 易错点：只对“可重试”的错误重试（超时、429、5xx），4xx 参数错误重试没有意义；重试要有次数上限和指数退避。
|
||
- 面试常问：重试为什么可能导致重复写入（要配合幂等键）；什么是测试替身（mock / fake / stub），为什么测试不能依赖真实外部服务。
|
||
|
||
### 要建的文件
|
||
|
||
- `day6_http.py`、`tests/test_day6.py`
|
||
|
||
### 运行方式
|
||
|
||
```bash
|
||
uv run pytest tests/test_day6.py -v
|
||
```
|
||
|
||
### 完成标准
|
||
|
||
- 测试通过；能讲清重试策略和指数退避。
|
||
|
||
---
|
||
|
||
## 第 7 天：FastAPI（Uvicorn、依赖注入、中间件、鉴权、生命周期、SSE）
|
||
|
||
### 学习内容
|
||
|
||
- 说明：
|
||
- **FastAPI** 基于类型标注自动做参数解析、校验和接口文档（`/docs`）；**Uvicorn** 是运行它的 ASGI 服务器（相当于 Go 里的 `http.Server`）。Flask 是更老、更简单的同步框架，概念相通。
|
||
- **依赖注入** `Depends`：把“取当前用户”“拿数据库连接”这类公共逻辑写成函数，接口参数里声明即可，测试时可以替换（`app.dependency_overrides`）。
|
||
- **中间件**：和 Gin 一样包在请求外层，常用于日志、耗时、请求 ID。
|
||
- **鉴权**：用 `HTTPBearer` 取令牌，用 PyJWT 校验，失败抛 `HTTPException(401)`。
|
||
- **lifespan**：服务启动和关闭时执行的代码（创建 / 关闭连接池、启动后台任务）。
|
||
- **SSE**：响应类型 `text/event-stream`，每条消息是 `id: ...\nevent: ...\ndata: ...\n\n`；浏览器 `EventSource` 断线会自动重连并带上 `Last-Event-ID`。下面先手写格式理解协议；新版 FastAPI 自带 `fastapi.sse.EventSourceResponse`（我验证时的 0.161 版本已有），用法以官方文档为准。
|
||
- **后台任务和连接分开**：任务用 `asyncio.create_task` 在后台跑，SSE 只负责“读进度”，关掉页面任务也不停（这就是第 12 周 Agent 项目的核心思路）。
|
||
- 案例：
|
||
|
||
```python
|
||
# 文件：py-learning/app/__init__.py
|
||
```
|
||
|
||
```python
|
||
# 文件：py-learning/app/main.py
|
||
import asyncio
|
||
import json
|
||
import os
|
||
import time
|
||
import uuid
|
||
from contextlib import asynccontextmanager
|
||
from typing import Annotated
|
||
|
||
import jwt
|
||
from fastapi import Depends, FastAPI, HTTPException, Request
|
||
from fastapi.responses import StreamingResponse
|
||
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
||
from pydantic import BaseModel, Field
|
||
|
||
SECRET = os.environ.get("JWT_SECRET", "dev-only-insecure-secret-change-me-32b") # 生产环境必须通过环境变量设置
|
||
jobs: dict[str, dict] = {}
|
||
background: set[asyncio.Task] = set() # 保存任务引用，防止被垃圾回收
|
||
|
||
|
||
@asynccontextmanager
|
||
async def lifespan(app: FastAPI):
|
||
print("启动：创建连接池等")
|
||
yield
|
||
for t in background:
|
||
t.cancel()
|
||
print("关闭：清理资源")
|
||
|
||
|
||
app = FastAPI(lifespan=lifespan)
|
||
|
||
|
||
@app.middleware("http")
|
||
async def timing(request: Request, call_next):
|
||
start = time.perf_counter()
|
||
response = await call_next(request)
|
||
response.headers["X-Cost-Ms"] = f"{(time.perf_counter() - start) * 1000:.1f}"
|
||
return response
|
||
|
||
|
||
bearer = HTTPBearer(auto_error=False)
|
||
|
||
|
||
def current_user(cred: Annotated[HTTPAuthorizationCredentials | None, Depends(bearer)]) -> str:
|
||
if cred is None:
|
||
raise HTTPException(401, "未登录")
|
||
try:
|
||
return jwt.decode(cred.credentials, SECRET, algorithms=["HS256"])["sub"]
|
||
except jwt.PyJWTError:
|
||
raise HTTPException(401, "令牌无效或已过期")
|
||
|
||
|
||
class JobIn(BaseModel):
|
||
steps: int = Field(default=5, ge=1, le=20)
|
||
|
||
|
||
@app.get("/health")
|
||
def health():
|
||
return {"status": "ok"}
|
||
|
||
|
||
@app.post("/jobs", status_code=201)
|
||
async def create_job(body: JobIn, user: Annotated[str, Depends(current_user)]):
|
||
job_id = uuid.uuid4().hex[:8]
|
||
jobs[job_id] = {"owner": user, "events": [], "done": False}
|
||
|
||
async def run():
|
||
for i in range(1, body.steps + 1):
|
||
await asyncio.sleep(0.05) # 模拟每一步的工作
|
||
jobs[job_id]["events"].append({"step": i})
|
||
jobs[job_id]["done"] = True
|
||
|
||
task = asyncio.create_task(run())
|
||
background.add(task)
|
||
task.add_done_callback(background.discard)
|
||
return {"job_id": job_id}
|
||
|
||
|
||
@app.get("/jobs/{job_id}/events")
|
||
async def job_events(job_id: str, request: Request):
|
||
if job_id not in jobs:
|
||
raise HTTPException(404, "任务不存在")
|
||
last = int(request.headers.get("last-event-id", 0)) # 断线重连时从这里继续
|
||
|
||
async def stream():
|
||
nonlocal last
|
||
while True:
|
||
if await request.is_disconnected():
|
||
break # 客户端断开只影响这条连接，后台任务继续跑
|
||
events = jobs[job_id]["events"]
|
||
while last < len(events):
|
||
last += 1
|
||
yield f"id: {last}\nevent: progress\ndata: {json.dumps(events[last - 1])}\n\n"
|
||
if jobs[job_id]["done"]:
|
||
yield "event: end\ndata: {}\n\n"
|
||
break
|
||
await asyncio.sleep(0.05)
|
||
|
||
return StreamingResponse(stream(), media_type="text/event-stream", headers={"Cache-Control": "no-cache"})
|
||
```
|
||
|
||
```python
|
||
# 文件：py-learning/tests/test_app.py
|
||
import time
|
||
|
||
import jwt
|
||
from fastapi.testclient import TestClient
|
||
|
||
from app.main import SECRET, app
|
||
|
||
|
||
def token(sub: str = "u1") -> str:
|
||
return jwt.encode({"sub": sub, "exp": int(time.time()) + 60}, SECRET, algorithm="HS256")
|
||
|
||
|
||
def test_auth_and_sse():
|
||
with TestClient(app) as client: # with 会触发 lifespan
|
||
assert client.post("/jobs", json={"steps": 3}).status_code == 401
|
||
r = client.post("/jobs", json={"steps": 3}, headers={"Authorization": f"Bearer {token()}"})
|
||
assert r.status_code == 201
|
||
job_id = r.json()["job_id"]
|
||
with client.stream("GET", f"/jobs/{job_id}/events") as s:
|
||
body = "".join(s.iter_text())
|
||
assert body.count("event: progress") == 3 and "event: end" in body
|
||
with client.stream("GET", f"/jobs/{job_id}/events", headers={"Last-Event-ID": "2"}) as s:
|
||
body = "".join(s.iter_text())
|
||
assert body.count("event: progress") == 1 # 只补发第 3 条
|
||
```
|
||
|
||
**易错点 / 面试常问（补充）**
|
||
|
||
- 易错点：在 `async def` 接口里调用同步阻塞的库（如 `requests`、同步数据库驱动、`time.sleep`）会卡住整个事件循环；要么换异步库，要么把接口写成普通 `def`（FastAPI 会放到线程池执行）。
|
||
- 易错点：`BackgroundTasks` 适合发邮件这类小任务，进程重启就丢；长任务要用持久化的作业表或队列（第 12 周会做）。
|
||
- 易错点：用 `TestClient` 测试有 lifespan 的应用时，要写成 `with TestClient(app) as client:`，否则启动和关闭代码不会执行。
|
||
- 面试常问：FastAPI 为什么快（基于 Starlette 和异步 I/O，Uvicorn 是 ASGI 服务器）；依赖注入有什么好处（复用、可测试、可替换）；SSE 和 WebSocket 的区别（SSE 单向、基于普通 HTTP、浏览器自动重连）。
|
||
- 文档：FastAPI 更新较快，`fastapi.sse` 等新功能的用法以官方文档为准：https://fastapi.tiangolo.com/zh/
|
||
|
||
### 要建的文件
|
||
|
||
- `app/__init__.py`、`app/main.py`、`tests/test_app.py`
|
||
|
||
### 运行方式
|
||
|
||
```bash
|
||
uv run uvicorn app.main:app --reload # 打开 http://localhost:8000/docs
|
||
uv run pytest -v
|
||
```
|
||
|
||
### 完成标准
|
||
|
||
- 测试通过；在浏览器控制台用 `new EventSource(...)` 订阅（`EventSource` 不能带请求头，练习时可以临时去掉鉴权或改用 `fetch` 读流），关掉页面后再打开，能补齐事件。
|
||
- 提交推送；在总纲进度记录里勾掉第 4 周。
|
||
|
||
---
|
||
|
||
<!-- week-topics:start -->
|
||
|
||
## 本周相关专题
|
||
|
||
> 本节由 `tools/gen_week_topics.py` 根据专题文件的 `week` 字段自动生成，请勿手改；想调整就修改对应专题文件的 frontmatter 后重新生成。
|
||
|
||
- **AI 编程练习（每周）**：本周至少一个功能用 AI 编程工具（Cursor / Claude Code / Codex 等）完成，按 [AI 编程：工作流](../topics/ai/ai-coding/01-workflow-rules-skills.md) 写清需求和规则，按 [验证与证据](../topics/ai/ai-coding/02-verify-evidence.md) 保存测试结果和会话记录（脱敏后），第 15 周整理成“AI 编程 + 验证方法”的应聘案例。
|
||
|
||
| 技术栈 | 课程 / 场景 / 面试题 |
|
||
| --- | --- |
|
||
| [asyncio 异步编程](../topics/python/asyncio/index.md) | 课程：[协程、任务和事件循环](../topics/python/asyncio/01-coroutines-tasks.md)；课程：[并发控制](../topics/python/asyncio/02-concurrency-control.md)；[面试题](../topics/python/asyncio/interview.md) |
|
||
| [FastAPI](../topics/python/fastapi/index.md) | 课程：[路由和参数校验](../topics/python/fastapi/01-routing-validation.md)；课程：[依赖注入、鉴权、中间件和统一错误处理](../topics/python/fastapi/02-deps-auth-errors.md)；[面试题](../topics/python/fastapi/interview.md) |
|
||
| [Python 语言](../topics/python/python/index.md) | 课程：[给 JS 开发者的 Python](../topics/python/python/01-syntax-for-js.md)；课程：[类型标注、dataclass、异常和上下文管理器、pytest](../topics/python/python/02-typing-errors-testing.md)；课程：[工程化](../topics/python/python/03-project-packaging.md)；场景：[订单对账脚本](../topics/python/python/scenarios/order-reconciliation.md)；[面试题](../topics/python/python/interview.md) |
|
||
|
||
<!-- week-topics:end -->
|
||
|
||
## 本周自查清单
|
||
|
||
- [ ] 会用 uv，知道 pip / venv / poetry 的对应关系
|
||
- [ ] 熟悉推导式、解包、f-string、可变默认参数的坑
|
||
- [ ] 会写类型标注、dataclass、Protocol，会用 pyright、ruff、pytest
|
||
- [ ] 会用自定义异常、`raise from`、上下文管理器、logging
|
||
- [ ] 理解 asyncio：`create_task`、`gather`、超时、取消、`Queue`、异步生成器、`to_thread`；能讲 GIL
|
||
- [ ] 会用 pydantic 校验外部数据和模型输出，会用 pydantic-settings
|
||
- [ ] 会用 httpx 异步请求、分类重试、`MockTransport` 测试
|
||
- [ ] 会写 FastAPI：依赖注入、中间件、JWT 鉴权、lifespan、SSE、后台任务
|