Add endpoint to set spares definition #413

inventory_management_system_api/core/exceptions.py

@@ -82,6 +82,12 @@ class PartOfItemError(DatabaseError):
     """
 
 
+class PartOfSettingError(DatabaseError):
+    """
+    Exception raised when attempting to delete an entity being referred to by a setting
+    """
+
+
 class DatabaseIntegrityError(DatabaseError):
     """
     Exception raised when something is found in the database that shouldn't have been

inventory_management_system_api/main.py

@@ -17,6 +17,7 @@
     catalogue_item,
     item,
     manufacturer,
+    setting,
     system,
     unit,
     usage_status,
@@ -92,6 +93,7 @@ def get_router_dependencies() -> list:
 app.include_router(system.router, dependencies=router_dependencies)
 app.include_router(unit.router, dependencies=router_dependencies)
 app.include_router(usage_status.router, dependencies=router_dependencies)
+app.include_router(setting.router, dependencies=router_dependencies)
 
 
 @app.get("/")

inventory_management_system_api/models/manufacturer.py

@@ -1,5 +1,5 @@
 """
-Module for defining the database models for representing manufacturer.
+Module for defining the database models for representing manufacturers.
 """
 
 from typing import Optional

inventory_management_system_api/models/setting.py

@@ -0,0 +1,65 @@
+"""
+Module for defining the database models for representing settings.
+"""
+
+from abc import ABC, abstractmethod
+from typing import ClassVar
+
+from pydantic import BaseModel, Field
+
+from inventory_management_system_api.models.custom_object_id_data_types import CustomObjectIdField, StringObjectIdField
+from inventory_management_system_api.models.usage_status import UsageStatusOut
+
+
+class SettingInBase(BaseModel, ABC):
+    """
+    Base input database model for a setting.
+    """
+
+    @property
+    @staticmethod
+    @abstractmethod
+    def SETTING_ID() -> str:  # pylint: disable=invalid-name
+        """ID of the setting. Ensures this value can be obtained from the class type itself as a static variable."""
+
+
+class SettingOutBase(SettingInBase):
+    """
+    Base output database model for a setting.
+    """
+
+    id: StringObjectIdField = Field(alias="_id")
+
+
+class SparesDefinitionUsageStatusIn(BaseModel):
+    """
+    Input database model for a usage status in a spares definition.
+    """
+
+    id: CustomObjectIdField
+
+
+class SparesDefinitionUsageStatusOut(BaseModel):
+    """
+    Output database model for a usage status in a spares definition.
+    """
+
+    id: StringObjectIdField
+
+
+class SparesDefinitionIn(SettingInBase):
+    """
+    Input database model for a spares definition.
+    """
+
+    SETTING_ID: ClassVar[str] = "spares_definition"
+
+    usage_statuses: list[SparesDefinitionUsageStatusIn]
+
+
+class SparesDefinitionOut(SparesDefinitionIn, SettingOutBase):
+    """
+    Output database model for a spares definition.
+    """
+
+    usage_statuses: list[UsageStatusOut]

inventory_management_system_api/models/system.py

@@ -1,5 +1,5 @@
 """
-Module for defining the database models for representing a System
+Module for defining the database models for representing systems.
 """
 
 from typing import Optional

inventory_management_system_api/repositories/setting.py

@@ -0,0 +1,106 @@
+"""
+Module for providing a repository for managing settings in a MongoDB database.
+"""
+
+import logging
+from typing import Optional, Type, TypeVar
+
+from pymongo.client_session import ClientSession
+from pymongo.collection import Collection
+
+from inventory_management_system_api.core.database import DatabaseDep
+from inventory_management_system_api.models.setting import SettingInBase, SettingOutBase, SparesDefinitionOut
+
+logger = logging.getLogger()
+
+# Template types for models inheriting from SettingIn/OutBase so this repo can be used generically for multiple settings
+SettingInBaseT = TypeVar("SettingInBaseT", bound=SettingInBase)
+SettingOutBaseT = TypeVar("SettingOutBaseT", bound=SettingOutBase)
+
+
+# Aggregation pipeline for getting the spares definition complete with usage status data
+SPARES_DEFINITION_GET_AGGREGATION_PIPELINE: list = [
+    # Only perform this on the relevant document
+    {"$match": {"_id": SparesDefinitionOut.SETTING_ID}},
+    # Deconstruct the usage statuses so can go through them one by one
+    {"$unwind": "$usage_statuses"},
+    # Find and store actual usage status data as 'statusDetails'
+    {
+        "$lookup": {
+            "from": "usage_statuses",
+            "localField": "usage_statuses.id",
+            "foreignField": "_id",
+            "as": "statusDetails",
+        }
+    },
+    {"$unwind": "$statusDetails"},
+    # Merge the two sets of documents together
+    {"$addFields": {"usage_statuses": {"$mergeObjects": ["$usage_statuses", "$statusDetails"]}}},
+    # Remove the temporary 'statusDetails' field as no longer needed
+    {"$unset": "statusDetails"},
+    # Reconstruct the original document by merging with the original fields
+    {
+        "$group": {
+            "_id": "$_id",
+            "usage_statuses": {"$push": "$usage_statuses"},
+            "otherFields": {"$first": "$$ROOT"},
+        }
+    },
+    {"$replaceRoot": {"newRoot": {"$mergeObjects": ["$otherFields", {"usage_statuses": "$usage_statuses"}]}}},
+]
+
+
+class SettingRepo:
+    """
+    Repository for managing settings in a MongoDB database.
+    """
+
+    def __init__(self, database: DatabaseDep) -> None:
+        """
+        Initialize the `SettingRepo` with a MongoDB database instance.
+
+        :param database: The database to use.
+        """
+        self._database = database
+        self._settings_collection: Collection = self._database.settings
+
+    def upsert(
+        self, setting: SettingInBaseT, out_model_type: Type[SettingOutBaseT], session: ClientSession = None
+    ) -> SettingOutBaseT:
+        """
+        Update or insert a setting in a MongoDB database depending on whether it already exists.
+
+        :param setting: Setting containing the fields to be updated. Also contains the ID for lookup.
+        :param out_model_type: The output type of the setting's model.
+        :param session: PyMongo ClientSession to use for database operations.
+        :return: The updated setting.
+        """
+
+        logger.info("Assigning setting with ID: %s in the database", setting.SETTING_ID)
+        self._settings_collection.update_one(
+            {"_id": setting.SETTING_ID}, {"$set": setting.model_dump(by_alias=True)}, upsert=True, session=session
+        )
+
+        return self.get(out_model_type=out_model_type, session=session)
+
+    def get(self, out_model_type: Type[SettingOutBaseT], session: ClientSession = None) -> Optional[SettingOutBaseT]:
+        """
+        Retrieve a setting from a MongoDB database.
+
+        :param out_model_type: The output type of the setting's model. Also contains the ID for lookup.
+        :param session: PyMongo ClientSession to use for database operations.
+        :return: Retrieved setting or `None` if not found.
+        """
+
+        if out_model_type is SparesDefinitionOut:
+            # The spares definition contains a list of usage statuses - use an aggregate query here to obtain
+            # the actual usage status entities instead of just their stored ID
+
+            result = list(self._settings_collection.aggregate(SPARES_DEFINITION_GET_AGGREGATION_PIPELINE))
+            setting = result[0] if len(result) > 0 else None
+        else:
+            setting = self._settings_collection.find_one({"_id": out_model_type.SETTING_ID}, session=session)
+
+        if setting is not None:
+            return out_model_type(**setting)
+        return None

inventory_management_system_api/repositories/usage_status.py

@@ -10,10 +10,15 @@
 
 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, MissingRecordError, PartOfItemError
+from inventory_management_system_api.core.exceptions import (
+    DuplicateRecordError,
+    MissingRecordError,
+    PartOfItemError,
+    PartOfSettingError,
+)
+from inventory_management_system_api.models.setting import SparesDefinitionIn
 from inventory_management_system_api.models.usage_status import UsageStatusIn, UsageStatusOut
 
-
 logger = logging.getLogger()
 
 
@@ -31,6 +36,7 @@ def __init__(self, database: DatabaseDep) -> None:
         self._database = database
         self._usage_statuses_collection: Collection = self._database.usage_statuses
         self._items_collection: Collection = self._database.items
+        self._settings_collection: Collection = self._database.settings
 
     def create(self, usage_status: UsageStatusIn, session: ClientSession = None) -> UsageStatusOut:
         """
@@ -89,8 +95,10 @@ def delete(self, usage_status_id: str, session: ClientSession = None) -> None:
         :raises MissingRecordError: if supplied usage status ID does not exist in the database
         """
         usage_status_id = CustomObjectId(usage_status_id)
-        if self._is_usage_status_in_item(str(usage_status_id), session=session):
+        if self._is_usage_status_in_item(usage_status_id, session=session):
             raise PartOfItemError(f"The usage status with ID {str(usage_status_id)} is a part of an Item")
+        if self._is_usage_status_in_setting(usage_status_id, session=session):
+            raise PartOfSettingError(f"The usage status with ID {str(usage_status_id)} is used in the settings")
 
         logger.info("Deleting usage status with ID %s from the database", usage_status_id)
         result = self._usage_statuses_collection.delete_one({"_id": usage_status_id}, session=session)
@@ -115,11 +123,25 @@ def _is_duplicate_usage_status(
         return usage_status is not None
 
     def _is_usage_status_in_item(self, usage_status_id: str, session: ClientSession = None) -> bool:
-        """Checks to see if any of the items in the database have a specific usage status ID
+        """Checks to see if any of the items in the database have a specific usage status ID.
 
-        :param usage_status_id: The ID of the usage status that is looked for
-        :param session: PyMongo ClientSession to use for database operations
-        :return: `True` if 1 or more items have the usage status ID, `False` otherwise
+        :param usage_status_id: The ID of the usage status that is looked for.
+        :param session: PyMongo ClientSession to use for database operations.
+        :return: `True` if 1 or more items have the usage status ID, `False` otherwise.
         """
-        usage_status_id = CustomObjectId(usage_status_id)
         return self._items_collection.find_one({"usage_status_id": usage_status_id}, session=session) is not None
+
+    def _is_usage_status_in_setting(self, usage_status_id: CustomObjectId, session: ClientSession = None) -> bool:
+        """Checks to see if any of the settings in the database refer to a specific usage status ID.
+
+        :param usage_status_id: The ID of the usage status that is looked for.
+        :param session: PyMongo ClientSession to use for database operations.
+        :return: `True` if 1 or more items have the usage status ID, `False` otherwise.
+        """
+        return (
+            self._settings_collection.find_one(
+                {"_id": SparesDefinitionIn.SETTING_ID, "usage_statuses": {"$elemMatch": {"id": usage_status_id}}},
+                session=session,
+            )
+            is not None
+        )

inventory_management_system_api/routers/v1/setting.py

@@ -0,0 +1,39 @@
+"""
+Module for providing an API router which defines routes for managing settings using the `SettingService` service.
+"""
+
+import logging
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, HTTPException, status
+
+from inventory_management_system_api.core.exceptions import InvalidObjectIdError, MissingRecordError
+from inventory_management_system_api.schemas.setting import SparesDefinitionPutSchema, SparesDefinitionSchema
+from inventory_management_system_api.services.setting import SettingService
+
+logger = logging.getLogger()
+
+router = APIRouter(prefix="/v1/settings", tags=["settings"])
+
+SettingServiceDep = Annotated[SettingService, Depends(SettingService)]
+
+
+@router.put(
+    path="/spares_definition",
+    summary="Set the definition of a spare",
+    response_description="Spares definition updated successfully",
+)
+def set_spares_definition(
+    spares_definition: SparesDefinitionPutSchema, setting_service: SettingServiceDep
+) -> SparesDefinitionSchema:
+    # pylint: disable=missing-function-docstring
+    logger.info("Setting spares definition")
+    logger.debug("Spares definition data: %s", spares_definition)
+
+    try:
+        updated_spares_definition = setting_service.set_spares_definition(spares_definition)
+        return SparesDefinitionSchema(**updated_spares_definition.model_dump())
+    except (MissingRecordError, InvalidObjectIdError) as exc:
+        message = "A specified usage status does not exist"
+        logger.exception(message)
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=message) from exc

inventory_management_system_api/routers/v1/system.py

@@ -1,5 +1,5 @@
 """
-Module for providing an API router which defines routes for managing Systems using the `SystemService`
+Module for providing an API router which defines routes for managing systems using the `SystemService`
 service.
 """
 

inventory_management_system_api/routers/v1/usage_status.py

@@ -13,6 +13,7 @@
     InvalidObjectIdError,
     MissingRecordError,
     PartOfItemError,
+    PartOfSettingError,
 )
 from inventory_management_system_api.schemas.usage_status import UsageStatusPostSchema, UsageStatusSchema
 from inventory_management_system_api.services.usage_status import UsageStatusService
@@ -101,3 +102,7 @@ def delete_usage_status(
         message = "The specified usage status is part of an Item"
         logger.exception(message)
         raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=message) from exc
+    except PartOfSettingError as exc:
+        message = "The specified usage status is part of a setting"
+        logger.exception(message)
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=message) from exc

inventory_management_system_api/schemas/catalogue_category.py

@@ -96,15 +96,15 @@ def check_valid_allowed_values(
         cls, allowed_values: Optional[AllowedValuesSchema], property_data: dict[str, Any]
     ) -> None:
         """
-        Checks allowed_values against its parent property raising an error if its invalid
+        Checks `allowed_values` against its parent property raising an error if its invalid
 
-        :param allowed_values: The value of the `allowed_values` field.
+        :param `allowed_values`: The value of the `allowed_values` field.
         :param property_data: Property data to validate the allowed values against.
         :raises ValueError:
-            - If the allowed_values has been given a value and the property type is a `boolean`
-            - If the allowed_values is of type 'list' and 'values' contains any with a different type to the property
-              type
-            - If the allowed_values is of type 'list' and 'values' contains any duplicates
+            - If the `allowed_values` has been given a value and the property type is a `boolean`.
+            - If the `allowed_values` is of type `list` and `values` contains any with a different type to the property
+              type.
+            - If the `allowed_values` is of type `list` and `values` contains any duplicates.
         """
         if allowed_values is not None and "type" in property_data:
             # Ensure the type is not boolean
@@ -143,7 +143,7 @@ def validate_allowed_values(
         """
         Validator for the `allowed_values` field.
 
-        Ensures the allowed_values are valid given the rest of the property schema.
+        Ensures the `allowed_values` are valid given the rest of the property schema.
 
         :param allowed_values: The value of the `allowed_values` field.
         :param info: Validation info from pydantic.