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

Question

I want to make an HTTP endpoint in FastAPI that requires a specific header, produces a custom response code when the header is absent, as well as shows the header as required in the OpenAPI docs generated by FastAPI.

For example, if I make this endpoint to require some-custom-header:

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

when a client request lacks some-custom-header, the server will produce a response with error code 422 ("unprocessable entity"). However I'd like to be able to change that to 401 ("unauthorized").

I thought a possible solution would be to use Header(None), and do a test for None in the function body, but, unfortunately, this results in the OpenAPI docs indicating that the header is optional.

Answer 1

Option 1

If you didn't mind having the Header showing as Optional in OpenAPI, it would be as easy as follows:

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

However, since you would like the Header to appear as required in OpenAPI, you should override the default exception handler. 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, you can access the errors as shown in the above link, so that you can check whether your custom Header is included in the errors (meaning that is missing from the request, or is not of str type), and hence, return your custom response. Example below:

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}

Option 3

An alternative solution to Option 2 would be to use 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.

Example below:

Note: If you mount the sub-application (i.e., subapi in the example below) at the path "/", as shown below, you won't be able to see the routes of subapi at 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. Depending on the requirements of the app, one might choose between the three options listed in this answer.

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)