Fastapi: Custom Exception not being catch python-FastAPI

[rishabhsiitk]

I am using FastAPI version 0.52.0. I am trying to raise Custom Exception from a file which is not being catch by exception handler. I am also using middleware.
Below are the code snippets
Custom Exception class

class CustomException(Exception):
    def __init__(self, code: int, message: str):
        self.code = code
        self.message = message

endpoint

@app.put("/check")
def check(user_data: UserData):
    start = time.time()
    PutUserData().process(user_data)
    print('\n'.join([
        "Web Method Completed",
        f"\tTotal Time: {time.time() - start}"]))
    return JSONResponse(status_code=status.HTTP_201_CREATED, content={"message": "OK"})

Exception Handler

@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
    return JSONResponse(
        status_code=exc.code,
        content={"message": f"Exception Occurred! Reason -> {exc.message}"},
    )

Middleware

@app.middleware("http")
async def add_metric(request: Request, call_next):
    response = await call_next(request)
    print("Response: ", response.status_code)
    return response

I am raising CustomException from PutUserData().process() with custom status code and some message which is not being processed and API is responding with 201 status code. In my use case i would need to raise CustomException with different status code based on situations. Please help me on this.

[rishabhsiitk]

@ycd Can you please help me on this ?

[Mause]

Your code seems to do exactly what you want already?

import time
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from starlette import status
from fastapi.testclient import TestClient
from pydantic import BaseModel

app = FastAPI()


class UserData(BaseModel):
    pass


class PutUserData:
    def process(self, user_data):
        raise CustomException(200, 'Whatever')

# your code starts here

class CustomException(Exception):
    def __init__(self, code: int, message: str):
        self.code = code
        self.message = message


@app.put("/check")
def check(user_data: UserData):
    start = time.time()
    PutUserData().process(user_data)
    print('\n'.join(["Web Method Completed", f"\tTotal Time: {time.time() - start}"]))
    return JSONResponse(status_code=status.HTTP_201_CREATED, content={"message": "OK"})


@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
    return JSONResponse(
        status_code=exc.code,
        content={"message": f"Exception Occurred! Reason -> {exc.message}"},
    )


@app.middleware("http")
async def add_metric(request: Request, call_next):
    response = await call_next(request)
    print("Response: ", response.status_code)
    return response

# your code ends here

tc = TestClient(app)

response = tc.put('/check', json={})
assert response.json()['message'].startswith('Exception Occurred!')  # this assertion passes

[ycd]

Like @Mause mentioned your code looks fine if it is your actual use case, are you trying to achieve something else?

[AlvaroLQueiroz]

Hello guys, i have a similar problem. I need raise a custom exception inside a middleware, but my exception handler do not catch the error.

from datetime import time

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.testclient import TestClient
from starlette import status
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.types import ASGIApp

app = FastAPI()


# your code starts here

class CustomException(Exception):
    def __init__(self, code: int, message: str):
        self.code = code
        self.message = message


class PartnerAvailabilityMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        raise CustomException(
            status.HTTP_503_SERVICE_UNAVAILABLE,
            'Partner services is unavailable.'
        )
        return await call_next(request)


@app.get("/test")
def test():
    start = time.time()
    print('\n'.join(["Web Method Completed", f"\tTotal Time: {time.time() - start}"]))
    return JSONResponse(status_code=status.HTTP_200_OK, content={"message": "OK"})


app.add_middleware(PartnerAvailabilityMiddleware)


@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
    return JSONResponse(
        status_code=exc.code,
        content={"message": f"Exception Occurred! Reason -> {exc.message}"},
    )

# your code ends here

tc = TestClient(app)

response = tc.get('/test')
assert response.json()['message'].startswith('Exception Occurred!')  # this assertion fail

FastAPI==0.61.1
Starlette==0.13.6
Python==3.8

[sdklab007]

+1 facing the same issue as @AlvaroLQueiroz

Have someone found the solution?

[Mause]

You can add the exception handler middleware twice to handle middleware exceptions:

import time
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.testclient import TestClient
from starlette import status
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.exceptions import ExceptionMiddleware
from starlette.types import ASGIApp

app = FastAPI()


# your code starts here



class CustomException(Exception):
    def __init__(self, code: int, message: str):
        self.code = code
        self.message = message


class PartnerAvailabilityMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        raise CustomException(
            status.HTTP_503_SERVICE_UNAVAILABLE, 'Partner services is unavailable.'
        )
        return await call_next(request)


@app.get("/test")
def test():
    start = time.time()
    print('\n'.join(["Web Method Completed", f"\tTotal Time: {time.time() - start}"]))
    return JSONResponse(status_code=status.HTTP_200_OK, content={"message": "OK"})


@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
    return JSONResponse(
        status_code=exc.code,
        content={"message": f"Exception Occurred! Reason -> {exc.message}"},
    )


# your code ends here

app.add_middleware(PartnerAvailabilityMiddleware)
app.add_middleware(ExceptionMiddleware, handlers=app.exception_handlers)  # this is the change

tc = TestClient(app)

response = tc.get('/test')
assert response.json()['message'].startswith('Exception Occurred!')  # this assertion now passes