FastAPI 核心机制教学文档

面向读者：已掌握基础路由（GET/POST、路径参数、Pydantic 模型），准备攻克进阶机制的你。 学习目标：对每个机制做到「看得懂原理 → 写得出代码 → 知道何时该用、何时不该用」。 使用方法：先看第 0 章全景图建立全局感 → 每章按 7 步学 → 做小练习 → 对照附录答案 → 最后过自查清单。

环境准备
第 0 章开篇：请求生命周期全景图
第 1 章 Middleware + request.state
第 2 章 APIRouter
第 3 章 CORS
第 4 章 lifespan
第 5 章 BackgroundTasks
第 6 章三个关联要点：async def vs def、asyncio.gather、exception_handler
第 7 章综合实战 + 自查清单
附录：练习参考答案

环境准备

本教程所有示例均为最小可运行代码，开始前请先准备环境：

bash

pip install fastapi "uvicorn[standard]" httpx

Python 版本：3.9+（示例在 3.10+ 验证通过）
每章代码建议保存为 main.py，用 uvicorn main:app --reload 启动
验证安装：uvicorn --version 能输出版本号即可

第 0 章开篇：请求生命周期全景图

先建立全局感。这 10 个知识点不是孤立的 API，它们分布在「请求的一生」的不同阶段。请盯着这张图看一分钟：

text

进程启动
   │
   ▼
┌──────────────────────────────────────────────────────────┐
│ lifespan（第4章）启动段：加载 LLM 客户端、连接池、配置     │ ← 进程级，执行一次
└──────────────────────────────────────────────────────────┘
   │
   ▼  请求到达（循环处理每个请求）
   │
┌──────────────────────── Middleware 洋葱圈 ─────────────────────────┐
│                                                                      │
│  外层 ── CORS 中间件（第3章）：浏览器跨域安检，最先见请求             │
│  内层 ── 自定义中间件（第1章）：鉴权 / 计时 / trace_id                │
│            └─ 写入 request.state（第1章）：本请求专属「便签」         │
│                 │                                                    │
│                 ▼                                                    │
│      APIRouter 路由匹配（第2章）：/v1/chat、/v1/admin…                │
│            │                                                         │
│            ▼                                                         │
│      路由函数（第6.1章）：async def → 事件循环；def → 线程池          │
│            │                                                         │
│            ├─ 并发 I/O：asyncio.gather（第6.2章）同时调多个工具       │
│            ├─ 出错：raise HTTPException ──► exception_handler（第6.3章）│
│            └─ 正常：返回响应 ──► 中间件后段（加响应头）──► 客户端      │
│                                                                      │
└──────────────────────────────────────────────────────────────────────┘
   │
   ▼  响应已发出
┌──────────────────────────────────────────────────────────┐
│ BackgroundTasks（第5章）：写审计日志、发通知……客户端不等    │ ← 响应之后执行
└──────────────────────────────────────────────────────────┘
   │
   ▼  进程退出
┌──────────────────────────────────────────────────────────┐
│ lifespan 关闭段：close() 连接、释放资源                     │ ← 进程级，执行一次
└──────────────────────────────────────────────────────────┘

一句话总结每个阶段的归属：

生命周期阶段	对应知识点	作用域
进程启动 / 退出	lifespan	全局一次
请求入口	CORS、Middleware、request.state	每个请求
路由处理	APIRouter、async def vs def、asyncio.gather	每个请求
错误路径	HTTPException、exception_handler	每个请求
响应之后	BackgroundTasks	每个请求

本教程重点讲解你不熟悉的 5 个核心机制（第 1–5 章），第 6 章紧凑补上其余 3 个要点。

第 1 章 Middleware + request.state

1️⃣ 一句话本质

中间件是包裹所有路由的「洋葱皮」：请求进来先经过它，响应出去也要经过它；request.state 是它贴在请求上的「便签」，供后面的路由读取。

2️⃣ 生活类比

写字楼前台 + 访客证。每位访客（请求）进楼前在前台登记、领一张访客证（request.state 贴上 trace_id）；离开时前台再盖一个「离开时间」章（写响应头）。前台不关心访客去哪个房间办事，但每一趟进出都从它手里过。

3️⃣ 图解：洋葱模型

text

请求 ──► ┌──────────────────────────┐ ──► 路由函数
         │ 中间件A前段 → 中间件B前段 │
响应 ◄── │ 中间件B后段 ← 中间件A后段 │ ◄── 路由函数
         └──────────────────────────┘

call_next(request) 是分水岭：它之前是请求进来的路，它之后是响应出去的路。

4️⃣ 最小可运行代码

python

import time, uuid
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def trace_and_time(request: Request, call_next):
    # ① 请求进来：贴便签、开始计时
    request.state.trace_id = str(uuid.uuid4())
    start = time.perf_counter()

    # ② 放行，进入路由（路由抛出的异常也会从这里冒出来）
    response = await call_next(request)

    # ③ 响应出去：再加工
    response.headers["X-Trace-ID"] = request.state.trace_id
    response.headers["X-Process-Time"] = f"{time.perf_counter() - start:.4f}s"
    return response

@app.get("/hello")
async def hello(request: Request):
    # 路由里直接读中间件留下的便签
    return {"msg": "hello", "trace_id": request.state.trace_id}

运行：uvicorn main:app --reload，然后 curl -i http://127.0.0.1:8000/hello，看响应头里的 X-Trace-ID 和 X-Process-Time。

5️⃣ 要点拆解

@app.middleware("http") 是 Starlette BaseHTTPMiddleware 的函数式简写，日常够用。
call_next(request) 只能调一次；它返回的 response 必须 return 回去，否则请求永远挂起。
request.state 的生命周期 = 这一个请求，请求结束即销毁。跨请求共享请用 app.state（见第 4 章）。
执行顺序：后注册的先执行（最后 add 的在最外层）。所以 CORS 通常最后 add，成为最外层（第 3 章会再讲）。
中间件能看到路由抛出的异常（在 call_next 外面 try/except），但业务异常处理不推荐放这里——那是 exception_handler 的活（第 6.3 节）。

6️⃣ 常见坑

⚠️ 坑1：忘记 return response → 客户端一直等到超时。这是新手最高频的错误。
⚠️ 坑2：在中间件里写阻塞代码（如同步 requests 调用）→ 所有请求被一起拖慢。中间件里同样要遵守第 6.1 节的 async 原则。
⚠️ 坑3：拿 request.state 存跨请求数据 → 每个请求的 state 相互独立，存「用户会话」这类数据必然取不到，还会造成误解。
⚠️ 坑4：与 BackgroundTasks 混用的兼容问题 → Starlette 官方文档列了注意事项（背景任务与 BaseHTTPMiddleware 后段代码的时序问题）。遇到怪问题，改用下面的纯 ASGI 中间件写法：

python

# 纯 ASGI 中间件：与 BackgroundTasks 混用有兼容问题时，用它替代 @app.middleware("http")
from starlette.types import ASGIApp, Scope, Receive, Send, Message

class TraceMiddleware:
    def __init__(self, app: ASGIApp):
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        async def send_with_header(message: Message) -> None:
            # 响应头需要在包装后的 send 里加上去
            if message["type"] == "http.response.start":
                message.setdefault("headers", []).append((b"x-trace-id", b"demo"))
            await send(message)

        await self.app(scope, receive, send_with_header)

app.add_middleware(TraceMiddleware)

7️⃣ 小练习

写一个「简易 API Key 鉴权」中间件：检查请求头 X-API-Key，不匹配直接返回 403 JSON（不调用 call_next）；匹配则在 request.state 里存用户名，路由里读出来返回。参考答案见附录。

第 2 章 APIRouter

1️⃣ 一句话本质

把路由按业务域拆成多个模块文件，再在主应用里「组装」回去——项目的「分房术」。

2️⃣ 生活类比

公司部门划分。一个公司（app）有客服部、技术部、财务部（各一个 router），各自处理自己的业务；前台总机按分机号（include_router）把来电转过去。部门内部怎么干，总部不关心。

3️⃣ 图解：推荐目录结构

text

app/
├── main.py            # 组装入口：include_router
└── routers/
    ├── __init__.py
    ├── chat.py        # /v1/chat 相关（prefix + tags）
    ├── tools.py       # /v1/tools 相关
    └── admin.py       # /v1/admin 相关

4️⃣ 最小可运行代码

routers/chat.py：

python

from fastapi import APIRouter

router = APIRouter(prefix="/v1/chat", tags=["chat"])

@router.get("/history")
async def get_history():
    return {"messages": []}

@router.post("")
async def chat():
    return {"reply": "hi"}

main.py：

python

from fastapi import FastAPI
from routers import chat

app = FastAPI()
app.include_router(chat.router)

访问 GET /v1/chat/history、POST /v1/chat；打开 /docs 会看到 chat 分组——tags 的功劳。

5️⃣ 要点拆解

prefix：该模块所有路由的统一前缀，避免每个路由手写一遍。
tags：Swagger 文档（/docs）的分组名，API 文档整洁的关键。
dependencies：给整个模块挂统一依赖（如鉴权），例如 dependencies=[Depends(verify_api_key)]。
include_router 还支持再叠加 prefix、全局 dependencies、全局 responses 等。
拆分标准：单个文件路由超过约 10 个，或出现明显业务域边界，就拆。

6️⃣ 常见坑

⚠️ 坑1：prefix 写错或遗漏 → 全部 404，排查半天。建议团队统一 /v1/xxx 风格。
⚠️ 坑2：重复鉴权 → router 里写的 dependencies 和 include_router 里写的 dependencies 是叠加关系，两边都写会执行两次。
⚠️ 坑3：tags 命名不一致 → 文档里出现散乱分组。约定好命名规范。
⚠️ 坑4：忘记 import 子模块 → include_router 报 NameError。用 routers/__init__.py 统一导出可减少这类错误。

7️⃣ 小练习

把第 1 章的 /hello 改造成 routers/hello.py，prefix="/greet"、tags=["greeting"]，在 main.py 里组装。要求：/docs 能看到 greeting 分组，访问 /greet/hello 正常。参考答案见附录。

第 3 章 CORS

1️⃣ 一句话本质

浏览器为了保护用户，默认禁止网页 JS 跨域调用 API；CORS 是服务器声明「我允许哪些网站来调我」的机制。

2️⃣ 生活类比

小区门禁。住户（服务器资源）在门卫处登记「允许访客名单」（allow_origins）；访客（网页请求）到了，门卫查名单才放行。注意：拦人的是「浏览器这个门卫」，不是服务器。

3️⃣ 图解

text

浏览器页面 http://localhost:3000
      │  JS 发起 fetch 到 http://localhost:8000
      ▼
┌───────────────────────────────┐
│  浏览器安全机制（门卫）          │
│  检查响应头 Access-Control-*   │
│  通过  → JS 拿到数据           │
│  不通过 → Console 报 CORS 错误  │
└───────────────────────────────┘
      │
      ▼
服务器（需配置 CORSMiddleware 返回放行响应头）

4️⃣ 最小可运行代码

python

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "https://myfront.com"],  # 允许的前端域名
    allow_credentials=True,    # 允许带 Cookie
    allow_methods=["*"],       # 允许的 HTTP 方法
    allow_headers=["*"],       # 允许的请求头
)

@app.get("/data")
async def get_data():
    return {"data": "hello"}

前端页面（放在 http://localhost:3000 下）：

javascript

fetch("http://localhost:8000/data", { credentials: "include" })
  .then(r => r.json())
  .then(d => console.log("成功:", d))
  .catch(e => console.error("CORS 失败:", e));

5️⃣ 要点拆解

allow_origins：允许的前端来源列表。注意 http://localhost:3000 和 http://127.0.0.1:3000 是不同来源。
allow_methods / allow_headers：浏览器发「预检请求」（OPTIONS）时先问服务器允许什么，这两个参数就是答案。
allow_credentials：允许跨域带 Cookie / Authorization 头。
CORS 由浏览器强制执行：curl、Postman 不会报 CORS 错，只有浏览器里的 JS 会。
中间件顺序：CORS 应最后 add_middleware（最外层），否则预检 OPTIONS 可能先被你的鉴权中间件拦成 401。

6️⃣ 常见坑

⚠️ 坑1：用 curl 测试「没问题」就上线 → 浏览器一开就报错。务必在浏览器 Console 里验证。
⚠️ 坑2：allow_origins=["*"] 与 allow_credentials=True 同用 → 虽然 Starlette 会自动把 * 换成请求来源让它「能跑」，但这等于向任意网站开放带凭证访问，是安全风险。要带凭证，就写死具体域名列表。
⚠️ 坑3：混淆来源 → localhost vs 127.0.0.1、http vs https、端口不同，全算不同来源。
⚠️ 坑4：中间件顺序错误 → 预检请求被鉴权中间件拦下，前端报错但后端日志看起来「没有请求进来」。

7️⃣ 小练习

写一个 HTML 页面（含上面的 fetch 代码），用 python -m http.server 3000 托管，验证跨域成功；然后把 allow_origins 改成只允许 http://localhost:5500，重启后刷新页面，观察 Console 报错。参考答案见附录。

第 4 章 lifespan

1️⃣ 一句话本质

应用「出生时」和「临终前」各做一次的事：启动时初始化资源，关闭时释放资源。yield 之前是启动段，yield 之后是关闭段。

2️⃣ 生活类比

餐厅开业与打烊。开门前：进货、开炉子（加载模型、建连接池）；营业中：正常接单（处理请求）；打烊后：关炉子、清点库存（关闭连接、释放资源）。

3️⃣ 图解

text

@asynccontextmanager
async def lifespan(app):
    │
    ├─ 启动段：加载 LLM 客户端 / DB 连接池 / 配置     ← 进程级，一次
    │
    ├─ yield  ←—— 应用运行期（持续处理所有请求）
    │
    └─ 关闭段：close() 连接 / 清理临时文件 / 上报指标  ← 进程级，一次

4️⃣ 最小可运行代码

python

from contextlib import asynccontextmanager
from fastapi import FastAPI

class FakeLLM:
    def __init__(self):
        print("LLM 客户端已加载")
    async def close(self):
        print("LLM 客户端已关闭")

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动段：全局只执行一次
    app.state.llm = FakeLLM()
    print("启动完成，开始服务")
    yield
    # 关闭段：Ctrl+C 或进程退出时执行
    await app.state.llm.close()
    print("已优雅关闭")

app = FastAPI(lifespan=lifespan)

@app.get("/status")
async def status():
    return {"llm_loaded": app.state.llm is not None}

uvicorn main:app 启动时看控制台出现「启动完成」；Ctrl+C 停止时看到「已优雅关闭」。

5️⃣ 要点拆解

替代已废弃的 @app.on_event("startup"/"shutdown") 写法。
yield 前 = 启动段，yield 后 = 关闭段。注意：如果 yield 之前抛异常，yield 之后不会执行（asynccontextmanager 语义）。
app.state vs request.state（高频混淆点）：
- app.state：全局一份，放连接池、模型客户端、配置；
- request.state：每请求独立（第 1 章）。名字像，作用域天差地别。
常见用途：初始化 httpx.AsyncClient、数据库连接池、加载 ML 模型；关闭时全部释放。

6️⃣ 常见坑

⚠️ 坑1：把「每个请求都不同的数据」放 lifespan → 数据串线。lifespan 是全局的。
⚠️ 坑2：启动段做耗时 / 可能失败的操作且不处理 → 应用起不来或异常静默。启动段要 try/except 并打日志。
⚠️ 坑3：忘记关闭资源（httpx 客户端、连接池）→ 连接泄漏，长时间运行后服务器假死。
⚠️ 坑4：开发时用 --reload → 每次改代码进程重启，启动段反复执行。开发时别在启动段放太重的东西，或接受这一点。

7️⃣ 小练习

写一个 lifespan：启动时创建 httpx.AsyncClient 存到 app.state.client；一个路由用它请求 https://httpbin.org/delay/1 并返回耗时；关闭时 await client.aclose()。参考答案见附录。

第 5 章 BackgroundTasks

1️⃣ 一句话本质

把「客户端不需要等待结果」的活儿，挂到响应发出之后再执行：先还响应，后干杂活。

2️⃣ 生活类比

餐厅打包。客人点完餐拿到小票（响应）就可以走了；厨房「稍后」把电子发票发到客人手机（后台任务）。客人不用站在柜台等发票。

3️⃣ 图解：时序

text

客户端 ──请求──► 路由函数 ──add_task(写审计日志)──► 响应立即返回客户端
                       │
                       ▼   （响应发出之后）
                  FastAPI 依次执行后台任务：
                  ① 写审计日志  ② 发通知邮件
                       │
                       ▼
                  任务完成（客户端早已拿到响应）

4️⃣ 最小可运行代码

python

import asyncio
from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

async def write_audit_log(user: str, content: str):
    await asyncio.sleep(1)          # 模拟写日志的耗时
    print(f"[审计] {user}: {content}")

@app.post("/ask")
async def ask(background_tasks: BackgroundTasks, user: str = "guest"):
    answer = "你好，我是 Agent"        # 实际业务
    background_tasks.add_task(write_audit_log, user, answer)  # 挂后台
    return {"reply": answer}          # 响应立刻返回，不等日志写完

curl 测试：响应秒回；约 1 秒后服务器控制台打印审计日志。

5️⃣ 要点拆解

add_task(func, *args, **kwargs)：func 可以是 async 函数，也可以是普通函数。
执行时机：响应体发送给客户端之后、请求生命周期结束之前。
在路由参数里声明 background_tasks: BackgroundTasks 即可，FastAPI 自动注入。
边界（重要）：它是「进程内」的轻量机制——没有重试、没有持久化，进程一挂任务就丢。
- 适合：写审计日志、发通知、埋点上报等「丢了也不致命」的活；
- 不适合：转账、长任务、必须可靠执行的事 → 上 Celery / ARQ / Dramatiq 等任务队列。

6️⃣ 常见坑

⚠️ 坑1：拿它当消息队列用 → 任务丢失无感知。分清「尽力而为」与「必须可靠」。
⚠️ 坑2：任务里做重活 → 后台任务依然吃服务器资源，照样拖慢后续请求。
⚠️ 坑3：与 @app.middleware("http")（BaseHTTPMiddleware）混用 → Starlette 有已知的时序注意事项（响应后中间件后段代码与后台任务的执行顺序问题）。复杂场景用纯 ASGI 中间件。
⚠️ 坑4：误以为 async 后台任务「在别的进程跑」 → async 任务仍在事件循环里，阻塞代码照样卡。

7️⃣ 小练习

写一个 GET /visit 接口：把 someone visited 写入本地文件 audit.log（后台任务），响应立即返回。连续 curl 三次，观察响应速度与文件写入。参考答案见附录。

第 6 章三个关联要点：async def vs def、asyncio.gather、exception_handler

这一章紧凑讲清其余三个要点——它们是理解前 5 章的必要背景，也是拉开差距的地方。

6.1 async def vs def —— 决定代码跑在哪

一句话本质：async def 路由跑在事件循环上；def 路由被 FastAPI 自动扔进线程池（默认约 40 个线程）。

python

import time, httpx
from fastapi import FastAPI

app = FastAPI()

@app.get("/slow-sync")      # 里面有阻塞代码 → 用 def，FastAPI 放线程池
def slow_sync():
    time.sleep(2)
    return {"ok": True}

@app.get("/fast-async")     # 真正的异步 I/O → 用 async def + await
async def fast_async():
    async with httpx.AsyncClient() as client:
        r = await client.get("https://httpbin.org/delay/1")
    return {"status": r.status_code}

核心要点：async def 里有阻塞代码（同步数据库驱动、requests、time.sleep）→ 卡死整个事件循环，所有请求一起等。
经验法则：只有异步 I/O → async def；有阻塞 / CPU 密集 → def。不是 async def 一定更快，关键看里面有没有阻塞。
坑：「路由里调 OpenAI SDK，用 async def 就快了」——错。SDK 是同步的，反而把事件循环堵死。

6.2 asyncio.gather —— 把串行等待变并发等待

一句话本质：多个 await 同时进行，总耗时 = 最慢的那个，而不是相加。

python

# 串行：2 + 3 = 5 秒
weather = await fetch_weather("北京")
news    = await fetch_news("北京")

# 并发：max(2, 3) = 3 秒
weather, news = await asyncio.gather(
    fetch_weather("北京"),
    fetch_news("北京"),
    return_exceptions=True,   # 单个失败不拖垮全体，结果里是异常对象
)

前提：底层是真正的异步 I/O（httpx.AsyncClient 等）。全是阻塞调用时 gather 不加速——这又回到 6.1。
Agent 场景：一个请求里并行调多个工具 / 多个模型，用它直接砍掉一大半延迟。

6.3 exception_handler —— 把内部异常翻译成对外响应

一句话本质：业务代码 raise 自定义异常，在出口统一翻译成规范的 JSON 错误。

python

from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()

class QuotaExceeded(Exception):
    def __init__(self, msg: str):
        self.msg = msg

@app.exception_handler(QuotaExceeded)
async def quota_handler(request, exc: QuotaExceeded):
    return JSONResponse(status_code=429, content={"code": 429, "message": exc.msg})

@app.get("/ask")
async def ask():
    raise QuotaExceeded("今日额度已用完")   # 业务里直接抛，出口统一翻译

与 HTTPException 的分工：简单错误直接 raise HTTPException(status_code=404, detail=...)；需要统一格式 / 自定义异常类时用 exception_handler。
高频用法：① 全站统一错误格式 {"code": ..., "message": ...}；② 自定义 RequestValidationError 处理器，改写默认 422 校验错误格式。
坑：exception_handler 捕获的是「从路由抛出的异常」；中间件里提前 return 的错误不经过它。

第 7 章综合实战 + 自查清单

7.1 综合实战：把全部知识点拼成一个 Agent API 服务

main.py：

python

import time, uuid
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from routers import chat, admin

@asynccontextmanager
async def lifespan(app: FastAPI):
    # ④ lifespan：启动时加载 LLM 客户端
    app.state.llm = "fake-llm-client"
    print("llm client loaded")
    yield
    # ④ 关闭时释放
    print("llm client closed")

app = FastAPI(lifespan=lifespan)

# ① middleware + request.state：鉴权 + trace_id（先注册 → 内层）
@app.middleware("http")
async def guard(request: Request, call_next):
    if request.url.path.startswith("/v1") and request.headers.get("x-api-key") != "secret":
        return JSONResponse(status_code=403, content={"code": 403, "message": "invalid api key"})
    request.state.trace_id = str(uuid.uuid4())
    start = time.perf_counter()
    response = await call_next(request)
    response.headers["x-trace-id"] = request.state.trace_id
    response.headers["x-cost"] = f"{time.perf_counter() - start:.4f}s"
    return response

# ③ CORS：最后 add → 最外层，先处理预检 OPTIONS 请求
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# ② APIRouter：模块化拆分
app.include_router(chat.router)
app.include_router(admin.router)

# ⑥ exception_handler：全站统一错误格式
@app.exception_handler(HTTPException)
async def http_exc_handler(request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"code": exc.status_code, "message": str(exc.detail)},
    )

routers/chat.py：

python

import asyncio
from fastapi import APIRouter, BackgroundTasks, HTTPException

router = APIRouter(prefix="/v1/chat", tags=["chat"])

async def fetch_tool(name: str):
    await asyncio.sleep(1)                 # 模拟调用一个工具
    return f"{name}: ok"

def audit(content: str):                   # 同步函数也可以做后台任务
    with open("audit.log", "a") as f:
        f.write(content + "\n")

@router.post("")
async def chat(background_tasks: BackgroundTasks, q: str):
    if not q:
        raise HTTPException(status_code=400, detail="q is required")  # ⑥ 统一格式
    # ⑥ asyncio.gather：并发调两个工具（1 秒完成，串行要 2 秒）
    r1, r2 = await asyncio.gather(
        fetch_tool("weather"),
        fetch_tool("search"),
        return_exceptions=True,
    )
    # ⑤ BackgroundTasks：响应返回后写审计日志
    background_tasks.add_task(audit, f"asked: {q}")
    return {"reply": f"{q} -> {r1} | {r2}"}

routers/admin.py：

python

from fastapi import APIRouter

router = APIRouter(prefix="/v1/admin", tags=["admin"])

@router.get("/status")
async def status():
    return {"ok": True}

验证方式：uvicorn main:app；curl 带 / 不带 x-api-key 对比；用第 3 章练习里的 HTML 页面从 3000 端口调用验证 CORS。（routers/ 目录下记得建一个空的 __init__.py，与第 2 章的目录结构保持一致。）

⚠️ 注意：本实战同时使用了 @app.middleware("http")（BaseHTTPMiddleware）与 BackgroundTasks——正是第 1 章坑4 / 第 5 章坑3 警告的组合。教学演示可以这样写；生产环境若遇到后台任务时序异常，请按第 5 章坑3 改用纯 ASGI 中间件（写法见第 1 章坑4 的示例）。

7.2 自查清单（全部能答上 = 真正掌握）

Middleware + request.state

[ ] 不看代码，说出 call_next(request) 前后两段分别是什么时机
[ ] 能解释 request.state 和 app.state 的区别
[ ] 知道中间件执行顺序（谁在最外层）

APIRouter

[ ] 能说出 prefix / tags / dependencies 各自作用
[ ] 能画出推荐的项目目录结构
[ ] 知道什么时候该拆 router

CORS

[ ] 能解释为什么 curl 测不出 CORS 问题
[ ] 知道 allow_credentials=True 时 allow_origins 的正确写法
[ ] 知道 localhost 和 127.0.0.1 算不同来源

lifespan

[ ] 能说出 yield 前 / yield 后分别执行什么
[ ] 知道它替代了哪个废弃写法
[ ] 能举出 2 个必须放 lifespan 的资源

BackgroundTasks

[ ] 能说出后台任务的执行时机
[ ] 能说清它和 Celery 的边界（什么该用它、什么不该）
[ ] 知道它与 BaseHTTPMiddleware 的兼容注意事项

三个关联要点

[ ] 能一句话说清 async def 和 def 的执行位置差异
[ ] 知道 asyncio.gather 加速的前提条件
[ ] 知道 exception_handler 与直接 raise HTTPException 的分工

附录：练习参考答案

第 1 章练习：API Key 鉴权中间件

python

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

@app.middleware("http")
async def api_key_auth(request: Request, call_next):
    if request.headers.get("x-api-key") != "my-secret":
        return JSONResponse(status_code=403, content={"detail": "invalid api key"})
    request.state.api_user = "edgar"
    return await call_next(request)

@app.get("/me")
async def me(request: Request):
    return {"user": request.state.api_user}

第 2 章练习：拆出 greet 模块

routers/hello.py：

python

from fastapi import APIRouter

router = APIRouter(prefix="/greet", tags=["greeting"])

@router.get("/hello")
async def hello():
    return {"msg": "hello"}

main.py：

python

from fastapi import FastAPI
from routers import hello

app = FastAPI()
app.include_router(hello.router)

第 3 章练习：CORS 验证流程

python

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

完整 test.html：

html

<!DOCTYPE html>
<html>
<body>
<script>
fetch("http://localhost:8000/data", { credentials: "include" })
  .then(r => r.json())
  .then(d => console.log("成功:", d))
  .catch(e => console.error("CORS 失败:", e));
</script>
</body>
</html>

验证：目录下放 test.html，执行 python -m http.server 3000，浏览器打开 http://localhost:3000/test.html，Console 打印「成功」；再把 allow_origins 改成 ["http://localhost:5500"] 并重启后端，刷新页面，Console 出现 CORS 报错。

第 4 章练习：httpx 客户端生命周期

python

import time, httpx
from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.client = httpx.AsyncClient(timeout=10)
    yield
    await app.state.client.aclose()

app = FastAPI(lifespan=lifespan)

@app.get("/delay")
async def delay():
    t0 = time.perf_counter()
    r = await app.state.client.get("https://httpbin.org/delay/1")
    return {"status": r.status_code, "cost": round(time.perf_counter() - t0, 3)}

第 5 章练习：审计日志后台写入

python

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def write_audit(line: str):
    with open("audit.log", "a") as f:
        f.write(line + "\n")

@app.get("/visit")
async def visit(background_tasks: BackgroundTasks):
    background_tasks.add_task(write_audit, "someone visited")
    return {"ok": True}

验证：连续 curl http://127.0.0.1:8000/visit 三次，响应都是立即返回；随后 cat audit.log 看到三行记录。

文档完。建议学习顺序：通读 → 做每章小练习 → 对照附录 → 完成综合实战 → 过自查清单。

FastAPI 核心机制教学文档 ​

目录 ​

环境准备 ​

第 0 章 开篇：请求生命周期全景图 ​

第 1 章 Middleware + request.state ​

1️⃣ 一句话本质 ​

2️⃣ 生活类比 ​

3️⃣ 图解：洋葱模型 ​

4️⃣ 最小可运行代码 ​

5️⃣ 要点拆解 ​

6️⃣ 常见坑 ​

7️⃣ 小练习 ​

第 2 章 APIRouter ​

1️⃣ 一句话本质 ​

2️⃣ 生活类比 ​

3️⃣ 图解：推荐目录结构 ​

4️⃣ 最小可运行代码 ​

5️⃣ 要点拆解 ​

6️⃣ 常见坑 ​

7️⃣ 小练习 ​

第 3 章 CORS ​

1️⃣ 一句话本质 ​

2️⃣ 生活类比 ​

3️⃣ 图解 ​

4️⃣ 最小可运行代码 ​

5️⃣ 要点拆解 ​

6️⃣ 常见坑 ​

7️⃣ 小练习 ​

第 4 章 lifespan ​

1️⃣ 一句话本质 ​

2️⃣ 生活类比 ​

3️⃣ 图解 ​

4️⃣ 最小可运行代码 ​

5️⃣ 要点拆解 ​

6️⃣ 常见坑 ​

7️⃣ 小练习 ​

第 5 章 BackgroundTasks ​

1️⃣ 一句话本质 ​

2️⃣ 生活类比 ​

3️⃣ 图解：时序 ​

4️⃣ 最小可运行代码 ​

5️⃣ 要点拆解 ​

6️⃣ 常见坑 ​

7️⃣ 小练习 ​

第 6 章 三个关联要点：async def vs def、asyncio.gather、exception_handler ​

6.1 async def vs def —— 决定代码跑在哪 ​

6.2 asyncio.gather —— 把串行等待变并发等待 ​

6.3 exception_handler —— 把内部异常翻译成对外响应 ​

第 7 章 综合实战 + 自查清单 ​

7.1 综合实战：把全部知识点拼成一个 Agent API 服务 ​

7.2 自查清单（全部能答上 = 真正掌握） ​

附录：练习参考答案 ​

第 1 章练习：API Key 鉴权中间件 ​

第 2 章练习：拆出 greet 模块 ​

第 3 章练习：CORS 验证流程 ​

第 4 章练习：httpx 客户端生命周期 ​

第 5 章练习：审计日志后台写入 ​

FastAPI 核心机制教学文档

目录

环境准备

第 0 章开篇：请求生命周期全景图

第 1 章 Middleware + request.state

1️⃣ 一句话本质

2️⃣ 生活类比

3️⃣ 图解：洋葱模型

4️⃣ 最小可运行代码

5️⃣ 要点拆解

6️⃣ 常见坑

7️⃣ 小练习

第 2 章 APIRouter

1️⃣ 一句话本质

2️⃣ 生活类比

3️⃣ 图解：推荐目录结构

4️⃣ 最小可运行代码

5️⃣ 要点拆解

6️⃣ 常见坑

7️⃣ 小练习

第 3 章 CORS

1️⃣ 一句话本质

2️⃣ 生活类比

3️⃣ 图解

4️⃣ 最小可运行代码

5️⃣ 要点拆解

6️⃣ 常见坑

7️⃣ 小练习

第 4 章 lifespan

1️⃣ 一句话本质

2️⃣ 生活类比

3️⃣ 图解

4️⃣ 最小可运行代码

5️⃣ 要点拆解

6️⃣ 常见坑

7️⃣ 小练习

第 5 章 BackgroundTasks

1️⃣ 一句话本质

2️⃣ 生活类比

3️⃣ 图解：时序

4️⃣ 最小可运行代码

5️⃣ 要点拆解

6️⃣ 常见坑

7️⃣ 小练习

第 6 章三个关联要点：async def vs def、asyncio.gather、exception_handler

6.1 async def vs def —— 决定代码跑在哪

6.2 asyncio.gather —— 把串行等待变并发等待

6.3 exception_handler —— 把内部异常翻译成对外响应

第 7 章综合实战 + 自查清单

7.1 综合实战：把全部知识点拼成一个 Agent API 服务

7.2 自查清单（全部能答上 = 真正掌握）

附录：练习参考答案

第 1 章练习：API Key 鉴权中间件

第 2 章练习：拆出 greet 模块

第 3 章练习：CORS 验证流程

第 4 章练习：httpx 客户端生命周期

第 5 章练习：审计日志后台写入