Refactor error handling for systems #74

Author: joelvdavies

inventory_management_system_api/core/exceptions.py

@@ -2,8 +2,42 @@
 Module for custom exception classes.
 """
 
+from typing import Optional
 
-class DatabaseError(Exception):
+from fastapi import status
+
+
+class BaseAPIException(Exception):
+    """Base exception for API errors."""
+
+    # Status code to return if this exception is raised
+    status_code: int
+
+    # Generic detail of the exception (That may be returned in a response)
+    response_detail: str
+
+    detail: str
+
+    def __init__(self, detail: str, response_detail: Optional[str] = None, status_code: Optional[int] = None):
+        """
+        Initialise the exception.
+
+        :param detail: Specific detail of the exception (just like Exception would take - this will only be logged
+                       and not returned in a response).
+        :param response_detail: Generic detail of the exception that will be returned in a response.
+        :param status_code: Status code that will be returned in a response.
+        """
+        super().__init__(detail)
+
+        self.detail = detail
+
+        if response_detail is not None:
+            self.response_detail = response_detail
+        if status_code is not None:
+            self.status_code = status_code
+
+
+class DatabaseError(BaseAPIException):
     """
     Database related error.
     """
@@ -46,22 +80,29 @@ class MissingMandatoryProperty(Exception):
 
 
 class DuplicateRecordError(DatabaseError):
-    """
-    The record being added to the database is a duplicate.
-    """
+    """The record being added to the database is a duplicate."""
+
+    status_code = status.HTTP_409_CONFLICT
+    response_detail = "Duplicate record found"
 
 
 class InvalidObjectIdError(DatabaseError):
     """
     The provided value is not a valid ObjectId.
     """
 
+    status_code = status.HTTP_422_UNPROCESSABLE_ENTITY
+    response_detail = "Invalid ID given"
+
 
 class MissingRecordError(DatabaseError):
     """
     A specific database record was requested but could not be found.
     """
 
+    status_code = status.HTTP_404_NOT_FOUND
+    response_detail = "Requested record was not found"
+
 
 class ChildElementsExistError(DatabaseError):
     """

inventory_management_system_api/main.py

@@ -11,6 +11,7 @@
 from fastapi.responses import JSONResponse
 
 from inventory_management_system_api.core.config import config
+from inventory_management_system_api.core.exceptions import BaseAPIException
 from inventory_management_system_api.core.logger_setup import setup_logger
 from inventory_management_system_api.routers.v1 import (
     catalogue_category,
@@ -29,6 +30,22 @@
 logger.info("Logging now setup")
 
 
+@app.exception_handler(BaseAPIException)
+async def custom_base_api_exception_handler(_: Request, exc: BaseAPIException) -> JSONResponse:
+    """
+    Custom exception handler for FastAPI to handle `BaseAPIException`'s.
+
+    This handler ensures that these exceptions return the appropriate response code and generalised detail
+    while logging any specific detail.
+
+    :param _: Unused.
+    :param exc: The exception object representing the `BaseAPIException`.
+    :return: A JSON response with exception details.
+    """
+    logger.exception(exc.detail)
+    return JSONResponse(content={"detail": exc.response_detail}, status_code=exc.status_code)
+
+
 @app.exception_handler(Exception)
 async def custom_general_exception_handler(_: Request, exc: Exception) -> JSONResponse:
     """

inventory_management_system_api/repositories/system.py

@@ -10,11 +10,7 @@
 
 from inventory_management_system_api.core.custom_object_id import CustomObjectId
 from inventory_management_system_api.core.database import DatabaseDep
-from inventory_management_system_api.core.exceptions import (
-    DuplicateRecordError,
-    InvalidActionError,
-    MissingRecordError,
-)
+from inventory_management_system_api.core.exceptions import DuplicateRecordError, InvalidActionError, MissingRecordError
 from inventory_management_system_api.models.system import SystemIn, SystemOut
 from inventory_management_system_api.repositories import utils
 from inventory_management_system_api.schemas.breadcrumbs import BreadcrumbsGetSchema
@@ -53,10 +49,17 @@ def create(self, system: SystemIn, session: Optional[ClientSession] = None) -> S
         """
         parent_id = str(system.parent_id) if system.parent_id else None
         if parent_id and not self.get(parent_id, session=session):
-            raise MissingRecordError(f"No parent system found with ID: {parent_id}")
+            raise MissingRecordError(
+                f"No parent system found with ID: {parent_id}",
+                "The specified parent system does not exist",
+                status_code=422,
+            )
 
         if self._is_duplicate_system(parent_id, system.code, session=session):
-            raise DuplicateRecordError("Duplicate system found within the parent system")
+            raise DuplicateRecordError(
+                "Duplicate system found within the parent system",
+                "A system with the same name already exists within the parent system",
+            )
 
         logger.info("Inserting the new system into the database")
         result = self._systems_collection.insert_one(system.model_dump(), session=session)

inventory_management_system_api/routers/v1/system.py

@@ -6,7 +6,7 @@
 import logging
 from typing import Annotated, Optional
 
-from fastapi import APIRouter, Depends, HTTPException, Path, Request, Query, status
+from fastapi import APIRouter, Depends, HTTPException, Path, Query, Request, status
 
 from inventory_management_system_api.core.config import config
 from inventory_management_system_api.core.exceptions import (
@@ -40,17 +40,9 @@ def create_system(system: SystemPostSchema, system_service: SystemServiceDep) ->
     # pylint: disable=missing-function-docstring
     logger.info("Creating a new system")
     logger.debug("System data: %s", system)
-    try:
-        system = system_service.create(system)
-        return SystemSchema(**system.model_dump())
-    except (MissingRecordError, InvalidObjectIdError) as exc:
-        message = "The specified parent system does not exist"
-        logger.exception(message)
-        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=message) from exc
-    except DuplicateRecordError as exc:
-        message = "A system with the same name already exists within the parent system"
-        logger.exception(message)
-        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=message) from exc
+
+    system = system_service.create(system)
+    return SystemSchema(**system.model_dump())
 
 
 @router.get(path="", summary="Get systems", response_description="List of systems")

inventory_management_system_api/services/system.py

@@ -8,7 +8,12 @@
 from fastapi import Depends
 
 from inventory_management_system_api.core.config import config
-from inventory_management_system_api.core.exceptions import MissingRecordError, ChildElementsExistError
+from inventory_management_system_api.core.custom_object_id import CustomObjectId
+from inventory_management_system_api.core.exceptions import (
+    ChildElementsExistError,
+    InvalidObjectIdError,
+    MissingRecordError,
+)
 from inventory_management_system_api.core.object_storage_api_client import ObjectStorageAPIClient
 from inventory_management_system_api.models.system import SystemIn, SystemOut
 from inventory_management_system_api.repositories.system import SystemRepo
@@ -42,8 +47,8 @@ def create(self, system: SystemPostSchema) -> SystemOut:
         parent_id = system.parent_id
 
         code = utils.generate_code(system.name, "system")
-        return self._system_repository.create(
-            SystemIn(
+        try:
+            system_in = SystemIn(
                 parent_id=parent_id,
                 description=system.description,
                 name=system.name,
@@ -52,7 +57,11 @@ def create(self, system: SystemPostSchema) -> SystemOut:
                 importance=system.importance,
                 code=code,
             )
-        )
+        except InvalidObjectIdError as exc:
+            # Provide more specific detail
+            exc.response_detail = "The specified parent system does not exist"
+            raise exc
+        return self._system_repository.create(system_in)
 
     def get(self, system_id: str) -> Optional[SystemOut]:
         """