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

by

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. 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

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

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 "https://stackoverflow.com/" 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 "https://stackoverflow.com/" 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("https://stackoverflow.com/")
async def main():
    return {'message': 'Hello from main API'}
    

subapi = FastAPI()
   
@subapi.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(content={'401': 'Unauthorized'}, status_code=401)

    
@subapi.get("https://stackoverflow.com/")
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:
                raise HTTPException(status_code=401, detail={'401': 'Unauthorized'})
                            
        return custom_route_handler


app = FastAPI()
router = APIRouter(route_class=ValidationErrorHandlingRoute)


@app.get("https://stackoverflow.com/")
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}

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

More Related Contents:

  1. How to customise error response for a specific route in FastAPI?
  2. How to add both file and JSON body in a FastAPI POST request?
  3. How to customise error response in FastAPI?
  4. How to download a large file using FastAPI?
  5. How to exclude Optional unset values from a Pydantic model using FastAPI?
  6. How to use a Pydantic model with Form data in FastAPI?
  7. FastAPI – GET Request with Pydantic List field
  8. FastAPI runs api-calls in serial instead of parallel fashion
  9. How can I send an HTTP request from my FastAPI app to another site (API)?
  10. Sharing python objects across multiple workers
  11. Render NumPy array in FastAPI
  12. How to return a numpy array as an image using FastAPI?
  13. FastAPI asynchronous background tasks blocks other requests?
  14. How to Upload a large File (≥3GB) to FastAPI backend?
  15. FastAPI is very slow in returning a large amount of JSON data
  16. How do I get my FastAPI application’s console log in JSON format with a different structure and different fields?
  17. How to send a FastAPI response without redirecting the user to another page?
  18. How to save an uploaded image to FastAPI using Python Imaging Library (PIL)?
  19. How to use url_for() to pass path and query data to a route using FastAPI and Jinja2?
  20. Value is not a valid dict when posting JSON data through Postman to FastAPI backend
  21. How to return a custom 404 Not Found page using FastAPI?
  22. How to upload a CSV file in FastAPI and convert it into JSON?
  23. How to document default None/null in OpenAPI/Swagger using FastAPI?
  24. FastAPI StreamingResponse not streaming with generator function
  25. How to upload a CSV file in FastAPI and convert it into a Pandas Dataframe?
  26. How to return a csv file/Pandas DataFrame in JSON format using FastAPI?
  27. How to access FastAPI backend from a different machine/IP on the same local network?
  28. Using FastAPI in a sync way, how can I get the raw body of a POST request?
  29. How to forward FastAPI requests to another server?
  30. How to send JSON or Form data to FastAPI backend through Axios JavaScript HTTP request?

Leave a Comment