FastAPI中间件深入指南：执行时机、写法与避坑实践

前阵子在给一个内部管理系统补接口层统一能力的时候，发现很多同行对FastAPI中间件的理解还停留在“复制一段CORS代码”的阶段。一旦要加登录态解析、耗时统计、接口频控，就开始往每个路由函数里复制粘贴，或者干脆自己写个装饰器包一层。这种写法不是不行，但项目一多、路由一多，维护成本就直线上升。FastAPI中间件这东西，说简单也简单，说深也深，它决定了你在哪里处理“所有请求都要做的那件事”，值得单独写一篇把它讲透。

这篇文章适合这么几类人看：已经会用FastAPI写接口，但还没系统搞清楚中间件执行时机的人；用中间件只是抄过官方CORS示例，想自己写一个的人；以及项目里已经堆了五六个中间件，开始分不清执行顺序、踩了坑的人。我会从请求链路讲起，后面直接给可落地的代码和踩坑结论，尽量不绕弯。

1. FastAPI中间件到底夹在哪一层：先建立正确的心智模型

1.1 一次HTTP请求经过FastAPI时发生了什么

很多初学者以为FastAPI中间件就是“路由前面的一个函数”，这个理解太粗了。要搞清楚中间件，得先知道FastAPI本身其实是构建在Starlette之上的，而Starlette是一个ASGI框架，中间件本质上是ASGI应用的一层层包装。

一次请求从浏览器或者后端服务发出来，到你的接口函数执行完，大致会经过这条链：

1. Uvicorn这类ASGI服务器收到HTTP请求，把请求转成scope、receive、send三个对象。
2. 调用FastAPI应用实例，也就是那个ASGI app。
3. app内部并不是直接去找路由，而是先经过一层层中间件包装后的对象。
4. 中间件做完自己的事情，调用下一层，直到进入路由匹配。
5. 路由命中后，FastAPI执行依赖注入、校验参数，最后调用你的端点函数。
6. 返回的Response沿着原路逆向一层层传出来，每一层中间件在拿到响应之后，还可以做后置处理。

这里最关键的一点是：中间件在路由分发之前执行，在路由处理完之后也会执行。这也是为什么它天然适合处理CORS、日志、鉴权这类横切逻辑。

我用一个生活化类比：路由函数是具体某个窗口的办事员，中间件是大厅入口的安检和出口的盖章。你进出大厅不管办什么事，都必须经过那道闸机，但闸机完全不知道你在几号窗口办了什么事，它只管放行和记录。

1.2 中间件能横切什么、不能横切什么

先想清楚边界，才不会在写的时候产生“中间件怎么这也不能干”的挫败感。

我列一个简单的对照表，基本覆盖日常开发会遇到的情况：

这里面最容易让人误判的是“我想在中间件里拿到路径参数”。比如请求“/user/123”，你想在中间件里知道id=123，从而判断这个用户能不能操作这个资源。很遗憾，中间件执行时路由还没开始匹配，path参数还没有被解析出来。虽然scope里的path字段能看到原始路径字符串，但路由匹配结果要等Router去算。你要么在中间件里自己写解析逻辑，要么把这个判断放回依赖注入里做，后者才符合FastAPI的设计习惯。

2. 从零手写两个中间件：装饰器版本和纯ASGI版本

2.1 最快上手的@app.middleware("http")写法

如果你只想快速给所有接口加一个统一功能，最简单的方式是用FastAPI自带的装饰器。

from fastapi import FastAPI, Request import time app = FastAPI() @app.middleware("http") async def add_process_time_header(request: Request, call_next): start = time.perf_counter() response = await call_next(request) cost_ms = (time.perf_counter() - start) * 1000 response.headers["X-Process-Time"] = f"{cost_ms:.2f}ms" return response

这段代码里，call_next是一个可调用对象，它接收request，返回真正的Response。你在await call_next(request)之前写的逻辑，是请求从中间件往下传之前做的；之后写的逻辑，是响应一路返回之后做的。

这个写法本质上是Starlette封装好的快捷方式，适合逻辑不超过三五十行的场景。但要注意，装饰器函数必须是async，因为ASGI本身就是异步模型，很多新手在这里写成普通def，结果发现请求根本不经过中间件，因为FastAPI只认async函数。

2.2 更进一步：继承BaseHTTPMiddleware

装饰器写法简单，但当你需要把中间件做成一个可复用组件，或者参数比较多的时候，建议改成类，继承starlette.middleware.base.BaseHTTPMiddleware。

import logging import time from starlette.middleware.base import BaseHTTPMiddleware from starlette.requests import Request logger = logging.getLogger("access") class AccessLogMiddleware(BaseHTTPMiddleware): def __init__(self, app, *, header_name: str = "X-Process-Time"): super().__init__(app) self.header_name = header_name async def dispatch(self, request: Request, call_next): start = time.perf_counter() response = await call_next(request) cost_ms = (time.perf_counter() - start) * 1000 response.headers[self.header_name] = f"{cost_ms:.2f}ms" logger.info( "%s %s -> %d cost=%.2fms", request.method, request.url.path, response.status_code, cost_ms, ) return response

然后在应用里挂载：

app = FastAPI() app.add_middleware(AccessLogMiddleware, header_name="X-Cost")

类式写法把逻辑封装成独立单元，代码组织会清爽很多。如果你在多个项目里复用同一个中间件，还可以把它做成一个单独的py模块，甚至发布成小工具包。

2.3 还不满足：直接写一个ASGI中间件

Starlette的BaseHTTPMiddleware虽然不是性能瓶颈，但它引入了额外的一层抽象，会丢失一些底层ASGI的能力。当你需要精确控制每个ASGI事件，或者想绕开它带来的流式响应缓冲问题时，可以手写纯ASGI中间件。

class CustomASGIMiddleware: def __init__(self, app): self.app = app async def __call__(self, scope, receive, send): if scope["type"] != "http": await self.app(scope, receive, send) return async def send_wrapper(message): if message["type"] == "http.response.start": headers = message.get("headers", []) headers = list(headers) headers.append((b"x-custom-asgi", b"1")) message["headers"] = headers await send(message) await self.app(scope, receive, send_wrapper)

这个写法看起来有点绕，但理解后其实很直观。scope是本次请求的上下文，receive是异步拿请求体的函数，send是异步发送响应的函数。你只需要在真正调用底层app前做点事，然后包装send拦截响应开始时的事件，把自定义头加进headers里再放行。

它的好处是全程不构造Response对象，也不会主动去读响应体，对大响应、SSE、文件下载这种场景特别友好。代价是代码更接近底层，代码可读性差一点，出错了也没那么多官方中间件给你兜底。

3. 高频场景落地：CORS、日志、鉴权、限流的正手和反手

3.1 CORS跨域：最容易被忽略的两个参数

跨域是前后端分离项目绕不开的问题。FastAPI自带CORS中间件，绝大多数情况下直接复用就行，不需要自己造轮子。

from fastapi.middleware.cors import CORSMiddleware app.add_middleware( CORSMiddleware, allow_origins=["https://example.com", "https://admin.example.com"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], )

这里面有一个经验教训。allow_origins不能设成["*"]的同时把allow_credentials设成True，因为浏览器规范不允许“全来源”和“带上Cookie凭证”同时存在。如果你要支持带登录Cookie的跨域请求，allow_origins必须明确列出具体域名。很多开发本地调试没问题，一发到测试环境发现浏览器一直报跨域，排查半天就是这个配置组合的问题。

另外提醒一点：CORS中间件要在请求跨域的Options预检阶段就返回200，所以它应当注册在很外层。后面我会讲中间件注册顺序，这里先留个印象。

3.2 访问日志与响应耗时：把request.state用起来

我习惯在每个服务里加一个访问日志中间件，记录谁在什么时候调了什么接口、花了多久、返回了什么状态码。这个中间件的代码量和上面AccessLogMiddleware类似，但有一个进阶玩法是用request.state。

FastAPI的Request对象上有一个state属性，相当于一个架子，可以往上面挂任意属性。中间件可以把解析出来的信息塞进request.state，后面依赖注入或者端点函数里能直接读。

@app.middleware("http") async def request_id_middleware(request: Request, call_next): request_id = request.headers.get("X-Request-ID") if not request_id: request_id = uuid.uuid4().hex[:16] request.state.request_id = request_id response = await call_next(request) response.headers["X-Request-ID"] = request_id return response

然后在接口里：

@app.get("/ping") async def ping(request: Request): return {"request_id": request.state.request_id}

这样做的好处是，一个请求的生命周期内，request_id是同一个，前端传过来的追踪ID在日志里能贯穿整个链路，排障的时候非常有用。

3.3 Token校验：中间件只做粗筛，依赖做细查

很多人一上来就把用户查库的逻辑放进中间件，这是个不太合适的用法。中间件没有依赖注入，你没法用FastAPI的Depends机制拿到DB Session，只能手动创建连接，还得自己管理事务。更麻烦的是中间件对所有路由生效，而你某些路由可能根本不需要鉴权。

我的建议是分工：中间件只做“粗筛”，比如检查Authorization头存不存在、格式对不对、token是否过期前的基本解析；真正的用户身份查询、权限判断放在依赖注入里做。

@app.middleware("http") async def token_extract_middleware(request: Request, call_next): auth_header = request.headers.get("Authorization", "") if auth_header.startswith("Bearer "): request.state.token = auth_header[7:] else: request.state.token = None return await call_next(request)

然后在依赖里读取token，同时可以做白名单逻辑：

from fastapi import Depends, HTTPException, Request WHITE_LIST = {"/healthz", "/docs", "/openapi.json"} def get_current_user(request: Request): if request.url.path in WHITE_LIST: return None token = getattr(request.state, "token", None) if token is None: raise HTTPException(status_code=401, detail="missing token") user = query_user_by_token(token) # 这里可以放心查库 return user

这样“哪些接口要鉴权”的逻辑就下放到依赖里，接口可以灵活地加Depends(get_current_user)，而不是只能被中间件一刀切。

3.4 简单限流：单机版怎么做，多机部署要注意什么

限流是另一个代表性的横切需求。我用过一个单机滑动窗口版本，简单够用，适合内部系统。

import time from collections import defaultdict from starlette.requests import Request from starlette.responses import JSONResponse class SimpleRateLimitMiddleware: def __init__(self, app, *, max_requests: int = 30, window_seconds: int = 10): self.app = app self.max_requests = max_requests self.window_seconds = window_seconds self.records = defaultdict(list) async def __call__(self, scope, receive, send): if scope["type"] != "http": await self.app(scope, receive, send) return request = Request(scope) ip = request.headers.get("x-forwarded-for", request.client.host).split(",")[0].strip() now = time.monotonic() recent = [t for t in self.records[ip] if now - t < self.window_seconds] self.records[ip] = recent if len(recent) >= self.max_requests: response = JSONResponse({"detail": "too many requests"}, status_code=429) await response(scope, receive, send) return self.records[ip].append(now) await self.app(scope, receive, send)

这里有个注意点：如果服务背后挂了Nginx或网关，request.client.host拿到的一直是网关的内网IP，没办法区分真实客户端。所以我先取x-forwarded-for，再取第一个IP，这是因为标准链路中真实客户端IP在最前面。如果你直接用默认的client.host，等于所有请求共享一个桶，限流就失效了。

另一个关键问题是单机内存计数在多Worker、多实例部署下完全不准确，Uvicorn开两个Worker进程，每个进程各有一个计数器，10秒内实际能放行的请求量会翻倍。生产环境真要限流，建议把计数逻辑迁移到Redis，中间件只负责调Redis接口。

3.5 统一响应包装：能写，但我劝你别这么干

统一响应体是很多前端团队的要求，比如所有成功响应都长这样：{"code": 0, "message": "ok", "data": ...}。这个需求在FastAPI里有好几种实现方式，但“用中间件读body再包装”是最不推荐的一种。

反面教材写法大概长这样：

import json @app.middleware("http") async def wrap_response_middleware(request: Request, call_next): response = await call_next(request) if response.status_code >= 400: return response body = b"" async for chunk in response.body_iterator: body += chunk data = json.loads(body) new_body = json.dumps({"code": 0, "message": "ok", "data": data}) return JSONResponse(content=json.loads(new_body), status_code=response.status_code)

表面上看能用，但你一旦这么写，会遇到三个问题。第一，如果接口返回的是文件流、视频流、SSE，body_iterator会继续吐二进制数据，你把它当成JSON解析会直接报错，或者包装出来的内容毫无意义。第二，即使全是JSON响应，你也把整个响应体读进了内存，大接口响应很容易让内存飙升。第三，fastapi里有些接口返回StreamingResponse，期望的是边生成边推送，你这一读等于把流式效果彻底废掉。

我的经验是用异常处理器加统一响应模型，或者干脆在路由层约定返回结构，而不是在中间件里做这件事。真要处理异常，FastAPI有现成的ExceptionHandler机制，可以给自定义异常返回统一错误格式，这个方案比中间件干净得多。

4. 中间件的边界：和异常处理、流式响应、BackgroundTasks的相处之道

4.1 call_next会不会抛出HTTPException

这是一个非常好的问题，也是我早期踩过的坑。

在FastAPI里，路由函数抛出HTTPException时，这个异常并不是直接抛给你的中间件，而是由Router层抛出，被Starlette的ExceptionMiddleware捕获并转成一个标准的JSONResponse。而你自己注册的中间件，是挂在ExceptionMiddleware之外的。也就是说，你在中间件里await call_next(request)，拿到的是一个正常的Response对象，状态码可能是404、400，但不会抛HTTPException。

所以你在中间件里不需要专门写“捕获HTTPException再转成响应”的逻辑，因为底层已经处理好了。反过来，如果你在中间件里自己抛了一个普通Exception，且没有catch，这个异常会往外抛，最终被最外层的ServerErrorMiddleware捕获，返回一个500响应。中间件里的异常兜底通常这么写：

@app.middleware("http") async def error_catch_middleware(request: Request, call_next): try: return await call_next(request) except Exception: # 打日志、上报监控 return JSONResponse(status_code=500, content={"detail": "internal error"})

要注意这个兜底是最后的手段，它捕获不到已经被ExceptionMiddleware处理的HTTPException，也不能在响应已经开始发送之后再去改状态码。

4.2 在中间件里读body，等于亲手毁掉流式响应

中间件里可以拿到response.body_iterator，但不代表你应该去读它。BaseHTTPMiddleware返回的Response对象，body_iterator是异步迭代器，一旦你把它完整消费掉，后面的流式推送就没了。

我实际遇到过一个事故：某个导出的Excel接口，数据量大概几十万行，服务端使用StreamingResponse边查边写。上了统一响应包装中间件之后，导出文件变得非常大，而且客户端迟迟收不到响应，最后超时。排查原因就是中间件把整个流式响应缓冲成了body，再重新包装成JSONResponse，Excel内容已经变成字节串被塞进JSON里了。

如果你只是想统计响应大小，正确的做法是改用纯ASGI中间件，在send_wrapper里累计每个http.response.body消息的长度，而不是去读取body_iterator。

4.3 注册顺序决定执行顺序：后添加的先执行

我见过不少项目，中间件挂了好几个，但没人说得清谁先执行。Starlette对用户中间件的处理方式是“后添加的先执行”，有点像一个洋葱从外往里包，后包的在外面。

比如：

app.add_middleware(CORSMiddleware, ...) app.add_middleware(AccessLogMiddleware, ...)

执行顺序是AccessLogMiddleware先收到请求，它调call_next后才进入CORS中间件，再往下才到路由。因为AccessLogMiddleware是后添加的，它实际处在外层。

这个顺序对业务是有影响的。比如你把CORS放在最外层，那么浏览器发出的OPTIONS预检请求会先经过CORS中间件，CORS中间件能直接返回预检响应，后面的日志中间件不会记录到OPTIONS请求。反过来，如果日志在最外层，日志中间件会先看到OPTIONS请求并记录，然后才交给CORS处理。这不是对错问题，而是你需要根据自己的排查需求决定。

建议是：把全局异常兜底和请求ID相关的中间件放最外层，业务相关的往下放，CORS放在偏外层的位置。越靠外，越意味着“所有请求都必须先经过这里”。

4.4 中间件 vs 依赖注入：什么时候用哪个

我在开发中一般用一条很简单的规则来判断：如果这件事跟具体接口的入参和返回值强相关，放到依赖注入；如果这件事是所有请求无差别都要做的横切动作，放到中间件。

举个例子，用户鉴权里“这个token是谁、有没有权限操作这个资源”属于接口业务逻辑，应当放依赖；而“请求里有没有Authorization头、格式对不对”属于横切动作，可以放中间件。再比如，DB Session的获取放依赖，因为你只有进入具体路由之后才知道查哪个库、用什么事务。

所以中间件不是万能的，也不是越强大越好。能不用中间件解决的，尽量别用，保持这个判断标准能让你少写很多麻烦代码。

5. 工程化补充：测试、热更新和我的中间件清单

5.1 用TestClient把中间件变成可回归的用例

中间件一旦多了，很担心某天改一处把另一处搞挂。所以中间件也要写测试。

FastAPI自带的TestClient基于httpx，可以直接调用整个应用，包括所有中间件。

from fastapi.testclient import TestClient def test_request_id_middleware(): client = TestClient(app) resp = client.get("/ping") assert resp.status_code == 200 assert "X-Request-ID" in resp.headers assert "X-Process-Time" in resp.headers

如果你用纯ASGI中间件，也可以直接用httpx的ASGITransport：

import httpx async def test_asgi_middleware(): transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: resp = await client.get("/ping") assert resp.headers.get("x-custom-asgi") == "1"

我建议把每个中间件的“核心行为”拆成独立测试用例，注册顺序改了、参数变了，测试就会提醒你。中间件这类全局逻辑如果出了bug，影响面是整个服务，值得这个投入。

5.2 一个容易忽略的环境问题：uv创建虚拟环境与热更新

既然聊到了FastAPI项目落地，顺便提两个跟环境相关的点。

现在新建FastAPI项目，我基本都用uv来管理虚拟环境，比手动维护requirements.txt顺滑很多：

uv venv uv pip install "fastapi[standard]" uv run uvicorn main:app --reload

有些人在PyCharm里直接装fastapi装不上，大概率是没激活虚拟环境、或者默认pip源访问慢，解决方案就是在终端里先用uv建好虚拟环境，再把PyCharm解释器指过去。这个方式能绕开大部分安装失败报错。

另一个高频问题是“fastapi启动不热更新”。如果你看到服务起来了，但改代码之后没有自动重载，先确认启动命令里有没有加--reload。注意如果一个进程是用uvicorn main:app --reload启动，另一个进程又在跑普通uvicorn main:app，端口会冲突，也可能让你误以为热更新失效。再有就是确认watch目录对不对，默认会监听当前工作目录，如果你的代码在别的目录，要加--reload-dir去指定。

5.3 我目前的通用中间件清单

最后分享一个比较克制的中间件配置。之前踩过很多坑之后，我发现中间件并不是越多越好，每多一层洋葱皮，请求链路就多一次跳转和潜在风险。

我现在内部基础服务一般只保留这几样：

真正的用户鉴权、数据库操作、业务校验全部放到依赖注入里，中间件保持轻薄、可测试。

中间件这东西，刚学的时候觉得它是救命稻草，想什么功能都往里塞；用多了反而觉得它应该尽量少、尽量薄。判断标准很简单：如果这个逻辑不是“所有请求都必须经历”的横切动作，就把它放下层，能不用中间件就不用中间件。这样你的服务在排查问题的时候，链路才能一眼看透。