如何在FastAPI的GET /v1/health路由为security.HTTPBearer[]添加作用域?
FastAPI中为HTTPBearer路由添加作用域的解决方案
核心思路
HTTPBearer本身仅负责Bearer令牌的提取,不直接支持作用域配置,但可以通过FastAPI的内置工具或手动配置实现作用域验证与OpenAPI文档的同步更新。以下是两种可行方案:
方案1:结合SecurityScopes实现自动验证与文档生成
通过SecurityScopes和自定义依赖,既可以完成作用域的合法性校验,又能让OpenAPI自动生成包含作用域的文档。
from fastapi import FastAPI, Depends, HTTPException, status, Security from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from fastapi.security.scopes import SecurityScopes from pydantic import BaseModel from typing import Annotated class HealthCheck(BaseModel): status: str = "ok" v1_router = FastAPI().router # 初始化HTTPBearer实例 security = HTTPBearer() def verify_required_scopes( security_scopes: SecurityScopes, credentials: HTTPAuthorizationCredentials = Depends(security) ): # 此处替换为你的实际令牌解析逻辑,示例假设令牌内容为逗号分隔的作用域字符串 token_scopes = credentials.credentials.split(",") # 校验所需作用域是否全部存在 for required_scope in security_scopes.scopes: if required_scope not in token_scopes: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="权限不足", headers={"WWW-Authenticate": f'Bearer scope="{security_scopes.scope_str}"'}, ) @v1_router.get( "/health", tags=["healthcheck"], summary="Perform a Health Check", response_description="Return HTTP Status Code 200 (OK)", status_code=status.HTTP_200_OK, response_model=HealthCheck, # 直接在路由参数中指定作用域,自动同步到OpenAPI文档 security=[{"HTTPBearer": ["read", "me"]}] ) def get_health( credentials: Annotated[HTTPAuthorizationCredentials, Security(verify_required_scopes, scopes=["read", "me"])] ) -> HealthCheck: return HealthCheck()
方案2:手动配置OpenAPI+自定义作用域校验
如果不需要自动文档生成逻辑,也可以手动修改OpenAPI Schema,同时在路由内完成作用域校验。
from fastapi import FastAPI, Depends, HTTPException, status from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from pydantic import BaseModel from typing import Annotated class HealthCheck(BaseModel): status: str = "ok" v1_router = FastAPI().router security = HTTPBearer() @v1_router.get( "/health", tags=["healthcheck"], summary="Perform a Health Check", response_description="Return HTTP Status Code 200 (OK)", status_code=status.HTTP_200_OK, response_model=HealthCheck, ) def get_health(credentials: Annotated[HTTPAuthorizationCredentials, Depends(security)]) -> HealthCheck: # 自定义作用域校验逻辑 required_scopes = {"read", "me"} # 替换为你的令牌解析逻辑,示例仅做演示 token_scopes = set(credentials.credentials.split(",")) if not required_scopes.issubset(token_scopes): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="缺少必要的权限作用域" ) return HealthCheck() # 手动修改OpenAPI Schema,添加作用域配置 def custom_openapi(): if v1_router.openapi_schema: return v1_router.openapi_schema openapi_schema = v1_router.openapi() # 定位到目标路由并更新security字段 openapi_schema["paths"]["/v1/health"]["get"]["security"] = [{"HTTPBearer": ["read", "me"]}] v1_router.openapi_schema = openapi_schema return openapi_schema v1_router.openapi = custom_openapi
修改后的OpenAPI JSON效果
两种方案最终都会生成符合需求的OpenAPI配置:
"/v1/health": { "get": { "tags": [ "healthcheck" ], "summary": "Perform a Health Check", "operationId": "get_health_v1_health_get", "responses": { "200": { "description": "Return HTTP Status Code 200 (OK)", "content": { "application/json": { "schema": { "$ref": "#/components/schemas/HealthCheck" } } } } }, "security": [ { "HTTPBearer": ["read", "me"] } ] } }
内容的提问来源于stack exchange,提问作者Shuzheng
相关产品推荐
相关产品推荐

