How to customise error response for a specific route in 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. In other words, I would like to customise the RequestValidationError for that specific route in my API.

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/Swagger UI autodocs, 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, which contains the body it received with invalid data.

Since RequestValidationError is a sub-class of Pydantic's ValidationError, you can access the errors as shown in the link above, so that you can check whether your custom Header is included in the errors (if so, that means that is either missing from the request, or is not of str type), and hence, return your custom error response. If your custom Header (i.e., some_custom_header in the example below) is the only parameter in that specific endpoint, then it is not necessary to perform the check described above (and demosntrated below), as if a RequestValidationError was raised, it would be only for that parameter.

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

Another solution would be to use Sub-Application(s) (inspired by the discussion here). You could create a sub app (or more if needed) and mount it to the main app—which would include the route(s) that require the custom Header; hence, overriding the exception_handler for RequestValidationError in that sub app would only apply to those routes, without having to check for the request.url.path, as demonstrated in the previous solution—and have the main app with the remaining routes as usual. 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

Note: If you mounted the sub-application (i.e., subapi in the example below) using the '/' path, you wouldn't be able to access the routes of subapi at http://127.0.0.1:8000/docs, as the API docs on that page will only include the routes of the main app. Also, it would interfere with the '/' route of the main API (if such a route exists in the main API), and since endpoints' order matters in FastAPI, issuing a request to http://127.0.0.1:8000/ would actually call the corresponding route of the main API (as demonstrated below). Thus, you would rather mount subapi using a different path, e.g., '/sub', as demonstrated below, and access the sub API docs at http://127.0.0.1:8000/sub/docs. A Python requests example is also given below, demonstrating how to test the app.

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)

Test the example above

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

A further solution would be to use an APIRouter with a custom APIRoute class, as demonstrated in Option 2 of this answer, and handle the request inside a try-except block (which will be used to catch RequestValidationError exceptions), as described in FastAPI's documentation. If an exception occurs, you can handle the error as desired, and return a custom respone.

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) 

Test the example above

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