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 theRequestValidationError
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
Option 1
If you didn't mind having the Header
showing as Optional
in
[OpenAPI/Swagger UI autodocs](https://fastapi.tiangolo.com/tutorial/first-
steps/#interactive-api-docs), 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
](https://fastapi.tiangolo.com/tutorial/handling-
errors/#override-the-default-exception-handlers). Thus, you need to override
the RequestValidationError
, which [contains the body it received with
invalid data](https://fastapi.tiangolo.com/tutorial/handling-errors/#use-the-
requestvalidationerror-body).
Since RequestValidationError
is a sub-class of Pydantic's
[ValidationError
](https://pydantic-docs.helpmanual.io/usage/models/#error-
handling), 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](https://fastapi.tiangolo.com/advanced/sub-
applications/#mounting-a-fastapi-application):
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
](https://fastapi.tiangolo.com/tutorial/bigger-
applications/#apirouter) with a [custom APIRoute
class](https://fastapi.tiangolo.com/advanced/custom-request-and-route/#custom-
apiroute-class-in-a-router), 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](https://fastapi.tiangolo.com/advanced/custom-request-and-
route/#accessing-the-request-body-in-an-exception-handler). 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())