Merge pull request #78 from ral-facilities/endpoint-for-editing-catal…

…ogue-items-#9

Partially edit catalogue items

inventory_management_system_api/repositories/catalogue_category.py

@@ -97,7 +97,7 @@ def get(self, catalogue_category_id: str) -> Optional[CatalogueCategoryOut]:
             return CatalogueCategoryOut(**catalogue_category)
         return None
 
-    def update(self, catalogue_category_id: str, catalogue_category: CatalogueCategoryIn):
+    def update(self, catalogue_category_id: str, catalogue_category: CatalogueCategoryIn) -> CatalogueCategoryOut:
         """
         Update a catalogue category by its ID in a MongoDB database.
 

inventory_management_system_api/repositories/catalogue_item.py

@@ -10,7 +10,6 @@
 
 from inventory_management_system_api.core.custom_object_id import CustomObjectId
 from inventory_management_system_api.core.database import get_database
-from inventory_management_system_api.core.exceptions import DuplicateRecordError
 from inventory_management_system_api.models.catalogue_item import CatalogueItemOut, CatalogueItemIn
 
 logger = logging.getLogger()
@@ -34,15 +33,9 @@ def create(self, catalogue_item: CatalogueItemIn) -> CatalogueItemOut:
         """
         Create a new catalogue item in a MongoDB database.
 
-        The method checks if a duplicate catalogue item is found within the catalogue category.
-
         :param catalogue_item: The catalogue item to be created.
         :return: The created catalogue item.
-        :raises DuplicateRecordError: If a duplicate catalogue item is found within the catalogue category.
         """
-        if self._is_duplicate_catalogue_item(str(catalogue_item.catalogue_category_id), catalogue_item.name):
-            raise DuplicateRecordError("Duplicate catalogue item found within the catalogue category")
-
         logger.info("Inserting the new catalogue item into the database")
         result = self._collection.insert_one(catalogue_item.dict())
         catalogue_item = self.get(str(result.inserted_id))
@@ -62,6 +55,23 @@ def get(self, catalogue_item_id: str) -> Optional[CatalogueItemOut]:
             return CatalogueItemOut(**catalogue_item)
         return None
 
+    def update(self, catalogue_item_id: str, catalogue_item: CatalogueItemIn) -> CatalogueItemOut:
+        """
+        Update a catalogue item by its ID in a MongoDB database.
+
+        :param catalogue_item_id: The ID of the catalogue item to update.
+        :param catalogue_item: The catalogue item containing the update data.
+        :return: The updated catalogue item.
+        """
+        catalogue_item_id = CustomObjectId(catalogue_item_id)
+        # pylint: disable=fixme
+        # TODO - (when the relevant item logic is implemented) check if catalogue item has children elements if the
+        #  `catalogue_category_id` is being updated.
+        logger.info("Updating catalogue item with ID: %s in the database", catalogue_item_id)
+        self._collection.update_one({"_id": catalogue_item_id}, {"$set": catalogue_item.dict()})
+        catalogue_item = self.get(str(catalogue_item_id))
+        return catalogue_item
+
     def list(self, catalogue_category_id: Optional[str]) -> List[CatalogueItemOut]:
         """
         Retrieve all catalogue items from a MongoDB.
@@ -83,15 +93,3 @@ def list(self, catalogue_category_id: Optional[str]) -> List[CatalogueItemOut]:
 
         catalogue_items = self._collection.find(query)
         return [CatalogueItemOut(**catalogue_item) for catalogue_item in catalogue_items]
-
-    def _is_duplicate_catalogue_item(self, catalogue_category_id: str, name: str) -> bool:
-        """
-        Check if a catalogue item with the same name already exists within the catalogue category.
-
-        :param catalogue_category_id: The ID of the catalogue category to check for duplicates in.
-        :return: `True` if a duplicate catalogue item is found, `False` otherwise.
-        """
-        logger.info("Checking if catalogue item with name '%s' already exists within the category", name)
-        catalogue_category_id = CustomObjectId(catalogue_category_id)
-        count = self._collection.count_documents({"catalogue_category_id": catalogue_category_id, "name": name})
-        return count > 0

inventory_management_system_api/routers/v1/catalogue_item.py

@@ -10,12 +10,15 @@
 from inventory_management_system_api.core.exceptions import (
     MissingRecordError,
     InvalidObjectIdError,
-    DuplicateRecordError,
     NonLeafCategoryError,
     InvalidCatalogueItemPropertyTypeError,
     MissingMandatoryCatalogueItemProperty,
 )
-from inventory_management_system_api.schemas.catalogue_item import CatalogueItemSchema, CatalogueItemPostRequestSchema
+from inventory_management_system_api.schemas.catalogue_item import (
+    CatalogueItemSchema,
+    CatalogueItemPostRequestSchema,
+    CatalogueItemPatchRequestSchema,
+)
 from inventory_management_system_api.services.catalogue_item import CatalogueItemService
 
 logger = logging.getLogger()
@@ -52,18 +55,15 @@ def get_catalogue_item(
 ) -> CatalogueItemSchema:
     # pylint: disable=missing-function-docstring
     logger.info("Getting catalogue item with ID: %s", catalogue_item_id)
+    message = "A catalogue item with such ID was not found"
     try:
         catalogue_item = catalogue_item_service.get(catalogue_item_id)
         if not catalogue_item:
-            raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND, detail="The requested catalogue item was not found"
-            )
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=message)
         return CatalogueItemSchema(**catalogue_item.dict())
     except InvalidObjectIdError as exc:
         logger.exception("The ID is not a valid ObjectId value")
-        raise HTTPException(
-            status_code=status.HTTP_404_NOT_FOUND, detail="The requested catalogue item was not found"
-        ) from exc
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=message) from exc
 
 
 @router.post(
@@ -85,13 +85,47 @@ def create_catalogue_item(
         logger.exception(str(exc))
         raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc
     except (MissingRecordError, InvalidObjectIdError) as exc:
-        message = "The specified catalogue category ID does not exist in the database"
+        message = "The specified catalogue category ID 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 catalogue item with the same name already exists within the catalogue category"
+    except NonLeafCategoryError as exc:
+        message = "Adding a catalogue item to a non-leaf catalogue category is not allowed"
         logger.exception(message)
         raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=message) from exc
+
+
+@router.patch(
+    path="/{catalogue_item_id}",
+    summary="Update a catalogue item partially by ID",
+    response_description="Catalogue item updated successfully",
+)
+def partial_update_catalogue_item(
+    catalogue_item: CatalogueItemPatchRequestSchema,
+    catalogue_item_id: str = Path(description="The ID of the catalogue item to update"),
+    catalogue_item_service: CatalogueItemService = Depends(),
+) -> CatalogueItemSchema:
+    # pylint: disable=missing-function-docstring
+    logger.info("Partially updating catalogue item with ID: %s", catalogue_item_id)
+    logger.debug("Catalogue item data: %s", catalogue_item)
+    try:
+        updated_catalogue_item = catalogue_item_service.update(catalogue_item_id, catalogue_item)
+        return CatalogueItemSchema(**updated_catalogue_item.dict())
+    except (InvalidCatalogueItemPropertyTypeError, MissingMandatoryCatalogueItemProperty) as exc:
+        logger.exception(str(exc))
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc
+    except (MissingRecordError, InvalidObjectIdError) as exc:
+        if (
+            catalogue_item.catalogue_category_id
+            and catalogue_item.catalogue_category_id in str(exc)
+            or "catalogue category" in str(exc).lower()
+        ):
+            message = "The specified catalogue category ID does not exist"
+            logger.exception(message)
+            raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=message) from exc
+
+        message = "A catalogue item with such ID was not found"
+        logger.exception(message)
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=message) from exc
     except NonLeafCategoryError as exc:
         message = "Adding a catalogue item to a non-leaf catalogue category is not allowed"
         logger.exception(message)

inventory_management_system_api/schemas/catalogue_item.py

@@ -47,6 +47,19 @@ class CatalogueItemPostRequestSchema(BaseModel):
     manufacturer: ManufacturerSchema = Field(description="The details of the manufacturer")
 
 
+class CatalogueItemPatchRequestSchema(CatalogueItemPostRequestSchema):
+    """
+    Schema model for a catalogue item update request.
+    """
+
+    catalogue_category_id: Optional[str] = Field(
+        description="The ID of the catalogue category that the catalogue item belongs to"
+    )
+    name: Optional[str] = Field(description="The name of the catalogue item")
+    description: Optional[str] = Field(description="The catalogue item description")
+    manufacturer: Optional[ManufacturerSchema] = Field(description="The details of the manufacturer")
+
+
 class CatalogueItemSchema(CatalogueItemPostRequestSchema):
     """
     Schema model for a catalogue item response.

inventory_management_system_api/services/catalogue_category.py

@@ -113,32 +113,22 @@ def update(
         if not stored_catalogue_category:
             raise MissingRecordError(f"No catalogue category found with ID: {catalogue_category_id}")
 
-        if "name" in update_data and update_data["name"] != stored_catalogue_category.name:
-            stored_catalogue_category.name = update_data["name"]
-            stored_catalogue_category.code = utils.generate_code(stored_catalogue_category.name, "catalogue category")
-            stored_catalogue_category.path = utils.generate_path(
-                stored_catalogue_category.parent_path, stored_catalogue_category.code, "catalogue category"
+        if "name" in update_data and catalogue_category.name != stored_catalogue_category.name:
+            update_data["code"] = utils.generate_code(catalogue_category.name, "catalogue category")
+            update_data["path"] = utils.generate_path(
+                stored_catalogue_category.parent_path, update_data["code"], "catalogue category"
             )
 
-        if "parent_id" in update_data and update_data["parent_id"] != stored_catalogue_category.parent_id:
-            stored_catalogue_category.parent_id = update_data["parent_id"]
-            parent_catalogue_category = (
-                self.get(stored_catalogue_category.parent_id) if stored_catalogue_category.parent_id else None
-            )
-            stored_catalogue_category.parent_path = parent_catalogue_category.path if parent_catalogue_category else "/"
-            stored_catalogue_category.path = utils.generate_path(
-                stored_catalogue_category.parent_path, stored_catalogue_category.code, "catalogue category"
-            )
+        if "parent_id" in update_data and catalogue_category.parent_id != stored_catalogue_category.parent_id:
+            parent_catalogue_category = self.get(catalogue_category.parent_id) if catalogue_category.parent_id else None
+            update_data["parent_path"] = parent_catalogue_category.path if parent_catalogue_category else "/"
+            code = update_data["code"] if "code" in update_data else stored_catalogue_category.code
+            update_data["path"] = utils.generate_path(update_data["parent_path"], code, "catalogue category")
 
             if parent_catalogue_category and parent_catalogue_category.is_leaf:
                 raise LeafCategoryError("Cannot add catalogue category to a leaf parent catalogue category")
 
-        if "is_leaf" in update_data:
-            stored_catalogue_category.is_leaf = update_data["is_leaf"]
-
-        if "catalogue_item_properties" in update_data:
-            stored_catalogue_category.catalogue_item_properties = update_data["catalogue_item_properties"]
-
+        stored_catalogue_category = stored_catalogue_category.copy(update=update_data)
         return self._catalogue_category_repository.update(
             catalogue_category_id, CatalogueCategoryIn(**stored_catalogue_category.dict())
         )