diff --git a/catalog.json b/catalog.json index 31bc6c4..bdfc901 100644 --- a/catalog.json +++ b/catalog.json @@ -425,42 +425,48 @@ "domain": "python", "title": "asyncio 异步编程", "index": "topics/python/asyncio/index.md", - "lessons": 0, - "scenarios": 0, + "lessons": 2, + "scenarios": 1, "interview_questions": { - "基础": 0, - "进阶": 0, - "场景题": 0 + "基础": 4, + "进阶": 3, + "场景题": 3 }, - "examples": [] + "examples": [ + "topics/python/asyncio/examples/fanout" + ] }, { "stack": "fastapi", "domain": "python", "title": "FastAPI", "index": "topics/python/fastapi/index.md", - "lessons": 0, - "scenarios": 0, + "lessons": 3, + "scenarios": 1, "interview_questions": { - "基础": 0, - "进阶": 0, - "场景题": 0 + "基础": 4, + "进阶": 3, + "场景题": 3 }, - "examples": [] + "examples": [ + "topics/python/fastapi/examples/chat-api" + ] }, { "stack": "python", "domain": "python", "title": "Python 语言", "index": "topics/python/python/index.md", - "lessons": 0, - "scenarios": 0, + "lessons": 3, + "scenarios": 1, "interview_questions": { - "基础": 0, - "进阶": 0, - "场景题": 0 + "基础": 5, + "进阶": 4, + "场景题": 3 }, - "examples": [] + "examples": [ + "topics/python/python/examples/reconcile" + ] } ], "items": [ @@ -3777,6 +3783,84 @@ "第 7 天：开始投递，建立复盘循环" ] }, + { + "id": "python.asyncio.concurrency-control", + "title": "并发控制：gather / TaskGroup、Semaphore、超时、取消、阻塞调用", + "type": "lesson", + "domain": "python", + "stack": [ + "asyncio" + ], + "tags": [ + "TaskGroup", + "Semaphore", + "timeout", + "取消", + "to_thread", + "背压" + ], + "level": "intermediate", + "prerequisites": [ + "python.asyncio.coroutines-tasks" + ], + "related": [ + "python.asyncio.scenario.concurrent-llm-calls", + "python.asyncio.interview", + "backend.go.errors-concurrency" + ], + "est_hours": 3, + "week": [ + 4, + 5, + 12 + ], + "order": 2, + "path": "topics/python/asyncio/02-concurrency-control.md", + "sections": [ + "解释说明", + "案例", + "代码示例", + "易错点", + "面试常问" + ] + }, + { + "id": "python.asyncio.coroutines-tasks", + "title": "协程、任务和事件循环：和 JS Promise 对照", + "type": "lesson", + "domain": "python", + "stack": [ + "asyncio" + ], + "tags": [ + "协程", + "事件循环", + "Task", + "await", + "async for" + ], + "level": "basic", + "prerequisites": [ + "python.python.syntax-for-js" + ], + "related": [ + "python.asyncio.concurrency-control", + "python.asyncio.interview" + ], + "est_hours": 2, + "week": [ + 4 + ], + "order": 1, + "path": "topics/python/asyncio/01-coroutines-tasks.md", + "sections": [ + "解释说明", + "案例", + "代码示例", + "易错点", + "面试常问" + ] + }, { "id": "python.asyncio.index", "title": "asyncio 异步编程", @@ -3785,7 +3869,12 @@ "stack": [ "asyncio" ], - "tags": [], + "tags": [ + "asyncio", + "协程", + "并发", + "超时" + ], "level": null, "prerequisites": [], "related": [], @@ -3797,7 +3886,134 @@ ], "order": null, "path": "topics/python/asyncio/index.md", - "sections": [] + "sections": [ + "学习路线", + "企业场景", + "面试题", + "示例代码", + "在 16 周计划中", + "官方资料" + ] + }, + { + "id": "python.asyncio.interview", + "title": "asyncio 面试题", + "type": "interview", + "domain": "python", + "stack": [ + "asyncio" + ], + "tags": [ + "asyncio", + "协程", + "并发", + "面试" + ], + "level": "intermediate", + "prerequisites": [], + "related": [ + "python.asyncio.coroutines-tasks", + "python.asyncio.concurrency-control" + ], + "est_hours": null, + "week": [ + 4, + 16 + ], + "order": null, + "path": "topics/python/asyncio/interview.md", + "sections": [ + "基础", + "进阶", + "场景题" + ], + "questions": { + "基础": 4, + "进阶": 3, + "场景题": 3 + } + }, + { + "id": "python.asyncio.scenario.concurrent-llm-calls", + "title": "并发调用多个模型和外部接口：限并发、超时、重试、部分失败", + "type": "scenario", + "domain": "python", + "stack": [ + "asyncio", + "fastapi" + ], + "tags": [ + "并发", + "大模型调用", + "超时", + "重试", + "限流", + "部分失败" + ], + "level": "intermediate", + "prerequisites": [ + "python.asyncio.concurrency-control" + ], + "related": [ + "python.asyncio.interview", + "backend.go.scenario.batch-third-party-calls" + ], + "est_hours": 2, + "week": [ + 5, + 12 + ], + "order": null, + "path": "topics/python/asyncio/scenarios/concurrent-llm-calls.md", + "sections": [ + "业务背景与问题", + "约束", + "方案设计", + "关键代码", + "取舍", + "易错点", + "面试怎么讲" + ] + }, + { + "id": "python.fastapi.deps-auth-errors", + "title": "依赖注入、鉴权、中间件和统一错误处理", + "type": "lesson", + "domain": "python", + "stack": [ + "fastapi" + ], + "tags": [ + "Depends", + "鉴权", + "JWT", + "中间件", + "异常处理器", + "多租户" + ], + "level": "intermediate", + "prerequisites": [ + "python.fastapi.routing-validation" + ], + "related": [ + "python.fastapi.streaming-testing", + "backend.gin.middleware-auth", + "python.fastapi.interview" + ], + "est_hours": 3, + "week": [ + 4, + 5 + ], + "order": 2, + "path": "topics/python/fastapi/02-deps-auth-errors.md", + "sections": [ + "解释说明", + "案例", + "代码示例", + "易错点", + "面试常问" + ] }, { "id": "python.fastapi.index", @@ -3807,7 +4023,12 @@ "stack": [ "fastapi" ], - "tags": [], + "tags": [ + "FastAPI", + "Web 框架", + "pydantic", + "SSE" + ], "level": null, "prerequisites": [], "related": [], @@ -3819,7 +4040,174 @@ ], "order": null, "path": "topics/python/fastapi/index.md", - "sections": [] + "sections": [ + "学习路线", + "企业场景", + "面试题", + "示例代码", + "在 16 周计划中", + "官方资料" + ] + }, + { + "id": "python.fastapi.interview", + "title": "FastAPI 面试题", + "type": "interview", + "domain": "python", + "stack": [ + "fastapi" + ], + "tags": [ + "FastAPI", + "pydantic", + "依赖注入", + "面试" + ], + "level": "intermediate", + "prerequisites": [], + "related": [ + "python.fastapi.routing-validation", + "python.fastapi.deps-auth-errors", + "python.fastapi.streaming-testing" + ], + "est_hours": null, + "week": [ + 4, + 5, + 16 + ], + "order": null, + "path": "topics/python/fastapi/interview.md", + "sections": [ + "基础", + "进阶", + "场景题" + ], + "questions": { + "基础": 4, + "进阶": 3, + "场景题": 3 + } + }, + { + "id": "python.fastapi.routing-validation", + "title": "路由和参数校验：路径 / 查询 / 请求体、pydantic 模型、响应模型", + "type": "lesson", + "domain": "python", + "stack": [ + "fastapi" + ], + "tags": [ + "路由", + "pydantic", + "参数校验", + "响应模型", + "OpenAPI" + ], + "level": "basic", + "prerequisites": [ + "python.python.typing-errors-testing" + ], + "related": [ + "python.fastapi.deps-auth-errors", + "backend.gin.routing-binding", + "python.fastapi.interview" + ], + "est_hours": 2, + "week": [ + 4 + ], + "order": 1, + "path": "topics/python/fastapi/01-routing-validation.md", + "sections": [ + "解释说明", + "案例", + "代码示例", + "易错点", + "面试常问" + ] + }, + { + "id": "python.fastapi.scenario.streaming-chat-api", + "title": "大模型对话 API：鉴权、统一错误、SSE 流式、客户端断开即停止", + "type": "scenario", + "domain": "python", + "stack": [ + "fastapi", + "asyncio", + "nginx" + ], + "tags": [ + "对话 API", + "SSE", + "鉴权", + "错误处理", + "可测试性" + ], + "level": "intermediate", + "prerequisites": [ + "python.fastapi.streaming-testing" + ], + "related": [ + "python.fastapi.interview", + "devops.nginx.scenario.ai-gateway-sse-canary" + ], + "est_hours": 2, + "week": [ + 5, + 12 + ], + "order": null, + "path": "topics/python/fastapi/scenarios/streaming-chat-api.md", + "sections": [ + "业务背景与问题", + "约束", + "方案设计", + "关键代码", + "取舍", + "易错点", + "面试怎么讲" + ] + }, + { + "id": "python.fastapi.streaming-testing", + "title": "流式输出（SSE）、生命周期、测试和部署", + "type": "lesson", + "domain": "python", + "stack": [ + "fastapi" + ], + "tags": [ + "SSE", + "StreamingResponse", + "lifespan", + "TestClient", + "pytest", + "uvicorn" + ], + "level": "intermediate", + "prerequisites": [ + "python.fastapi.deps-auth-errors" + ], + "related": [ + "python.fastapi.scenario.streaming-chat-api", + "devops.nginx.https-stream-limit", + "python.fastapi.interview" + ], + "est_hours": 3, + "week": [ + 5, + 12 + ], + "order": 3, + "path": "topics/python/fastapi/03-streaming-testing.md", + "sections": [ + "解释说明", + "案例", + "代码示例", + "易错点", + "面试常问" + ] }, { "id": "python.python.index", @@ -3829,7 +4217,12 @@ "stack": [ "python" ], - "tags": [], + "tags": [ + "Python", + "语法", + "类型标注", + "pytest" + ], "level": null, "prerequisites": [], "related": [], @@ -3839,7 +4232,206 @@ ], "order": null, "path": "topics/python/python/index.md", - "sections": [] + "sections": [ + "学习路线", + "企业场景", + "面试题", + "示例代码", + "在 16 周计划中", + "官方资料" + ] + }, + { + "id": "python.python.interview", + "title": "Python 面试题", + "type": "interview", + "domain": "python", + "stack": [ + "python" + ], + "tags": [ + "Python", + "基础", + "面试" + ], + "level": "basic", + "prerequisites": [], + "related": [ + "python.python.syntax-for-js", + "python.python.typing-errors-testing", + "python.python.project-packaging" + ], + "est_hours": null, + "week": [ + 4, + 16 + ], + "order": null, + "path": "topics/python/python/interview.md", + "sections": [ + "基础", + "进阶", + "场景题" + ], + "questions": { + "基础": 5, + "进阶": 4, + "场景题": 3 + } + }, + { + "id": "python.python.project-packaging", + "title": "工程化：虚拟环境、uv / pip、项目结构、导入、日志和配置", + "type": "lesson", + "domain": "python", + "stack": [ + "python" + ], + "tags": [ + "虚拟环境", + "uv", + "pip", + "pyproject.toml", + "导入", + "logging", + "配置" + ], + "level": "basic", + "prerequisites": [ + "python.python.syntax-for-js" + ], + "related": [ + "python.fastapi.routing-validation", + "python.python.interview" + ], + "est_hours": 2, + "week": [ + 4 + ], + "order": 3, + "path": "topics/python/python/03-project-packaging.md", + "sections": [ + "解释说明", + "案例", + "代码示例", + "易错点", + "面试常问" + ] + }, + { + "id": "python.python.scenario.order-reconciliation", + "title": "订单对账脚本：从“能跑的脚本”到可测试、可维护的工具", + "type": "scenario", + "domain": "python", + "stack": [ + "python" + ], + "tags": [ + "对账", + "CSV", + "Decimal", + "数据处理", + "测试" + ], + "level": "basic", + "prerequisites": [ + "python.python.typing-errors-testing" + ], + "related": [ + "python.python.interview", + "backend.mq.reliability" + ], + "est_hours": 2, + "week": [ + 4 + ], + "order": null, + "path": "topics/python/python/scenarios/order-reconciliation.md", + "sections": [ + "业务背景与问题", + "约束", + "方案设计", + "关键代码", + "取舍", + "易错点", + "面试怎么讲" + ] + }, + { + "id": "python.python.syntax-for-js", + "title": "给 JS 开发者的 Python：语法、数据结构、推导式、函数、类", + "type": "lesson", + "domain": "python", + "stack": [ + "python" + ], + "tags": [ + "语法", + "list", + "dict", + "推导式", + "解包", + "类" + ], + "level": "basic", + "prerequisites": [], + "related": [ + "python.python.typing-errors-testing", + "python.python.interview" + ], + "est_hours": 3, + "week": [ + 4 + ], + "order": 1, + "path": "topics/python/python/01-syntax-for-js.md", + "sections": [ + "解释说明", + "案例", + "代码示例", + "易错点", + "面试常问" + ] + }, + { + "id": "python.python.typing-errors-testing", + "title": "类型标注、dataclass、异常和上下文管理器、pytest", + "type": "lesson", + "domain": "python", + "stack": [ + "python" + ], + "tags": [ + "类型标注", + "dataclass", + "Protocol", + "异常", + "with", + "生成器", + "pytest" + ], + "level": "basic", + "prerequisites": [ + "python.python.syntax-for-js" + ], + "related": [ + "python.python.scenario.order-reconciliation", + "python.asyncio.coroutines-tasks", + "python.python.interview" + ], + "est_hours": 3, + "week": [ + 4 + ], + "order": 2, + "path": "topics/python/python/02-typing-errors-testing.md", + "sections": [ + "解释说明", + "案例", + "代码示例", + "易错点", + "面试常问" + ] }, { "id": "scenarios.index", diff --git a/topics/python/asyncio/01-coroutines-tasks.md b/topics/python/asyncio/01-coroutines-tasks.md new file mode 100644 index 0000000..4300163 --- /dev/null +++ b/topics/python/asyncio/01-coroutines-tasks.md @@ -0,0 +1,112 @@ +--- +id: python.asyncio.coroutines-tasks +title: 协程、任务和事件循环：和 JS Promise 对照 +type: lesson +domain: python +stack: [asyncio] +tags: [协程, 事件循环, Task, await, async for] +level: basic +prerequisites: [python.python.syntax-for-js] +est_hours: 2 +related: [python.asyncio.concurrency-control, python.asyncio.interview] +week: [4] +order: 1 +--- + +# 协程、任务和事件循环：和 JS Promise 对照 + +## 解释说明 + +| JavaScript | Python asyncio | 区别 | +| --- | --- | --- | +| 事件循环由运行时自动提供 | `asyncio.run(main())` 显式启动 | Python 脚本默认是同步的 | +| `async function f()` | `async def f():` | | +| 调用 async 函数**立即开始执行**，返回 Promise | 调用 `f()` 只得到**协程对象，不会执行** | 必须 `await` 或包装成任务 | +| `Promise.all([...])` | `await asyncio.gather(...)` 或 `TaskGroup` | | +| `Promise.race` | `asyncio.wait(..., return_when=FIRST_COMPLETED)` | | +| `setTimeout` 包装的延时 | `await asyncio.sleep(1)` | **不要用 `time.sleep`**，会阻塞整个循环 | +| 后台执行不等待 | `asyncio.create_task(coro)` | 要保存任务引用，否则可能被垃圾回收 | +| `for await (const x of stream)` | `async for x in stream:` | 流式读取模型输出常用 | +| `AbortController` | `task.cancel()`、`asyncio.timeout()` | | + +**核心概念** + +- **协程（coroutine）**：`async def` 定义的函数调用后得到的对象，在 `await` 处可以暂停，把控制权交还给事件循环。 +- **任务（Task）**：被事件循环调度执行的协程。`create_task` 之后，它会在当前协程下一次 `await` 时开始运行。 +- **事件循环**：单线程，不断地挑选“已经可以继续执行的任务”来运行。所以**任何一段不 await 的长时间计算或阻塞调用，都会让所有任务停下来**。 + +**适用场景**：IO 密集（网络请求、数据库、文件）。CPU 密集计算不会因为 asyncio 变快。 + +## 案例 + +RAG 接口需要同时做三件事：向量检索、关键词检索、查询用户权限。顺序执行约 300ms + 200ms + 100ms = 600ms；用 `gather` 并发执行，总耗时约等于最慢的那个，300ms。 + +## 代码示例 + +```python +import asyncio +import time + + +async def vector_search(q: str) -> list[str]: + await asyncio.sleep(0.3) # 模拟网络 IO + return [f"向量结果：{q}"] + + +async def keyword_search(q: str) -> list[str]: + await asyncio.sleep(0.2) + return [f"关键词结果：{q}"] + + +async def user_scopes(user_id: int) -> set[str]: + await asyncio.sleep(0.1) + return {"hr", "public"} + + +async def main() -> None: + # 1. 顺序执行：约 0.6 秒 + start = time.perf_counter() + await vector_search("年假") + await keyword_search("年假") + await user_scopes(1) + print(f"顺序：{time.perf_counter() - start:.2f}s") + + # 2. 并发执行：约 0.3 秒；结果顺序和参数顺序一致 + start = time.perf_counter() + vec, kw, scopes = await asyncio.gather(vector_search("年假"), keyword_search("年假"), user_scopes(1)) + print(f"并发：{time.perf_counter() - start:.2f}s", vec, kw, scopes) + + # 3. 只调用不 await：协程不会执行，并且会有 "coroutine was never awaited" 警告 + coro = vector_search("x") + print(type(coro).__name__) # coroutine + coro.close() # 演示用：手动关闭，避免警告 + + # 4. 后台任务：先启动，做别的事，再取结果 + task = asyncio.create_task(vector_search("后台")) + await asyncio.sleep(0) # 让出一次控制权，任务开始执行 + print("先做别的事") + print(await task) + + # 5. 异步迭代：模拟模型流式输出 + async def stream_tokens(text: str): + for ch in text: + await asyncio.sleep(0.01) + yield ch + + print("".join([t async for t in stream_tokens("流式输出")])) + + +asyncio.run(main()) +``` + +## 易错点 + +- **忘记 `await`**：拿到的是协程对象，函数根本没执行，常见表现是“结果是 ``”。 +- **在 async 函数里用 `time.sleep`、`requests.get`、同步数据库驱动**：整个服务卡住。用 `asyncio.sleep`、`httpx.AsyncClient`、异步驱动，或者 `asyncio.to_thread` 包装。 +- **`create_task` 后不保存引用**：任务可能被垃圾回收而中途消失；也看不到它抛出的异常。 +- **在已经运行的事件循环里再调用 `asyncio.run`**（比如在 Jupyter、FastAPI 的 async 路由里）：会报错，直接 `await` 即可。 +- **以为 asyncio 是多线程**：它是单线程的协作式调度，共享变量在两个 `await` 之间是安全的，但跨 `await` 仍可能被别的任务修改。 + +## 面试常问 + +见 [asyncio 面试题](interview.md)：协程和线程的区别、事件循环原理、asyncio 适合什么场景、为什么不能在协程里调用阻塞函数。 diff --git a/topics/python/asyncio/02-concurrency-control.md b/topics/python/asyncio/02-concurrency-control.md new file mode 100644 index 0000000..b85e0e3 --- /dev/null +++ b/topics/python/asyncio/02-concurrency-control.md @@ -0,0 +1,138 @@ +--- +id: python.asyncio.concurrency-control +title: 并发控制：gather / TaskGroup、Semaphore、超时、取消、阻塞调用 +type: lesson +domain: python +stack: [asyncio] +tags: [TaskGroup, Semaphore, timeout, 取消, to_thread, 背压] +level: intermediate +prerequisites: [python.asyncio.coroutines-tasks] +est_hours: 3 +related: [python.asyncio.scenario.concurrent-llm-calls, python.asyncio.interview, backend.go.errors-concurrency] +week: [4, 5, 12] +order: 2 +--- + +# 并发控制：gather / TaskGroup、Semaphore、超时、取消、阻塞调用 + +## 解释说明 + +**等待一组任务的三种方式** + +| 写法 | 一个失败时 | 适合 | +| --- | --- | --- | +| `gather(*coros)` | 第一个异常立即抛给调用方，**其他任务继续运行**（不会被取消） | 简单的并发 | +| `gather(*coros, return_exceptions=True)` | 异常作为结果返回，不抛出 | 允许部分失败 | +| `async with asyncio.TaskGroup() as tg:`（3.11+） | **取消其余任务**，异常合并为 `ExceptionGroup` 抛出 | 要么全部成功、要么全部取消 | + +**限制并发**：`asyncio.Semaphore(n)`，`async with sem:` 包住真正的调用。没有限制时，一次性并发 1000 个请求会触发下游限流、耗尽连接池。（对照 Go：带缓冲的 channel 当信号量，或 `errgroup.SetLimit`。） + +**超时** + +- `async with asyncio.timeout(秒):`（3.11+）或 `await asyncio.wait_for(coro, 秒)`；超时抛 `TimeoutError`，内部的任务会被取消。 +- 分清**单次调用超时**和**整体截止时间**（例如接口必须 10 秒内返回，内部每次模型调用最多 5 秒）。 + +**取消** + +- `task.cancel()` 会在任务下一个 `await` 点抛出 `CancelledError`。 +- 捕获 `CancelledError` 做清理后**必须重新抛出**，否则取消被吞掉，调用方以为任务还在正常运行。 +- `except Exception` 不会捕获 `CancelledError`（它继承自 `BaseException`），这是有意设计的。 + +**阻塞调用**：必须用同步库（某些 SDK、文件解析、CPU 计算）时，`await asyncio.to_thread(func, 参数)` 放到线程池执行，不阻塞事件循环。CPU 密集且耗时长的用进程池。 + +**背压**：生产者比消费者快时，用有界的 `asyncio.Queue(maxsize=n)`，队列满时 `put` 会等待。 + +## 案例 + +批量为 2000 篇文档生成摘要：用 Semaphore 限制最多 8 个并发请求（符合模型服务商的速率限制）；每次调用 30 秒超时；限流错误（429）指数退避重试；单篇失败记录下来，最后汇总重跑，而不是让整个批次失败。 + +## 代码示例 + +```python +import asyncio + + +async def fetch(i: int) -> int: + await asyncio.sleep(0.05 * (i % 3)) + if i == 4: + raise ValueError(f"任务 {i} 失败") + return i * 10 + + +async def main() -> None: + # 1. gather + return_exceptions：部分失败也拿到所有结果 + results = await asyncio.gather(*(fetch(i) for i in range(6)), return_exceptions=True) + print([r if not isinstance(r, Exception) else f"错误：{r}" for r in results]) + + # 2. TaskGroup：一个失败，其余被取消，异常以 ExceptionGroup 抛出 + try: + async with asyncio.TaskGroup() as tg: + tasks = [tg.create_task(fetch(i)) for i in range(6)] + except* ValueError as eg: # except* 处理 ExceptionGroup（3.11+） + print("TaskGroup 失败：", [str(e) for e in eg.exceptions]) + print("被取消的任务数：", sum(t.cancelled() for t in tasks)) + + # 3. Semaphore 限制并发 + sem = asyncio.Semaphore(2) + running = 0 + + async def limited(i: int) -> None: + nonlocal running + async with sem: + running += 1 + assert running <= 2 + await asyncio.sleep(0.01) + running -= 1 + + await asyncio.gather(*(limited(i) for i in range(10))) + print("Semaphore：同时运行的任务从未超过 2 个") + + # 4. 超时 + try: + async with asyncio.timeout(0.1): + await asyncio.sleep(1) + except TimeoutError: + print("超时") + + # 5. 取消：清理后重新抛出 + async def worker() -> None: + try: + await asyncio.sleep(10) + except asyncio.CancelledError: + print("收到取消，释放资源") + raise # 必须重新抛出 + + t = asyncio.create_task(worker()) + await asyncio.sleep(0.01) + t.cancel() + try: + await t + except asyncio.CancelledError: + print("任务已取消：", t.cancelled()) + + # 6. 同步阻塞函数放到线程里执行 + def parse_pdf_sync(path: str) -> str: + import time + time.sleep(0.1) # 模拟同步库的阻塞调用 + return f"{path} 的文本" + + print(await asyncio.to_thread(parse_pdf_sync, "manual.pdf")) + + +asyncio.run(main()) +``` + +把这些组合起来（并发上限 + 单次超时 + 重试 + 部分失败 + 整体截止时间）的完整实现和测试见 [examples/fanout](examples/fanout/)。 + +## 易错点 + +- **用 `gather` 以为失败时会取消其他任务**：不会，其他任务继续跑（可能继续消耗模型额度）。需要“一个失败全部取消”用 TaskGroup。 +- **吞掉 `CancelledError`**：任务无法被取消，服务关闭时卡住。 +- **没有并发上限**：一次性创建成千上万个请求，下游限流、连接池耗尽、内存上涨。 +- **只设整体超时不设单次超时**（或反过来）：一个慢请求占满全部时间，或者总时长失控。 +- **重试不区分错误类型**：参数错误（400）重试多少次都一样失败，只重试超时、429、5xx。 +- **在 `to_thread` 里跑 CPU 密集任务**：受 GIL 限制，没有并行效果，用进程池。 + +## 面试常问 + +见 [asyncio 面试题](interview.md)：gather 和 TaskGroup 的区别、怎么限制并发、超时和取消如何实现、怎么在异步代码中调用同步库。 diff --git a/topics/python/asyncio/examples/fanout/fanout.py b/topics/python/asyncio/examples/fanout/fanout.py new file mode 100644 index 0000000..cf6091f --- /dev/null +++ b/topics/python/asyncio/examples/fanout/fanout.py @@ -0,0 +1,100 @@ +"""并发调用多个下游（例如大模型 / 检索 / 第三方 API）：并发上限、单次超时、有限重试、部分失败、整体截止时间。 + +只用标准库。运行：python fanout.py；测试：python -m pytest -q test_fanout.py +""" +from __future__ import annotations + +import asyncio +import random +import time +from collections.abc import Awaitable, Callable +from dataclasses import dataclass +from typing import Generic, TypeVar + +T = TypeVar("T") + + +class RetryableError(Exception): + """可以重试的错误（限流 429、服务暂时不可用 503 等）。""" + + +@dataclass +class Result(Generic[T]): + key: str + value: T | None = None + error: str | None = None + attempts: int = 0 + + +async def call_with_retry(fn: Callable[[], Awaitable[T]], *, timeout: float, retries: int, + base_delay: float = 0.05) -> tuple[T, int]: + """单次调用带超时；遇到超时或可重试错误时指数退避重试。返回 (结果, 尝试次数)。""" + for attempt in range(1, retries + 2): + try: + async with asyncio.timeout(timeout): # Python 3.11+ + return await fn(), attempt + except (TimeoutError, RetryableError): + if attempt > retries: + raise + await asyncio.sleep(base_delay * 2 ** (attempt - 1) * (1 + random.random() / 2)) + raise AssertionError("不会执行到这里") + + +async def fan_out(jobs: dict[str, Callable[[], Awaitable[T]]], *, concurrency: int, timeout: float, + retries: int = 1, deadline: float | None = None) -> list[Result[T]]: + """并发执行所有任务，最多 concurrency 个同时进行；单个失败不影响其他任务。 + deadline：整体截止时间（秒），到时未完成的任务被取消并标记为超时。""" + sem = asyncio.Semaphore(concurrency) + results = {k: Result[T](key=k) for k in jobs} + + async def run(key: str, fn: Callable[[], Awaitable[T]]) -> None: + async def counted() -> T: # 记录尝试次数（失败时也能看到试了几次） + results[key].attempts += 1 + return await fn() + + async with sem: + try: + results[key].value, _ = await call_with_retry(counted, timeout=timeout, retries=retries) + except asyncio.CancelledError: + results[key].error = "整体超时，已取消" + raise # 取消必须继续向上传播 + except Exception as e: # 单个任务失败只记录，不影响其他任务 + results[key].error = f"{type(e).__name__}: {e}" if str(e) else type(e).__name__ + + tasks = [asyncio.create_task(run(k, fn)) for k, fn in jobs.items()] + done, pending = await asyncio.wait(tasks, timeout=deadline) + for t in pending: + t.cancel() + await asyncio.gather(*pending, return_exceptions=True) # 等待取消完成，避免任务泄漏 + return [results[k] for k in jobs] + + +# ---------------- 演示 ---------------- +def fake_llm(name: str, delay: float, fail_times: int = 0) -> Callable[[], Awaitable[str]]: + state = {"left": fail_times} + + async def call() -> str: + await asyncio.sleep(delay) + if state["left"] > 0: + state["left"] -= 1 + raise RetryableError("429 Too Many Requests") + return f"{name} 的回答" + + return call + + +async def demo() -> None: + jobs = { + "摘要": fake_llm("摘要", 0.10), + "分类": fake_llm("分类", 0.10, fail_times=1), # 第一次被限流，重试后成功 + "翻译": fake_llm("翻译", 2.00), # 太慢，单次超时 + "关键词": fake_llm("关键词", 0.05), + } + start = time.perf_counter() + for r in await fan_out(jobs, concurrency=2, timeout=0.5, retries=1): + print(f"{r.key}: {r.value or r.error}（尝试 {r.attempts} 次）") + print(f"总耗时 {time.perf_counter() - start:.2f}s") + + +if __name__ == "__main__": + asyncio.run(demo()) diff --git a/topics/python/asyncio/examples/fanout/test_fanout.py b/topics/python/asyncio/examples/fanout/test_fanout.py new file mode 100644 index 0000000..6ff497a --- /dev/null +++ b/topics/python/asyncio/examples/fanout/test_fanout.py @@ -0,0 +1,60 @@ +import asyncio +import time + +from fanout import RetryableError, fake_llm, fan_out + + +def run(coro): + return asyncio.run(coro) + + +def test_concurrency_limit(): + active, peak = 0, 0 + + def job(): + async def call(): + nonlocal active, peak + active += 1 + peak = max(peak, active) + await asyncio.sleep(0.05) + active -= 1 + return "ok" + return call + + res = run(fan_out({str(i): job() for i in range(10)}, concurrency=3, timeout=1)) + assert peak == 3 and all(r.value == "ok" for r in res) + + +def test_retry_then_success(): + [r] = run(fan_out({"a": fake_llm("a", 0.01, fail_times=1)}, concurrency=1, timeout=1, retries=1)) + assert r.value == "a 的回答" and r.attempts == 2 + + +def test_partial_failure(): + async def boom(): + raise ValueError("参数错误") # 不可重试的错误：只尝试一次 + + res = run(fan_out({"ok": fake_llm("ok", 0.01), "bad": boom, "slow": fake_llm("slow", 1)}, + concurrency=3, timeout=0.1, retries=1)) + by = {r.key: r for r in res} + assert by["ok"].value == "ok 的回答" + assert by["bad"].error == "ValueError: 参数错误" + assert by["slow"].error == "TimeoutError" + + +def test_retry_exhausted(): + [r] = run(fan_out({"x": fake_llm("x", 0.01, fail_times=5)}, concurrency=1, timeout=1, retries=2)) + assert r.value is None and "RetryableError" in r.error + + +def test_overall_deadline_cancels(): + start = time.perf_counter() + res = run(fan_out({"fast": fake_llm("fast", 0.01), "slow": fake_llm("slow", 5)}, + concurrency=2, timeout=10, deadline=0.2)) + assert time.perf_counter() - start < 1 + by = {r.key: r for r in res} + assert by["fast"].value and by["slow"].error == "整体超时，已取消" + + +def test_retryable_error_is_exception(): + assert issubclass(RetryableError, Exception) diff --git a/topics/python/asyncio/index.md b/topics/python/asyncio/index.md index 4259028..f19f88b 100644 --- a/topics/python/asyncio/index.md +++ b/topics/python/asyncio/index.md @@ -4,12 +4,38 @@ title: asyncio 异步编程 type: index domain: python stack: [asyncio] -tags: [] +tags: [asyncio, 协程, 并发, 超时] week: [4, 5, 12] --- # asyncio 异步编程 -协程、任务、并发控制、超时与取消。 +AI 应用的大部分时间在“等”：等模型生成、等检索、等数据库。asyncio 让一个进程同时等很多件事。你熟悉 JS 的 `async/await` 和 Promise，概念几乎可以直接迁移；不同点在于 Python **需要显式运行事件循环**、**同步阻塞调用会卡住整个循环**、以及任务取消和超时的写法。 -> 本专题建设中：课程、企业场景、面试题和示例代码会陆续补充。 +## 学习路线 + +| 顺序 | 课程 | 级别 | 预计 | +| --- | --- | --- | --- | +| 1 | [协程、任务和事件循环：和 JS Promise 对照](01-coroutines-tasks.md) | 基础 | 2 小时 | +| 2 | [并发控制：gather / TaskGroup、Semaphore、超时、取消、阻塞调用](02-concurrency-control.md) | 进阶 | 3 小时 | + +## 企业场景 + +- [并发调用多个模型和外部接口：限并发、超时、重试、部分失败](scenarios/concurrent-llm-calls.md) + +## 面试题 + +- [asyncio 面试题](interview.md) + +## 示例代码 + +- [examples/fanout](examples/fanout/)：并发上限、单次超时、指数退避重试、部分失败、整体截止时间取消，附 6 个测试，只用标准库 + +## 在 16 周计划中 + +- [第 4 周：Python 速成（asyncio 是重点）](../../../plan/week04-python.md)、[第 5 周：RAG 服务](../../../plan/week05-rag-1.md)、[第 12 周：Agent 持久化](../../../plan/week12-agent-m3-persistence.md) + +## 官方资料 + +- asyncio 文档（中文）：https://docs.python.org/zh-cn/3/library/asyncio.html +- 协程与任务：https://docs.python.org/zh-cn/3/library/asyncio-task.html diff --git a/topics/python/asyncio/interview.md b/topics/python/asyncio/interview.md new file mode 100644 index 0000000..ac217cd --- /dev/null +++ b/topics/python/asyncio/interview.md @@ -0,0 +1,59 @@ +--- +id: python.asyncio.interview +title: asyncio 面试题 +type: interview +domain: python +stack: [asyncio] +tags: [asyncio, 协程, 并发, 面试] +level: intermediate +related: [python.asyncio.coroutines-tasks, python.asyncio.concurrency-control] +week: [4, 16] +--- + +# asyncio 面试题 + +## 基础 + +### 【基础】协程、线程、进程有什么区别？ + +**参考答案**：进程有独立的内存空间，切换和通信开销大，能利用多核；线程共享进程内存，由操作系统抢占式调度，在 CPython 中受 GIL 限制不能并行执行 Python 代码；协程是用户态的、协作式调度的执行单元，在 `await` 处主动让出控制权，切换开销极小，单线程即可同时处理大量 IO 等待。 + +### 【基础】asyncio 适合什么场景？不适合什么场景？ + +**参考答案**：适合 IO 密集、并发连接多的场景：调用大模型 API、HTTP 请求、数据库查询、WebSocket / SSE 长连接。不适合 CPU 密集计算（单线程，无法加速），这类任务用多进程或交给 C 扩展库。 + +### 【基础】调用 `async def` 函数但没有 `await` 会怎样？ + +**参考答案**：只会创建协程对象，函数体不会执行，Python 会给出 “coroutine was never awaited” 警告。这一点和 JS 不同——JS 的 async 函数调用后立即开始执行。 + +### 【基础】为什么在协程里不能使用 `time.sleep` 或 `requests`？ + +**参考答案**：事件循环是单线程的，阻塞调用会让整个线程停住，所有其他协程都无法运行，服务的并发能力退化为 1。应该用 `asyncio.sleep`、`httpx.AsyncClient` 等异步库；只能用同步库时用 `asyncio.to_thread` 放到线程池执行。 + +## 进阶 + +### 【进阶】`asyncio.gather` 和 `TaskGroup` 有什么区别？ + +**参考答案**：`gather` 在某个任务抛异常时立即把异常抛给调用方，但其余任务继续运行；加 `return_exceptions=True` 则把异常作为结果返回。`TaskGroup`（3.11+）是结构化并发：一个任务失败时取消组内其他任务，等它们全部结束后，以 `ExceptionGroup` 抛出所有异常，不会留下“孤儿任务”。 + +### 【进阶】怎么限制并发数？ + +**参考答案**：用 `asyncio.Semaphore(n)`，在每个任务的实际调用外层 `async with sem:`；或者用固定数量的 worker 从 `asyncio.Queue` 取任务。注意多个请求要共享同一个 Semaphore；多进程 / 多实例时需要 Redis 等全局限流。 + +### 【进阶】asyncio 的取消机制是怎样的？ + +**参考答案**：`task.cancel()` 请求取消，任务在下一个 `await` 点收到 `CancelledError`。任务可以捕获它做清理（关闭连接、记录状态），但应该重新抛出，否则取消失效。`asyncio.timeout` / `wait_for` 超时也是通过取消内部任务实现的。`CancelledError` 继承自 `BaseException`，所以 `except Exception` 不会误吞它。 + +## 场景题 + +### 【场景题】需要对 1 万条文本调用大模型生成摘要，服务商限制每分钟 600 次请求，怎么设计？ + +**参考答案**：用 Semaphore 或 worker 池限制并发，再加速率控制（令牌桶，每秒约 10 次）；每次调用设超时；对 429、超时、5xx 做带抖动的指数退避重试，429 响应如果带 `Retry-After` 就按它等待；单条失败记录下来，最后汇总重跑；记录进度（已完成的 ID）以便中断后续跑；结果写入数据库而不是全放内存。规模更大时用消息队列分发给多个 worker。 + +### 【场景题】FastAPI 接口里调用了一个同步的 SDK，发现服务在压测时吞吐很低，为什么？怎么改？ + +**参考答案**：在 `async def` 路由里直接调用同步 SDK，阻塞了事件循环，所有请求排队。改法：换成 SDK 的异步版本；或者 `await asyncio.to_thread(sdk.call, ...)`；或者把路由定义为普通 `def`，FastAPI 会把它放到线程池执行。然后重新压测确认。 + +### 【场景题】一个异步任务在服务关闭时总是卡住，可能是什么原因？ + +**参考答案**：任务里捕获了 `CancelledError` 却没有重新抛出（或者用了裸 `except:`），导致取消无效；或者任务在执行不响应取消的阻塞调用（同步 IO、`to_thread` 中的线程无法被强制停止）；或者有后台任务没有被追踪和取消。修复：正确传播取消、给阻塞调用设超时、在应用的生命周期关闭阶段统一取消并等待后台任务。 diff --git a/topics/python/asyncio/scenarios/concurrent-llm-calls.md b/topics/python/asyncio/scenarios/concurrent-llm-calls.md new file mode 100644 index 0000000..0d7037c --- /dev/null +++ b/topics/python/asyncio/scenarios/concurrent-llm-calls.md @@ -0,0 +1,111 @@ +--- +id: python.asyncio.scenario.concurrent-llm-calls +title: 并发调用多个模型和外部接口：限并发、超时、重试、部分失败 +type: scenario +domain: python +stack: [asyncio, fastapi] +tags: [并发, 大模型调用, 超时, 重试, 限流, 部分失败] +level: intermediate +prerequisites: [python.asyncio.concurrency-control] +est_hours: 2 +related: [python.asyncio.interview, backend.go.scenario.batch-third-party-calls] +week: [5, 12] +--- + +# 并发调用多个模型和外部接口：限并发、超时、重试、部分失败 + +## 业务背景与问题 + +客服工单系统里，每张新工单需要并行做四件事：摘要（大模型）、分类（大模型）、情绪识别（小模型服务）、查客户等级（CRM 接口）。第一版是顺序调用，单张工单处理要 8~15 秒；改成直接 `gather` 之后变快了，但遇到新问题： + +- 早高峰一次进来几百张工单，同时发出上千个请求，模型服务商返回大量 429（限流）； +- 某一个接口卡住时，整张工单一直没有结果； +- 任何一个调用报错，整个 `gather` 抛异常，其余已经成功的结果也丢了。 + +## 约束 + +- 模型服务商有速率限制（每分钟请求数和 token 数）。 +- 工单处理结果允许部分缺失（例如情绪识别失败，不影响分配客服），但必须标明缺失项。 +- 每张工单的处理总时长不超过 20 秒，超时的部分以后台任务补齐。 + +## 方案设计 + +```mermaid +flowchart LR + T[新工单] --> F[fan_out] + F -->|Semaphore 限并发| S[摘要 LLM] + F --> C[分类 LLM] + F --> E[情绪服务] + F --> R[CRM 接口] + S & C & E & R --> M[合并结果：成功的字段 + 失败原因] + M --> D[分配客服] + M -->|有失败项| Q[补偿队列：稍后重试] +``` + +1. **全局并发上限**：所有模型调用共享一个 `Semaphore`（按服务商限额设置），而不是每张工单各自限制。多进程 / 多实例部署时，进一步用 Redis 做全局限流。 +2. **单次超时 + 整体截止时间**：每次调用 5 秒超时；每张工单整体 20 秒，到时取消未完成的调用。 +3. **有选择地重试**：只对超时、429、5xx 做指数退避重试（加随机抖动），最多 1~2 次；参数错误不重试。 +4. **部分失败**：每个子任务的结果单独记录成功值或错误原因，不让一个失败影响全部；失败项进入补偿队列。 +5. **可观测**：记录每个子任务的耗时、重试次数、失败原因，用于调整并发和超时参数。 + +## 关键代码 + +完整实现和测试见 [examples/fanout](../examples/fanout/)： + +```python +async def fan_out(jobs, *, concurrency, timeout, retries=1, deadline=None): + sem = asyncio.Semaphore(concurrency) + results = {k: Result(key=k) for k in jobs} + + async def run(key, fn): + async with sem: + try: + results[key].value, _ = await call_with_retry(fn, timeout=timeout, retries=retries) + except asyncio.CancelledError: + results[key].error = "整体超时，已取消" + raise # 取消要继续传播 + except Exception as e: # 单个失败只记录 + results[key].error = f"{type(e).__name__}: {e}" + + tasks = [asyncio.create_task(run(k, fn)) for k, fn in jobs.items()] + done, pending = await asyncio.wait(tasks, timeout=deadline) + for t in pending: + t.cancel() + await asyncio.gather(*pending, return_exceptions=True) + return [results[k] for k in jobs] +``` + +演示输出（并发 2、单次超时 0.5 秒、重试 1 次）： + +```text +摘要: 摘要 的回答（尝试 1 次） +分类: 分类 的回答（尝试 2 次） ← 第一次 429，重试成功 +翻译: TimeoutError（尝试 2 次） ← 太慢，两次都超时 +关键词: 关键词 的回答（尝试 1 次） +``` + +## 取舍 + +| 方案 | 优点 | 缺点 | +| --- | --- | --- | +| 顺序调用 | 简单 | 慢 | +| 裸 `gather` | 快、代码少 | 无并发上限、一个失败全部失败 | +| **Semaphore + 超时 + 重试 + 部分失败（本方案）** | 可控、稳定 | 代码多一些，参数要调 | +| 消息队列 + 独立 worker | 削峰、可水平扩展、失败可重放 | 架构更复杂，结果是异步的 | + +请求量继续增大时，把“工单 → 子任务”拆成队列消息，由 worker 池消费，Semaphore 换成 Redis 全局限流。 + +## 易错点 + +- 每个请求各自创建一个 Semaphore：等于没有限制。 +- 重试不加抖动：所有任务在同一时刻重试，再次触发限流。 +- `except Exception` 之后不区分错误类型就重试：参数错误被重试多次，浪费额度。 +- 捕获 `CancelledError` 后不重新抛出：整体截止时间失效。 +- 截止时间到了只是“不等了”，没有取消任务：后台还在消耗模型额度。 +- 只记录“失败”不记录原因：无法判断是限流、超时还是参数问题。 + +## 面试怎么讲 + +> “工单要同时调用几个模型和接口。我用 asyncio 并发，但加了全局 Semaphore 控制并发数以符合服务商的限流；每次调用有超时，只对超时和 429 做带抖动的指数退避重试；每个子任务单独记录结果和错误，单个失败不影响其他，失败项进补偿队列；整体有截止时间，到时取消剩余调用。处理时间从顺序调用的 8 到 15 秒降到了约等于最慢的那个调用。” + +注意：面试中提到的耗时数据要来自你自己项目的实际测量，不要编造。和 Go 版本的同类场景对照：[批量调用第三方接口（Go）](../../../backend/go/scenarios/batch-third-party-calls.md)。 diff --git a/topics/python/fastapi/01-routing-validation.md b/topics/python/fastapi/01-routing-validation.md new file mode 100644 index 0000000..da8df8e --- /dev/null +++ b/topics/python/fastapi/01-routing-validation.md @@ -0,0 +1,116 @@ +--- +id: python.fastapi.routing-validation +title: 路由和参数校验：路径 / 查询 / 请求体、pydantic 模型、响应模型 +type: lesson +domain: python +stack: [fastapi] +tags: [路由, pydantic, 参数校验, 响应模型, OpenAPI] +level: basic +prerequisites: [python.python.typing-errors-testing] +est_hours: 2 +related: [python.fastapi.deps-auth-errors, backend.gin.routing-binding, python.fastapi.interview] +week: [4] +order: 1 +--- + +# 路由和参数校验：路径 / 查询 / 请求体、pydantic 模型、响应模型 + +## 解释说明 + +- **参数来源由类型和位置决定**：路径里出现的参数 → 路径参数；简单类型（int、str）→ 查询参数；pydantic 模型 → JSON 请求体；`Header()`、`Cookie()` 显式声明。 +- **校验**：`Field(ge=1, le=100, min_length=1, pattern=...)`、`Query(...)`、`Path(...)`；`Literal` 限定枚举值；校验失败自动返回 **422**，带出错字段和原因。 +- **响应模型**：返回类型标注或 `response_model=`，会过滤掉模型之外的字段——防止把密码哈希等内部字段返回出去。 +- **自动文档**：启动后访问 `/docs`（Swagger UI），前端联调直接用。 +- **`async def` 还是 `def`**：路由里调用的都是异步库 → `async def`；调用同步阻塞库 → 普通 `def`（FastAPI 会放到线程池执行）。**不要在 `async def` 里调用阻塞函数。** +- **APIRouter**：按模块拆分路由，`app.include_router(router, prefix="/v1/docs")`。 + +和 Gin 对照：`c.Param` ↔ 路径参数，`c.Query` ↔ 查询参数，`ShouldBindJSON` + `binding` 标签 ↔ pydantic 模型 + `Field`。 + +## 案例 + +知识库的文档接口：创建文档（标题 1~200 字、标签最多 10 个、可见范围只能是 `private / team / public`）、按 ID 查询、分页列表（页码 ≥ 1，每页 1~100）。用户传了 `"visibility": "everyone"`，FastAPI 自动返回 422 和具体字段，不用手写校验代码。 + +## 代码示例 + +```python +from datetime import datetime, timezone +from typing import Annotated, Literal + +from fastapi import APIRouter, FastAPI, HTTPException, Path, Query +from fastapi.testclient import TestClient +from pydantic import BaseModel, Field + + +class DocCreate(BaseModel): + title: str = Field(min_length=1, max_length=200) + content: str = Field(min_length=1) + tags: list[str] = Field(default_factory=list, max_length=10) + visibility: Literal["private", "team", "public"] = "team" + + +class DocOut(BaseModel): # 响应模型：只暴露这些字段 + id: int + title: str + tags: list[str] + visibility: str + created_at: datetime + + +class Page(BaseModel): + items: list[DocOut] + total: int + + +router = APIRouter(prefix="/v1/docs", tags=["文档"]) +DB: dict[int, dict] = {} + + +@router.post("", status_code=201) +async def create_doc(body: DocCreate) -> DocOut: + doc_id = len(DB) + 1 + DB[doc_id] = {**body.model_dump(), "id": doc_id, "owner_secret": "内部字段", + "created_at": datetime.now(timezone.utc)} + return DocOut(**DB[doc_id]) # 只取模型里声明的字段 + + +@router.get("/{doc_id}") +async def get_doc(doc_id: Annotated[int, Path(ge=1)]) -> DocOut: + if doc_id not in DB: + raise HTTPException(status_code=404, detail="文档不存在") + return DocOut(**DB[doc_id]) + + +@router.get("") +async def list_docs(page: Annotated[int, Query(ge=1)] = 1, + size: Annotated[int, Query(ge=1, le=100)] = 20, + tag: str | None = None) -> Page: + docs = [d for d in DB.values() if tag is None or tag in d["tags"]] + start = (page - 1) * size + return Page(items=[DocOut(**d) for d in docs[start:start + size]], total=len(docs)) + + +app = FastAPI() +app.include_router(router) + +c = TestClient(app) +r = c.post("/v1/docs", json={"title": "年假制度", "content": "……", "tags": ["hr"]}) +print(r.status_code, r.json()["id"], "owner_secret" in r.json()) # 201 1 False +r = c.post("/v1/docs", json={"title": "", "content": "x", "visibility": "everyone"}) +print(r.status_code, [e["loc"] for e in r.json()["detail"]]) # 422 [['body', 'title'], ['body', 'visibility']] +print(c.get("/v1/docs", params={"size": 500}).status_code) # 422 +print(c.get("/v1/docs/99").json()) # {'detail': '文档不存在'} +print(c.get("/v1/docs", params={"tag": "hr"}).json()["total"]) # 1 +``` + +## 易错点 + +- **在 `async def` 路由里调用同步数据库驱动或 `requests`**：阻塞事件循环，吞吐骤降。 +- **直接返回 ORM 对象或字典而不声明响应模型**：内部字段（密码哈希、内部备注）泄露。 +- **可变默认值**：`tags: list[str] = []` 在 pydantic 里是安全的（会复制），但养成 `default_factory` 的习惯更清晰。 +- **路由顺序**：`/docs/{doc_id}` 写在 `/docs/stats` 之前，`stats` 会被当成 doc_id 去解析，返回 422。固定路径放在前面。 +- **422 的错误格式直接给前端**：默认格式偏技术化，通常用异常处理器转换成团队统一的格式（见下一课）。 +- **pydantic v1 和 v2 写法混用**：v2 用 `model_dump()`、`model_validate()`、`field_validator`；网上很多旧教程是 v1 写法（`dict()`、`validator`）。 + +## 面试常问 + +见 [FastAPI 面试题](interview.md)：FastAPI 为什么快、参数校验原理、async def 和 def 的区别、响应模型的作用。 diff --git a/topics/python/fastapi/02-deps-auth-errors.md b/topics/python/fastapi/02-deps-auth-errors.md new file mode 100644 index 0000000..31abfdf --- /dev/null +++ b/topics/python/fastapi/02-deps-auth-errors.md @@ -0,0 +1,134 @@ +--- +id: python.fastapi.deps-auth-errors +title: 依赖注入、鉴权、中间件和统一错误处理 +type: lesson +domain: python +stack: [fastapi] +tags: [Depends, 鉴权, JWT, 中间件, 异常处理器, 多租户] +level: intermediate +prerequisites: [python.fastapi.routing-validation] +est_hours: 3 +related: [python.fastapi.streaming-testing, backend.gin.middleware-auth, python.fastapi.interview] +week: [4, 5] +order: 2 +--- + +# 依赖注入、鉴权、中间件和统一错误处理 + +## 解释说明 + +**依赖注入 `Depends`**：把“获取当前用户”“获取数据库会话”“获取模型客户端”写成函数，路由参数里声明 `Depends(函数)`，FastAPI 自动调用并把结果传进来。 + +- 依赖可以嵌套（获取当前用户 → 依赖解析 token → 依赖读取配置）。 +- 带 `yield` 的依赖可以做资源清理（请求结束后关闭数据库会话）。 +- 同一个请求里同一个依赖默认只执行一次（有缓存）。 +- **测试时用 `app.dependency_overrides[原依赖] = 假实现`** 替换，这是 FastAPI 最好用的测试手段。 + +**鉴权**：常见两种——给服务调用方的 API Key（请求头 `X-API-Key`），给用户的 JWT（`Authorization: Bearer ...`）。写成依赖：校验失败抛 `HTTPException(401)`；权限不足抛 403。多租户系统里，依赖返回“当前租户”，后续查询都带上租户条件。 + +**中间件**：每个请求都要经过的逻辑——请求 ID、耗时日志、CORS。`@app.middleware("http")` 或添加 Starlette 中间件类。业务相关的校验（鉴权、权限）用依赖更合适，因为能精确控制作用于哪些路由。 + +**统一错误处理**：`@app.exception_handler(异常类型)`。 + +- `HTTPException` → 统一格式 `{"code", "message", "request_id"}`； +- `RequestValidationError` → 把 422 转成统一格式，列出字段和原因； +- 兜底 `Exception` → 记录完整日志，返回 500 和通用提示，**不返回内部细节**。 + +和 Gin 对照：`Depends` 类似在中间件里 `c.Set("user", ...)` 然后在 handler 里取，但它有类型、按路由声明、可以在测试中替换。 + +## 案例 + +AI 服务给多个业务系统调用：每个系统一个 API Key，对应一个租户；每个请求日志都带 `request_id` 和租户；参数错误返回统一格式让调用方能程序化处理；内部异常（数据库连接串、堆栈）只进日志。 + +## 代码示例 + +```python +from collections.abc import Iterator +from typing import Annotated + +from fastapi import Depends, FastAPI, Header, HTTPException, Request +from fastapi.responses import JSONResponse +from fastapi.testclient import TestClient + + +class Session: # 模拟数据库会话 + def __init__(self) -> None: + self.closed = False + + def query_docs(self, tenant: str) -> list[str]: + return [f"{tenant} 的文档 {i}" for i in range(2)] + + +opened: list[Session] = [] + + +def get_session() -> Iterator[Session]: # 带 yield 的依赖：请求结束后清理 + s = Session() + opened.append(s) + try: + yield s + finally: + s.closed = True + + +KEYS = {"key-a": "tenant-a", "key-b": "tenant-b"} + + +def current_tenant(x_api_key: Annotated[str | None, Header()] = None) -> str: + if not x_api_key or x_api_key not in KEYS: + raise HTTPException(status_code=401, detail="API Key 无效") + return KEYS[x_api_key] + + +def require_admin(tenant: Annotated[str, Depends(current_tenant)]) -> str: # 依赖嵌套 + if tenant != "tenant-a": + raise HTTPException(status_code=403, detail="没有管理权限") + return tenant + + +app = FastAPI() + + +@app.exception_handler(HTTPException) +async def http_error(request: Request, exc: HTTPException): + return JSONResponse({"code": exc.status_code, "message": exc.detail}, status_code=exc.status_code) + + +@app.get("/documents") +def list_docs(tenant: Annotated[str, Depends(current_tenant)], + db: Annotated[Session, Depends(get_session)]) -> list[str]: + return db.query_docs(tenant) # 查询一律带租户条件 + + +@app.delete("/cache") +def clear_cache(tenant: Annotated[str, Depends(require_admin)]) -> dict: + return {"cleared_by": tenant} + + +c = TestClient(app) +print(c.get("/documents").json()) # {'code': 401, 'message': 'API Key 无效'} +print(c.get("/documents", headers={"X-API-Key": "key-b"}).json()) # ['tenant-b 的文档 0', 'tenant-b 的文档 1'] +print(opened[-1].closed) # True：请求结束后会话已关闭 +print(c.delete("/cache", headers={"X-API-Key": "key-b"}).status_code) # 403 + +# 测试里替换依赖：不需要真实的 API Key 和数据库 +app.dependency_overrides[current_tenant] = lambda: "tenant-test" +print(c.get("/documents").json()) # ['tenant-test 的文档 0', ...] +app.dependency_overrides.clear() +``` + +请求 ID 中间件、422 统一格式、500 兜底的完整写法见 [examples/chat-api/app.py](examples/chat-api/app.py)。 + +## 易错点 + +- **鉴权写在中间件里对所有路径生效**：健康检查、文档页面也被拦住；用依赖按路由声明更精确（或在路由器级别 `dependencies=[Depends(...)]`）。 +- **多租户只在入口校验，查询时忘了带租户条件**：越权读到别的租户数据。把租户作为数据访问层的必传参数。 +- **兜底异常处理器把 `str(exc)` 返回给调用方**：泄露数据库地址、内部路径。 +- **测试后不清理 `dependency_overrides`**：影响其他测试。 +- **API Key 明文存储、用 `==` 比较**：存哈希，用 `secrets.compare_digest` 比较以防时序攻击。 +- **自定义路由用了 `/docs`、`/redoc`、`/openapi.json`**：和 FastAPI 自带的文档页面冲突（本课示例最初就踩了这个坑，`GET /docs` 返回的是 Swagger 页面）。换路径，或者 `FastAPI(docs_url=None)` 关闭/改名自带文档——生产环境通常也应该关闭或加鉴权。 +- **在依赖里做很重的操作**（每个请求都查一次配置中心）：考虑缓存或在应用启动时加载。 + +## 面试常问 + +见 [FastAPI 面试题](interview.md)：Depends 的原理和用途、中间件和依赖的区别、怎么统一错误格式、怎么做多租户隔离。 diff --git a/topics/python/fastapi/03-streaming-testing.md b/topics/python/fastapi/03-streaming-testing.md new file mode 100644 index 0000000..3eff8d5 --- /dev/null +++ b/topics/python/fastapi/03-streaming-testing.md @@ -0,0 +1,134 @@ +--- +id: python.fastapi.streaming-testing +title: 流式输出（SSE）、生命周期、测试和部署 +type: lesson +domain: python +stack: [fastapi] +tags: [SSE, StreamingResponse, lifespan, TestClient, pytest, uvicorn] +level: intermediate +prerequisites: [python.fastapi.deps-auth-errors] +est_hours: 3 +related: [python.fastapi.scenario.streaming-chat-api, devops.nginx.https-stream-limit, python.fastapi.interview] +week: [5, 12] +order: 3 +--- + +# 流式输出（SSE）、生命周期、测试和部署 + +## 解释说明 + +**SSE 流式输出** + +- 大模型逐字生成，用 `StreamingResponse(异步生成器, media_type="text/event-stream")` 边生成边发送。 +- 每个事件的格式：`data: 内容\n\n`（以空行结束）；可以加 `event: 名字` 区分事件类型（如 `error`）。内容里有换行时用 JSON 编码。 +- **响应头一旦发出就不能再修改状态码**：生成中途出错，只能在流里发送一个错误事件，前端据此提示。 +- **客户端断开**：用 `await request.is_disconnected()` 检查，断开就停止调用模型，节省费用。 +- 经过 Nginx 时要关闭代理缓冲（响应头 `X-Accel-Buffering: no`），见 [Nginx 流式代理](../../devops/nginx/02-https-stream-limit.md)。 +- 前端：`EventSource` 只支持 GET；POST 请求体（对话消息）通常用 `fetch` + `ReadableStream` 自己解析 SSE。 + +**生命周期 lifespan**：应用启动时创建共享资源（HTTP 客户端连接池、数据库连接池、加载模型配置），关闭时释放。不要每个请求新建 `httpx.AsyncClient`。 + +**测试** + +- `TestClient(app)`：同步写法测试，内部运行事件循环，最常用； +- `httpx.AsyncClient(transport=ASGITransport(app=app))`：在异步测试中使用； +- `app.dependency_overrides`：替换模型客户端、数据库、鉴权； +- 测试里**永远不要调用真实的模型 API**——慢、花钱、结果不稳定。 + +**部署**：`uvicorn app:app --host 0.0.0.0 --port 8000 --workers 4`；或 gunicorn + uvicorn worker。每个 worker 是独立进程，内存里的状态（缓存、限流计数）不共享，需要共享的放 Redis。容器化见 [Docker 专题](../../devops/docker/index.md)。 + +## 案例 + +RAG 问答接口：lifespan 中创建共享的 `httpx.AsyncClient`（连接池、超时）；问答接口先检索，然后流式输出答案，最后发一个 `event: sources` 事件携带引用来源；测试用假的检索器和模型客户端，验证事件顺序。 + +## 代码示例 + +```python +import asyncio +import json +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager + +import httpx +from fastapi import FastAPI, Request +from fastapi.responses import StreamingResponse +from fastapi.testclient import TestClient + + +@asynccontextmanager +async def lifespan(app: FastAPI): + # 启动：创建共享资源（连接池复用，设置超时） + app.state.http = httpx.AsyncClient(timeout=httpx.Timeout(30, connect=5)) + yield + # 关闭：释放资源 + await app.state.http.aclose() + + +app = FastAPI(lifespan=lifespan) + + +async def fake_answer(question: str) -> AsyncIterator[str]: + for word in ["根据", "制度", "，", "年假", "为", "5", "天", "。"]: + await asyncio.sleep(0.01) + yield word + + +def sse(data, event: str | None = None) -> str: + head = f"event: {event}\n" if event else "" + return f"{head}data: {json.dumps(data, ensure_ascii=False)}\n\n" + + +@app.get("/ask") +async def ask(q: str, request: Request): + async def events() -> AsyncIterator[str]: + async for token in fake_answer(q): + if await request.is_disconnected(): + return # 客户端断开，停止生成 + yield sse({"delta": token}) + yield sse([{"title": "员工手册", "section": "3.2"}], event="sources") + yield "data: [DONE]\n\n" + + return StreamingResponse(events(), media_type="text/event-stream", + headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}) + + +# 测试：with 语句会触发 lifespan 的启动和关闭 +with TestClient(app) as client: + assert isinstance(app.state.http, httpx.AsyncClient) + with client.stream("GET", "/ask", params={"q": "年假几天"}) as r: + body = "".join(r.iter_text()) + blocks = body.strip().split("\n\n") + answer = "".join(json.loads(b[6:])["delta"] for b in blocks if b.startswith("data: {")) + print(answer) # 根据制度，年假为5天。 + print(blocks[-2].splitlines()[0], blocks[-1]) # event: sources data: [DONE] +print(app.state.http.is_closed) # True：关闭时已释放 +``` + +pytest 中的异步测试写法（需要 `pytest-asyncio` 或 `anyio` 插件）： + +```python +import httpx +import pytest + + +@pytest.mark.anyio +async def test_health(): + from app import app # 你的应用 + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + r = await client.get("/healthz") + assert r.status_code == 200 +``` + +## 易错点 + +- **流式响应中途出错仍想返回 500**：做不到，响应头已发出。在流里发送错误事件。 +- **不检查客户端断开**：用户关掉页面，后端仍在生成并计费。 +- **每个请求创建新的 HTTP 客户端**：无法复用连接，延迟高，还可能耗尽文件描述符。 +- **用 `TestClient(app)` 但不用 `with`**：lifespan 不会执行，`app.state` 上的资源不存在。 +- **多 worker 部署后，内存缓存 / 限流 / 会话状态各进程不一致**：共享状态放 Redis。 +- **测试调用真实模型**：CI 不稳定且产生费用。 + +## 面试常问 + +见 [FastAPI 面试题](interview.md)：怎么实现流式输出、SSE 和 WebSocket 的区别、怎么测试依赖外部服务的接口、部署时 worker 数怎么定。 diff --git a/topics/python/fastapi/examples/chat-api/app.py b/topics/python/fastapi/examples/chat-api/app.py new file mode 100644 index 0000000..8c284ac --- /dev/null +++ b/topics/python/fastapi/examples/chat-api/app.py @@ -0,0 +1,152 @@ +# requires: fastapi, httpx +"""流式对话 API：鉴权、参数校验、请求 ID、统一错误格式、SSE 流式输出、依赖注入替换模型客户端。 + +运行：uvicorn app:app --reload（或 python app.py 打印一次演示） +测试：python -m pytest -q test_app.py +模型客户端是假的（逐字返回固定文本），换成真实 SDK 时只需要实现同样的 stream 方法。 +""" +from __future__ import annotations + +import asyncio +import json +import logging +import time +import uuid +from collections.abc import AsyncIterator +from typing import Annotated, Literal, Protocol + +from fastapi import Depends, FastAPI, Header, HTTPException, Request +from fastapi.exceptions import RequestValidationError +from fastapi.responses import JSONResponse, StreamingResponse +from pydantic import BaseModel, Field + +log = logging.getLogger("chat-api") + + +# ---------- 模型 ---------- +class Message(BaseModel): + role: Literal["system", "user", "assistant"] + content: str = Field(min_length=1, max_length=4000) + + +class ChatRequest(BaseModel): + messages: list[Message] = Field(min_length=1, max_length=50) + temperature: float = Field(default=0.3, ge=0, le=2) + stream: bool = True + + +class ChatResponse(BaseModel): + request_id: str + content: str + + +# ---------- 依赖 ---------- +class LLMClient(Protocol): + def stream(self, messages: list[Message], temperature: float) -> AsyncIterator[str]: ... + + +class FakeLLM: + async def stream(self, messages: list[Message], temperature: float) -> AsyncIterator[str]: + answer = f"你问的是：{messages[-1].content}。这是一个演示回答。" + for ch in answer: + await asyncio.sleep(0.005) + yield ch + + +def get_llm() -> LLMClient: + return FakeLLM() + + +API_KEYS = {"demo-key": "tenant-a"} # 演示用；真实项目从数据库或密钥服务读取，并做哈希比对 + + +def current_tenant(x_api_key: Annotated[str | None, Header()] = None) -> str: + tenant = API_KEYS.get(x_api_key or "") + if tenant is None: + raise HTTPException(status_code=401, detail="API Key 无效") + return tenant + + +# ---------- 应用 ---------- +app = FastAPI(title="chat-api") + + +@app.middleware("http") +async def request_id_and_timing(request: Request, call_next): + rid = request.headers.get("X-Request-ID") or uuid.uuid4().hex[:16] + request.state.request_id = rid + start = time.perf_counter() + response = await call_next(request) + response.headers["X-Request-ID"] = rid + log.info("%s %s %s %.1fms", request.method, request.url.path, response.status_code, + (time.perf_counter() - start) * 1000) + return response + + +def error_body(request: Request, code: str, message: str, details=None) -> dict: + body = {"code": code, "message": message, "request_id": getattr(request.state, "request_id", "")} + if details is not None: + body["details"] = details + return body + + +@app.exception_handler(HTTPException) +async def http_error(request: Request, exc: HTTPException): + return JSONResponse(error_body(request, f"HTTP_{exc.status_code}", str(exc.detail)), status_code=exc.status_code) + + +@app.exception_handler(RequestValidationError) +async def validation_error(request: Request, exc: RequestValidationError): + details = [{"field": ".".join(str(p) for p in e["loc"][1:]), "message": e["msg"]} for e in exc.errors()] + return JSONResponse(error_body(request, "INVALID_PARAMS", "参数校验失败", details), status_code=422) + + +@app.exception_handler(Exception) +async def unhandled_error(request: Request, exc: Exception): + log.exception("未处理的异常") # 详细信息只进日志，不返回给调用方 + return JSONResponse(error_body(request, "INTERNAL", "服务器内部错误"), status_code=500) + + +@app.get("/healthz") +async def healthz() -> dict: + return {"status": "ok"} + + +@app.post("/v1/chat", response_model=None) +async def chat(req: ChatRequest, request: Request, + tenant: Annotated[str, Depends(current_tenant)], + llm: Annotated[LLMClient, Depends(get_llm)]): + rid = request.state.request_id + if not req.stream: + content = "".join([t async for t in llm.stream(req.messages, req.temperature)]) + return ChatResponse(request_id=rid, content=content) + + async def events() -> AsyncIterator[str]: + try: + async for token in llm.stream(req.messages, req.temperature): + if await request.is_disconnected(): # 客户端断开就停止生成，节省模型费用 + log.info("客户端已断开，停止生成 request_id=%s", rid) + return + yield f"data: {json.dumps({'delta': token}, ensure_ascii=False)}\n\n" + yield "data: [DONE]\n\n" + except Exception: + log.exception("生成失败 request_id=%s", rid) + # 响应头已经发出，不能再改状态码，只能在流里发送错误事件 + yield f"event: error\ndata: {json.dumps({'message': '生成失败', 'request_id': rid}, ensure_ascii=False)}\n\n" + + return StreamingResponse(events(), media_type="text/event-stream", + headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}) + + +@app.get("/boom") +async def boom(): + raise RuntimeError("数据库连接串 postgres://secret@...") # 演示：内部细节不会泄露给调用方 + + +if __name__ == "__main__": + from fastapi.testclient import TestClient + + with TestClient(app) as c: + r = c.post("/v1/chat", headers={"X-API-Key": "demo-key"}, + json={"messages": [{"role": "user", "content": "年假怎么算"}], "stream": False}) + print(r.status_code, r.json()) diff --git a/topics/python/fastapi/examples/chat-api/test_app.py b/topics/python/fastapi/examples/chat-api/test_app.py new file mode 100644 index 0000000..15b7467 --- /dev/null +++ b/topics/python/fastapi/examples/chat-api/test_app.py @@ -0,0 +1,90 @@ +# requires: fastapi, httpx +import json +from collections.abc import AsyncIterator + +from fastapi.testclient import TestClient + +from app import Message, app, get_llm + +AUTH = {"X-API-Key": "demo-key"} +BODY = {"messages": [{"role": "user", "content": "你好"}]} + + +class ScriptedLLM: + """测试替身：按脚本返回 token，可以模拟中途失败。""" + + def __init__(self, tokens: list[str], fail_at: int | None = None) -> None: + self.tokens, self.fail_at = tokens, fail_at + self.calls: list[list[Message]] = [] + + async def stream(self, messages: list[Message], temperature: float) -> AsyncIterator[str]: + self.calls.append(messages) + for i, t in enumerate(self.tokens): + if i == self.fail_at: + raise RuntimeError("上游模型 500") + yield t + + +def client_with(llm) -> TestClient: + app.dependency_overrides[get_llm] = lambda: llm + return TestClient(app, raise_server_exceptions=False) + + +def parse_sse(text: str) -> list[tuple[str, str]]: + events = [] + for block in text.strip().split("\n\n"): + name, data = "message", "" + for line in block.splitlines(): + if line.startswith("event: "): + name = line[7:] + elif line.startswith("data: "): + data = line[6:] + events.append((name, data)) + return events + + +def teardown_function(): + app.dependency_overrides.clear() + + +def test_auth_required(): + r = client_with(ScriptedLLM(["x"])).post("/v1/chat", json=BODY) + assert r.status_code == 401 and r.json()["code"] == "HTTP_401" and r.json()["request_id"] + + +def test_validation_error_format(): + r = client_with(ScriptedLLM(["x"])).post("/v1/chat", headers=AUTH, + json={"messages": [{"role": "robot", "content": ""}], "temperature": 5}) + assert r.status_code == 422 + fields = {d["field"] for d in r.json()["details"]} + assert {"messages.0.role", "messages.0.content", "temperature"} <= fields + + +def test_non_stream(): + llm = ScriptedLLM(["你", "好"]) + r = client_with(llm).post("/v1/chat", headers={**AUTH, "X-Request-ID": "rid-1"}, json={**BODY, "stream": False}) + assert r.status_code == 200 + assert r.json() == {"request_id": "rid-1", "content": "你好"} + assert r.headers["X-Request-ID"] == "rid-1" + assert llm.calls[0][0].content == "你好" + + +def test_stream_sse(): + r = client_with(ScriptedLLM(["A", "B", "C"])).post("/v1/chat", headers=AUTH, json=BODY) + assert r.status_code == 200 and r.headers["content-type"].startswith("text/event-stream") + events = parse_sse(r.text) + assert [json.loads(d)["delta"] for n, d in events[:-1]] == ["A", "B", "C"] + assert events[-1] == ("message", "[DONE]") + + +def test_stream_error_event(): + r = client_with(ScriptedLLM(["A", "B"], fail_at=1)).post("/v1/chat", headers=AUTH, json=BODY) + events = parse_sse(r.text) + assert events[0][0] == "message" and events[-1][0] == "error" + assert json.loads(events[-1][1])["message"] == "生成失败" + + +def test_internal_error_hides_details(): + r = client_with(ScriptedLLM([])).get("/boom") + assert r.status_code == 500 + assert r.json()["message"] == "服务器内部错误" and "postgres" not in r.text diff --git a/topics/python/fastapi/index.md b/topics/python/fastapi/index.md index e0c3b37..4e53162 100644 --- a/topics/python/fastapi/index.md +++ b/topics/python/fastapi/index.md @@ -4,12 +4,39 @@ title: FastAPI type: index domain: python stack: [fastapi] -tags: [] +tags: [FastAPI, Web 框架, pydantic, SSE] week: [4, 5, 12] --- # FastAPI -路由与校验、依赖注入、中间件与鉴权、流式输出、测试。 +Python AI 服务最常用的 Web 框架：基于类型标注自动做参数校验和生成 OpenAPI 文档，原生支持 async 和流式响应。你熟悉的 Express / Koa、以及本计划里的 Gin，概念都能对应上：路由 ↔ 路由，中间件 ↔ 中间件，`Depends` ↔ 依赖注入（Gin 里通常手动传）。 -> 本专题建设中：课程、企业场景、面试题和示例代码会陆续补充。 +## 学习路线 + +| 顺序 | 课程 | 级别 | 预计 | +| --- | --- | --- | --- | +| 1 | [路由和参数校验：路径 / 查询 / 请求体、pydantic 模型、响应模型](01-routing-validation.md) | 基础 | 2 小时 | +| 2 | [依赖注入、鉴权、中间件和统一错误处理](02-deps-auth-errors.md) | 进阶 | 3 小时 | +| 3 | [流式输出（SSE）、生命周期、测试和部署](03-streaming-testing.md) | 进阶 | 3 小时 | + +## 企业场景 + +- [大模型对话 API：鉴权、统一错误、SSE 流式、客户端断开即停止](scenarios/streaming-chat-api.md) + +## 面试题 + +- [FastAPI 面试题](interview.md) + +## 示例代码 + +- [examples/chat-api](examples/chat-api/)：完整的流式对话 API（API Key 鉴权、请求 ID、统一错误格式、SSE、依赖替换）+ 6 个测试；模型客户端是假实现，不需要 API Key + +## 在 16 周计划中 + +- [第 4 周：FastAPI 基础](../../../plan/week04-python.md)、[第 5 周：RAG 服务](../../../plan/week05-rag-1.md)、[第 12 周：Agent 服务化](../../../plan/week12-agent-m3-persistence.md) + +## 官方资料 + +- FastAPI 文档（有中文版）：https://fastapi.tiangolo.com/zh/ +- pydantic v2：https://docs.pydantic.dev/latest/ diff --git a/topics/python/fastapi/interview.md b/topics/python/fastapi/interview.md new file mode 100644 index 0000000..1fb877b --- /dev/null +++ b/topics/python/fastapi/interview.md @@ -0,0 +1,59 @@ +--- +id: python.fastapi.interview +title: FastAPI 面试题 +type: interview +domain: python +stack: [fastapi] +tags: [FastAPI, pydantic, 依赖注入, 面试] +level: intermediate +related: [python.fastapi.routing-validation, python.fastapi.deps-auth-errors, python.fastapi.streaming-testing] +week: [4, 5, 16] +--- + +# FastAPI 面试题 + +## 基础 + +### 【基础】FastAPI 有什么特点？为什么 AI 服务常用它？ + +**参考答案**：基于 Starlette（ASGI）和 pydantic，原生支持 async，适合 IO 密集的模型调用；通过类型标注自动完成参数解析和校验、生成 OpenAPI 文档；依赖注入方便组织鉴权、数据库会话和测试替换；支持流式响应和 WebSocket。AI 生态（模型 SDK、LangChain、向量库客户端）以 Python 为主，用 FastAPI 封装服务很自然。 + +### 【基础】FastAPI 怎么区分路径参数、查询参数和请求体？ + +**参考答案**：在路径模板中出现的参数是路径参数；简单类型（int、str、bool 等）的其他参数是查询参数；pydantic 模型类型的参数是请求体；请求头、Cookie 用 `Header()`、`Cookie()` 显式声明。也可以用 `Path()`、`Query()`、`Body()` 显式指定并添加校验规则。 + +### 【基础】路由函数用 `async def` 和 `def` 有什么区别？ + +**参考答案**：`async def` 在事件循环中直接执行，里面只能调用非阻塞的异步代码；普通 `def` 会被放到线程池中执行，适合调用同步阻塞库。在 `async def` 中调用阻塞函数会卡住整个事件循环，是最常见的性能问题。 + +### 【基础】响应模型有什么作用？ + +**参考答案**：声明接口返回的数据结构：自动过滤掉未声明的字段（防止泄露内部字段），校验和序列化返回数据，并体现在 OpenAPI 文档中，前端可以据此生成类型。 + +## 进阶 + +### 【进阶】`Depends` 的原理和常见用途？ + +**参考答案**：FastAPI 在处理请求时解析路由函数签名，递归解析依赖树，按顺序调用依赖函数并把返回值注入参数；同一请求内相同依赖默认只调用一次；带 `yield` 的依赖可以在请求结束后执行清理。用途：鉴权和获取当前用户 / 租户、数据库会话、分页参数、共享客户端；测试时通过 `dependency_overrides` 替换。 + +### 【进阶】中间件和依赖都能做鉴权，怎么选？ + +**参考答案**：中间件作用于所有请求，适合与业务无关的横切逻辑：请求 ID、日志、耗时、CORS。依赖按路由或路由器声明，能拿到类型化的结果（当前用户）并注入处理函数，能在测试中替换，适合鉴权、权限、租户。通常鉴权用依赖。 + +### 【进阶】怎么实现大模型的流式输出？要注意什么？ + +**参考答案**：用 `StreamingResponse` 包装异步生成器，`media_type="text/event-stream"`，每个事件 `data: ...\n\n`。注意：响应头发出后不能改状态码，错误要以事件形式发送；检查客户端断开及时停止生成；内容用 JSON 编码避免换行破坏格式；经过 Nginx 时关闭缓冲；前端 POST 请求需要用 fetch 读取流。 + +## 场景题 + +### 【场景题】压测发现 FastAPI 服务 QPS 很低，CPU 也不高，可能是什么原因？ + +**参考答案**：最常见的是在 `async def` 路由里调用了阻塞操作（同步数据库驱动、`requests`、`time.sleep`、同步 SDK），事件循环被阻塞。其他原因：每个请求新建 HTTP 客户端或数据库连接、连接池太小、下游（模型、数据库）本身慢、只启动了一个 worker。排查：看慢请求的耗时分布，用 py-spy 看事件循环线程在做什么。 + +### 【场景题】一个接口依赖大模型和向量数据库，怎么写单元测试？ + +**参考答案**：把模型客户端和检索器抽象成接口（Protocol），通过依赖注入获取；测试中用 `app.dependency_overrides` 替换为假实现，返回固定内容或模拟异常；用 `TestClient` 调用接口，断言状态码、响应格式、流式事件顺序、错误处理。真实模型的效果评估另外做（评测集），不放在单元测试里。 + +### 【场景题】给多个业务系统提供 AI 能力，怎么防止某个系统把额度用光？ + +**参考答案**：每个调用方一个 API Key，鉴权依赖解析出租户；在服务端按租户做限流（Redis 滑动窗口 / 令牌桶）和配额（按日 / 月的 token 数）；记录每次调用的 token 用量写入数据库；超过配额返回 429 并给出提示；对异常增长告警。Nginx 层可以再加一层按 IP 的粗粒度限流。 diff --git a/topics/python/fastapi/scenarios/streaming-chat-api.md b/topics/python/fastapi/scenarios/streaming-chat-api.md new file mode 100644 index 0000000..8deb96e --- /dev/null +++ b/topics/python/fastapi/scenarios/streaming-chat-api.md @@ -0,0 +1,112 @@ +--- +id: python.fastapi.scenario.streaming-chat-api +title: 大模型对话 API：鉴权、统一错误、SSE 流式、客户端断开即停止 +type: scenario +domain: python +stack: [fastapi, asyncio, nginx] +tags: [对话 API, SSE, 鉴权, 错误处理, 可测试性] +level: intermediate +prerequisites: [python.fastapi.streaming-testing] +est_hours: 2 +related: [python.fastapi.interview, devops.nginx.scenario.ai-gateway-sse-canary] +week: [5, 12] +--- + +# 大模型对话 API：鉴权、统一错误、SSE 流式、客户端断开即停止 + +## 业务背景与问题 + +公司要把大模型能力封装成内部统一的对话服务，给客服系统、OA、知识库等多个业务调用，而不是每个业务各自直连模型服务商。要求： + +- 各业务系统用自己的 API Key 调用，按租户统计用量； +- 支持流式（打字机效果）和非流式两种模式； +- 错误格式统一，调用方能程序化处理； +- 出了问题能用请求 ID 串起调用方和服务端的日志； +- 未来可以切换模型服务商而不影响调用方。 + +## 约束 + +- 第一版只有两三个人开发，要求简单可靠。 +- 模型调用按 token 计费，用户关闭页面后不应继续生成。 +- 模型 API Key 只保存在这个服务里，不下发给业务系统。 + +## 方案设计 + +```mermaid +sequenceDiagram + participant B as 业务系统 + participant N as Nginx + participant A as 对话服务（FastAPI） + participant L as 模型服务商 + B->>N: POST /v1/chat（X-API-Key, X-Request-ID） + N->>A: 转发（关闭缓冲） + A->>A: 依赖：校验 API Key → 租户；pydantic 校验参数 + A->>L: 流式请求 + loop 每个 token + L-->>A: delta + A-->>B: data: {"delta": ...} + A->>A: 客户端断开？→ 停止 + end + A-->>B: data: [DONE]（或 event: error） +``` + +1. **分层**：路由层（校验、鉴权、响应格式）→ 服务层（组装提示词、计量）→ 模型客户端（`LLMClient` 协议，可替换服务商）。 +2. **鉴权**：`current_tenant` 依赖，API Key → 租户；后续计量、限流都按租户。 +3. **统一错误**：`HTTPException`、`RequestValidationError`、兜底异常都转成 `{code, message, request_id}`；兜底异常只记日志不返回细节。 +4. **请求 ID**：中间件读取或生成 `X-Request-ID`，写进响应头和日志。 +5. **流式**：SSE；中途出错发送 `event: error`；客户端断开立即停止。 +6. **可测试**：模型客户端通过依赖注入，测试中替换为脚本化的假实现。 + +## 关键代码 + +完整代码和测试见 [examples/chat-api](../examples/chat-api/)： + +```python +@app.post("/v1/chat", response_model=None) +async def chat(req: ChatRequest, request: Request, + tenant: Annotated[str, Depends(current_tenant)], + llm: Annotated[LLMClient, Depends(get_llm)]): + rid = request.state.request_id + if not req.stream: + content = "".join([t async for t in llm.stream(req.messages, req.temperature)]) + return ChatResponse(request_id=rid, content=content) + + async def events(): + try: + async for token in llm.stream(req.messages, req.temperature): + if await request.is_disconnected(): # 用户走了，停止生成 + return + yield f"data: {json.dumps({'delta': token}, ensure_ascii=False)}\n\n" + yield "data: [DONE]\n\n" + except Exception: + log.exception("生成失败 request_id=%s", rid) + yield f"event: error\ndata: {json.dumps({'message': '生成失败', 'request_id': rid}, ensure_ascii=False)}\n\n" + + return StreamingResponse(events(), media_type="text/event-stream", + headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}) +``` + +测试覆盖：未带 API Key 返回 401 统一格式；参数错误返回 422 并列出 `messages.0.role` 等字段；非流式返回完整内容并透传请求 ID；流式事件顺序正确并以 `[DONE]` 结束；模型中途失败时发送 `error` 事件；内部异常不泄露连接串。 + +## 取舍 + +| 选择 | 优点 | 缺点 | +| --- | --- | --- | +| **SSE（本方案）** | 基于 HTTP，简单，代理和网关支持好 | 单向；`EventSource` 不支持 POST，前端要用 fetch 解析 | +| WebSocket | 双向，适合语音、实时协作 | 连接管理和负载均衡更复杂 | +| 统一对话服务（本方案） | 密钥集中、可计量、可切换服务商 | 多一跳；需要保证高可用 | +| 业务直连模型服务商 | 少一跳 | 密钥分散、无法统一计量和治理 | +| 直接使用 LangChain 等框架封装 | 功能多 | 对一个转发 + 治理的服务来说偏重，版本变化快 | + +## 易错点 + +- 流式中途异常想改状态码：响应头已发出，只能发错误事件。 +- 不检查断开：页面关了模型还在生成。 +- 统一错误处理器漏掉 `RequestValidationError`：422 的格式和其他错误不一致。 +- 兜底异常返回 `str(e)`：可能包含连接串、内部地址。 +- Nginx 没关缓冲：流式变整段（见 [AI 应用网关场景](../../../devops/nginx/scenarios/ai-gateway-sse-canary.md)）。 +- 只在内存里计量：多 worker / 多实例数据不准，要写数据库或 Redis。 + +## 面试怎么讲 + +> “我做了一个内部统一的对话服务，FastAPI 实现。鉴权用依赖注入把 API Key 映射到租户，便于计量和限流；所有错误转成统一的 code/message/request_id 格式，内部异常只记日志；流式用 SSE，中途出错发 error 事件，客户端断开就停止生成以节省 token；模型客户端抽象成协议，测试里用假实现替换，覆盖了鉴权、校验、流式和异常场景。” diff --git a/topics/python/python/01-syntax-for-js.md b/topics/python/python/01-syntax-for-js.md new file mode 100644 index 0000000..7392dfc --- /dev/null +++ b/topics/python/python/01-syntax-for-js.md @@ -0,0 +1,132 @@ +--- +id: python.python.syntax-for-js +title: 给 JS 开发者的 Python：语法、数据结构、推导式、函数、类 +type: lesson +domain: python +stack: [python] +tags: [语法, list, dict, 推导式, 解包, 类] +level: basic +est_hours: 3 +related: [python.python.typing-errors-testing, python.python.interview] +week: [4] +order: 1 +--- + +# 给 JS 开发者的 Python：语法、数据结构、推导式、函数、类 + +## 解释说明 + +| JavaScript / TypeScript | Python | 备注 | +| --- | --- | --- | +| `const` / `let` | 直接赋值 `x = 1` | 没有声明关键字；常量靠全大写约定 | +| `{}` 代码块 | 缩进（4 个空格） | 缩进就是语法 | +| `null` / `undefined` | `None` | 只有一个“空” | +| `true` / `false` | `True` / `False` | 首字母大写 | +| `&&` `||` `!` | `and` `or` `not` | | +| `===` | `==`（值相等）、`is`（同一对象） | 判断 None 用 `is None` | +| 模板字符串 `` `a${b}` `` | f-string `f"a{b}"` | | +| Array | `list` | `xs.append(1)`、`len(xs)`、切片 `xs[1:3]`、`xs[-1]` 取最后一个 | +| Object / Map | `dict` | `d["k"]`（不存在会抛 KeyError）、`d.get("k", 默认值)` | +| Set | `set` | `a & b` 交集、`a | b` 并集 | +| 不可变数组 | `tuple` | `(1, 2)`，常用于返回多个值 | +| `arr.map / filter` | 推导式 | `[x * 2 for x in xs if x > 0]` | +| 解构 `const [a, b] = arr` | 解包 `a, b = arr` | `first, *rest = xs` | +| 展开 `{...a, ...b}` | `{**a, **b}` 或 `a | b` | | +| 箭头函数 | `lambda x: x * 2` | 只能写一个表达式，复杂逻辑用 `def` | +| `try / catch` | `try / except` | | +| `for (const x of xs)` | `for x in xs:` | `for i, x in enumerate(xs)` 带下标 | +| `class` + `constructor` | `class` + `__init__` | 方法第一个参数是 `self` | + +**真值判断**：空字符串、`0`、`None`、空列表 / 字典 / 集合都是假，和 JS 类似；但 Python 里 `[]` 是假，JS 里 `[]` 是真。 + +**函数参数**：位置参数、关键字参数 `f(name="a")`、默认值、`*args`（剩余位置参数）、`**kwargs`（剩余关键字参数）、`*` 之后的参数只能用关键字传。 + +## 案例 + +把 JS 里一段常见的数据处理“翻译”成 Python：按用户聚合订单金额并取前三。 + +```js +// JS +const totals = {}; +for (const o of orders) totals[o.user] = (totals[o.user] ?? 0) + o.amount; +const top3 = Object.entries(totals).sort((a, b) => b[1] - a[1]).slice(0, 3); +``` + +## 代码示例 + +```python +from collections import Counter, defaultdict + +orders = [ + {"user": "alice", "amount": 30, "status": "paid"}, + {"user": "bob", "amount": 50, "status": "paid"}, + {"user": "alice", "amount": 20, "status": "refunded"}, + {"user": "carol", "amount": 70, "status": "paid"}, + {"user": "alice", "amount": 40, "status": "paid"}, +] + +# 推导式 = map + filter +paid = [o for o in orders if o["status"] == "paid"] +amounts = [o["amount"] for o in paid] +print(sum(amounts), max(amounts)) # 190 70 + +# defaultdict：不存在的 key 自动给默认值（类似 ?? 0） +totals: defaultdict[str, int] = defaultdict(int) +for o in paid: + totals[o["user"]] += o["amount"] +top3 = sorted(totals.items(), key=lambda kv: kv[1], reverse=True)[:3] +print(top3) # [('alice', 70), ('carol', 70), ('bob', 50)] + +# Counter：计数专用 +print(Counter(o["status"] for o in orders).most_common()) # [('paid', 4), ('refunded', 1)] + +# 字典推导式、解包、合并 +by_user = {u: t for u, t in totals.items() if t >= 60} +first, *rest = sorted(totals) +defaults = {"page": 1, "size": 20} +params = defaults | {"size": 50} # Python 3.9+，相当于 {...defaults, size: 50} +print(by_user, first, rest, params) + + +# 函数：默认值、关键字参数、只能用关键字传的参数 +def paginate(items: list, *, page: int = 1, size: int = 20) -> list: + start = (page - 1) * size + return items[start:start + size] + + +print(paginate(list(range(100)), page=3, size=10)) # [20, ..., 29] + + +# 类 +class Cart: + def __init__(self, owner: str) -> None: + self.owner = owner + self.items: list[tuple[str, int]] = [] + + def add(self, sku: str, qty: int = 1) -> "Cart": + self.items.append((sku, qty)) + return self + + def __len__(self) -> int: # 让 len(cart) 可用 + return sum(q for _, q in self.items) + + def __repr__(self) -> str: # 打印时的样子，类似 toString + return f"Cart({self.owner!r}, {len(self)} 件)" + + +print(Cart("alice").add("pen", 2).add("book")) # Cart('alice', 3 件) +``` + +## 易错点 + +- **可变默认参数**：`def f(xs=[])` 的默认列表只创建一次，多次调用共享。写成 `def f(xs=None): xs = xs or []`（或 `if xs is None`）。 +- **`dict["k"]` 不存在会抛异常**，不像 JS 返回 `undefined`。用 `.get()`。 +- **整数除法**：`7 / 2 == 3.5`，`7 // 2 == 3`。 +- **浮点数算金额**：`0.1 + 0.2 != 0.3`，金额用 `Decimal` 或整数“分”。 +- **`is` 和 `==` 混用**：比较值用 `==`，只有 `None`、`True`、`False` 用 `is`。 +- **循环里修改正在遍历的列表**：结果出乎意料，遍历副本或生成新列表。 +- **缩进混用 Tab 和空格**：直接报错。编辑器统一 4 个空格。 + +## 面试常问 + +见 [Python 面试题](interview.md)：可变与不可变类型、可变默认参数、`is` 和 `==`、深拷贝和浅拷贝、推导式和生成器。 diff --git a/topics/python/python/02-typing-errors-testing.md b/topics/python/python/02-typing-errors-testing.md new file mode 100644 index 0000000..de2f695 --- /dev/null +++ b/topics/python/python/02-typing-errors-testing.md @@ -0,0 +1,147 @@ +--- +id: python.python.typing-errors-testing +title: 类型标注、dataclass、异常和上下文管理器、pytest +type: lesson +domain: python +stack: [python] +tags: [类型标注, dataclass, Protocol, 异常, with, 生成器, pytest] +level: basic +prerequisites: [python.python.syntax-for-js] +est_hours: 3 +related: [python.python.scenario.order-reconciliation, python.asyncio.coroutines-tasks, python.python.interview] +week: [4] +order: 2 +--- + +# 类型标注、dataclass、异常和上下文管理器、pytest + +## 解释说明 + +**类型标注**：和 TypeScript 不同，Python 的类型标注**运行时不检查**，只给编辑器和类型检查器（mypy、pyright）用；但 pydantic、FastAPI 会在运行时读取标注做校验。 + +- 基本写法：`def f(name: str, tags: list[str] | None = None) -> dict[str, int]:` +- `Protocol`：结构化类型（鸭子类型），类似 TS 的 interface——只要有对应的方法就算实现了，不需要继承。 +- `TypedDict`：给字典描述结构；`Literal["a", "b"]`：字面量联合类型。 + +**dataclass**：自动生成 `__init__`、`__repr__`、`__eq__` 的数据类，`frozen=True` 不可变，`slots=True` 省内存。需要运行时校验（接口入参、配置）时用 pydantic。 + +**异常**：`try / except 具体异常 / else（没出错时执行）/ finally（总会执行）`；自定义异常继承 `Exception`；`raise 新异常 from 原异常` 保留原因链。**不要裸写 `except:`**，至少写 `except Exception`，并且要记录或重新抛出。 + +**上下文管理器**：`with open(...) as f:` 离开代码块时自动关闭，类似 JS 的 `try/finally`。自己写用 `@contextmanager`。 + +**生成器**：函数里用 `yield`，按需逐个产生值，不一次性把所有数据读进内存——处理大文件、分页拉取数据时很有用。 + +**pytest**：文件名 `test_*.py`，函数名 `test_*`，直接用 `assert`；`pytest.raises` 断言异常；`@pytest.mark.parametrize` 表驱动；fixture（`tmp_path`、`monkeypatch`、`caplog` 等）提供测试依赖。 + +## 案例 + +对账脚本里：记录用 `@dataclass(frozen=True)`；数据来源（本地 CSV、渠道 API）定义成 `Protocol`，测试时传假实现；每一行解析错误抛自定义异常并带上行号；读取大文件用生成器逐行处理。完整代码见[订单对账场景](scenarios/order-reconciliation.md)。 + +## 代码示例 + +```python +import logging +from collections.abc import Iterator +from contextlib import contextmanager +from dataclasses import dataclass, field +from decimal import Decimal +from typing import Protocol +import time + +log = logging.getLogger(__name__) + + +@dataclass(frozen=True, slots=True) +class Order: + id: str + amount: Decimal + tags: tuple[str, ...] = () # 不可变默认值；可变的要用 field(default_factory=list) + + +@dataclass +class Batch: + name: str + orders: list[Order] = field(default_factory=list) + + +class OrderSource(Protocol): # 任何有 fetch 方法的类都满足这个协议 + def fetch(self) -> Iterator[Order]: ... + + +class FakeSource: + def fetch(self) -> Iterator[Order]: + yield Order("A1", Decimal("9.90")) + yield Order("A2", Decimal("20.00"), ("vip",)) + + +class ImportError_(Exception): # 自定义异常（避免和内置 ImportError 重名） + def __init__(self, line: int, reason: str) -> None: + super().__init__(f"第 {line} 行：{reason}") + self.line = line + + +def parse_amount(line: int, raw: str) -> Decimal: + try: + return Decimal(raw) + except ArithmeticError as e: # decimal.InvalidOperation 是 ArithmeticError 的子类 + raise ImportError_(line, f"金额格式错误 {raw!r}") from e + + +@contextmanager +def timer(name: str) -> Iterator[None]: + start = time.perf_counter() + try: + yield + finally: + log.info("%s 耗时 %.1f ms", name, (time.perf_counter() - start) * 1000) + + +def total(source: OrderSource) -> Decimal: + with timer("汇总"): + return sum((o.amount for o in source.fetch()), Decimal("0")) + + +logging.basicConfig(level=logging.INFO) +print(total(FakeSource())) # 29.90 +try: + parse_amount(3, "12,00") +except ImportError_ as e: + print(e, "| 原因：", type(e.__cause__).__name__) +``` + +对应的 pytest 测试： + +```python +# 文件：test_orders.py（和上面的代码放在 orders.py 中） +from decimal import Decimal + +import pytest + +from orders import FakeSource, ImportError_, parse_amount, total + + +def test_total(): + assert total(FakeSource()) == Decimal("29.90") + + +@pytest.mark.parametrize("raw, ok", [("1.5", True), ("abc", False), ("", False)]) +def test_parse_amount(raw, ok): + if ok: + assert parse_amount(1, raw) == Decimal(raw) + else: + with pytest.raises(ImportError_, match="第 1 行"): + parse_amount(1, raw) +``` + +## 易错点 + +- **以为类型标注会在运行时报错**：`def f(x: int)` 传字符串照样运行。需要校验用 pydantic。 +- **dataclass 用可变默认值**：`items: list = []` 会直接报错，用 `field(default_factory=list)`。 +- **`except Exception: pass`**：错误被悄悄吞掉，排查时毫无线索。 +- **`raise` 新异常时不加 `from`**：丢失原始原因（虽然 Python 会显示“During handling...”，但语义不清楚）。 +- **测试依赖真实网络、真实时间**：不稳定。用 Protocol + 假实现、`monkeypatch` 替换。 +- **生成器只能遍历一次**：第二次遍历是空的，需要多次使用就转成 list。 + +## 面试常问 + +见 [Python 面试题](interview.md)：类型标注的作用、dataclass 和 pydantic 的区别、生成器和迭代器、上下文管理器原理、异常处理规范。 diff --git a/topics/python/python/03-project-packaging.md b/topics/python/python/03-project-packaging.md new file mode 100644 index 0000000..4b4b9c9 --- /dev/null +++ b/topics/python/python/03-project-packaging.md @@ -0,0 +1,114 @@ +--- +id: python.python.project-packaging +title: 工程化：虚拟环境、uv / pip、项目结构、导入、日志和配置 +type: lesson +domain: python +stack: [python] +tags: [虚拟环境, uv, pip, pyproject.toml, 导入, logging, 配置] +level: basic +prerequisites: [python.python.syntax-for-js] +est_hours: 2 +related: [python.fastapi.routing-validation, python.python.interview] +week: [4] +order: 3 +--- + +# 工程化：虚拟环境、uv / pip、项目结构、导入、日志和配置 + +## 解释说明 + +| npm 生态 | Python 生态 | +| --- | --- | +| `package.json` | `pyproject.toml`（项目元数据和依赖） | +| `package-lock.json` | `uv.lock` / `poetry.lock`；老项目用 `requirements.txt` 锁版本 | +| `node_modules`（每个项目一份） | **虚拟环境** `.venv`（每个项目一份，需要“激活”或用工具代为执行） | +| `npm install` | `uv sync` / `pip install -r requirements.txt` | +| `npx` | `uv run` / `python -m 模块` | +| nvm | `uv python install 3.12` / pyenv | + +**为什么一定要用虚拟环境**：不同项目依赖的版本不同，全局安装会互相覆盖，系统 Python 也可能被破坏。 + +**uv**：近年流行的包管理工具，速度快，集成了虚拟环境、依赖锁定和 Python 版本管理。团队已有 pip / poetry 也完全可以，原理相同。命令以官方文档为准：https://docs.astral.sh/uv/ + +**项目结构**（推荐 `src` 布局或简单的包目录）： + +```text +ai-service/ +├── pyproject.toml +├── app/ +│ ├── __init__.py +│ ├── main.py # FastAPI 入口 +│ ├── config.py # 配置 +│ └── rag/ +│ ├── __init__.py +│ └── retriever.py +└── tests/ + └── test_retriever.py +``` + +**导入**：用包的绝对导入 `from app.rag.retriever import Retriever`；从项目根目录用 `python -m app.main` 或 `pytest` 运行，避免“直接运行子目录里的文件导致找不到包”。 + +**日志**：用 `logging` 而不是 `print`；每个模块 `log = logging.getLogger(__name__)`；在入口统一配置格式和级别；生产环境输出 JSON 便于收集。 + +**配置**：从环境变量读取（十二要素原则），用 pydantic-settings 做类型校验和默认值；`.env` 只在本地使用、不提交。 + +## 案例 + +AI 服务项目：`uv init` 创建项目，`uv add fastapi uvicorn httpx pydantic-settings`，`uv add --dev pytest ruff`；CI 里 `uv sync --frozen` 严格按锁文件安装；配置类读取 `LLM_API_KEY`、`DATABASE_URL`，缺失时启动即失败，而不是运行到一半才报错。 + +## 代码示例 + +```bash +# 用 uv（推荐） +uv init ai-service && cd ai-service +uv add fastapi "uvicorn[standard]" httpx pydantic-settings +uv add --dev pytest ruff +uv run pytest -q +uv run ruff check . + +# 用标准库 venv + pip（同样有效） +python3 -m venv .venv +source .venv/bin/activate # Windows：.venv\Scripts\activate +pip install fastapi uvicorn httpx pydantic-settings +pip freeze > requirements.txt +``` + +配置和日志（`app/config.py`）： + +```python +import logging + +from pydantic import Field, SecretStr +from pydantic_settings import BaseSettings, SettingsConfigDict + + +class Settings(BaseSettings): + model_config = SettingsConfigDict(env_file=".env", env_prefix="APP_") + + database_url: str = "postgresql://app:app@localhost:5432/app" + llm_api_key: SecretStr = Field(default=SecretStr("")) # 打印时显示为 ********** + llm_timeout_seconds: float = 30.0 + log_level: str = "INFO" + + +def setup_logging(level: str) -> None: + logging.basicConfig(level=level, format="%(asctime)s %(levelname)s %(name)s %(message)s") + + +settings = Settings() # 读取环境变量 APP_DATABASE_URL 等 +setup_logging(settings.log_level) +logging.getLogger(__name__).info("配置加载完成：%s", settings.model_dump(exclude={"llm_api_key"})) +``` + +## 易错点 + +- **不用虚拟环境，`sudo pip install`**：破坏系统 Python（很多 Linux 发行版已经禁止这样做）。 +- **`requirements.txt` 不锁版本**：今天能装好，下个月依赖升级后跑不起来。 +- **在子目录里直接 `python retriever.py`**：`from app...` 找不到包。从项目根目录用 `-m` 运行。 +- **文件名和标准库或第三方库同名**（如 `random.py`、`fastapi.py`）：导入的是你的文件，报奇怪的错误。 +- **到处 `print` 调试并留在代码里**：用 logging 并按级别控制。 +- **把密钥写进代码或日志**：用 `SecretStr`，日志里排除敏感字段。 + +## 面试常问 + +见 [Python 面试题](interview.md)：虚拟环境的作用、依赖如何锁定、GIL 是什么。 diff --git a/topics/python/python/examples/reconcile/reconcile.py b/topics/python/python/examples/reconcile/reconcile.py new file mode 100644 index 0000000..59f3ef4 --- /dev/null +++ b/topics/python/python/examples/reconcile/reconcile.py @@ -0,0 +1,120 @@ +"""订单对账：比较“我方订单”和“支付渠道账单”两份 CSV，输出差异。 + +运行：python reconcile.py（使用内置示例数据）或 python reconcile.py 我方.csv 渠道.csv +测试：python -m pytest -q test_reconcile.py +""" +from __future__ import annotations + +import csv +import io +import logging +import sys +from collections.abc import Iterable, Iterator +from dataclasses import dataclass +from decimal import Decimal, InvalidOperation +from enum import Enum +from pathlib import Path + +log = logging.getLogger("reconcile") + + +class DiffKind(str, Enum): + MISSING_IN_CHANNEL = "渠道缺失" # 我方有、渠道没有：可能没支付成功或渠道漏单 + MISSING_IN_OURS = "我方缺失" # 渠道有、我方没有：掉单（钱收了订单没更新） + AMOUNT_MISMATCH = "金额不一致" + + +@dataclass(frozen=True, slots=True) +class Record: + order_id: str + amount: Decimal # 金额一律用 Decimal，不用 float + + +@dataclass(frozen=True, slots=True) +class Diff: + kind: DiffKind + order_id: str + ours: Decimal | None + channel: Decimal | None + + +class BadRow(ValueError): + pass + + +def parse_rows(rows: Iterable[dict[str, str]], id_col: str, amount_col: str, cents: bool = False) -> Iterator[Record]: + for line_no, row in enumerate(rows, start=2): # 第 1 行是表头 + try: + oid = row[id_col].strip() + raw = row[amount_col].strip() + amount = Decimal(raw) / 100 if cents else Decimal(raw) + except (KeyError, InvalidOperation) as e: + raise BadRow(f"第 {line_no} 行格式错误：{row}") from e + if not oid: + raise BadRow(f"第 {line_no} 行缺少订单号") + yield Record(oid, amount.quantize(Decimal("0.01"))) + + +def to_map(records: Iterable[Record]) -> dict[str, Record]: + result: dict[str, Record] = {} + for r in records: + if r.order_id in result: + log.warning("订单号重复：%s，后出现的记录将被忽略", r.order_id) + continue + result[r.order_id] = r + return result + + +def reconcile(ours: Iterable[Record], channel: Iterable[Record]) -> list[Diff]: + a, b = to_map(ours), to_map(channel) + diffs: list[Diff] = [] + for oid in sorted(a.keys() | b.keys()): # 集合并集，排序后输出稳定 + x, y = a.get(oid), b.get(oid) + if y is None: + diffs.append(Diff(DiffKind.MISSING_IN_CHANNEL, oid, x.amount, None)) + elif x is None: + diffs.append(Diff(DiffKind.MISSING_IN_OURS, oid, None, y.amount)) + elif x.amount != y.amount: + diffs.append(Diff(DiffKind.AMOUNT_MISMATCH, oid, x.amount, y.amount)) + return diffs + + +def write_report(diffs: list[Diff], out) -> None: + w = csv.writer(out) + w.writerow(["差异类型", "订单号", "我方金额", "渠道金额"]) + for d in diffs: + w.writerow([d.kind.value, d.order_id, d.ours if d.ours is not None else "", d.channel if d.channel is not None else ""]) + + +SAMPLE_OURS = """order_id,amount +A001,99.00 +A002,19.90 +A003,250.00 +A004,10.00 +""" +SAMPLE_CHANNEL = """out_trade_no,total_fee +A001,9900 +A002,1999 +A004,1000 +A005,500 +""" + + +def main(argv: list[str]) -> int: + logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") + if len(argv) == 3: + ours_text = Path(argv[1]).read_text(encoding="utf-8-sig") # utf-8-sig：兼容 Excel 导出的 BOM + channel_text = Path(argv[2]).read_text(encoding="utf-8-sig") + else: + ours_text, channel_text = SAMPLE_OURS, SAMPLE_CHANNEL + ours = parse_rows(csv.DictReader(io.StringIO(ours_text)), "order_id", "amount") + channel = parse_rows(csv.DictReader(io.StringIO(channel_text)), "out_trade_no", "total_fee", cents=True) + diffs = reconcile(ours, channel) + write_report(diffs, sys.stdout) + log.info("共发现 %d 条差异", len(diffs)) + return 1 if diffs else 0 + + +if __name__ == "__main__": + code = main(sys.argv) + sys.exit(0 if len(sys.argv) != 3 else code) # 示例数据本来就有差异，演示模式下返回 0 diff --git a/topics/python/python/examples/reconcile/test_reconcile.py b/topics/python/python/examples/reconcile/test_reconcile.py new file mode 100644 index 0000000..a5a1227 --- /dev/null +++ b/topics/python/python/examples/reconcile/test_reconcile.py @@ -0,0 +1,40 @@ +import csv +import io +from decimal import Decimal + +import pytest + +from reconcile import SAMPLE_CHANNEL, SAMPLE_OURS, BadRow, DiffKind, Record, parse_rows, reconcile + + +def load(text: str, id_col: str, amount_col: str, cents: bool = False) -> list[Record]: + return list(parse_rows(csv.DictReader(io.StringIO(text)), id_col, amount_col, cents)) + + +def test_sample(): + diffs = reconcile(load(SAMPLE_OURS, "order_id", "amount"), + load(SAMPLE_CHANNEL, "out_trade_no", "total_fee", cents=True)) + assert [(d.kind, d.order_id) for d in diffs] == [ + (DiffKind.AMOUNT_MISMATCH, "A002"), + (DiffKind.MISSING_IN_CHANNEL, "A003"), + (DiffKind.MISSING_IN_OURS, "A005"), + ] + assert diffs[0].ours == Decimal("19.90") and diffs[0].channel == Decimal("19.99") + + +def test_decimal_not_float(): + # 0.1 + 0.2 用 float 不等于 0.3，用 Decimal 才精确 + assert 0.1 + 0.2 != 0.3 + assert Decimal("0.1") + Decimal("0.2") == Decimal("0.3") + + +@pytest.mark.parametrize("text", ["order_id,amount\nA1,abc\n", "order_id,amount\n,1.00\n", "id,amount\nA1,1\n"]) +def test_bad_rows(text): + with pytest.raises(BadRow): + load(text, "order_id", "amount") + + +def test_duplicate_ids_logged(caplog): + recs = [Record("A1", Decimal("1.00")), Record("A1", Decimal("2.00"))] + assert reconcile(recs, [Record("A1", Decimal("1.00"))]) == [] + assert "订单号重复" in caplog.text diff --git a/topics/python/python/index.md b/topics/python/python/index.md index c0ebe0e..ce4239d 100644 --- a/topics/python/python/index.md +++ b/topics/python/python/index.md @@ -4,12 +4,41 @@ title: Python 语言 type: index domain: python stack: [python] -tags: [] +tags: [Python, 语法, 类型标注, pytest] week: [4] --- # Python 语言 -给 JS 开发者的 Python：语法、类型、异常、包管理、测试。 +AI 应用 / Agent 岗位的主力语言。你已经会 JavaScript / TypeScript，学 Python 的重点不是从零学编程，而是**对照差异**：语法、数据结构、类型标注、异常、包管理和测试的“Python 写法”，以及一些和 JS 直觉相反的坑（可变默认参数、浮点金额、导入路径）。 -> 本专题建设中：课程、企业场景、面试题和示例代码会陆续补充。 +## 学习路线 + +| 顺序 | 课程 | 级别 | 预计 | +| --- | --- | --- | --- | +| 1 | [给 JS 开发者的 Python：语法、数据结构、推导式、函数、类](01-syntax-for-js.md) | 基础 | 3 小时 | +| 2 | [类型标注、dataclass、异常和上下文管理器、pytest](02-typing-errors-testing.md) | 基础 | 3 小时 | +| 3 | [工程化：虚拟环境、uv / pip、项目结构、导入、日志和配置](03-project-packaging.md) | 基础 | 2 小时 | + +## 企业场景 + +- [订单对账脚本：从“能跑的脚本”到可测试、可维护的工具](scenarios/order-reconciliation.md) + +## 面试题 + +- [Python 面试题](interview.md) + +## 示例代码 + +- [examples/reconcile](examples/reconcile/)：CSV 对账（Decimal 金额、dataclass、生成器、日志、异常）+ pytest 测试，只用标准库 + +## 在 16 周计划中 + +- [第 4 周：Python 速成 + FastAPI 基础](../../../plan/week04-python.md) + +## 官方资料 + +- Python 官方教程（中文）：https://docs.python.org/zh-cn/3/tutorial/ +- typing 模块：https://docs.python.org/zh-cn/3/library/typing.html +- pytest：https://docs.pytest.org/ +- uv：https://docs.astral.sh/uv/ diff --git a/topics/python/python/interview.md b/topics/python/python/interview.md new file mode 100644 index 0000000..7b1d8c4 --- /dev/null +++ b/topics/python/python/interview.md @@ -0,0 +1,93 @@ +--- +id: python.python.interview +title: Python 面试题 +type: interview +domain: python +stack: [python] +tags: [Python, 基础, 面试] +level: basic +related: [python.python.syntax-for-js, python.python.typing-errors-testing, python.python.project-packaging] +week: [4, 16] +--- + +# Python 面试题 + +## 基础 + +### 【基础】Python 中哪些类型是可变的，哪些是不可变的？有什么影响？ + +**参考答案**：不可变：int、float、str、tuple、frozenset、bytes；可变：list、dict、set、自定义对象（默认）。影响：函数参数传的是对象引用，函数里修改可变对象会影响调用方；只有不可变（可哈希）的对象才能做 dict 的 key 和 set 的元素。 + +### 【基础】为什么不要用可变对象作为函数默认参数？ + +**参考答案**：默认值在函数定义时只创建一次，之后每次调用共享同一个对象。`def f(xs=[])` 多次调用会在同一个列表上累加。改为默认 `None`，在函数内部创建新列表。 + +### 【基础】`is` 和 `==` 的区别？ + +**参考答案**：`==` 比较值是否相等（调用 `__eq__`），`is` 比较是否是同一个对象（身份）。判断 `None` 用 `is None`；比较值永远用 `==`，不要依赖小整数缓存之类的实现细节。 + +### 【基础】浅拷贝和深拷贝的区别？ + +**参考答案**：浅拷贝（`list(x)`、`x.copy()`、`copy.copy`）只复制最外层容器，内部元素仍是同一引用；深拷贝（`copy.deepcopy`）递归复制所有层级。嵌套结构修改内层元素时，浅拷贝会影响原对象。 + +### 【基础】列表推导式和生成器表达式有什么区别？ + +**参考答案**：`[x for x in xs]` 立即生成完整列表，占用内存；`(x for x in xs)` 生成器表达式按需逐个产生，内存占用恒定，但只能遍历一次。传给 `sum`、`max` 等函数时用生成器表达式即可。 + +## 进阶 + +### 【进阶】什么是 GIL？对 AI 应用开发有什么影响？ + +**参考答案**：GIL（全局解释器锁）使 CPython 同一时刻只有一个线程执行 Python 字节码，所以多线程无法利用多核做 CPU 密集计算；但 IO 等待时会释放 GIL，所以多线程 / asyncio 对 IO 密集任务有效。AI 应用大多是 IO 密集（调用模型 API、查数据库、检索），用 asyncio 很合适；CPU 密集（本地向量计算、文本解析）用多进程，或者依赖 numpy 等在 C 层释放 GIL 的库。Python 3.13 起提供了实验性的无 GIL 构建，以官方文档为准。 + +### 【进阶】装饰器是什么？写一个统计耗时的装饰器。 + +**参考答案**：装饰器是接收函数并返回新函数的函数，用来在不修改原函数的情况下增加功能（日志、重试、缓存、鉴权）。 + +```python +import functools +import time + + +def timed(fn): + @functools.wraps(fn) # 保留原函数的名字和文档 + def wrapper(*args, **kwargs): + start = time.perf_counter() + try: + return fn(*args, **kwargs) + finally: + print(f"{fn.__name__} 耗时 {(time.perf_counter() - start) * 1000:.1f} ms") + return wrapper + + +@timed +def work(n: int) -> int: + return sum(range(n)) + + +work(1_000_000) +``` + +**追问**：异步函数怎么写装饰器？→ wrapper 也要定义成 `async def` 并 `await fn(...)`。 + +### 【进阶】dataclass、pydantic、TypedDict 怎么选？ + +**参考答案**：内部数据结构用 dataclass（轻量、无运行时校验）；外部输入（接口请求体、配置、大模型返回的 JSON）用 pydantic，它会在运行时校验和转换类型并给出清晰的错误；只是给已有的字典加类型提示（例如 JSON 数据直接传来传去）用 TypedDict。 + +### 【进阶】上下文管理器的原理是什么？ + +**参考答案**：实现了 `__enter__` 和 `__exit__` 方法的对象。`with` 进入时调用 `__enter__`，离开代码块（无论正常还是异常）时调用 `__exit__`，保证资源释放；`__exit__` 返回 True 会吞掉异常。也可以用 `contextlib.contextmanager` 装饰生成器函数实现，异步版本是 `async with` 和 `__aenter__` / `__aexit__`。 + +## 场景题 + +### 【场景题】要处理一个 10GB 的日志文件，统计每个接口的请求数，怎么写？ + +**参考答案**：不能一次读入内存。用 `with open(...)` 逐行迭代（文件对象本身就是惰性的），解析出接口路径后用 `Counter` 计数；正则预编译；如果需要更快，可以按字节范围切分文件用多进程并行统计后合并。输出前 N 个用 `most_common`。 + +### 【场景题】一段 Python 代码在本机能跑，在服务器上报 `ModuleNotFoundError`，怎么排查？ + +**参考答案**：确认服务器上是否激活了正确的虚拟环境（`which python`、`python -c "import sys; print(sys.executable)"`）；依赖是否按锁文件安装；运行目录和方式是否不同（直接运行子目录的文件 vs 从根目录 `python -m`）；是否有本地文件和包同名遮蔽了真正的包；Python 版本是否一致。 + +### 【场景题】你从前端转 Python，怎么保证写出来的 Python 代码质量？ + +**参考答案**：用 ruff 做 lint 和格式化，用 mypy / pyright 做类型检查，pytest 写测试并在 CI 中运行；遵循 PEP 8 和项目约定；数据边界用 pydantic 校验；对照 JS 的经验注意 Python 特有的坑（可变默认参数、浮点金额、导入路径）。可以举一个自己写过的、带测试的小工具为例。 diff --git a/topics/python/python/scenarios/order-reconciliation.md b/topics/python/python/scenarios/order-reconciliation.md new file mode 100644 index 0000000..842478d --- /dev/null +++ b/topics/python/python/scenarios/order-reconciliation.md @@ -0,0 +1,87 @@ +--- +id: python.python.scenario.order-reconciliation +title: 订单对账脚本：从“能跑的脚本”到可测试、可维护的工具 +type: scenario +domain: python +stack: [python] +tags: [对账, CSV, Decimal, 数据处理, 测试] +level: basic +prerequisites: [python.python.typing-errors-testing] +est_hours: 2 +related: [python.python.interview, backend.mq.reliability] +week: [4] +--- + +# 订单对账脚本：从“能跑的脚本”到可测试、可维护的工具 + +## 业务背景与问题 + +财务每天要核对“系统里的已支付订单”和“支付渠道的账单”。原来由一位同事手工在 Excel 里 VLOOKUP，经常漏看；后来有人写了一个 50 行的 Python 脚本，但：金额用 float 比较，出现“19.9 和 19.90 不一致”的误报；渠道账单金额单位是“分”，脚本没处理；文件里有一行格式错误整个脚本崩溃，不知道是哪一行；没人敢改。 + +## 约束 + +- 输入是两份 CSV（渠道账单可能是 Excel 导出的，带 BOM）。 +- 输出差异清单给财务：渠道缺失（我方有渠道无）、我方缺失（掉单）、金额不一致。 +- 每天运行一次，后续可能接入定时任务和告警。 +- 先不引入 pandas 等重型依赖（服务器上只有标准 Python）。 + +## 方案设计 + +1. **数据模型**：`Record(order_id, amount: Decimal)`，用 `dataclass(frozen=True)`；金额统一转成两位小数的 `Decimal`，渠道的“分”在解析时除以 100。 +2. **解析和比较分离**：`parse_rows`（生成器，逐行解析，错误带行号）→ `to_map`（处理重复订单号并告警）→ `reconcile`（纯函数，集合运算找差异）→ `write_report`（输出 CSV）。纯函数便于测试。 +3. **输入兼容**：`utf-8-sig` 读取，兼容 Excel 的 BOM；列名作为参数传入，不同渠道只是配置不同。 +4. **可观测**：logging 输出统计；有差异时返回非零退出码，方便定时任务据此告警。 +5. **测试**：示例数据、错误格式、重复订单号、Decimal 精度都有用例。 + +## 关键代码 + +完整代码和测试见 [examples/reconcile](../examples/reconcile/)： + +```python +def reconcile(ours: Iterable[Record], channel: Iterable[Record]) -> list[Diff]: + a, b = to_map(ours), to_map(channel) + diffs: list[Diff] = [] + for oid in sorted(a.keys() | b.keys()): # 两边订单号的并集，排序后输出稳定 + x, y = a.get(oid), b.get(oid) + if y is None: + diffs.append(Diff(DiffKind.MISSING_IN_CHANNEL, oid, x.amount, None)) + elif x is None: + diffs.append(Diff(DiffKind.MISSING_IN_OURS, oid, None, y.amount)) + elif x.amount != y.amount: + diffs.append(Diff(DiffKind.AMOUNT_MISMATCH, oid, x.amount, y.amount)) + return diffs +``` + +运行示例数据的输出： + +```text +差异类型,订单号,我方金额,渠道金额 +金额不一致,A002,19.90,19.99 +渠道缺失,A003,250.00, +我方缺失,A005,,5.00 +``` + +## 取舍 + +| 方案 | 优点 | 缺点 | +| --- | --- | --- | +| **标准库 csv + dict（本方案）** | 无依赖、逻辑清晰、易测试 | 数据量到几千万行时内存吃紧 | +| pandas | 一两行完成合并对比，适合分析 | 依赖重；类型和精度要小心（默认 float） | +| 数据库里对账（导入后 SQL FULL OUTER JOIN） | 数据量大也能处理 | 需要导入流程 | +| 实时对账（支付回调 + 定时查询渠道） | 发现问题更早 | 复杂度高 | + +数据量变大后，可以把两份数据导入 PostgreSQL，用 `FULL OUTER JOIN` 找差异——思路完全一样。 + +## 易错点 + +- 用 float 比较金额：`19.9 != 19.90` 的问题其实来自字符串比较，`0.1 + 0.2 != 0.3` 来自 float，两者都要避免——统一转成 `Decimal` 并量化到两位小数。 +- 忘了单位换算（元 / 分）。 +- 一行坏数据导致整个任务失败且不知道是哪一行：错误信息带行号；也可以选择“记录坏行、继续处理、最后汇总”。 +- 订单号重复时静默覆盖：要告警。 +- 文件编码：Excel 导出的 CSV 可能带 BOM 或是 GBK 编码。 + +## 面试怎么讲 + +> “我把一个手工对账流程改成了 Python 工具：金额统一用 Decimal 并处理元和分的换算；解析用生成器逐行处理，错误带行号；比较逻辑是纯函数，用集合运算找出渠道缺失、我方缺失和金额不一致三类差异；有差异时返回非零退出码，接入定时任务告警。写了 pytest 覆盖精度、坏数据和重复订单号。” + +前端转岗时，这类“数据处理 + 工程化”的小工具是证明 Python 能力的好素材，比只写语法练习更有说服力。