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