新增第 4 周 Python 速成+FastAPI 和第 7 周前端工程化查漏（Vite/Webpack/TS/Pinia/Router/SSR/ESLint/pnpm monorepo/Vitest），补充易错点和面试常问
This commit is contained in:
Shikui Wang committed
1 parent
b79280e6d4
commit
27c1965070
3 files changed
+2515
-233
No files matched your search
@@ -0,0 +1,862 @@
|
||||
# 第 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 周。
|
||||
|
||||
---
|
||||
|
||||
## 本周自查清单
|
||||
|
||||
- [ ] 会用 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、后台任务
|
||||
@@ -1,233 +0,0 @@
|
||||
# 转岗计划 · 第 6 周学习计划（Python 速成：给 JS 开发者）
|
||||
|
||||
> 所属阶段：第二段“Python 与 AI 应用”的第一周。Python 是求职的**主语言**，后面的 RAG 项目（第 7~8 周）和多 Agent 项目（第 11~14 周）都用它。总纲见同目录 `README.md`，下周计划见 `rag-project.md`。
|
||||
|
||||
目标：一周内把写 AI 应用服务需要的 Python 补齐：基础语法、类型标注、asyncio、pydantic、httpx、pytest、FastAPI 和 SSE。不追求全面，只学项目里马上会用到的。
|
||||
|
||||
学法：你有 7 年 JS 和一个月 Go 的底子，每个知识点都先想“JS / Go 里对应的是什么”，再看 Python 的写法。每天的练习文件第一段写一张小对照表。
|
||||
|
||||
本周只学 Python，不开 RAG 和 Agent 项目（戒律：同一周不同时开新框架和新功能）。招聘里 FastAPI 和 Flask 都常见，本计划用 FastAPI；Flask 概念更少，会了 FastAPI 再看 Flask 半天就能上手。
|
||||
|
||||
参考资料：
|
||||
- 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/
|
||||
- FastAPI（中文）：https://fastapi.tiangolo.com/zh/
|
||||
- httpx：https://www.python-httpx.org/
|
||||
|
||||
---
|
||||
|
||||
## 最终项目结构
|
||||
|
||||
```
|
||||
py-learning/
|
||||
├── pyproject.toml # uv 管理依赖
|
||||
├── .python-version
|
||||
├── day1_basics.py
|
||||
├── day2_typing.py
|
||||
├── day3_errors_files.py
|
||||
├── day4_async.py
|
||||
├── day5_models.py
|
||||
├── day6_http.py
|
||||
├── fastapi_demo/
|
||||
│ ├── __init__.py
|
||||
│ ├── main.py # FastAPI 应用
|
||||
│ └── static/
|
||||
│ └── sse.html # 用 EventSource 订阅 SSE 的测试页
|
||||
└── tests/
|
||||
├── test_day2.py
|
||||
├── test_day5.py
|
||||
├── test_day6.py
|
||||
└── test_fastapi_demo.py
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 第 1 天：环境和基础语法
|
||||
|
||||
**学习内容**
|
||||
- 用 uv 管理 Python 版本和依赖（相当于 `nvm + npm` 合在一起）：`curl -LsSf https://astral.sh/uv/install.sh | sh`。
|
||||
- 缩进代替大括号；`None` / `True` / `False`；`list`、`dict`、`tuple`、`set`。
|
||||
- f-string（相当于模板字符串）、切片 `a[1:3]`、解包 `a, *rest = items`。
|
||||
- 列表推导式和字典推导式（相当于 `map` + `filter`）。
|
||||
- 函数：默认参数、关键字参数、`*args`、`**kwargs`；默认参数不要用可变对象（`def f(x=[])` 是经典坑）。
|
||||
- `if __name__ == "__main__":` 的作用。
|
||||
|
||||
**要建的文件**
|
||||
- `py-learning/`：运行 `uv init py-learning && cd py-learning && uv python pin 3.12`，生成 `pyproject.toml` 和 `.python-version`。
|
||||
- `day1_basics.py`：
|
||||
- 文件开头用注释写一张 JS ↔ Python 对照表（至少 10 行：`const/let`、`===`、`null/undefined`、`arr.map`、`arr.filter`、对象展开、解构、模板字符串、`for...of`、箭头函数）。
|
||||
- `def word_count(text: str) -> dict[str, int]`：和第 1 周 Go 的 `WordCount` 做同一件事，对比代码长短。
|
||||
- `def unique(items: list[str]) -> list[str]`：去重并保持原顺序。
|
||||
- 演示可变默认参数的坑和正确写法（`def f(x=None)`）。
|
||||
|
||||
**运行方式**：`uv run python day1_basics.py`
|
||||
|
||||
**完成标准**：能说清楚 Python 的 `list` 和 `tuple` 区别，以及为什么默认参数不能用 `[]`。
|
||||
|
||||
---
|
||||
|
||||
## 第 2 天：类型标注、dataclass 和 Protocol
|
||||
|
||||
**学习内容**
|
||||
- 类型标注只是提示，运行时不检查（和 TS 一样），要靠检查工具：`uv add --dev pyright ruff pytest`。
|
||||
- `list[str]`、`dict[str, int]`、`X | None`、`Literal["a", "b"]`、`TypedDict`。
|
||||
- `@dataclass`：相当于带构造函数的结构体。
|
||||
- `Protocol`：Python 的“隐式接口”，和 Go 的 `interface` 思路一样，Agent 项目里 LLM 客户端和存储层都会用它。
|
||||
- 类、`self`、`@property`、`@classmethod`。
|
||||
|
||||
**要建的文件**
|
||||
- `day2_typing.py`：
|
||||
- `TaskStatus = Literal["todo", "doing", "done"]`
|
||||
- `@dataclass class Task`：`id: int`、`title: str`、`status: TaskStatus = "todo"`、`tags: list[str] = field(default_factory=list)`。
|
||||
- `class TaskStore(Protocol)`：`add(self, task: Task) -> None`、`list(self) -> list[Task]`。
|
||||
- `class MemoryTaskStore`：不继承 `TaskStore`，但实现同样的方法（体会“隐式实现”）。
|
||||
- `def count_by_status(tasks: list[Task]) -> dict[TaskStatus, int]`
|
||||
- `tests/test_day2.py`：测试 `count_by_status`；用 `pytest.mark.parametrize` 写表格驱动测试（对应 Go 的表格驱动测试）。
|
||||
|
||||
**运行方式**：`uv run pytest -v`、`uv run pyright day2_typing.py`、`uv run ruff check .`
|
||||
|
||||
**完成标准**：故意把 `status` 写成 `"deleted"`，pyright 能报错；能说清楚 `Protocol` 和 Go 接口的相同点。
|
||||
|
||||
---
|
||||
|
||||
## 第 3 天：异常、上下文管理器和文件
|
||||
|
||||
**学习内容**
|
||||
- `try / except / else / finally`，`raise ... from err`（相当于 Go 的 `%w` 包装错误）。
|
||||
- 自定义异常层级：一个基类，下面分具体错误。
|
||||
- `with` 语句（上下文管理器）：自动关闭文件、连接，相当于 Go 的 `defer`。
|
||||
- `pathlib.Path`、`json.loads` / `json.dumps`（注意 `ensure_ascii=False` 才能正常输出中文）。
|
||||
- `logging` 基本用法。
|
||||
|
||||
**要建的文件**
|
||||
- `day3_errors_files.py`：
|
||||
- `class AppError(Exception)`，带 `code: str` 属性；子类 `NotFoundError`、`ValidationError`。
|
||||
- `def load_tasks(path: Path) -> list[dict]`：文件不存在返回空列表；JSON 格式错误时 `raise ValidationError(...) from err`。
|
||||
- `def save_tasks(path: Path, tasks: list[dict]) -> None`：先写临时文件再 `replace`，避免写到一半崩溃导致文件损坏（Agent 项目里“写状态要原子”就是这个道理）。
|
||||
- 用 `contextlib.contextmanager` 写一个 `timer(name)`，在 `with timer("load"):` 结束时打印耗时。
|
||||
|
||||
**运行方式**：`uv run python day3_errors_files.py`
|
||||
|
||||
**完成标准**：能说清楚 `raise X from err` 保留了什么信息，以及“先写临时文件再改名”为什么更安全。
|
||||
|
||||
---
|
||||
|
||||
## 第 4 天：asyncio（本周最重要的一天）
|
||||
|
||||
**学习内容**
|
||||
- `async def` / `await` 和 JS 很像，但要有事件循环：入口用 `asyncio.run(main())`。
|
||||
- 和 JS 的区别：调用 `async` 函数只得到协程对象，**不会自动开始执行**，要 `await` 或 `asyncio.create_task`。
|
||||
- `asyncio.gather`（相当于 `Promise.all`）、`asyncio.wait_for` / `asyncio.timeout`（超时）、`asyncio.Queue`（相当于 Go 的 channel）、`asyncio.Event`。
|
||||
- 取消：`task.cancel()` 会在任务里抛出 `CancelledError`，要在 `finally` 里清理，**不要吞掉这个异常**。
|
||||
- 异步生成器 `async def ... yield`，以及 `async for`：SSE 推送就是靠它。
|
||||
- 不能在 `async` 函数里调用会阻塞的同步代码（比如 `time.sleep`、同步的数据库驱动），否则整个服务卡住；实在要用就 `await asyncio.to_thread(...)`。
|
||||
|
||||
**要建的文件**
|
||||
- `day4_async.py`：
|
||||
- `async def fake_llm(name: str, delay: float) -> str`：`await asyncio.sleep(delay)` 后返回结果，模拟一次模型调用。
|
||||
- `async def run_parallel()`：用 `gather` 同时调 3 个 `fake_llm`，打印总耗时，和顺序调用对比。
|
||||
- `async def run_with_timeout()`：给一个 3 秒的调用设 1 秒超时，捕获 `TimeoutError` 并打印。
|
||||
- `async def producer_consumer()`：一个生产者往 `Queue` 放 5 个任务，两个消费者处理，全部处理完后退出（`queue.join()`）。
|
||||
- `async def ticker(n: int)`：异步生成器，每 0.5 秒 `yield` 一个数字；用 `async for` 消费，消费到第 3 个时取消，观察 `finally` 是否执行。
|
||||
- 文件开头写对照表：`Promise.all` ↔ `gather`、`setTimeout` ↔ `asyncio.sleep`、Go 的 `go f()` ↔ `create_task`、Go 的 channel ↔ `Queue`、Go 的 `context` 取消 ↔ `task.cancel()`。
|
||||
|
||||
**运行方式**：`uv run python day4_async.py`
|
||||
|
||||
**完成标准**：能说清楚“调用 async 函数但没 await”会发生什么；能解释为什么在 async 服务里用同步阻塞调用会拖垮所有请求。
|
||||
|
||||
---
|
||||
|
||||
## 第 5 天：pydantic 数据模型
|
||||
|
||||
**学习内容**
|
||||
- pydantic 是“带运行时校验的类型”，作用类似前端的 zod：外部进来的数据（请求体、模型输出的 JSON）都要先过一遍校验。
|
||||
- `BaseModel`、`Field(min_length=..., max_length=...)`、默认值、`model_validate`、`model_validate_json`、`model_dump`、`model_dump_json`。
|
||||
- `Literal` 字段和“按某个字段区分类型”的联合类型（discriminated union）。
|
||||
- `ValidationError` 里能拿到哪个字段、为什么错。
|
||||
- `pydantic-settings`：从环境变量和 `.env` 读配置。
|
||||
|
||||
**要建的文件**
|
||||
- `uv add pydantic pydantic-settings`
|
||||
- `day5_models.py`：
|
||||
- `class Decision(BaseModel)`：`action: Literal["dispatch", "finish", "ask_human"]`、`target: str | None = None`、`instruction: str = ""`、`reason: str`。（这就是第 11~12 周多 Agent 项目里协调者输出的雏形）
|
||||
- `def parse_decision(raw: str) -> Decision`：模型常常会在 JSON 外面包一层 ```` ```json ```` 代码块或多说几句话，先提取出第一个 `{...}` 再 `model_validate_json`；失败时抛出自定义的 `BadModelOutput`，错误信息里带上原始文本的前 200 个字。
|
||||
- `class Settings(BaseSettings)`：`llm_base_url: str`、`llm_api_key: str = ""`、`llm_model: str`、`llm_timeout_s: float = 30`，从 `.env` 读取。
|
||||
- `tests/test_day5.py`：参数化测试 `parse_decision`：正常 JSON、带代码块的 JSON、前后有废话的 JSON、`action` 拼错、缺少 `reason`、完全不是 JSON。
|
||||
|
||||
**运行方式**：`uv run pytest tests/test_day5.py -v`
|
||||
|
||||
**完成标准**：6 种输入都按预期通过或报错；能说清楚“为什么模型输出必须当成不可信的外部输入来校验”。
|
||||
|
||||
---
|
||||
|
||||
## 第 6 天：httpx 异步请求和测试
|
||||
|
||||
**学习内容**
|
||||
- `httpx.AsyncClient`：设置 `base_url`、`timeout`、请求头；用 `async with` 管理生命周期，复用连接。
|
||||
- 状态码和异常：`raise_for_status()`、`httpx.TimeoutException`、`httpx.ConnectError`。
|
||||
- 测试时不真的发请求：`httpx.MockTransport` 可以按请求返回任意响应，用来模拟超时、429、500。
|
||||
- pytest 跑异步测试：`uv add --dev pytest-asyncio`，在 `pyproject.toml` 里设置 `asyncio_mode = "auto"`。
|
||||
|
||||
**要建的文件**
|
||||
- `uv add httpx`
|
||||
- `day6_http.py`：
|
||||
- `class HttpError(Exception)`，带 `code` 和 `retryable: bool`。
|
||||
- `async def get_json(client: httpx.AsyncClient, url: str, retries: int = 2) -> dict`：超时和 5xx、429 时等待后重试（等待时间逐次翻倍），401 / 400 不重试，最后抛出带错误码的 `HttpError`。
|
||||
- `main()` 里请求一个公开接口（例如 `https://httpbin.org/json`）打印结果。
|
||||
- `tests/test_day6.py`，用 `MockTransport` 测试：
|
||||
- 200 直接返回。
|
||||
- 前两次 503、第三次 200，最终成功，并断言一共请求了 3 次。
|
||||
- 401 不重试，直接抛 `code == "AUTH"`。
|
||||
- 一直超时，最终抛 `code == "TIMEOUT"`。
|
||||
|
||||
**运行方式**：`uv run python day6_http.py`、`uv run pytest tests/test_day6.py -v`
|
||||
|
||||
**完成标准**：四个测试都通过；能说清楚哪些错误值得重试、哪些不值得。
|
||||
|
||||
---
|
||||
|
||||
## 第 7 天：FastAPI 基础和 SSE
|
||||
|
||||
**学习内容**
|
||||
- FastAPI 路由、路径参数、查询参数，请求体直接用 pydantic 模型（对照第 2 周的 Gin）。
|
||||
- 依赖注入 `Depends`。
|
||||
- `lifespan`：服务启动和关闭时执行的代码（第 13 周用它在启动时恢复未完成的任务）。
|
||||
- 后台任务：用 `asyncio.create_task` 启动，**要把任务对象保存起来**，否则可能被垃圾回收。
|
||||
- SSE 格式：`id: 1\nevent: progress\ndata: {...}\n\n`；用 `StreamingResponse(生成器, media_type="text/event-stream")` 返回；浏览器用 `EventSource` 订阅，断线会自动重连并带上 `Last-Event-ID` 请求头。
|
||||
- 检测客户端断开：`await request.is_disconnected()`。
|
||||
|
||||
**要建的文件**
|
||||
- `uv add fastapi uvicorn`
|
||||
- `fastapi_demo/main.py`：
|
||||
- `GET /health`。
|
||||
- `POST /jobs`：请求体 `{"n": 10}`，用 `create_task` 启动一个后台任务，每秒把进度追加到内存里的 `jobs[job_id]["events"]` 列表，返回 `{"job_id": ...}`。
|
||||
- `GET /jobs/{job_id}`：返回当前状态和已产生的事件数。
|
||||
- `GET /jobs/{job_id}/events`：SSE。先补发 `Last-Event-ID` 之后的事件，再等待新事件，任务完成后发 `event: done` 并结束；每 15 秒没有新事件就发一行 `: ping` 心跳。
|
||||
- **关键实验**：浏览器关掉后，后台任务要继续跑完（因为任务和 SSE 连接是分开的）。
|
||||
- `fastapi_demo/static/sse.html`：一个按钮启动任务，用 `EventSource` 显示进度。
|
||||
- `tests/test_fastapi_demo.py`：用 `fastapi.testclient.TestClient` 测 `/health` 和创建任务。
|
||||
|
||||
**运行方式**：`uv run uvicorn fastapi_demo.main:app --reload`，浏览器打开测试页；或者 `curl -N localhost:8000/jobs/<id>/events`。
|
||||
|
||||
**完成标准**
|
||||
- 启动任务后关掉页面，再打开、重新订阅，能看到任务已经继续往前跑，而且能补齐中间漏掉的事件。
|
||||
- 能说清楚“编排为什么不能只活在浏览器里”——这就是重点 C，也是第 13 周要做的核心功能的缩小版。
|
||||
- `uv run ruff check . && uv run pytest` 通过，提交并推送到 `zhuangang` 仓库（路径 `py-learning/`）。
|
||||
- 在总纲 `README.md` 的进度记录里勾掉第 6 周。
|
||||
|
||||
---
|
||||
|
||||
## 本周自查清单
|
||||
|
||||
- [ ] 会用 uv 建项目、加依赖、跑脚本
|
||||
- [ ] 熟悉列表推导式、解包、f-string，知道可变默认参数的坑
|
||||
- [ ] 会写类型标注、dataclass、Protocol，会用 pyright 检查
|
||||
- [ ] 会用自定义异常和 `raise ... from`
|
||||
- [ ] 理解 asyncio：协程、`create_task`、`gather`、超时、取消、`Queue`、异步生成器
|
||||
- [ ] 会用 pydantic 校验外部数据（包括模型输出）
|
||||
- [ ] 会用 httpx 发异步请求、按错误类型决定是否重试，会用 `MockTransport` 测试
|
||||
- [ ] 会写 FastAPI 接口和 SSE，理解后台任务和 SSE 连接是两回事
|
||||
|
||||
下周预告：第 7~8 周做 RAG 知识库问答项目（FastAPI + Chroma），见 `rag-project.md`。本周写的 httpx 重试、pydantic 解析、SSE 都会直接用上。
|
||||
File diff suppressed because it is too large.
Load diff
Reference in new issue
Block a user