008、中间件详解：跨域、日志、认证与自定义中间件开发

008、中间件详解：跨域、日志、认证与自定义中间件开发

之前排查线上问题，发现某个接口的响应时间比测试环境慢了近200ms。抓包一看，每个请求头部都带着一堆调试信息，这才想起来测试阶段加了个调试中间件忘记移除。这种“临时方案变永久”的坑，搞后端的朋友应该都遇到过。今天咱们就深入聊聊FastAPI中间件——这个既能帮你快速解决问题，也可能悄悄埋雷的双刃剑。

从那个200ms的坑说起

先还原一下当时的错误代码：

@app.middleware("http")asyncdefdebug_middleware(request:Request,call_next):# 这里踩过坑：生产环境千万别这样写！start_time=time.time()request.state.debug_info={"start":start_time}response=awaitcall_next(request)process_time=time.time()-start_time response.headers["X-Process-Time"]=str(process_time)response.headers["X-Debug-Data"]=json.dumps(request.state.debug_info)# 更坑的是这里：日志打到控制台，IO阻塞直接拖慢响应print(f"Request took{process_time:.3f}s")returnresponse

问题出在哪？第一，生产环境暴露调试信息存在安全风险；第二，同步的print语句在异步上下文中可能阻塞事件循环；第三，往header里塞大体积JSON数据影响网络传输。三个问题叠加，那200ms的延迟就这么来了。

跨域中间件：别只会copy-paste

CORS配置大概是复制最多的中间件代码了。但很多人只是机械地复制，并不理解每个参数的含义：

fromfastapi.middleware.corsimportCORSMiddleware app.add_middleware(CORSMiddleware,allow_origins=["https://frontend.myapp.com","http://localhost:3000",# 注意：这里别写正则，列表里必须是完整域名],allow_credentials=True,# 允许携带cookie时要设为Trueallow_methods=["*"],# 生产环境建议显式列出方法allow_headers=["*"],# 同样建议显式列出需要的headerexpose_headers=["X-Custom-Header"],# 前端能访问的额外headermax_age=600,# 预检请求缓存时间，单位秒)

有个细节容易忽略：当allow_credentials=True时，allow_origins不能包含通配符"*"，必须明确列出每个域名。这是浏览器安全策略的要求，不是FastAPI的限制。

日志中间件：记录什么、怎么记录

日志中间件要平衡信息量和性能。我现在的生产方案是这样的：

importloggingimporttimefromuuidimportuuid4 logger=logging.getLogger("api")@app.middleware("http")asyncdeflogging_middleware(request:Request,call_next):request_id=str(uuid4())[:8]# 生成简短请求IDrequest.state.request_id=request_id start_time=time.perf_counter()# 用perf_counter更精确# 记录请求开始logger.info(f"req_start | id:{request_id}| "f"{request.method}{request.url.path}")try:response=awaitcall_next(request)exceptExceptionasexc:# 异常日志要包含请求ID，方便追踪logger.error(f"req_error | id:{request_id}| "f"error:{str(exc)[:100]}",exc_info=True)raiseprocess_time=time.perf_counter()-start_time# 结构化日志，方便后续分析log_data={"request_id":request_id,"method":request.method,"path":request.url.path,"status":response.status_code,"duration":round(process_time,4),"client":request.client.hostifrequest.clientelseNone,}# 根据响应状态选择日志级别ifresponse.status_code>=500:logger.error(f"req_end |{log_data}")elifresponse.status_code>=400:logger.warning(f"req_end |{log_data}")else:logger.info(f"req_end |{log_data}")# 响应头里加上请求ID，前端报错时可以传回来response.headers["X-Request-ID"]=request_idreturnresponse

关键点：日志要结构化，方便用ELK或Loki这类工具分析；异常日志一定要带exc_info=True才能拿到堆栈；性能敏感场景考虑异步日志处理器。

认证中间件：别把所有逻辑都塞进去

认证中间件最容易写得臃肿。记住它的核心职责只有一个：验证请求是否合法，然后把验证结果（如用户ID）放到合适的地方。

fromfastapiimportHTTPException,status@app.middleware("http")asyncdefauth_middleware(request:Request,call_next):# 1. 提取tokenauth_header=request.headers.get("Authorization")ifnotauth_headerornotauth_header.startswith("Bearer "):# 公共接口直接放行ifrequest.url.pathin["/docs","/openapi.json","/health"]:returnawaitcall_next(request)raiseHTTPException(status_code=status.HTTP_401_UNAUTHORIZED)token=auth_header[7:]# 去掉"Bearer "# 2. 验证token（这里只是示例，实际可能查Redis或数据库）try:user_id=verify_token(token)# 你的验证逻辑exceptTokenExpired:raiseHTTPException(status_code=status.HTTP_401_UNAUTHORIZED,detail="Token expired")exceptInvalidToken:raiseHTTPException(status_code=status.HTTP_401_UNAUTHORIZED,detail="Invalid token")# 3. 把用户信息存到request.state，别直接修改request对象request.state.user_id=user_id request.state.token=token# 4. 继续处理请求response=awaitcall_next(request)returnresponse

然后在路由里通过依赖注入获取用户信息：

fromfastapiimportDependsdefget_current_user(request:Request):ifnothasattr(request.state,"user_id"):raiseHTTPException(status_code=401)return{"user_id":request.state.user_id}@app.get("/user/profile")asyncdefget_profile(user:dict=Depends(get_current_user)):# 业务逻辑里直接用user信息return{"user_id":user["user_id"]}

这种设计的好处是：中间件只做验证，业务逻辑通过依赖注入获取验证结果，两者解耦。哪天你想把认证方式从Token换成JWT，只需要改中间件，业务代码完全不用动。

自定义中间件开发：几个实用模式

限流中间件

fromcollectionsimportdefaultdictimportasyncioclassRateLimiter:def__init__(self,requests_per_minute:int=60):self.requests_per_minute=requests_per_minute self.request_counts=defaultdict(list)asyncdef__call__(self,request:Request,call_next):client_ip=request.client.host now=time.time()# 清理一分钟前的记录self.request_counts[client_ip]=[tfortinself.request_counts[client_ip]ifnow-t<60]iflen(self.request_counts[client_ip])>=self.requests_per_minute:returnJSONResponse(status_code=429,content={"detail":"Too many requests"})self.request_counts[client_ip].append(now)returnawaitcall_next(request)# 使用方式app.add_middleware(RateLimiter,requests_per_minute=30)

注意：这个简单实现只适合单机部署。分布式环境要用Redis之类的共享存储，还要考虑滑动窗口等高级算法。

响应压缩中间件

importgzipfromtypingimportOptional@app.middleware("http")asyncdefcompression_middleware(request:Request,call_next):# 只压缩特定类型的响应compressible_types={"application/json","text/html","text/plain","application/javascript"}response=awaitcall_next(request)accept_encoding=request.headers.get("Accept-Encoding","")content_type=response.headers.get("Content-Type","")if("gzip"inaccept_encodingandany(ctincontent_typeforctincompressible_types)andlen(response.body)>512):# 小响应不值得压缩compressed=gzip.compress(response.body)response.body=compressed response.headers["Content-Encoding"]="gzip"response.headers["Content-Length"]=str(len(compressed))returnresponse

请求耗时告警中间件

WARNING_THRESHOLD=2.0# 2秒@app.middleware("http")asyncdefslow_request_warning(request:Request,call_next):start=time.perf_counter()response=awaitcall_next(request)duration=time.perf_counter()-startifduration>WARNING_THRESHOLD:# 发送到监控系统，不要直接打印awaitsend_to_monitoring({"type":"slow_request","path":request.url.path,"method":request.method,"duration":duration,"timestamp":time.time()})returnresponse

中间件的执行顺序问题

这是很多人困惑的地方。中间件的执行顺序取决于添加顺序，而且是“洋葱模型”：

# 先添加的先执行外层（请求阶段）# 后添加的先执行内层（响应阶段）app.add_middleware(MiddlewareA)# A请求 -> B请求 -> 路由 -> B响应 -> A响应app.add_middleware(MiddlewareB)# 实际执行顺序：# 1. A的请求处理# 2. B的请求处理# 3. 路由处理函数# 4. B的响应处理# 5. A的响应处理

所以如果你有多个中间件，要考虑清楚顺序。比如认证中间件应该放在最外层（最先添加），这样后续中间件就能用到认证结果；日志中间件也应该靠外，这样才能记录完整的处理时间。

个人经验与建议

1. 中间件要轻量：中间件在每个请求都会执行，这里面的任何低效代码都会被放大。避免在中间件里做复杂计算、同步IO或大内存操作。

2. 异常处理要谨慎：中间件里抛出的异常会直接返回给客户端，记得给友好的错误信息。但不要尝试在中间件里捕获所有业务异常——那是路由层的职责。

3. 状态管理要清晰：用request.state存放请求级数据，这是FastAPI提供的标准方式。别往request对象上随意添加属性，类型提示会失效。

4. 考虑使用Starlette中间件：FastAPI基于Starlette，所有Starlette中间件都能直接用。比如GZipMiddleware、SessionMiddleware，没必要自己重复造轮子。

5. 测试中间件要单独测：写单元测试模拟Request和call_next，确保中间件在各种边界条件下行为正确。特别是认证和限流中间件，一定要测满各种异常场景。

6. 生产环境记得关调试：我的那个200ms的坑就是教训。用环境变量控制中间件行为：

ifos.getenv("ENVIRONMENT")!="production":app.add_middleware(DebugMiddleware)

最后说个反直觉的观点：不是所有功能都应该做成中间件。如果某个逻辑只针对特定路由，用依赖注入更合适；如果需要在响应后继续执行（比如发送统计事件），考虑后台任务。中间件应该是横切关注点的解决方案，别把它当成万能工具箱。

下次写中间件时，不妨先问自己：这个逻辑真的是每个请求都需要吗？放在中间件真的是最优解吗？多问这两个问题，能帮你避开不少设计上的坑。