如何为FastAPI后端启用维护模式?
FastAPI 实现全局维护模式的最佳方案
下面是几种实用的实现方式,按推荐程度排序:
1. 全局HTTP中间件(最推荐)
通过中间件拦截所有进入的请求,直接返回维护提示,这是最简洁高效的方式,适合全局维护场景。
代码示例:
from fastapi import FastAPI, Request from fastapi.responses import JSONResponse import os app = FastAPI() # 从环境变量控制维护状态,生产环境可通过运维工具快速切换 MAINTENANCE_ENABLED = os.getenv("MAINTENANCE_ENABLED", "false").lower() == "true" MAINTENANCE_RECOVER_TIME = os.getenv("MAINTENANCE_RECOVER_TIME", "1小时后") @app.middleware("http") async def maintenance_interceptor(request: Request, call_next): if MAINTENANCE_ENABLED: return JSONResponse( status_code=503, # 符合HTTP规范的"服务不可用"状态码 content={ "code": 503, "msg": "服务器正在进行例行维护,暂时无法提供服务", "recover_time": MAINTENANCE_RECOVER_TIME } ) # 非维护模式,继续处理请求 response = await call_next(request) return response # 正常业务路由示例 @app.get("/user/profile") async def get_user_profile(): return {"id": 1, "name": "test"}
优势:
- 全局生效,无需修改现有路由代码
- 性能开销极小,仅做一次状态判断
- 配合环境变量,运维无需改动代码即可开启/关闭维护
2. 全局依赖项(灵活度高)
通过FastAPI的依赖系统,给所有路由添加维护检查逻辑,适合需要排除部分路由(如健康检查)的场景。
代码示例:
from fastapi import FastAPI, Depends, HTTPException import os app = FastAPI() MAINTENANCE_ENABLED = os.getenv("MAINTENANCE_ENABLED", "false").lower() == "true" MAINTENANCE_RECOVER_TIME = os.getenv("MAINTENANCE_RECOVER_TIME", "1小时后") # 定义维护检查依赖 async def check_maintenance_status(): if MAINTENANCE_ENABLED: raise HTTPException( status_code=503, detail=f"服务器维护中,预计{MAINTENANCE_RECOVER_TIME}恢复服务" ) # 给所有路由全局挂载该依赖 app.router.dependencies.append(Depends(check_maintenance_status)) # 正常业务路由 @app.get("/order/list") async def get_order_list(): return {"orders": []} # 排除维护检查的路由(如健康检查) @app.get("/health", dependencies=[]) async def health_check(): return {"status": "running"}
优势:
- 可灵活排除特定路由,不影响运维监控
- 符合FastAPI的依赖设计规范,扩展性强
3. 临时替换路由(应急场景)
如果是临时紧急维护,可直接替换整个路由列表,返回统一响应。这种方式适合快速上线,但灵活性较差。
代码示例:
from fastapi import FastAPI, Request from fastapi.responses import JSONResponse app = FastAPI() # 保存原路由(维护结束后恢复) original_routes = app.router.routes.copy() def enable_maintenance_mode(): # 清空现有路由,添加全局维护路由 app.router.routes.clear() @app.get("/{full_path:path}") @app.post("/{full_path:path}") @app.put("/{full_path:path}") @app.delete("/{full_path:path}") async def maintenance_response(request: Request): return JSONResponse( status_code=503, content={"msg": "服务器维护中,1小时后恢复"} ) def disable_maintenance_mode(): # 恢复原路由 app.router.routes = original_routes.copy() # 示例:启动时根据环境变量判断是否开启维护 import os if os.getenv("MAINTENANCE_ENABLED") == "true": enable_maintenance_mode()
注意事项:
- 建议配合环境变量使用,避免硬编码开关
- 维护结束后需恢复原路由,否则业务无法正常运行
通用注意点
- 必须使用503状态码,符合HTTP协议规范,客户端可识别为临时不可用
- 返回内容需明确告知用户维护原因和预计恢复时间,提升用户体验
- 生产环境建议结合配置中心或容器编排工具,实现无重启切换维护状态
内容的提问来源于stack exchange,提问作者Siqueler
相关产品推荐
相关产品推荐

