如何在FastAPI的类中间件中实现Bearer认证并集成到OpenAPI文档
如何在FastAPI的类中间件中实现Bearer认证并集成到OpenAPI文档
我明白你想要的效果:用类中间件统一处理大部分路由的Bearer认证,同时让OpenAPI文档(也就是/docs页面)能识别这个认证要求,支持在页面里输入token测试接口。你的现有中间件已经能完成认证逻辑,但缺少和OpenAPI的集成——这是因为FastAPI的文档生成系统是和它的安全依赖体系绑定的,而不是直接和中间件挂钩。
下面是一个完整的解决方案,既保留类中间件的统一认证逻辑,又能让OpenAPI文档正确显示认证选项:
第一步:在单独的auth.py文件中实现认证逻辑和安全方案
我们会把验证逻辑抽出来复用,同时创建一个FastAPI能识别的安全方案类,用于更新OpenAPI文档:
from fastapi import HTTPException, Request from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from starlette.responses import JSONResponse # 抽离验证逻辑,供中间件和安全方案复用 def validate_bearer_token(token: str, settings) -> bool: return token == settings.auth_token # 定义FastAPI能识别的安全方案,用于生成OpenAPI文档 class AuthBearer(HTTPBearer): def __init__(self, settings, auto_error: bool = True): super().__init__(auto_error=auto_error) self.settings = settings async def __call__(self, request: Request): credentials: HTTPAuthorizationCredentials = await super().__call__(request) if not credentials: raise HTTPException(status_code=401, detail="未提供认证信息") if credentials.scheme != "Bearer": raise HTTPException(status_code=401, detail="无效的认证方案,请使用Bearer") if not validate_bearer_token(credentials.credentials, self.settings): raise HTTPException(status_code=401, detail="无效的token") return credentials.credentials # 你的类中间件,负责实际的认证拦截 class AuthMiddleware: def __init__(self, settings): self.settings = settings async def __call__(self, request: Request, call_next): # 判断当前路径是否需要保护 if self.is_protected_url(request.method, request.url.path): auth_header = request.headers.get("Authorization") if not auth_header: return JSONResponse( content={"detail": "未提供认证信息"}, status_code=401, headers={"WWW-Authenticate": "Bearer"} ) try: scheme, token = auth_header.split(" ", 1) except ValueError: return JSONResponse( content={"detail": "无效的认证格式"}, status_code=401, headers={"WWW-Authenticate": "Bearer"} ) if scheme != "Bearer": return JSONResponse( content={"detail": "无效的认证方案,请使用Bearer"}, status_code=401, headers={"WWW-Authenticate": "Bearer"} ) if not validate_bearer_token(token, self.settings): return JSONResponse( content={"detail": "无效的token"}, status_code=401, headers={"WWW-Authenticate": "Bearer"} ) # 认证通过,继续处理请求 response = await call_next(request) return response def is_protected_url(self, method, url): # 这里实现你的路径保护逻辑,比如排除文档和公开接口 public_paths = ["/docs", "/openapi.json", "/public"] return url not in public_paths
第二步:在主文件中集成中间件和OpenAPI支持
这里有两种方式让OpenAPI文档识别认证要求,你可以根据自己的路由结构选择:
方式一:用路由组(APIRouter)统一添加安全依赖
如果你的受保护路由都可以归到一个或多个路由组里,这种方式最简洁:
from fastapi import FastAPI, APIRouter, Depends from auth import AuthMiddleware, AuthBearer from your_settings_module import get_settings # 导入你的配置 # 初始化配置和认证组件 settings = get_settings() auth_middleware = AuthMiddleware(settings) auth_scheme = AuthBearer(settings) app = FastAPI(title="带Bearer认证的API") # 注册中间件 app.middleware("http")(auth_middleware) # 创建受保护的路由组,添加安全依赖(用于OpenAPI文档) protected_router = APIRouter(dependencies=[Depends(auth_scheme)]) # 受保护的接口示例 @protected_router.get("/protected/data") def get_protected_data(): return {"message": "这是受保护的数据"} # 公开接口示例 @app.get("/public/info") def get_public_info(): return {"message": "这是公开信息"} # 将受保护路由组添加到主应用 app.include_router(protected_router)
这样,protected_router下的所有接口都会在OpenAPI文档中显示Bearer认证选项,而中间件会实际拦截并验证token。
方式二:自定义OpenAPI schema,批量添加认证要求
如果你的受保护路由比较分散,不想用路由组,可以直接修改OpenAPI的schema,给符合条件的路径添加认证要求:
from fastapi import FastAPI from fastapi.openapi.utils import get_openapi from auth import AuthMiddleware from your_settings_module import get_settings settings = get_settings() auth_middleware = AuthMiddleware(settings) app = FastAPI(title="带Bearer认证的API") app.middleware("http")(auth_middleware) # 自定义OpenAPI schema def custom_openapi(): if app.openapi_schema: return app.openapi_schema # 生成基础的OpenAPI schema openapi_schema = get_openapi( title=app.title, version=app.version, description=app.description, routes=app.routes, ) # 添加Bearer认证方案到schema openapi_schema["components"]["securitySchemes"] = { "BearerAuth": { "type": "http", "scheme": "bearer", "bearerFormat": "Token" # 这里可以换成你实际的token格式,比如JWT } } # 给所有受保护的路径添加认证要求 auth_middleware_instance = auth_middleware for path in openapi_schema["paths"]: # 复用中间件的is_protected_url判断逻辑 if auth_middleware_instance.is_protected_url("GET", path): # 这里可以根据需要调整method for method in openapi_schema["paths"][path]: openapi_schema["paths"][path][method]["security"] = [{"BearerAuth": []}] app.openapi_schema = openapi_schema return app.openapi_schema # 替换默认的openapi生成函数 app.openapi = custom_openapi # 示例接口 @app.get("/protected/data") def get_protected_data(): return {"message": "这是受保护的数据"} @app.get("/public/info") def get_public_info(): return {"message": "这是公开信息"}
这种方式会自动给所有is_protected_url返回True的路径添加认证要求,不需要给每个路由或路由组手动加依赖。
为什么要这样做?
- 中间件负责全局拦截和验证,避免在每个路由里重复写认证逻辑;
- 安全方案(AuthBearer)负责告诉FastAPI生成对应的OpenAPI文档,让/docs页面支持token输入;
- 抽离的验证逻辑保证了中间件和安全方案的验证规则一致,避免出现矛盾。
现在你启动应用后,访问/docs页面就能看到"Authorize"按钮,输入Bearer token后就可以测试受保护的接口了,同时中间件会在后台统一处理认证逻辑。
备注:内容来源于stack exchange,提问作者dmcblue
相关产品推荐
相关产品推荐

