feat: standardized ErrorResponse model

api/src/app.py

@@ -3,11 +3,16 @@
 
 import click
 from fastapi import APIRouter, FastAPI, Security
+from fastapi.encoders import jsonable_encoder
+from fastapi.exceptions import RequestValidationError
+from starlette import status
 from starlette.requests import Request
-from starlette.responses import Response
+from starlette.responses import JSONResponse, Response
 
 from authentication.authentication import auth_with_jwt
-from common.utils.logger import logger
+from common.exceptions import ErrorResponse
+from common.logger import logger
+from common.responses import responses
 from config import config
 from features.health_check import health_check_feature
 from features.todo import todo_feature
@@ -25,7 +30,7 @@ def create_app() -> FastAPI:
     authenticated_routes = APIRouter()
     authenticated_routes.include_router(todo_feature.router)
     authenticated_routes.include_router(whoami_feature.router)
-    app = FastAPI(title="Awesome Boilerplate", description="")
+    app = FastAPI(title="Awesome Boilerplate", description="", responses=responses)
     app.include_router(authenticated_routes, prefix=prefix, dependencies=[Security(auth_with_jwt)])
     app.include_router(public_routes, prefix=prefix)
 
@@ -39,6 +44,20 @@ async def add_process_time_header(request: Request, call_next: Callable) -> Resp
         response.headers["X-Process-Time"] = str(process_time)
         return response
 
+    # Intercept FastAPI builtin validation errors, so they can be returned on our standardized format.
+    @app.exception_handler(RequestValidationError)
+    async def validation_exception_handler(request: Request, exc: RequestValidationError):
+        return JSONResponse(
+            ErrorResponse(
+                status=status.HTTP_422_UNPROCESSABLE_ENTITY,
+                type="RequestValidationError",
+                message="The received values are invalid",
+                debug="The received values are invalid according to the endpoints model definition",
+                extra=jsonable_encoder({"detail": exc.errors(), "body": exc.body}),
+            ).dict(),
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+        )
+
     return app
 
 
@@ -55,6 +74,7 @@ def run():
         "app:create_app",
         host="0.0.0.0",  # nosec
         port=5000,
+        factory=True,
         reload=config.ENVIRONMENT == "local",
         log_level=config.LOGGER_LEVEL.lower(),
     )

api/src/authentication/authentication.py

@@ -7,7 +7,7 @@
 from authentication.mock_token_generator import mock_rsa_public_key
 from authentication.models import User
 from common.exceptions import credentials_exception
-from common.utils.logger import logger
+from common.logger import logger
 from config import config, default_user
 
 oauth2_scheme = OAuth2AuthorizationCodeBearer(

api/src/common/exceptions.py

@@ -1,35 +1,66 @@
-from typing import Optional
-
 from fastapi import HTTPException
-from starlette import status
+from pydantic import BaseModel
+from starlette import status as request_status
+
+
+# Pydantic models can not inherit from "Exception", but we use it for OpenAPI spec
+class ErrorResponse(BaseModel):
+    status: int = 500
+    type: str = "ApplicationException"
+    message: str = "The requested operation failed"
+    debug: str = "An unknown and unhandled exception occurred in the API"
+    extra: dict | None = None
 
 
 class ApplicationException(Exception):
-    def __init__(self, message: Optional[str] = "Something went wrong"):
-        super()
+    status: int = 500
+    type: str = "ApplicationException"
+    message: str = "The requested operation failed"
+    debug: str = "An unknown and unhandled exception occurred in the API"
+    extra: dict | None = None
+
+    def __init__(self, message: str = "The requested operation failed",
+                 debug: str = "An unknown and unhandled exception occurred in the API", extra: dict = None,
+                 status: int = 500, ):
+        self.status = status
+        self.type = self.__class__.__name__
         self.message = message
+        self.debug = debug
+        self.extra = extra
+
+    def dict(self):
+        return {"status": self.status, "type": self.type, "message": self.message, "debug": self.debug,
+            "extra": self.extra, }
+
 
-    def __str__(self):
-        return self.message
+class MissingPrivilegeException(ApplicationException):
+    def __init__(self, message: str = "You do not have the required permissions",
+            debug: str = "Action denied because of insufficient permissions", extra: dict = None, ):
+        super().__init__(message, debug, extra, request_status.HTTP_403_FORBIDDEN)
+        self.type = self.__class__.__name__
 
 
-class EntityNotFoundException(ApplicationException):
-    def __init__(self, message: str = None):
-        super().__init__(message)
+class NotFoundException(ApplicationException):
+    def __init__(self, message: str = "The requested resource could not be found",
+            debug: str = "The requested resource could not be found", extra: dict = None, ):
+        super().__init__(message, debug, extra, request_status.HTTP_404_NOT_FOUND)
+        self.type = self.__class__.__name__
 
 
-class BadRequest(ApplicationException):
-    def __init__(self, message: str = None):
-        super().__init__(message)
+class BadRequestException(ApplicationException):
+    def __init__(self, message: str = "Invalid data for the operation",
+            debug: str = "Unable to complete the requested operation with the given input values.",
+            extra: dict = None, ):
+        super().__init__(message, debug, extra, request_status.HTTP_400_BAD_REQUEST)
+        self.type = self.__class__.__name__
 
 
-class MissingPrivilegeException(Exception):
-    def __init__(self, message=None):
-        self.message = message if message else "Missing privileges to perform operation on the resource"
+class ValidationException(ApplicationException):
+    def __init__(self, message: str = "The received data is invalid",
+            debug: str = "Values are invalid for requested operation.", extra: dict = None, ):
+        super().__init__(message, debug, extra, request_status.HTTP_422_UNPROCESSABLE_ENTITY)
+        self.type = self.__class__.__name__
 
 
-credentials_exception = HTTPException(
-    status_code=status.HTTP_401_UNAUTHORIZED,
-    detail="Token validation failed",
-    headers={"WWW-Authenticate": "Bearer"},
-)
+credentials_exception = HTTPException(status_code=request_status.HTTP_401_UNAUTHORIZED,
+    detail="Token validation failed", headers={"WWW-Authenticate": "Bearer"}, )

api/src/common/utils/logger.py → api/src/common/logger.py

@@ -1,17 +1,17 @@
-"""
-Setup API and Uvicorn logger.
-"""
-
-import logging
-
-from config import config
-
-uvicorn_logger = logging.getLogger("uvicorn")
-
-logger = logging.getLogger("API")
-logger.setLevel(config.LOGGER_LEVEL.upper())
-formatter = logging.Formatter("%(levelname)s:%(asctime)s %(message)s")
-channel = logging.StreamHandler()
-channel.setFormatter(formatter)
-channel.setLevel(config.LOGGER_LEVEL.upper())
-logger.addHandler(channel)
+"""
+Setup API and Uvicorn logger.
+"""
+
+import logging
+
+from config import config
+
+uvicorn_logger = logging.getLogger("uvicorn")
+
+logger = logging.getLogger("API")
+logger.setLevel(config.LOGGER_LEVEL.upper())
+formatter = logging.Formatter("%(levelname)s:%(asctime)s %(message)s")
+channel = logging.StreamHandler()
+channel.setFormatter(formatter)
+channel.setLevel(config.LOGGER_LEVEL.upper())
+logger.addHandler(channel)

api/src/common/responses.py

@@ -1,45 +1,87 @@
 import functools
 import traceback
+from inspect import iscoroutinefunction
 from typing import Callable, Type, TypeVar
 
-from pydantic import ValidationError
 from requests import HTTPError
 from starlette import status
-from starlette.responses import PlainTextResponse, Response
+from starlette.responses import JSONResponse, Response
 
-from common.exceptions import BadRequest, EntityNotFoundException
-from common.utils.logger import logger
+from common.exceptions import (
+    ApplicationException,
+    BadRequestException,
+    ErrorResponse,
+    MissingPrivilegeException,
+    NotFoundException,
+    ValidationException,
+)
+from common.logger import logger
+
+responses = {
+    400: {"model": ErrorResponse, "content": {"application/json": {"example": BadRequestException().dict()}}},
+    401: {
+        "model": ErrorResponse,
+        "content": {
+            "application/json": {
+                "example": ErrorResponse(
+                    status=401, type="UnauthorizedException", message="Token validation failed"
+                ).dict()
+            }
+        },
+    },
+    403: {"model": ErrorResponse, "content": {"application/json": {"example": MissingPrivilegeException().dict()}}},
+    404: {"model": ErrorResponse, "content": {"application/json": {"example": NotFoundException().dict()}}},
+    422: {"model": ErrorResponse, "content": {"application/json": {"example": ValidationException().dict()}}},
+    500: {"model": ErrorResponse, "content": {"application/json": {"example": ApplicationException().dict()}}},
+}
 
 TResponse = TypeVar("TResponse", bound=Response)
 
+"""
+Function made to be used as a decorator for a route.
+It will execute the function, and return a successful response of the passed response class.
+If the execution fails, it will return a JSONResponse with a standardized error model.
+"""
+
 
-def create_response(response_class: Type[TResponse]) -> Callable[..., Callable[..., TResponse | PlainTextResponse]]:
-    def func_wrapper(func) -> Callable[..., TResponse | PlainTextResponse]:
+def create_response(response_class: Type[TResponse]) -> Callable[..., Callable[..., TResponse | JSONResponse]]:
+    def func_wrapper(func) -> Callable[..., TResponse | JSONResponse]:
         @functools.wraps(func)
-        def wrapper_decorator(*args, **kwargs) -> TResponse | PlainTextResponse:
+        async def wrapper_decorator(*args, **kwargs) -> TResponse | JSONResponse:
             try:
-                result = func(*args, **kwargs)
+                # Await function if needed
+                if not iscoroutinefunction(func):
+                    result = func(*args, **kwargs)
+                else:
+                    result = await func(*args, **kwargs)
                 return response_class(result, status_code=200)
-            except HTTPError as e:
-                logger.error(e)
-                return PlainTextResponse(str(e), status_code=e.response.status_code)
-            except ValidationError as e:
+            except HTTPError as http_error:
+                error_response = ErrorResponse(
+                    type="ExternalFetchException",
+                    status=http_error.response.status,
+                    message="Failed to fetch an external resource",
+                    debug=http_error.response,
+                )
+                logger.error(error_response)
+                return JSONResponse(error_response.dict(), status_code=error_response.status)
+            except ValidationException as e:
                 logger.error(e)
-                return PlainTextResponse(str(e), status_code=status.HTTP_422_UNPROCESSABLE_ENTITY)
-            except EntityNotFoundException as e:
+                return JSONResponse(e.dict(), status_code=status.HTTP_422_UNPROCESSABLE_ENTITY)
+            except NotFoundException as e:
                 logger.error(e)
-                return PlainTextResponse(str(e), status_code=status.HTTP_404_NOT_FOUND)
-            except BadRequest as e:
+                return JSONResponse(e.dict(), status_code=status.HTTP_404_NOT_FOUND)
+            except BadRequestException as e:
                 logger.error(e)
-                return PlainTextResponse(str(e), status_code=status.HTTP_400_BAD_REQUEST)
+                logger.error(e.dict())
+                return JSONResponse(e.dict(), status_code=status.HTTP_400_BAD_REQUEST)
+            except MissingPrivilegeException as e:
+                logger.warning(e)
+                return JSONResponse(e.dict(), status_code=status.HTTP_403_FORBIDDEN)
             except Exception as e:
                 traceback.print_exc()
                 logger.error(f"Unexpected unhandled exception: {e}")
-                return PlainTextResponse(
-                    "Unexpected unhandled exception",
-                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-                )
+                return JSONResponse(ErrorResponse().dict(), status_code=status.HTTP_500_INTERNAL_SERVER_ERROR)
 
-        return wrapper_decorator
+        return wrapper_decorator  # type: ignore
 
     return func_wrapper

api/src/features/todo/use_cases/get_todo_by_id.py

@@ -2,8 +2,8 @@
 
 from pydantic import BaseModel, Field
 
+from common.exceptions import NotFoundException
 from entities.TodoItem import TodoItem
-from features.todo.exceptions import TodoItemNotFoundError
 from infrastructure.repositories.TodoRepository import TodoRepository
 
 
@@ -20,5 +20,5 @@ def from_entity(todo_item: TodoItem) -> "GetTodoByIdResponse":
 def get_todo_by_id_use_case(id: str, todo_item_repository=TodoRepository()) -> GetTodoByIdResponse:
     todo_item = todo_item_repository.get(id)
     if not todo_item:
-        raise TodoItemNotFoundError
+        raise NotFoundException
     return GetTodoByIdResponse.from_entity(cast(TodoItem, todo_item))

api/src/infrastructure/clients/mongodb/MongoClient.py

@@ -4,7 +4,7 @@
 from pymongo import MongoClient
 from pymongo.errors import DuplicateKeyError
 
-from common.exceptions import EntityNotFoundException
+from common.exceptions import NotFoundException
 from config import config
 from infrastructure.clients.ClientInterface import ClientInterface
 
@@ -56,7 +56,7 @@ def list(self) -> List[dict]:
     def find_by_id(self, uid: str) -> Dict:
         document = self.handler[self.collection].find_one(filter={"_id": uid})
         if document is None:
-            raise EntityNotFoundException
+            raise NotFoundException
         else:
             return dict(document)
 

api/src/infrastructure/repositories/TodoRepository.py

@@ -1,7 +1,7 @@
 from typing import Optional
 
+from common.exceptions import NotFoundException
 from entities.TodoItem import TodoItem
-from features.todo.exceptions import TodoItemNotFoundError
 from features.todo.interfaces.TodoRepositoryInterface import TodoRepositoryInterface
 from infrastructure.clients.ClientInterface import ClientInterface
 from infrastructure.clients.mongodb.MongoClient import get_mongo_client
@@ -29,7 +29,7 @@ def delete_by_id(self, id: str):
     def get(self, id: str) -> TodoItem:
         todo_item = self.client.find_by_id(id)
         if todo_item is None:
-            raise TodoItemNotFoundError
+            raise NotFoundException
         return TodoItem.from_dict(todo_item)
 
     def create(self, todo_item: TodoItem) -> Optional[TodoItem]:

api/src/tests/unit/features/todo/conftest.py

@@ -3,8 +3,7 @@
 
 import pytest
 
-from common.exceptions import EntityNotFoundException
-from features.todo.exceptions import TodoItemNotFoundError
+from common.exceptions import NotFoundException
 from infrastructure.clients.ClientInterface import ClientInterface
 from infrastructure.repositories.TodoRepository import TodoRepository
 
@@ -27,7 +26,7 @@ def mock_create(document: Dict) -> Dict:
     def mock_find_by_id(id: str) -> Dict:
         if id in test_data.keys():
             return test_data[id]
-        raise TodoItemNotFoundError()
+        raise NotFoundException()
 
     def mock_get_all():
         document_list = []
@@ -43,7 +42,7 @@ def mock_delete(id: str) -> bool:
 
     def mock_update(id: str, document: Dict):
         if id not in list(test_data.keys()):
-            raise EntityNotFoundException()
+            raise NotFoundException()
         test_data[id] = document
         return test_data[id]