Custom Error Route Handling in Python and FastAPI
5 min read
I've been working on a lot of API design in Python and FastAPI recently. When structuring my code base, I try to keep my apis.py file as lean as possible, with the business logic encapsulated within a service level that is injected as a dependency to the API method. This makes things easy to test, and allows my API level to solely be responsible for returning a successful response or raising (and alerting) meaningful exceptions.
Here's an example of what my apis.py file might look like within the FastAPI framework. This an API endpoint for updating a game in my database. If you're not familiar with how FastAPI and Pydantic works, GameIn is the payload schema sent in the request, and GameOut is the schema of the response object. I've injected in my GameService as a dependency (get_game_service initialises the GameService with a Game repository that is responsible for connecting to the database).
@router.put("/{game_id}/", response_model=GameOut)
def update_game(
game_id: str,
game_in: GameIn,
service: GameService = Depends(get_game_service),
):
try:
return service.update(game_id=game_id, game_in=game_in)
except (ExceptionABC, ExceptionDEF) as e:
logger.warning(str(e))
raise HTTPException(
status_code=status.HTTP_412_PRECONDITION_FAILED, detail=str(e)
)
except ExceptionGHI as e:
logger.info(str(e))
raise HTTPException(
status_code=status.HTTP_412_PRECONDITION_FAILED, detail=str(e)
)
except Exception as e:
logger.error("Failed to update game. Reason: " + str(e))
raise_alert_in_production(
description=str(e),
)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Could not update game",
)This all works great, but what happens if you've say got 10 specific exceptions that you're looking to identify separately, and you've got 5 other API endpoints all looking to catch the same type of exceptions? You could copy and paste the try-except block for each endpoint, but this will result in a very long file with lots of repeated code. 😱
DRYing up exception handling in Python
This is where Python function decorators come in. We can define a reusable function game_error_handler that we use to decorate (i.e. wrap) all of the endpoints that require the same exception handling logic. You can read more about Python's functools library (for higher-order functions and operations on callable objects) and the wraps function here.
def game_error_handler(func):
@wraps(func)
def decorator(*args, **kwargs):
try:
return func(*args, **kwargs)
except (ExceptionABC, ExceptionDEF) as e:
logger.warning(str(e))
raise HTTPException(
status_code=status.HTTP_412_PRECONDITION_FAILED, detail=str(e)
)
except ExceptionGHI as e:
logger.info(str(e))
raise HTTPException(
status_code=status.HTTP_412_PRECONDITION_FAILED, detail=str(e)
)
except Exception as e:
logger.error("Failed to update game. Reason: " + str(e))
raise_alert_in_production(
description=str(e),
)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Could not update game",
)
return decorator
@router.put("/{game_id}/", response_model=GameOut)
@game_error_handler
def update_game(
game_id: str,
game_in: GameIn,
service: GameService = Depends(get_game_service),
):
return service.update(game_id=game_id, game_in=game_in)The above makes things nice, clean and scalable for if I need to add additional endpoints in the future. 🚀 This is all done in Python, so even if you're not using FastAPI, you should be able to do something similar in your framework of choice.
One step further with FastAPI and Pydantic
If you are using FastAPI and Pydantic however, we can add some additional logic to help us decode Pydantic validation errors which aren't always super intuitive. 😅 (If you're wondering what Pydantic is, it's a package that helps with data validation and settings management using Python type annotations, and works really nicely with FastAPI.)
In our example, we might expect the PUT payload for updating a game to include some very specific attributes, e.g. our GameIn schema might look something like this.
{
game_id: str
console: Literal["ps4", "ps5", "nintendo_switch"]
publication_date: date
on_sale: bool
}If the payload comes in, and the console value is "xbox", Pydantic will raise a validation error to highlight that the value is incompatible with what we expect. The example I've given above is manageable, but what if your schema includes lots of custom, complex validation requirements - how might we return Pydantic's validation errors in a more human-readable format?
A solution is to define your own custom error route handler.
import logging
from collections import defaultdict
from functools import wraps
from typing import Any, Callable, Coroutine
from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import RequestValidationError
from fastapi.routing import APIRoute
from starlette.responses import JSONResponse
logger = logging.getLogger(__name__)
class CustomErrorRouteHandler(APIRoute):
def get_route_handler(self) -> Callable[[Request], Coroutine[Any, Any, Response]]:
original_route_handler = super().get_route_handler()
async def custom_route_handler(request: Request) -> Response:
if request.method == "GET":
request_payload = request.query_params
else:
request_payload = await request.json()
logger.info(
f"Request received. URL: {request.url}. Payload: {request_payload} "
)
try:
return await original_route_handler(request)
except RequestValidationError as exc:
readable_error = await self.readable_errors_from_pydantic_message(
exc
)
return JSONResponse(
status_code=status.HTTP_400_BAD_REQUEST,
content=jsonable_encoder(
{"detail": "Invalid Request", "errors": reformatted_message}
),
)
return custom_route_handler
async def readable_errors_from_pydantic_message(self, exc):
readable_error = defaultdict(list)
for errors in exc.errors():
loc, msg = errors["loc"], errors["msg"]
filtered_loc = loc[1:] if loc[0] in ("body", "query", "path") else loc
field_string = ".".join(filtered_loc)
readable_error[field_string].append(msg)
return readable_error
router = APIRouter(route_class=CustomErrorRouteHandler)
def game_error_handler(func):
@wraps(func)
def decorator(*args, **kwargs):
try:
return func(*args, **kwargs)
except (ExceptionABC, ExceptionDEF) as e:
logger.warning(str(e))
raise HTTPException(
status_code=status.HTTP_412_PRECONDITION_FAILED, detail=str(e)
)
except ExceptionGHI as e:
logger.info(str(e))
raise HTTPException(
status_code=status.HTTP_412_PRECONDITION_FAILED, detail=str(e)
)
except Exception as e:
logger.error("Failed to update game. Reason: " + str(e))
raise_alert_in_production(
description=str(e),
)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Could not update game",
)
return decorator
@router.put("/{game_id}/", response_model=GameOut)
@game_error_handler
def update_game(
game_id: str,
game_in: GameIn,
service: GameService = Depends(get_game_service),
):
return service.update(game_id=game_id, game_in=game_in)Note that what we're doing here is overriding the logic used by the Request and APIRoute classes (not defining middleware). The specific method we're overriding is the get_route_handler where we're first inheriting the logic from the parent class APIRoute before checking to see if there are RequestValidationErrors from Pydantic. If so, we want to parse these errors into a more readable format before returning these as part of a 400 error response. You can check out the official FastAPI docs here.