第 9 章：中间件、CORS 与生命周期

前八章关注「单个请求怎么处理」。本章把视角拉到请求之外：请求进来先经过谁，应用启动时要准备什么，响应返回后还能做什么。

学习目标

理解中间件的洋葱模型，能预测多个中间件的执行顺序
掌握 @app.middleware("http") 与纯 ASGI 中间件的写法差异与取舍
能正确配置 CORSMiddleware，避开 allow_credentials 与通配符的冲突
掌握 lifespan 替代 on_event 的写法，完成资源建/销闭环
理解 BackgroundTasks 的适用边界，知道何时该换成任务队列

9.1 中间件是什么：洋葱模型

中间件（Middleware）夹在 ASGI 服务器与路由之间。请求从外向内穿过所有中间件才到路由函数，响应再从内向外穿回来：

这解释了中间件的三个能力：请求阶段读/改请求（注入请求 ID、鉴权、限流）；响应阶段读/改响应（加响应头、计时、压缩）；短路——不调用 call_next 直接返回，性能最好。

9.2 `@app.middleware("http")` 装饰器写法

python

import logging
import time
import uuid

from fastapi import FastAPI, Request

logger = logging.getLogger(__name__)
app = FastAPI()


@app.middleware("http")
async def request_context(request: Request, call_next):
    request_id = request.headers.get("X-Request-Id") or uuid.uuid4().hex
    request.state.request_id = request_id  # 供路由函数 / 异常处理器读取
    start = time.perf_counter()
    response = await call_next(request)
    elapsed_ms = (time.perf_counter() - start) * 1000
    response.headers["X-Request-Id"] = request_id
    response.headers["X-Process-Time-Ms"] = f"{elapsed_ms:.2f}"
    logger.info("%s %s -> %s %.2fms rid=%s", request.method, request.url.path,
                response.status_code, elapsed_ms, request_id)
    return response

路由函数里读取注入的值，推荐用依赖而不是直接摸 request.state：

python

from typing import Annotated

from fastapi import Depends


def get_request_id(request: Request) -> str:
    return getattr(request.state, "request_id", "-")


RequestIdDep = Annotated[str, Depends(get_request_id)]


@app.get("/whoami/")
async def whoami(request_id: RequestIdDep) -> dict[str, str]:
    return {"request_id": request_id}

💡 用 request.state 传值，它的生命周期与请求绑定，天然隔离并发；不要用全局变量或共享字典。

9.3 `BaseHTTPMiddleware` vs 纯 ASGI 中间件

维度	`@app.middleware("http")`	纯 ASGI 中间件
写法	`async def mw(request, call_next)`	`async def mw(app, scope, receive, send)`
心智负担	低，拿到 `Request` / `Response` 对象	高，直接操作 ASGI 三元组
性能	需包装请求/响应流，有额外开销	无额外包装，性能更好
读 body	消耗 `receive` 流，下游可能拿不到	可在 `receive` 上做无侵入的 tee
适用	加头、计时、日志等大多数场景	高频路径、需要读 body 或改流行为

纯 ASGI 中间件最小示例：

python

class TimingASGIMiddleware:
    """不包装 Request/Response，只观察消息流。"""

    def __init__(self, app) -> None:
        self.app = app

    async def __call__(self, scope, receive, send) -> None:
        if scope["type"] != "http":
            await self.app(scope, receive, send)  # 必须透传 lifespan / websocket
            return

        async def send_wrapper(message) -> None:
            if message["type"] == "http.response.start":
                message["headers"].append((b"x-powered-by", b"asgi-mw"))
            await send(message)

        await self.app(scope, receive, send_wrapper)

两个硬性约定：必须透传非 http 类型的 scope（lifespan / websocket），否则应用无法启动；响应头必须是 bytes 而非 str。

9.3.1 错误示范 vs 正确示范

❌ 不推荐：在 BaseHTTPMiddleware 里读 body，消费了 receive 流，导致下游路由函数读到空。

python

@app.middleware("http")
async def log_body_bad(request: Request, call_next):
    body = await request.body()   # ⚠️ 下游可能拿不到 body
    print("body:", body)
    return await call_next(request)

✅ 推荐：需要审计请求体时，在路由层用依赖读取（FastAPI 内部会缓存 body），或改用纯 ASGI 中间件做 tee。

python

async def body_snapshot(request: Request) -> bytes:
    return await request.body()


@app.post("/audit/")
async def audit(raw: bytes = Depends(body_snapshot)) -> dict[str, int]:
    return {"size": len(raw)}

9.4 中间件执行顺序：后添加的先执行

app.add_middleware() 是栈式的：后添加的在更外层。

python

app.add_middleware(TagMiddleware, tag="first")   # 先添加 → 内层
app.add_middleware(TagMiddleware, tag="second")  # 后添加 → 外层

text

请求方向：second → first → 路由
响应方向：路由 → first → second

⚠️ 常见误解是「后面的中间件更靠内」，事实相反。若把「注入用户身份」的中间件加在了「鉴权」之后，鉴权就在更内层，会拿到空身份。

9.5 `CORSMiddleware` 完整配置

跨域资源共享（Cross-Origin Resource Sharing，CORS）是浏览器强制的安全机制：前端页面与 API 不同源时，浏览器要求服务端显式授权。

python

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://blog.example.com", "https://admin.example.com"],
    allow_origin_regex=r"https://.*\.example\.com",
    allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
    allow_headers=["Authorization", "Content-Type", "X-Request-Id"],
    allow_credentials=True,
    expose_headers=["X-Request-Id", "X-Process-Time-Ms"],
    max_age=600,
)

参数	作用	注意
`allow_origins`	允许的来源白名单	生产环境显式列出，别用 `["*"]`
`allow_origin_regex`	用正则匹配来源	适合多子域，转义要正确
`allow_methods`	允许的 HTTP 方法	需要 `OPTIONS`（预检用）
`allow_headers`	允许客户端携带的请求头	自定义头必须列出，否则预检失败
`allow_credentials`	是否允许携带 Cookie / 认证信息	与 `allow_origins=["*"]` 互斥
`expose_headers`	允许 JS 读取的响应头	不列出的响应头前端读不到
`max_age`	预检结果缓存秒数	减少 OPTIONS 请求数量

9.5.1 重点坑：`allow_credentials=True` 不能配 `["*"]`

❌ 不推荐：

python

app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True,
                   allow_methods=["*"], allow_headers=["*"])

浏览器会拒绝：The value of the 'Access-Control-Allow-Origin' header in the response must not be the wildcard '*' when the request's credentials mode is 'include'.

✅ 推荐：需要带 Cookie 就显式列出来源（可由环境变量注入）；确实要放开所有来源就关掉凭证。

python

import os

cors_origins = [o.strip() for o in os.getenv("CORS_ORIGINS", "").split(",") if o.strip()]
app.add_middleware(CORSMiddleware, allow_origins=cors_origins, allow_credentials=True,
                   allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
                   allow_headers=["Authorization", "Content-Type"])

🔑 allow_origins=["*"] + allow_credentials=True 一旦被浏览器接受，等于允许任意站点以用户身份调用你的 API（CSRF 级风险）。即便某些 CDN 会「帮」你改成回显来源，也不应依赖这种行为。

9.5.2 预检请求（OPTIONS preflight）的触发条件

浏览器在下列任一情况下会先发 OPTIONS 预检：方法不是简单方法（GET / HEAD / POST 之外）；Content-Type 不是表单编码 / multipart/form-data / text/plain（例如 application/json）；携带了自定义请求头（如 Authorization）。预检通过后浏览器才发真实请求，因此 allow_methods 必须有 OPTIONS、allow_headers 不能漏自定义头。另注意预检请求不带 Cookie，鉴权中间件要对 OPTIONS 放行，否则预检直接 401/405，浏览器表现为 CORS 错误。

9.6 其他内置中间件

python

from fastapi.middleware.gzip import GZipMiddleware
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware

app.add_middleware(GZipMiddleware, minimum_size=1024, compresslevel=6)
app.add_middleware(TrustedHostMiddleware, allowed_hosts=["blog.example.com", "*.example.com"])
app.add_middleware(HTTPSRedirectMiddleware)  # http 请求 307 跳到 https

中间件	作用	注意
`GZipMiddleware`	压缩响应体	`minimum_size` 以下的响应不压缩；与流式响应（SSE）同用会破坏逐块推送
`TrustedHostMiddleware`	校验 `Host` 请求头	反代场景要包含反代转发的主机名
`HTTPSRedirectMiddleware`	HTTP 跳 HTTPS	需可靠的 `X-Forwarded-Proto`，否则可能死循环

💡 生产环境的压缩与 TLS 终止通常放在 Nginx / 负载均衡层，应用层只保留 TrustedHostMiddleware。同一件事别做两遍。

9.7 生命周期：`lifespan` 替代 `on_event`

@app.on_event("startup") / ("shutdown") 已废弃，改用 lifespan + @asynccontextmanager：

python

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from fastapi import FastAPI


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    # ---- 启动阶段 ----
    app.state.pool = await create_pool(dsn="postgresql://user:pwd@localhost:5432/app")
    try:
        yield  # ---- 应用运行中 ----
    finally:
        # ---- 关闭阶段：即使 yield 前抛异常也会执行 ----
        await app.state.pool.close()


app = FastAPI(title="Blog API", lifespan=lifespan)

yield 之后的代码在应用关闭时执行；try/finally 保证半初始化的资源也能被释放。取用资源的推荐方式仍是通过依赖（get_pool 内部读 request.app.state.pool），避免路由层直接操作 app.state。

⚠️ lifespan 中的异常会导致应用启动失败。这是正确行为——连接池建不起来就不该接流量。不要用 try/except 把启动异常吞掉，那只会把故障推迟到第一个请求。

9.8 `BackgroundTasks`：响应之后做事

python

from fastapi import BackgroundTasks


async def send_welcome_email(email: str, nickname: str) -> None:
    ...  # 真实实现应使用异步 SMTP / 第三方 API


def write_audit_log(email: str) -> None:
    ...  # 同步函数会被放到线程池执行，不会阻塞事件循环


@app.post("/signup/")
async def signup(form: SignupForm, tasks: BackgroundTasks) -> dict[str, str]:
    tasks.add_task(send_welcome_email, form.email, form.nickname)
    tasks.add_task(write_audit_log, form.email)
    return {"message": "注册成功，欢迎邮件稍后送达"}

参数注入方式：在路由函数签名里声明 tasks: BackgroundTasks（无默认值）即可，不需要 Depends。

维度	`BackgroundTasks`	Celery / RQ / Arq 类队列
运行位置	同进程内	独立 worker 进程
持久化	无，进程挂掉任务丢失	有（broker 持久化）
重试 / 定时	无内置	支持重试、延迟、定时
可观测性	只能靠日志	有队列长度、失败率等指标
适用	发通知、写审计、清缓存等「尽力而为」	订单处理、报表生成等需要可靠投递的任务

🔑 判断标准：任务丢了会不会造成业务损失？会 → 上队列；不会 → BackgroundTasks 足够。

最大的坑是后台任务用到已关闭的资源：依赖里的 yield 会话在响应发送后就关闭了，tasks.add_task(notify, session) 会在 notify 里报 session is closed。

✅ 推荐：后台任务内自己开资源、自己关，不要跨边界传会话。

python

async def notify(order_id: int) -> None:
    async with SessionFactory() as session:  # 自己的会话，自己的生命周期
        await session.execute(insert_audit_stmt(order_id))
        await session.commit()

9.9 综合示例：把四件事组织在一起

python

# 1) lifespan 先决定共享资源（建/销连接池，代码同 9.7 节）
app = FastAPI(title="Blog API", lifespan=lifespan)

# 2) CORS 放较外层，保证预检与错误响应也带 CORS 头
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://blog.example.com"],
    allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
    allow_headers=["Authorization", "Content-Type", "X-Request-Id"],
    allow_credentials=True,
    expose_headers=["X-Request-Id", "X-Process-Time-Ms"],
    max_age=600,
)


# 3) 请求 ID + 耗时日志（函数体同 9.2 节）
@app.middleware("http")
async def observability(request: Request, call_next):
    ...

装配顺序建议：先用 lifespan 决定有哪些共享资源（其他一切依赖它），再放 CORSMiddleware，最后放可观测性中间件。

常见坑与排查

现象	原因	解决
浏览器报 `must not be the wildcard '*' when credentials mode is 'include'`	`allow_credentials=True` 与 `allow_origins=["*"]` 冲突	显式列出来源（可用环境变量注入），或关掉 `allow_credentials`
预检 `OPTIONS` 返回 405	`allow_methods` 未包含 `OPTIONS`，或鉴权中间件拦截了 `OPTIONS`	在 `allow_methods` 加 `OPTIONS`；鉴权中间件对 `request.method == "OPTIONS"` 直接放行
前端读不到 `X-Request-Id`	响应头没有放进 `expose_headers`	把需要在 JS 中读取的响应头加入 `expose_headers`
路由函数拿到的 body 为空	上游 `BaseHTTPMiddleware` 里 `await request.body()` 消耗了 receive 流	改用纯 ASGI 中间件做 tee，或在路由层用依赖读取 body
路由里访问 `request.app.state.pool` 报 `AttributeError`	在 `lifespan` 之前（或未配置 `lifespan`）就注册了需要该资源的中间件	资源初始化统一放 `lifespan`；中间件内用 `getattr` 兜底并记录告警
后台任务报 `session is closed`	`BackgroundTasks` 在依赖的 `finally` 清理之后执行	后台任务内自建资源（`async with SessionFactory()`），不要跨边界传会话
开启 `GZipMiddleware` 后 SSE / 流式接口卡住不刷	压缩需要缓冲齐整块数据，破坏了逐块推送	对流式端点跳过压缩，或在 Nginx 层处理压缩
鉴权被绕过，非法请求也能进业务	中间件添加顺序错误：鉴权比身份注入更靠内	记住后添加的在外层；身份注入中间件应先添加（更内层）、鉴权后添加（更外层先执行）
应用启动直接失败并提示 lifespan 错误	`lifespan` 里初始化抛异常（连不上数据库等）	这是正确行为：修好依赖或加清晰日志，不要吞掉异常
纯 ASGI 中间件导致应用无法启动	未透传非 `http` 类型 scope（`lifespan` / `websocket`）	开头判断 `if scope["type"] != "http"` 并原样调用 `self.app(scope, receive, send)`

本章小结

要点	说明
洋葱模型	请求由外向内，响应由内向外；不调 `call_next` 即短路
装饰器写法	`@app.middleware("http")` 基于 `BaseHTTPMiddleware`，最常用
纯 ASGI	性能更好、能操作消息流，但必须透传 `lifespan` / `websocket` scope
添加顺序	`add_middleware` 是栈式，后添加的先执行
CORS 核心	`allow_origins` 显式列举；`allow_credentials=True` 禁用通配符
预检触发	非简单方法、非简单 `Content-Type`、自定义请求头三者之一
`expose_headers`	不列入的响应头 JS 读不到
`lifespan`	`@asynccontextmanager` + `yield`；异常导致启动失败是正确的
`try/finally`	保证半初始化资源也能被释放
`BackgroundTasks`	同进程、不持久化、不重试；丢了有损失的活儿交给队列
后台任务资源	自建自销，绝不传入依赖里的 session

练习题

写一个纯 ASGI 中间件，统计每个路径的请求次数与总耗时，并在应用关闭时通过 lifespan 打印汇总。
配置一份 CORS，要求：允许 https://*.example.com 全部子域、允许携带 Cookie、允许 Authorization 头、暴露 X-Request-Id，并说明为什么不能用 allow_origins=["*"]。
用 lifespan 完成「启动时连接 Redis、关闭时释放」，并在路由中通过 Depends 取用；故意让 create_redis 抛异常，观察应用启动行为。
实现「注册成功后异步发送欢迎邮件」：先用 BackgroundTasks 实现，再列出换成 Celery 后需要额外改动哪些部分（至少 4 项）。

下一章预告

现在应用有了生命周期、有了统一错误格式，但数据还是内存里的假数据。下一章接入真实数据库，用 SQLAlchemy 2.0 异步引擎把 lifespan 里的连接池换成生产可用实现。
👉 第 10 章：数据库集成（SQLAlchemy 2.0）

第 9 章：中间件、CORS 与生命周期 ​

学习目标 ​

9.1 中间件是什么：洋葱模型 ​

9.2 @app.middleware("http") 装饰器写法 ​

9.3 BaseHTTPMiddleware vs 纯 ASGI 中间件 ​

9.3.1 错误示范 vs 正确示范 ​

9.4 中间件执行顺序：后添加的先执行 ​

9.5 CORSMiddleware 完整配置 ​

9.5.1 重点坑：allow_credentials=True 不能配 ["*"] ​

9.5.2 预检请求（OPTIONS preflight）的触发条件 ​

9.6 其他内置中间件 ​

9.7 生命周期：lifespan 替代 on_event ​

9.8 BackgroundTasks：响应之后做事 ​

9.9 综合示例：把四件事组织在一起 ​

常见坑与排查 ​

本章小结 ​

练习题 ​

下一章预告 ​

第 9 章：中间件、CORS 与生命周期

学习目标

9.1 中间件是什么：洋葱模型

9.2 `@app.middleware("http")` 装饰器写法

9.3 `BaseHTTPMiddleware` vs 纯 ASGI 中间件

9.3.1 错误示范 vs 正确示范

9.4 中间件执行顺序：后添加的先执行

9.5 `CORSMiddleware` 完整配置

9.5.1 重点坑：`allow_credentials=True` 不能配 `["*"]`

9.5.2 预检请求（OPTIONS preflight）的触发条件

9.6 其他内置中间件

9.7 生命周期：`lifespan` 替代 `on_event`

9.8 `BackgroundTasks`：响应之后做事

9.9 综合示例：把四件事组织在一起

常见坑与排查

本章小结

练习题

下一章预告