Option 1
如果你不介意拥有Header显示为Optional in OpenAPI/Swagger UI 自动文档 https://fastapi.tiangolo.com/tutorial/first-steps/#interactive-api-docs，这将很简单，如下所示：
from fastapi import Header, HTTPException
@app.post("/")
def some_route(some_custom_header: Optional[str] = Header(None)):
if not some_custom_header:
raise HTTPException(status_code=401, detail="Unauthorized")
return {"some-custom-header": some_custom_header}
Option 2
但是，既然您想要Header出现为required在 OpenAPI 中，您应该覆盖默认的异常处理程序。当请求包含无效数据时，FastAPI 会在内部引发RequestValidationError https://fastapi.tiangolo.com/tutorial/handling-errors/#override-the-default-exception-handlers。因此，您需要重写RequestValidationError， 哪个包含收到的带有无效数据的正文 https://fastapi.tiangolo.com/tutorial/handling-errors/#use-the-requestvalidationerror-body.
Since RequestValidationError是 Pydantic 的子类ValidationError https://pydantic-docs.helpmanual.io/usage/models/#error-handling，您可以访问上面链接中显示的错误，以便检查您的自定义是否正确Header包含在错误中（如果是这样，则意味着请求中缺少该内容，或者不属于该错误）str类型），因此返回您的自定义错误响应。如果您的定制Header (i.e., some_custom_header在下面的示例中）是该特定端点中的唯一参数，那么它是not执行上述检查（并在下面演示）所必需的，就好像RequestValidationError被提出，它只会针对该参数。
Example
from fastapi import FastAPI, Request, Header, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from fastapi.encoders import jsonable_encoder
app = FastAPI()
routes_with_custom_exception = ['/']
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
if request.url.path in routes_with_custom_exception:
# check whether the error relates to the `some_custom_header` parameter
for err in exc.errors():
if err['loc'][0] == 'header' and err['loc'][1] == 'some-custom-header':
return JSONResponse(content={'401': 'Unauthorized'}, status_code=401)
return JSONResponse(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
content=jsonable_encoder({'detail': exc.errors(), 'body': exc.body}),
)
@app.get('/')
def some_route(some_custom_header: str = Header(...)):
return {'some-custom-header': some_custom_header}
Option 3
另一种解决方案是使用子应用程序 https://fastapi.tiangolo.com/advanced/sub-applications/（受到讨论的启发here https://github.com/tiangolo/fastapi/issues/1174）。您可以创建一个子应用程序（或更多，如果需要）并将其安装到主应用程序 - 其中将包括需要自定义的路由Header;因此，重写exception_handler for RequestValidationError在该子应用程序中仅适用于这些路线，而无需检查request.url.path，如之前的解决方案所示，并像往常一样让主应用程序具有剩余的路线。根据文档 https://fastapi.tiangolo.com/advanced/sub-applications/#mounting-a-fastapi-application:
安装 FastAPI 应用程序
“挂载”是指在一个应用程序中添加一个完全“独立”的应用程序。
特定路径，然后负责处理该路径下的所有内容
路径，与路径操作在该子应用程序中声明。
Example
Note:如果您安装了子应用程序（即subapi在下面的示例中）使用'/'路径，您将无法访问subapi at http://127.0.0.1:8000/docs http://127.0.0.1:8000/docs，因为该页面上的 API 文档仅包含主应用程序的路由。此外，它还会干扰'/'主 API 的路由（如果主 API 中存在这样的路由），并且因为FastAPI 中端点的顺序很重要 https://stackoverflow.com/a/74498663/17865804，发出请求http://127.0.0.1:8000/实际上会调用主API的相应路由（如下所示）。因此，您宁愿安装subapi使用不同的路径，例如'/sub'，如下所示，并访问子 API 文档：http://127.0.0.1:8000/sub/docs http://127.0.0.1:8000/sub/docs。下面还给出了一个 Python 请求示例，演示了如何测试应用程序。
from fastapi import FastAPI, Request, Header
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
app = FastAPI()
@app.get('/')
async def main():
return {'message': 'Hello from main API'}
subapi = FastAPI()
@subapi.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
# if there are other parameters defined in the endpoint other than
# `some_custom_header`, then perform a check, as demonstrated in Option 2
return JSONResponse(content={'401': 'Unauthorized'}, status_code=401)
@subapi.get('/')
async def sub_api_route(some_custom_header: str = Header(...)):
return {'some-custom-header': some_custom_header}
app.mount('/sub', subapi)
测试上面的例子
import requests
# Test main API
url = 'http://127.0.0.1:8000/'
r = requests.get(url=url)
print(r.status_code, r.json())
# Test sub API
url = 'http://127.0.0.1:8000/sub/'
r = requests.get(url=url)
print(r.status_code, r.json())
headers = {'some-custom-header': 'this is some custom header'}
r = requests.get(url=url, headers=headers)
print(r.status_code, r.json())
Option 4
进一步的解决方案是使用APIRouter https://fastapi.tiangolo.com/tutorial/bigger-applications/#apirouter with a custom APIRoute class https://fastapi.tiangolo.com/advanced/custom-request-and-route/#custom-apiroute-class-in-a-router，如选项 2 所示这个答案 https://stackoverflow.com/a/73464007/17865804，并在 a 内处理请求try-except块（将用于捕获RequestValidationError例外情况），如中所述FastAPI 的文档 https://fastapi.tiangolo.com/advanced/custom-request-and-route/#accessing-the-request-body-in-an-exception-handler。如果出现异常，您可以根据需要处理错误，并返回自定义响应。
Example
from fastapi import FastAPI, APIRouter, Response, Request, Header, HTTPException
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from fastapi.routing import APIRoute
from typing import Callable
class ValidationErrorHandlingRoute(APIRoute):
def get_route_handler(self) -> Callable:
original_route_handler = super().get_route_handler()
async def custom_route_handler(request: Request) -> Response:
try:
return await original_route_handler(request)
except RequestValidationError as e:
# if there are other parameters defined in the endpoint other than
# `some_custom_header`, then perform a check, as demonstrated in Option 2
raise HTTPException(status_code=401, detail='401 Unauthorized')
return custom_route_handler
app = FastAPI()
router = APIRouter(route_class=ValidationErrorHandlingRoute)
@app.get('/')
async def main():
return {'message': 'Hello from main API'}
@router.get('/custom')
async def custom_route(some_custom_header: str = Header(...)):
return {'some-custom-header': some_custom_header}
app.include_router(router)
测试上面的例子
import requests
# Test main API
url = 'http://127.0.0.1:8000/'
r = requests.get(url=url)
print(r.status_code, r.json())
# Test custom route
url = 'http://127.0.0.1:8000/custom'
r = requests.get(url=url)
print(r.status_code, r.json())
headers = {'some-custom-header': 'this is some custom header'}
r = requests.get(url=url, headers=headers)
print(r.status_code, r.json())
