中间件

async def my_middleware(request: Request, call_next):
    # ── 时机①「进」：请求还没到路由，可以看/改 request ──
    response = await call_next(request)
    # ── 时机②「出」：拿到下一层的响应，可以看/改 response ──
    return response

async def block_scanners(request: Request, call_next):
    if request.headers.get("user-agent", "").lower().startswith("scrapy"):
        return JSONResponse({"code": "40303", "message": "爬虫走开"}, status_code=403)
    return await call_next(request)

客户端请求
  ↓
ServerErrorMiddleware（Starlette 自带，最外层：兜住没人处理的异常 → 500）
  ↓
你注册的中间件（后注册的在外层，本章主角）
  ↓
ExceptionMiddleware（Starlette 自带，最内层：第 09 章的异常处理器住在这里）
  ↓
路由 → 依赖注入 → 校验 → 你的处理函数

@app.middleware("http")
async def 任意名字(request: Request, call_next: Callable[[Request], Awaitable[Response]]) -> Response:
    ...

from collections.abc import Awaitable, Callable

from fastapi import FastAPI, Request, Response

app = FastAPI(title="墨集 inkhub")


@app.middleware("http")
async def add_server_banner(
    request: Request, call_next: Callable[[Request], Awaitable[Response]]
) -> Response:
    response = await call_next(request)
    response.headers["X-Powered-By"] = "inkhub"
    return response

app.add_middleware(BaseHTTPMiddleware, dispatch=add_server_banner)

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.types import ASGIApp


class BannerMiddleware(BaseHTTPMiddleware):
    def __init__(self, app: ASGIApp, *, banner: str = "inkhub") -> None:
        super().__init__(app)
        self.banner = banner

    async def dispatch(
        self, request: Request, call_next: Callable[[Request], Awaitable[Response]]
    ) -> Response:
        response = await call_next(request)
        response.headers["X-Powered-By"] = self.banner
        return response


# 注册时把配置传进去
app.add_middleware(BannerMiddleware, banner="inkhub-editors")

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

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        await self.app(scope, receive, send)  # 直接操作协议层，没有 request/response 包装

rid = contextvars.ContextVar("rid", default="unset")

@app.middleware("http")
async def mw(request, call_next):
    resp = await call_next(request)
    resp.headers["X-Rid-After"] = rid.get()  # 读到的是 "unset"，不是端点设置的值！
    return resp

@app.get("/probe")
async def probe():
    rid.set("set-by-endpoint")
    return {"ok": True}

response.headers["X-Process-Time"] = 耗时字符串

import time
from collections.abc import Awaitable, Callable

from fastapi import Request, Response


@app.middleware("http")
async def timing_middleware(
    request: Request, call_next: Callable[[Request], Awaitable[Response]]
) -> Response:
    start = time.perf_counter()
    response = await call_next(request)
    elapsed_ms = (time.perf_counter() - start) * 1000
    response.headers["X-Process-Time"] = f"{elapsed_ms:.3f}"
    return response

curl -i http://127.0.0.1:8000/api/posts

HTTP/1.1 200 OK
x-process-time: 0.584        ← 这枚章
content-type: application/json

{"code":"0","data":[...],"message":"success"}

Access to fetch at 'http://127.0.0.1:8000/api/posts' from origin
'http://localhost:5173' has been blocked by CORS policy

OPTIONS /api/posts
Origin: http://localhost:5173
Access-Control-Request-Method: POST
Access-Control-Request-Headers: content-type

HTTP/1.1 200 OK
access-control-allow-origin: http://localhost:5173
access-control-allow-methods: GET, POST
access-control-allow-headers: Accept, Accept-Language, Authorization, Content-Language, Content-Type
access-control-max-age: 600

from starlette.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:5173", "http://127.0.0.1:5173"],
    allow_credentials=True,
    allow_methods=["GET", "POST"],
    allow_headers=["Authorization", "Content-Type"],
    expose_headers=["X-Process-Time"],
)

# 预检：三个请求头缺一不可，缺了就不是预检，只是个普通 OPTIONS
curl -i -X OPTIONS http://127.0.0.1:8000/api/posts \
  -H "Origin: http://localhost:5173" \
  -H "Access-Control-Request-Method: POST" \
  -H "Access-Control-Request-Headers: content-type"

# 普通请求带 Origin：看响应头回显
curl -i http://127.0.0.1:8000/api/posts -H "Origin: http://localhost:5173"

{"code": "0", "data": {"id": 1, "title": "中间件：FastAPI 的物业部"}, "message": "success"}

{"code": "40404", "data": null, "message": "文章不存在"}

import json

from fastapi.responses import JSONResponse


@app.middleware("http")
async def envelope_middleware(
    request: Request, call_next: Callable[[Request], Awaitable[Response]]
) -> Response:
    response = await call_next(request)

    # 放行判断①：非业务接口（/docs、/openapi.json、/healthz……）不动
    if not request.url.path.startswith("/api/"):
        return response

    # 放行判断②：非 JSON 响应（SSE 流、文件下载）原样放行
    if "json" not in response.headers.get("content-type", ""):
        return response

    # 第一段：读出响应体。body_iterator 只能消费一次，读完回不了头
    body = b""
    async for chunk in response.body_iterator:
        body += chunk

    # 第二段：复制原响应头（保住 X-Process-Time 等），删掉旧长度
    headers = dict(response.headers)
    headers.pop("content-length", None)

    payload = json.loads(body) if body else None

    # 第三段：错误分支——拍平，防二次包裹
    if response.status_code >= 400:
        err = payload if isinstance(payload, dict) else {}
        return JSONResponse(
            content={
                "code": err.get("code", str(response.status_code)),
                "data": None,
                "message": err.get("message", ""),
            },
            status_code=response.status_code,
            headers=headers,
        )

    # 第四段：成功分支——原数据整体放进 data
    return JSONResponse(
        content={"code": "0", "data": payload, "message": "success"},
        status_code=response.status_code,
        headers=headers,
    )

if isinstance(payload, dict) and "code" in payload and "data" in payload:
    return JSONResponse(content=payload, status_code=response.status_code, headers=headers)  # 已是信封，原样重建

# starlette/applications.py（简化）
def add_middleware(self, middleware_class, *args, **kwargs):
    self.user_middleware.insert(0, Middleware(middleware_class, *args, **kwargs))  # 插队首

def build_middleware_stack(self):
    middleware = [ServerError] + self.user_middleware + [Exception]
    app = self.router                       # 从路由开始
    for cls, args, kwargs in reversed(middleware):   # 从列表尾（最内层）往外包
        app = cls(app=app, *args, **kwargs)
    return app

trace: list[str] = []

@app.middleware("http")          # 代码里写在上面
async def mw_a(request, call_next):
    trace.append("a-in")
    resp = await call_next(request)
    trace.append("a-out")
    return resp

@app.middleware("http")          # 代码里写在下面
async def mw_b(request, call_next):
    trace.append("b-in")
    resp = await call_next(request)
    trace.append("b-out")
    return resp

trace: ['b-in', 'a-in', 'route', 'a-out', 'b-out']

@app.middleware("http")
async def my_middleware(request, call_next): ...
# 等价于
app.add_middleware(BaseHTTPMiddleware, dispatch=my_middleware)
# 也等价于：不写装饰器，拿到函数后手动注册
app.middleware("http")(my_middleware)

# ① 函数式（上面那种等价写法）
app.middleware("http")(my_middleware)

# ② BaseHTTPMiddleware 子类——可以传配置参数
app.add_middleware(BannerMiddleware, banner="inkhub-editors")

# ③ 纯 ASGI 中间件类——CORSMiddleware 这种，装饰器根本写不了
app.add_middleware(CORSMiddleware, allow_origins=["http://localhost:5173"])

import inspect

from fastapi import FastAPI

# 列表顺序 = 注册顺序 = 洋葱从内到外（第一个注册的在最内层）
MIDDLEWARES: list[tuple[type | Callable, dict]] = [
    (timing_middleware, {}),          # 最内层
    (envelope_middleware, {}),
    (CORSMiddleware, {"allow_origins": ["http://localhost:5173"], "expose_headers": ["X-Process-Time"]}),  # 最外层
]


def register_middleware(app: FastAPI) -> None:
    for mw, kwargs in MIDDLEWARES:
        if inspect.isclass(mw):
            app.add_middleware(mw, **kwargs)      # 类 → add_middleware
        else:
            app.middleware("http")(mw)            # 函数 → 装饰器等价注册

uv add fastapi "uvicorn[standard]"
python demo_middleware.py        # 内嵌 uvicorn，等价 uvicorn demo_middleware:app --port 8000

# ① 计时章 + 信封一起看
curl -i http://127.0.0.1:8000/api/posts

HTTP/1.1 200 OK
x-process-time: 0.584
content-type: application/json

{"code":"0","data":[{"id":1,"title":"中间件：FastAPI 的物业部","summary":"横切关注点一次性解决"},{"id":2,"title":"SQLAlchemy 异步会话的正确姿势","summary":null}],"message":"success"}

# ② 错误响应拍平（防二次包裹）：注意 data 是 null，code/message 来自 09 章格式
curl -i http://127.0.0.1:8000/api/posts/999

HTTP/1.1 404 Not Found
x-process-time: 0.477

{"code":"40404","data":null,"message":"文章不存在"}

# ③ SSE 流式放行：不进信封，逐条收到 data: 事件
curl -N http://127.0.0.1:8000/api/posts/1/stream

data: 墨

data: 集

data: inkhub

data: 中间件：FastAPI 的物业部

# ④ 文件下载放行：拿到的是原始 markdown，不是信封
curl -i http://127.0.0.1:8000/api/posts/export

# ⑤ 非 /api/ 路径放行：原样返回，但计时章还在
curl -i http://127.0.0.1:8000/healthz

# ⑥ CORS 预检：200 + Allow-* 头，且预检响应没被信封包裹（CORS 在最外层短路）
curl -i -X OPTIONS http://127.0.0.1:8000/api/posts \
  -H "Origin: http://localhost:5173" \
  -H "Access-Control-Request-Method: POST" \
  -H "Access-Control-Request-Headers: content-type"

# ⑦ 带 Origin 的普通请求：看 allow-origin 回显和 expose-headers
curl -i http://127.0.0.1:8000/api/posts -H "Origin: http://localhost:5173"

HTTP/1.1 200 OK
access-control-allow-methods: GET, POST
access-control-allow-origin: http://localhost:5173
access-control-allow-headers: Accept, Accept-Language, Authorization, Content-Language, Content-Type
access-control-max-age: 600
content-length: 2        ← 预检响应体只是 "OK"，没进信封