前阵子在给一个内部管理系统补接口层统一能力的时候发现很多同行对FastAPI中间件的理解还停留在“复制一段CORS代码”的阶段。一旦要加登录态解析、耗时统计、接口频控就开始往每个路由函数里复制粘贴或者干脆自己写个装饰器包一层。这种写法不是不行但项目一多、路由一多维护成本就直线上升。FastAPI中间件这东西说简单也简单说深也深它决定了你在哪里处理“所有请求都要做的那件事”值得单独写一篇把它讲透。这篇文章适合这么几类人看已经会用FastAPI写接口但还没系统搞清楚中间件执行时机的人用中间件只是抄过官方CORS示例想自己写一个的人以及项目里已经堆了五六个中间件开始分不清执行顺序、踩了坑的人。我会从请求链路讲起后面直接给可落地的代码和踩坑结论尽量不绕弯。1. FastAPI中间件到底夹在哪一层先建立正确的心智模型1.1 一次HTTP请求经过FastAPI时发生了什么很多初学者以为FastAPI中间件就是“路由前面的一个函数”这个理解太粗了。要搞清楚中间件得先知道FastAPI本身其实是构建在Starlette之上的而Starlette是一个ASGI框架中间件本质上是ASGI应用的一层层包装。一次请求从浏览器或者后端服务发出来到你的接口函数执行完大致会经过这条链Uvicorn这类ASGI服务器收到HTTP请求把请求转成scope、receive、send三个对象。调用FastAPI应用实例也就是那个ASGI app。app内部并不是直接去找路由而是先经过一层层中间件包装后的对象。中间件做完自己的事情调用下一层直到进入路由匹配。路由命中后FastAPI执行依赖注入、校验参数最后调用你的端点函数。返回的Response沿着原路逆向一层层传出来每一层中间件在拿到响应之后还可以做后置处理。这里最关键的一点是中间件在路由分发之前执行在路由处理完之后也会执行。这也是为什么它天然适合处理CORS、日志、鉴权这类横切逻辑。我用一个生活化类比路由函数是具体某个窗口的办事员中间件是大厅入口的安检和出口的盖章。你进出大厅不管办什么事都必须经过那道闸机但闸机完全不知道你在几号窗口办了什么事它只管放行和记录。1.2 中间件能横切什么、不能横切什么先想清楚边界才不会在写的时候产生“中间件怎么这也不能干”的挫败感。我列一个简单的对照表基本覆盖日常开发会遇到的情况能力中间件里能不能做常见用途读取和修改请求头能解析token、注入请求ID、CORS预检读取请求体不建议日志需求可以用纯ASGI中间件包装send轻易别读body修改响应头能添加X-Process-Time、安全响应头读取响应状态码能访问日志、监控报警读取和修改响应体能但危险统一响应包装流式响应会被破坏获取当前路由的路径参数拿不到路由参数在路由匹配后才填充中间件在匹配前执行做细粒度用户鉴权不推荐中间件里没有依赖注入查库很别扭给某个具体路由单独生效不能直接做中间件对全路由生效需要白名单逻辑自己控制这里面最容易让人误判的是“我想在中间件里拿到路径参数”。比如请求“/user/123”你想在中间件里知道id123从而判断这个用户能不能操作这个资源。很遗憾中间件执行时路由还没开始匹配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_nameX-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((bx-custom-asgi, b1)) 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_credentialsTrue, 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_code401, detailmissing 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_code429) 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(contentjson.loads(new_body), status_coderesponse.status_code)表面上看能用但你一旦这么写会遇到三个问题。第一如果接口返回的是文件流、视频流、SSEbody_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_code500, content{detail: internal error})要注意这个兜底是最后的手段它捕获不到已经被ExceptionMiddleware处理的HTTPException也不能在响应已经开始发送之后再去改状态码。4.2 在中间件里读body等于亲手毁掉流式响应中间件里可以拿到response.body_iterator但不代表你应该去读它。BaseHTTPMiddleware返回的Response对象body_iterator是异步迭代器一旦你把它完整消费掉后面的流式推送就没了。我实际遇到过一个事故某个导出的Excel接口数据量大概几十万行服务端使用StreamingResponse边查边写。上了统一响应包装中间件之后导出文件变得非常大而且客户端迟迟收不到响应最后超时。排查原因就是中间件把整个流式响应缓冲成了body再重新包装成JSONResponseExcel内容已经变成字节串被塞进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的获取放依赖因为你只有进入具体路由之后才知道查哪个库、用什么事务。维度中间件依赖注入作用范围所有路由按需声明能否访问路径参数不能能能否异步查库自己管理生命周期支持依赖生命周期代码侵入性低中典型场景CORS、日志、限流鉴权、DB会话、业务参数校验所以中间件不是万能的也不是越强大越好。能不用中间件解决的尽量别用保持这个判断标准能让你少写很多麻烦代码。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的ASGITransportimport httpx async def test_asgi_middleware(): transport httpx.ASGITransport(appapp) async with httpx.AsyncClient(transporttransport, base_urlhttp://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 我目前的通用中间件清单最后分享一个比较克制的中间件配置。之前踩过很多坑之后我发现中间件并不是越多越好每多一层洋葱皮请求链路就多一次跳转和潜在风险。我现在内部基础服务一般只保留这几样中间件作用放在哪层CORSMiddleware跨域处理最外层RequestIDMiddleware注入和透传请求ID外层AccessLogMiddleware访问日志与耗时统计外层TokenExtractMiddleware粗粒度token解析内层RateLimitMiddleware单机限流按需挂载真正的用户鉴权、数据库操作、业务校验全部放到依赖注入里中间件保持轻薄、可测试。中间件这东西刚学的时候觉得它是救命稻草想什么功能都往里塞用多了反而觉得它应该尽量少、尽量薄。判断标准很简单如果这个逻辑不是“所有请求都必须经历”的横切动作就把它放下层能不用中间件就不用中间件。这样你的服务在排查问题的时候链路才能一眼看透。