如何在使用 FastAPI 的请求中缺少 Header 时 return 自定义响应

Question

我想在需要特定 header 的 FastAPI 中创建一个 HTTP 端点，当 header 不存在时生成自定义 response 代码，以及在 FastAPI 生成的 OpenAPI 文档中将 header 显示为 必需的 。

例如，如果我让这个端点需要 some-custom-header:

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

当客户端请求缺少 some-custom-header 时，服务器将生成 response，错误代码为 422（“unprocessable entity"). However I'd like to be able to change that to 401 ("unauthorized”）。

我认为一个可能的解决方案是使用 Header(None)，并在函数 body 中对 None 进行测试，但不幸的是，这导致 OpenAPI 文档指示header 是 可选的 .

Answer 1

选项 1

如果您不介意 Header 在 OpenAPI 中显示为 Optional，它会像下面这样简单：

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

但是，由于您希望 Header 在 OpenAPI 中显示为 required，因此您应该覆盖默认的异常处理程序。 When a request contains invalid data, FastAPI internally raises a RequestValidationError. Thus, you need to override the RequestValidationError. The RequestValidationError contains the body it received with invalid data, and since RequestValidationError is a sub-class of Pydantic's ValidationError，你可以访问如上link所示的错误，这样你就可以检查你的自定义Header是否包含在错误中（意味着请求中缺少，或者不是 str 类型），因此，return 您的自定义响应。示例如下：

from fastapi import FastAPI, Request, Header, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from fastapi.encoders import jsonable_encoder

routes_with_custom_header = ["/"]

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    if request.url.path in routes_with_custom_header:
        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

选项 2 的替代解决方案是使用 Sub-Application(s) (inspired by the discussion here). You could have a main app - which would include all the routes/path operations that require the custom Header; hence, overriding the validation exception handler would apply to those routes only - and "mount" one (or more) sub-application(s) with the remaining routes. As per the documentation:

Mounting a FastAPI application

"Mounting" means adding a completely "independent" application in a specific path, that then takes care of handling everything under that path, with the path operations declared in that sub-application.

示例如下：

注意：如果在"/"路径下挂载sub-application（即下例中的subapi），如图在下方，您将无法在 http://127.0.0.1:8000/docs, as the API docs on that page will include only the routes for the main app. Thus, you would rather mount subapi at a different path, e.g., "/subapi", and access its docs at http://127.0.0.1:8000/subapi/docs 处看到 subapi 的路线。根据应用程序的要求，可以在此答案中列出的三个选项中进行选择。

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

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    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}    

subapi = FastAPI()

@subapi.get("/sub")
def read_sub(some_param: str):
    return {"message": "Hello World from sub API"}


app.mount("/", subapi)

如何在使用 FastAPI 的请求中缺少 Header 时 return 自定义响应

How to return a custom Response when a Header is absent from the Request using FastAPI

python

fastapi

选项 1

选项 2

选项 3