008、中间件详解跨域、日志、认证与自定义中间件开发之前排查线上问题发现某个接口的响应时间比测试环境慢了近200ms。抓包一看每个请求头部都带着一堆调试信息这才想起来测试阶段加了个调试中间件忘记移除。这种“临时方案变永久”的坑搞后端的朋友应该都遇到过。今天咱们就深入聊聊FastAPI中间件——这个既能帮你快速解决问题也可能悄悄埋雷的双刃剑。从那个200ms的坑说起先还原一下当时的错误代码app.middleware(http)asyncdefdebug_middleware(request:Request,call_next):# 这里踩过坑生产环境千万别这样写start_timetime.time()request.state.debug_info{start:start_time}responseawaitcall_next(request)process_timetime.time()-start_time response.headers[X-Process-Time]str(process_time)response.headers[X-Debug-Data]json.dumps(request.state.debug_info)# 更坑的是这里日志打到控制台IO阻塞直接拖慢响应print(fRequest took{process_time:.3f}s)returnresponse问题出在哪第一生产环境暴露调试信息存在安全风险第二同步的print语句在异步上下文中可能阻塞事件循环第三往header里塞大体积JSON数据影响网络传输。三个问题叠加那200ms的延迟就这么来了。跨域中间件别只会copy-pasteCORS配置大概是复制最多的中间件代码了。但很多人只是机械地复制并不理解每个参数的含义fromfastapi.middleware.corsimportCORSMiddleware app.add_middleware(CORSMiddleware,allow_origins[https://frontend.myapp.com,http://localhost:3000,# 注意这里别写正则列表里必须是完整域名],allow_credentialsTrue,# 允许携带cookie时要设为Trueallow_methods[*],# 生产环境建议显式列出方法allow_headers[*],# 同样建议显式列出需要的headerexpose_headers[X-Custom-Header],# 前端能访问的额外headermax_age600,# 预检请求缓存时间单位秒)有个细节容易忽略当allow_credentialsTrue时allow_origins不能包含通配符*必须明确列出每个域名。这是浏览器安全策略的要求不是FastAPI的限制。日志中间件记录什么、怎么记录日志中间件要平衡信息量和性能。我现在的生产方案是这样的importloggingimporttimefromuuidimportuuid4 loggerlogging.getLogger(api)app.middleware(http)asyncdeflogging_middleware(request:Request,call_next):request_idstr(uuid4())[:8]# 生成简短请求IDrequest.state.request_idrequest_id start_timetime.perf_counter()# 用perf_counter更精确# 记录请求开始logger.info(freq_start | id:{request_id}| f{request.method}{request.url.path})try:responseawaitcall_next(request)exceptExceptionasexc:# 异常日志要包含请求ID方便追踪logger.error(freq_error | id:{request_id}| ferror:{str(exc)[:100]},exc_infoTrue)raiseprocess_timetime.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_code500:logger.error(freq_end |{log_data})elifresponse.status_code400:logger.warning(freq_end |{log_data})else:logger.info(freq_end |{log_data})# 响应头里加上请求ID前端报错时可以传回来response.headers[X-Request-ID]request_idreturnresponse关键点日志要结构化方便用ELK或Loki这类工具分析异常日志一定要带exc_infoTrue才能拿到堆栈性能敏感场景考虑异步日志处理器。认证中间件别把所有逻辑都塞进去认证中间件最容易写得臃肿。记住它的核心职责只有一个验证请求是否合法然后把验证结果如用户ID放到合适的地方。fromfastapiimportHTTPException,statusapp.middleware(http)asyncdefauth_middleware(request:Request,call_next):# 1. 提取tokenauth_headerrequest.headers.get(Authorization)ifnotauth_headerornotauth_header.startswith(Bearer ):# 公共接口直接放行ifrequest.url.pathin[/docs,/openapi.json,/health]:returnawaitcall_next(request)raiseHTTPException(status_codestatus.HTTP_401_UNAUTHORIZED)tokenauth_header[7:]# 去掉Bearer # 2. 验证token这里只是示例实际可能查Redis或数据库try:user_idverify_token(token)# 你的验证逻辑exceptTokenExpired:raiseHTTPException(status_codestatus.HTTP_401_UNAUTHORIZED,detailToken expired)exceptInvalidToken:raiseHTTPException(status_codestatus.HTTP_401_UNAUTHORIZED,detailInvalid token)# 3. 把用户信息存到request.state别直接修改request对象request.state.user_iduser_id request.state.tokentoken# 4. 继续处理请求responseawaitcall_next(request)returnresponse然后在路由里通过依赖注入获取用户信息fromfastapiimportDependsdefget_current_user(request:Request):ifnothasattr(request.state,user_id):raiseHTTPException(status_code401)return{user_id:request.state.user_id}app.get(/user/profile)asyncdefget_profile(user:dictDepends(get_current_user)):# 业务逻辑里直接用user信息return{user_id:user[user_id]}这种设计的好处是中间件只做验证业务逻辑通过依赖注入获取验证结果两者解耦。哪天你想把认证方式从Token换成JWT只需要改中间件业务代码完全不用动。自定义中间件开发几个实用模式限流中间件fromcollectionsimportdefaultdictimportasyncioclassRateLimiter:def__init__(self,requests_per_minute:int60):self.requests_per_minuterequests_per_minute self.request_countsdefaultdict(list)asyncdef__call__(self,request:Request,call_next):client_iprequest.client.host nowtime.time()# 清理一分钟前的记录self.request_counts[client_ip][tfortinself.request_counts[client_ip]ifnow-t60]iflen(self.request_counts[client_ip])self.requests_per_minute:returnJSONResponse(status_code429,content{detail:Too many requests})self.request_counts[client_ip].append(now)returnawaitcall_next(request)# 使用方式app.add_middleware(RateLimiter,requests_per_minute30)注意这个简单实现只适合单机部署。分布式环境要用Redis之类的共享存储还要考虑滑动窗口等高级算法。响应压缩中间件importgzipfromtypingimportOptionalapp.middleware(http)asyncdefcompression_middleware(request:Request,call_next):# 只压缩特定类型的响应compressible_types{application/json,text/html,text/plain,application/javascript}responseawaitcall_next(request)accept_encodingrequest.headers.get(Accept-Encoding,)content_typeresponse.headers.get(Content-Type,)if(gzipinaccept_encodingandany(ctincontent_typeforctincompressible_types)andlen(response.body)512):# 小响应不值得压缩compressedgzip.compress(response.body)response.bodycompressed response.headers[Content-Encoding]gzipresponse.headers[Content-Length]str(len(compressed))returnresponse请求耗时告警中间件WARNING_THRESHOLD2.0# 2秒app.middleware(http)asyncdefslow_request_warning(request:Request,call_next):starttime.perf_counter()responseawaitcall_next(request)durationtime.perf_counter()-startifdurationWARNING_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的响应处理所以如果你有多个中间件要考虑清楚顺序。比如认证中间件应该放在最外层最先添加这样后续中间件就能用到认证结果日志中间件也应该靠外这样才能记录完整的处理时间。个人经验与建议中间件要轻量中间件在每个请求都会执行这里面的任何低效代码都会被放大。避免在中间件里做复杂计算、同步IO或大内存操作。异常处理要谨慎中间件里抛出的异常会直接返回给客户端记得给友好的错误信息。但不要尝试在中间件里捕获所有业务异常——那是路由层的职责。状态管理要清晰用request.state存放请求级数据这是FastAPI提供的标准方式。别往request对象上随意添加属性类型提示会失效。考虑使用Starlette中间件FastAPI基于Starlette所有Starlette中间件都能直接用。比如GZipMiddleware、SessionMiddleware没必要自己重复造轮子。测试中间件要单独测写单元测试模拟Request和call_next确保中间件在各种边界条件下行为正确。特别是认证和限流中间件一定要测满各种异常场景。生产环境记得关调试我的那个200ms的坑就是教训。用环境变量控制中间件行为ifos.getenv(ENVIRONMENT)!production:app.add_middleware(DebugMiddleware)最后说个反直觉的观点不是所有功能都应该做成中间件。如果某个逻辑只针对特定路由用依赖注入更合适如果需要在响应后继续执行比如发送统计事件考虑后台任务。中间件应该是横切关注点的解决方案别把它当成万能工具箱。下次写中间件时不妨先问自己这个逻辑真的是每个请求都需要吗放在中间件真的是最优解吗多问这两个问题能帮你避开不少设计上的坑。
008、中间件详解：跨域、日志、认证与自定义中间件开发
008、中间件详解跨域、日志、认证与自定义中间件开发之前排查线上问题发现某个接口的响应时间比测试环境慢了近200ms。抓包一看每个请求头部都带着一堆调试信息这才想起来测试阶段加了个调试中间件忘记移除。这种“临时方案变永久”的坑搞后端的朋友应该都遇到过。今天咱们就深入聊聊FastAPI中间件——这个既能帮你快速解决问题也可能悄悄埋雷的双刃剑。从那个200ms的坑说起先还原一下当时的错误代码app.middleware(http)asyncdefdebug_middleware(request:Request,call_next):# 这里踩过坑生产环境千万别这样写start_timetime.time()request.state.debug_info{start:start_time}responseawaitcall_next(request)process_timetime.time()-start_time response.headers[X-Process-Time]str(process_time)response.headers[X-Debug-Data]json.dumps(request.state.debug_info)# 更坑的是这里日志打到控制台IO阻塞直接拖慢响应print(fRequest took{process_time:.3f}s)returnresponse问题出在哪第一生产环境暴露调试信息存在安全风险第二同步的print语句在异步上下文中可能阻塞事件循环第三往header里塞大体积JSON数据影响网络传输。三个问题叠加那200ms的延迟就这么来了。跨域中间件别只会copy-pasteCORS配置大概是复制最多的中间件代码了。但很多人只是机械地复制并不理解每个参数的含义fromfastapi.middleware.corsimportCORSMiddleware app.add_middleware(CORSMiddleware,allow_origins[https://frontend.myapp.com,http://localhost:3000,# 注意这里别写正则列表里必须是完整域名],allow_credentialsTrue,# 允许携带cookie时要设为Trueallow_methods[*],# 生产环境建议显式列出方法allow_headers[*],# 同样建议显式列出需要的headerexpose_headers[X-Custom-Header],# 前端能访问的额外headermax_age600,# 预检请求缓存时间单位秒)有个细节容易忽略当allow_credentialsTrue时allow_origins不能包含通配符*必须明确列出每个域名。这是浏览器安全策略的要求不是FastAPI的限制。日志中间件记录什么、怎么记录日志中间件要平衡信息量和性能。我现在的生产方案是这样的importloggingimporttimefromuuidimportuuid4 loggerlogging.getLogger(api)app.middleware(http)asyncdeflogging_middleware(request:Request,call_next):request_idstr(uuid4())[:8]# 生成简短请求IDrequest.state.request_idrequest_id start_timetime.perf_counter()# 用perf_counter更精确# 记录请求开始logger.info(freq_start | id:{request_id}| f{request.method}{request.url.path})try:responseawaitcall_next(request)exceptExceptionasexc:# 异常日志要包含请求ID方便追踪logger.error(freq_error | id:{request_id}| ferror:{str(exc)[:100]},exc_infoTrue)raiseprocess_timetime.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_code500:logger.error(freq_end |{log_data})elifresponse.status_code400:logger.warning(freq_end |{log_data})else:logger.info(freq_end |{log_data})# 响应头里加上请求ID前端报错时可以传回来response.headers[X-Request-ID]request_idreturnresponse关键点日志要结构化方便用ELK或Loki这类工具分析异常日志一定要带exc_infoTrue才能拿到堆栈性能敏感场景考虑异步日志处理器。认证中间件别把所有逻辑都塞进去认证中间件最容易写得臃肿。记住它的核心职责只有一个验证请求是否合法然后把验证结果如用户ID放到合适的地方。fromfastapiimportHTTPException,statusapp.middleware(http)asyncdefauth_middleware(request:Request,call_next):# 1. 提取tokenauth_headerrequest.headers.get(Authorization)ifnotauth_headerornotauth_header.startswith(Bearer ):# 公共接口直接放行ifrequest.url.pathin[/docs,/openapi.json,/health]:returnawaitcall_next(request)raiseHTTPException(status_codestatus.HTTP_401_UNAUTHORIZED)tokenauth_header[7:]# 去掉Bearer # 2. 验证token这里只是示例实际可能查Redis或数据库try:user_idverify_token(token)# 你的验证逻辑exceptTokenExpired:raiseHTTPException(status_codestatus.HTTP_401_UNAUTHORIZED,detailToken expired)exceptInvalidToken:raiseHTTPException(status_codestatus.HTTP_401_UNAUTHORIZED,detailInvalid token)# 3. 把用户信息存到request.state别直接修改request对象request.state.user_iduser_id request.state.tokentoken# 4. 继续处理请求responseawaitcall_next(request)returnresponse然后在路由里通过依赖注入获取用户信息fromfastapiimportDependsdefget_current_user(request:Request):ifnothasattr(request.state,user_id):raiseHTTPException(status_code401)return{user_id:request.state.user_id}app.get(/user/profile)asyncdefget_profile(user:dictDepends(get_current_user)):# 业务逻辑里直接用user信息return{user_id:user[user_id]}这种设计的好处是中间件只做验证业务逻辑通过依赖注入获取验证结果两者解耦。哪天你想把认证方式从Token换成JWT只需要改中间件业务代码完全不用动。自定义中间件开发几个实用模式限流中间件fromcollectionsimportdefaultdictimportasyncioclassRateLimiter:def__init__(self,requests_per_minute:int60):self.requests_per_minuterequests_per_minute self.request_countsdefaultdict(list)asyncdef__call__(self,request:Request,call_next):client_iprequest.client.host nowtime.time()# 清理一分钟前的记录self.request_counts[client_ip][tfortinself.request_counts[client_ip]ifnow-t60]iflen(self.request_counts[client_ip])self.requests_per_minute:returnJSONResponse(status_code429,content{detail:Too many requests})self.request_counts[client_ip].append(now)returnawaitcall_next(request)# 使用方式app.add_middleware(RateLimiter,requests_per_minute30)注意这个简单实现只适合单机部署。分布式环境要用Redis之类的共享存储还要考虑滑动窗口等高级算法。响应压缩中间件importgzipfromtypingimportOptionalapp.middleware(http)asyncdefcompression_middleware(request:Request,call_next):# 只压缩特定类型的响应compressible_types{application/json,text/html,text/plain,application/javascript}responseawaitcall_next(request)accept_encodingrequest.headers.get(Accept-Encoding,)content_typeresponse.headers.get(Content-Type,)if(gzipinaccept_encodingandany(ctincontent_typeforctincompressible_types)andlen(response.body)512):# 小响应不值得压缩compressedgzip.compress(response.body)response.bodycompressed response.headers[Content-Encoding]gzipresponse.headers[Content-Length]str(len(compressed))returnresponse请求耗时告警中间件WARNING_THRESHOLD2.0# 2秒app.middleware(http)asyncdefslow_request_warning(request:Request,call_next):starttime.perf_counter()responseawaitcall_next(request)durationtime.perf_counter()-startifdurationWARNING_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的响应处理所以如果你有多个中间件要考虑清楚顺序。比如认证中间件应该放在最外层最先添加这样后续中间件就能用到认证结果日志中间件也应该靠外这样才能记录完整的处理时间。个人经验与建议中间件要轻量中间件在每个请求都会执行这里面的任何低效代码都会被放大。避免在中间件里做复杂计算、同步IO或大内存操作。异常处理要谨慎中间件里抛出的异常会直接返回给客户端记得给友好的错误信息。但不要尝试在中间件里捕获所有业务异常——那是路由层的职责。状态管理要清晰用request.state存放请求级数据这是FastAPI提供的标准方式。别往request对象上随意添加属性类型提示会失效。考虑使用Starlette中间件FastAPI基于Starlette所有Starlette中间件都能直接用。比如GZipMiddleware、SessionMiddleware没必要自己重复造轮子。测试中间件要单独测写单元测试模拟Request和call_next确保中间件在各种边界条件下行为正确。特别是认证和限流中间件一定要测满各种异常场景。生产环境记得关调试我的那个200ms的坑就是教训。用环境变量控制中间件行为ifos.getenv(ENVIRONMENT)!production:app.add_middleware(DebugMiddleware)最后说个反直觉的观点不是所有功能都应该做成中间件。如果某个逻辑只针对特定路由用依赖注入更合适如果需要在响应后继续执行比如发送统计事件考虑后台任务。中间件应该是横切关注点的解决方案别把它当成万能工具箱。下次写中间件时不妨先问自己这个逻辑真的是每个请求都需要吗放在中间件真的是最优解吗多问这两个问题能帮你避开不少设计上的坑。
