How do I integrate custom exception handling with the FastAPI exception handling?

Question

Python version 3.9, FastAPI version 0.78.0

I have a custom function that I use for application exception handling. When requests run into internal logic problems, i.e I want to send an HTTP response of 400 for some reason, I call a utility function.

@staticmethod
def raise_error(error: str, code: int) -> None:
    logger.error(error)
    raise HTTPException(status_code=code, detail=error)

Not a fan of this approach. So I look at

from fastapi import FastAPI, HTTPException, status
from fastapi.respones import JSONResponse

class ExceptionCustom(HTTPException):
    pass


def exception_404_handler(request: Request, exc: HTTPException):
    return JSONResponse(status_code=status.HTTP_404_NOT_FOUND, content={"message": "404"})


app.add_exception_handler(ExceptionCustom, exception_404_handler)

The problem I run into with the above approach is the inability to pass in the message as an argument.

Any thoughts on the whole topic?

Answer 1

You can define a custom Error exception class with a status code and a message as follows:

class Error(Exception):
    def __init__(self, status_code: int, message: str):
        self.status_code = status_code
        self.message = message

Next, create specific error classes that inherit from Error:

class InternalError(Error):
    def __init__(self, message: str):
        logging.error(f"internal error: {message}")
        super().__init__(500, message)

class UserNotFound(Error):
    def __init__(self, user_id: str):
        super().__init__(404, f"User {user_id} not found")

Then, define a custom exception handler in your FastAPI app:

@app.exception_handler(Error)
async def custom_exception_handler(request: Request, exc: Error):
    return JSONResponse(
        status_code=exc.status_code,
        content={"message": exc.message},
    )

Usage Example 1: Health Check

@app.get("/health")
async def health():
    raise InternalError("Service is unhealthy")

Response:

500

{
    "message": "Service is unhealthy"
}

Usage Example 2: Fetch User by ID

@app.get("/user/{user_id}")
async def get_user(user_id: str):
    raise UserNotFound(user_id)

Request:

GET /user/bob

Response:

404

{
    "message": "User bob not found"
}

This approach provides a clean, reusable, and structured way to handle errors in your FastAPI application.

Answer 2

Option 1

You could add custom exception handlers, and use attributes in your Exception class (i.e., MyException(Exception) in the example below), in order to pass a custom message or variables. The exception handler (in the example below, that is, my_exception_handler() with the @app.exception_handler(MyException) decorator) will handle the exception as you wish and return your custom message. For more options, please have a look at this related answer as well.

Working Example

In order to trigger the exception in the example below, call the /items/{item_id} endpoint using an item_id that is not present in the items dictionary.

from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse

app = FastAPI()
items = {"foo": "The Foo Wrestlers"}


class MyException(Exception):
    def __init__(self, item_id: str):
        self.item_id = item_id


@app.exception_handler(MyException)
async def my_exception_handler(request: Request, exc: MyException):
    return JSONResponse(status_code=status.HTTP_404_NOT_FOUND, 
        content={"message": f"Item for '{exc.item_id}' cannot be found." })


@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise MyException(item_id=item_id)
    return {"item": items[item_id]}

In case you wouldn't like using the @app.exception_handler() decorator, you could remove the decorator from the my_exception_handler() function and instead use the add_exception_handler() method to add the handler to the app instance. Example:

app.add_exception_handler(MyException, my_exception_handler)

Another way to add the exception handler to the app instance would be to use the exception_handlers parameter of the FastAPI class, as demonstrated in this answer. Related answers can also be found here and here.

Option 2

You could always use HTTPException to return HTTP responses with custom errors to the client (as well as add custom headers to the HTTP error).

Working Example

from fastapi import FastAPI, HTTPException

app = FastAPI()
items = {"foo": "The Foo Wrestlers"}


@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item": items[item_id]}

Answer 3

Your custom exception can have any custom attributes that you want. Let's say you write it this way:

class ExceptionCustom(HTTPException):
    pass 

in your custom handler, you can do something like

def exception_404_handler(request: Request, exc: HTTPException):
    return JSONResponse(status_code=status.HTTP_404_NOT_FOUND, content={"message": exc.detail})

Then, all you need to do is to raise the exception this way:

raise ExceptionCustom(status_code=404, detail='error message')

Note that you are creating a handler for this specific ExceptionCustom. If all you need is the message, you can write something more generic:

class MyHTTPException(HTTPException):
    pass

def my_http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(status_code=exc.status_code, content={"message": exc.detail})

app.add_exception_handler(MyHTTPException, my_http_exception_handler)

This way you can raise any exception, with any status code and any message and have the message in your JSON response.

There's a detailed explanation on FastAPI docs