I have been using FastAPI for almost five months extensively. Every day, I wake up to writing new APIs and test cases for our business function. The project is growing day by day, and we wanted to have a centralized error handler.
This blog post covers the hurdles I faced while implementing it and how I overcame them.
Setting the Stage
To walk through this experiment with me, you need a FastAPI app. Let's call this project Playground and our custom exception will be the Playground Exception
When you run the application with uvicorn app:app --reload and hit the API localhost:8000/ you will get a nicely formed JSON response with the error message and http status code 400.
Looks nice, easy, and simple, right? What's the problem?
Introducing Background Tasks
The project I was working on predominantly used background tasks to run the business logic. Given large data processing.
When an API throws an error in the background, the process exception_handler hook no longer catches them. Because by the time the exception reaches, the response is already generated.
The error is caused by starlette/middleware/exceptions.py and the ExceptionMiddleware class
The message print ("playground exception handled") was never printed in the stack trace, showing the handler wasn't called
The above exception was the direct cause of the following exception: and Caught handled exception, but response already started in stack trace shows that RuntimeError is a direct cause of mishandling the exception.
We need a better way that
Calls the exception handler for background task
But, does not print the crazy stack trace
Let's try different alternatives of exception handler hook to work around this error.
Version 1 - Capturing RuntimeError
Instead of handling global exceptions, how about we handle the RuntimeError?
Having both PlaygroundError and RuntimeError still, result in RuntimeError since it's the result of the ExceptionMiddleware unable to gracefully handle the PlaygroundError
When debugging an error after 6 months, how can you know if this is a result of a handled or unhandled exception
This will create logs that might trigger alerts from Datadog or Sentry.
Before considering alternative approaches, we have to ensure that we aren't doing anything wrong and there is no other way possible. For that, we need answers to the following two questions.
Why is this happening?
Going through the error logs deeper will bring out a few things.
At this point, I was happy with my solution. Everything was working smoothly and then came the test cases.
Even with the CustomExceptionHandlingMiddleware I couldn't get rid of RuntimeError: Caught handled exception, but response already started. the error. That is because the TestClient we use with FastAPI has raise_server_exceptions set to True by default.
We can, of course, set it to False but in the main project, we'd be constraining fellow developers to write code a certain way. We need a better way
Additional Handler?
How about adding extra logic to our custom middleware?
try: ...except PlaygroundError: ...exceptRuntimeErroras err:ifisinstance(err.__cause__, PlaygroundError):print("Error occured while making request to ACE")returnhandle_exception(request, err.__cause__)raise
This helps us handle the RuntimeErorr Gracefully that occurs as a result of unhandled custom exception
---
There is a Middleware to handle errors on background tasks and an exception-handling hook for synchronous APIs. However, this feels hacky and took a lot of time to figure out. FastAPI developers deserve better both in terms of documentation and errors.
Other Things I Considered But Didn't Do
Custom Background Task
In FastAPI all background task functions are wrapped around BackgroundTask class. We can extend that to handle a custom error. But that'd be constraining developer behavior for future development
FastAPI Style Middleware
If you dig through FastAPI documentation enough you will find it recommending app.add_middleware as a decorator or extending BaseHTTPMiddleware
Something like this
from fastapi import Requestfrom starlette.middleware.base import BaseHTTPMiddlewareclassMyMiddleware(BaseHTTPMiddleware):def__init__(self,app,some_attribute:str, ):super().__init__(app) self.some_attribute = some_attributeasyncdefdispatch(self,request: Request,call_next):# do something with the request object, for example content_type = request.headers.get('Content-Type')print(content_type)# process the request and get the response response =awaitcall_next(request)return response
This doesn't work because the exceptions we are dealing with happen at the Starlette middleware stack level. Doesn't matter how much I tried this particular case, the dispatch method was never reached. Maybe if I dig more I can find the why?