""" This file defines and handles all the errors The ExceptionHandlerRoute class is used by all the routes """ from enum import Enum from typing import Any, Coroutine, Type from fastapi import Request, Response from fastapi.routing import APIRoute from pydantic import BaseModel, ConfigDict from collections.abc import Callable from sqlalchemy.exc import NoResultFound, SQLAlchemyError class ProblemType(str, Enum): ITEM_NOT_FOUND = 'ITEM_NOT_FOUND' INTERNAL_SERVER_ERROR = 'INTERNAL_SERVER_ERROR' INVALID_CREDENTIALS = 'INVALID_CREDENTIALS' UNIQUE_CONSTRAINT = 'UNIQUE_CONSTRAINT' INVALID_REQUEST = 'INVALID_REQUEST' BLANK = 'BLANK' class ProblemDetail(BaseModel): """ RFC 7807 Problem Details for HTTP APIs """ type : ProblemType = ProblemType.BLANK status : int | None = None title : str | None = None detail : str | None = None model_config = ConfigDict(from_attributes=True) class ItemNotFoundException(Exception): def __init__(self, model: Type, key : Any = ""): self.model = model.__name__ self.key = str(key) class ExceptionHandlerRoute(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 SQLAlchemyError as exc: problem = ProblemDetail(status = 422) if type(exc) == NoResultFound: problem.type = ProblemType.ITEM_NOT_FOUND problem.title = f'{exc.model} Not Found' problem.detail = { 'key' : exc.key} problem.status = 404 elif type(exc.orig) == UniqueViolation: problem.detail = str(exc.orig).split('\n')[1].split(': ')[1] problem.type = ProblemType.UNIQUE_VALIDATION problem.title = 'Unique violation' elif type(exc.orig) == ForeignKeyViolation: problem.type = ProblemType.FOREIGN_KEY_VIOLATION problem.title = 'Foreign Key Violation' problem.detail = str(exc.orig).split('\n')[1].split(': ')[1] return JSONResponse( status_code = problem.status, content = problem.model_dump(exclude_none = True), headers = {'Content-Type': 'application/problem+json'} ) raise HTTPException(status_code = 422, detail = detail) except Exception as exc: problem = ProblemDetail(status = 500) problem.title = 'Internal Server Error please report the bug' problem.type = ProblemType.INTERNAL_SERVER_ERROR problem.detail = str(exc) if type(exc) == ItemNotFoundException: problem.type = ProblemType.ITEM_NOT_FOUND problem.title = f'{exc.model} Not Found' problem.detail = { 'key' : exc.key} problem.status = 404 if type(exc) == RequestValidationError: problem.type = ProblemType.INVALID_REQUEST problem.title = "Input validation error" problem.detail = exc.errors() problem.status = 422 if type(exc) == HTTPException: print(exc.status_code) problem.title = "Credientials error" problem.detail = exc.detail problem.status = exc.status_code if problem.status == 401: problem.type = ProblemType.INVALID_CREDENTIALS return JSONResponse( status_code = problem.status, content = problem.model_dump(exclude_none = True), headers = {'Content-Type': 'application/problem+json'} ) return custom_route_handler