如何在 FastAPI 中自定义特定路由的错误响应？

问题描述：

我想HTTP在 FastAPI 中创建一个端点，该端点需要特定的Header，response在Header不存在时生成自定义代码，并按照FastAPI 生成的 OpenAPI 文档中的要求Header显示。

例如，如果我使此端点要求some-custom-header：

@app.post("/")
async def fn(some_custom_header: str = Header(...)):
    pass

当客户端请求缺少 时some-custom-header，服务器将生成一个response错误代码为 的422 Unprocessable entity。但是我希望能够将其更改为401 Unauthorized。换句话说，我想在我的 API 中为该特定路由自定义。RequestValidationError

我认为一个可能的解决方案是使用Header(None)，并在函数体中对进行测试None，但不幸的是，这导致 OpenAPI 文档表明标题是可选的。

解决方案 1：

选项 1

如果您不介意Header像在OpenAPI/Swagger UI autodocsOptional中那样显示，那么它会很简单，如下所示：

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}

选项 2

但是，由于您希望在 OpenAPI 中Header按要求显示，因此您应该覆盖默认的异常处理程序。当请求包含无效数据时，FastAPI 会在内部引发RequestValidationError。因此，您需要覆盖RequestValidationError包含 收到的带有无效数据的主体的。

由于RequestValidationError是 Pydantic 的子类ValidationError，您可以访问错误，如上面的链接所示，以便您可以检查您的自定义是否Header包含在错误中（如果是，则意味着请求中缺少或不是类型str），因此，返回您的自定义错误响应。如果您的自定义Header（即，some_custom_header在下面的示例中）是该特定端点中的唯一参数，则无需执行上面描述的检查（并在下面演示），因为如果RequestValidationError引发了，它将仅针对该参数。

例子

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}

选项 3

另一个解决方案是使用子应用程序（受此处讨论的启发）。您可以创建一个子应用程序（如果需要，可以创建多个子应用程序）并将其挂载到主应用程序（其中将包括需要自定义的路由）Header；因此，覆盖该子应用程序中的exception_handlerforRequestValidationError仅适用于那些路由，而不必检查request.url.path，如上一个解决方案中所示——并让主应用程序像往常一样使用其余路由。根据文档：

挂载 FastAPI 应用程序

“挂载”意味着在特定路径中添加一个完全“独立”的应用程序，然后负责处理该路径下的所有内容，并使用该子应用程序中声明的路径操作。

例子

注意：如果您使用路径挂载了子应用程序（即subapi下面的示例）'/'，则无法访问http://127.0.0.1:8000/docssubapi处的路由，因为该页面上的 API 文档将仅包含主应用程序的路由。此外，它会干扰主 API 的路由（如果主 API 中存在这样的路由），并且由于端点的顺序在 FastAPI 中很重要，因此发出请求实际上会调用主 API 的相应路由（如下所示）。因此，您宁愿使用不同的路径挂载，例如，如下所示，并访问http://127.0.0.1:8000/sub/docs处的子 API 文档。下面还提供了一个 Python 请求示例，演示了如何测试应用程序。'/'`http://127.0.0.1:8000/subapi'/sub'`

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())

选项 4

另一种解决方案是使用APIRouter带有自定义APIRoute类的，如此答案的选项 2 中所示，并在块内处理请求try-except（将用于捕获RequestValidationError异常），如FastAPI 的文档中所述。 如果发生异常，您可以根据需要处理错误，并返回自定义响应。

例子

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())