Skip to content

Commit

Permalink
Merge pull request #62 from ral-facilities/endpoint-for-editing-catal…
Browse files Browse the repository at this point in the history
…ogue-category-#3

Partially edit catalogue categories
  • Loading branch information
VKTB authored Sep 29, 2023
2 parents b8a6d66 + 7d3dfb2 commit 3a1d6cb
Show file tree
Hide file tree
Showing 22 changed files with 2,266 additions and 330 deletions.
4 changes: 2 additions & 2 deletions inventory_management_system_api/core/custom_object_id.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,9 @@ def __init__(self, value: str):
:raises InvalidObjectIdError: If the string value is an invalid `ObjectId`.
"""
if not isinstance(value, str):
raise InvalidObjectIdError("ObjectId value must be a string")
raise InvalidObjectIdError(f"ObjectId value '{value}' must be a string")

if not ObjectId.is_valid(value):
raise InvalidObjectIdError("Invalid ObjectId value")
raise InvalidObjectIdError(f"Invalid ObjectId value '{value}'")

super().__init__(value)
2 changes: 1 addition & 1 deletion inventory_management_system_api/core/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -53,5 +53,5 @@ class MissingRecordError(DatabaseError):

class ChildrenElementsExistError(DatabaseError):
"""
Exception raised when attempting to delete a catalogue category that has children elements.
Exception raised when attempting to delete or update a catalogue category that has children elements.
"""
20 changes: 18 additions & 2 deletions inventory_management_system_api/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,10 +3,11 @@
"""
import logging

from fastapi import FastAPI, Request
from fastapi import FastAPI, Request, status
from fastapi.exception_handlers import request_validation_exception_handler
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse

from inventory_management_system_api.core.config import config
from inventory_management_system_api.core.logger_setup import setup_logger
Expand All @@ -20,8 +21,22 @@
logger.info("Logging now setup")


@app.exception_handler(Exception)
async def custom_general_exception_handler(_: Request, exc: Exception) -> JSONResponse:
"""
Custom exception handler for FastAPI to handle uncaught exceptions. It logs the error and returns an appropriate
response.
:param _: Unused
:param exc: The exception object that triggered this handler.
:return: A JSON response indicating that something went wrong.
"""
logger.exception(exc)
return JSONResponse(content={"detail": "Something went wrong"}, status_code=status.HTTP_500_INTERNAL_SERVER_ERROR)


@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
async def custom_validation_exception_handler(request: Request, exc: RequestValidationError) -> JSONResponse:
"""
Custom exception handler for FastAPI to handle `RequestValidationError`.
Expand All @@ -31,6 +46,7 @@ async def validation_exception_handler(request: Request, exc: RequestValidationE
:param request: The incoming HTTP request that caused the validation error.
:param exc: The exception object representing the validation error.
:return: A JSON response with validation error details.
"""
logger.exception(exc)
return await request_validation_exception_handler(request, exc)
Expand Down
72 changes: 27 additions & 45 deletions inventory_management_system_api/models/catalogue_category.py
Original file line number Diff line number Diff line change
@@ -1,52 +1,11 @@
"""
Module for defining the database models for representing catalogue categories.
"""
from typing import Optional, List
from typing import Optional, List, Dict, Any

from bson import ObjectId
from pydantic import BaseModel, Field
from pydantic import BaseModel, Field, validator

from inventory_management_system_api.core.custom_object_id import CustomObjectId


class CustomObjectIdField(ObjectId):
"""
Custom field for handling MongoDB ObjectId validation.
"""

@classmethod
def __get_validators__(cls):
yield cls.validate

@classmethod
def validate(cls, value: str) -> CustomObjectId:
"""
Validate if the string value is a valid `ObjectId`.
:param value: The string value to be validated.
:return: The validated `ObjectId`.
"""
return CustomObjectId(value)


class StringObjectIdField(str):
"""
Custom field for handling MongoDB ObjectId as string.
"""

@classmethod
def __get_validators__(cls):
yield cls.validate

@classmethod
def validate(cls, value: ObjectId) -> str:
"""
Convert the `ObjectId` value to string.
:param value: The `ObjectId` value to be converted.
:return: The converted `ObjectId` as a string.
"""
return str(value)
from inventory_management_system_api.models.custom_object_id_data_types import CustomObjectIdField, StringObjectIdField


class CatalogueItemProperty(BaseModel):
Expand All @@ -71,7 +30,30 @@ class CatalogueCategoryIn(BaseModel):
path: str
parent_path: str
parent_id: Optional[CustomObjectIdField] = None
catalogue_item_properties: List[CatalogueItemProperty]
catalogue_item_properties: List[CatalogueItemProperty] = []

@validator("catalogue_item_properties", pre=True, always=True)
@classmethod
def validate_catalogue_item_properties(
cls, catalogue_item_properties: List[CatalogueItemProperty] | None, values: Dict[str, Any]
) -> List[CatalogueItemProperty] | List:
"""
Validator for the `catalogue_item_properties` field that runs after field assignment but before type validation.
If the value is `None`, it replaces it with an empty list allowing for catalogue categories without catalogue
item properties to be created. If the category is a non-leaf category and if catalogue item properties are
supplied, it replaces it with an empty list because they cannot have properties.
:param catalogue_item_properties: The list of catalogue item properties.
:param values: The values of the model fields.
:return: The list of catalogue item properties or an empty list.
"""
if catalogue_item_properties is None or (
"is_leaf" in values and values["is_leaf"] is False and catalogue_item_properties
):
catalogue_item_properties = []

return catalogue_item_properties


class CatalogueCategoryOut(CatalogueCategoryIn):
Expand Down
33 changes: 30 additions & 3 deletions inventory_management_system_api/models/catalogue_item.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,9 +3,19 @@
"""
from typing import Optional, List, Any

from pydantic import BaseModel, Field
from pydantic import BaseModel, Field, validator

from inventory_management_system_api.models.catalogue_category import CustomObjectIdField, StringObjectIdField
from inventory_management_system_api.models.custom_object_id_data_types import CustomObjectIdField, StringObjectIdField


class Manufacturer(BaseModel):
"""
Model representing a catalogue item manufacturer.
"""

name: str
address: str
web_url: str


class Property(BaseModel):
Expand All @@ -26,7 +36,24 @@ class CatalogueItemIn(BaseModel):
catalogue_category_id: CustomObjectIdField
name: str
description: str
properties: List[Property]
properties: List[Property] = []
manufacturer: Manufacturer

@validator("properties", pre=True, always=True)
@classmethod
def validate_properties(cls, properties: List[Property] | None) -> List[Property] | List:
"""
Validator for the `properties` field that runs after field assignment but before type validation.
If the value is `None`, it replaces it with an empty list allowing for catalogue items without properties to be
created.
:param properties: The list of properties.
:return: The list of properties or an empty list.
"""
if properties is None:
properties = []
return properties


class CatalogueItemOut(CatalogueItemIn):
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,46 @@
"""
Module for defining custom `ObjectId` data type classes used by Pydantic models.
"""
from bson import ObjectId

from inventory_management_system_api.core.custom_object_id import CustomObjectId


class CustomObjectIdField(ObjectId):
"""
Custom data type for handling MongoDB ObjectId validation.
"""

@classmethod
def __get_validators__(cls):
yield cls.validate

@classmethod
def validate(cls, value: str) -> CustomObjectId:
"""
Validate if the string value is a valid `ObjectId`.
:param value: The string value to be validated.
:return: The validated `ObjectId`.
"""
return CustomObjectId(value)


class StringObjectIdField(str):
"""
Custom data type for handling MongoDB ObjectId as string.
"""

@classmethod
def __get_validators__(cls):
yield cls.validate

@classmethod
def validate(cls, value: ObjectId) -> str:
"""
Convert the `ObjectId` value to string.
:param value: The `ObjectId` value to be converted.
:return: The converted `ObjectId` as a string.
"""
return str(value)
74 changes: 62 additions & 12 deletions inventory_management_system_api/repositories/catalogue_category.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,15 +32,16 @@ def __init__(self, database: Database = Depends(get_database)) -> None:
:param database: The database to use.
"""
self._database = database
self._collection: Collection = self._database.catalogue_categories
self._catalogue_categories_collection: Collection = self._database.catalogue_categories
self._catalogue_items_collection: Collection = self._database.catalogue_items

def create(self, catalogue_category: CatalogueCategoryIn) -> CatalogueCategoryOut:
"""
Create a new catalogue category in a MongoDB database.
If a parent catalogue category is specified by `parent_id`, the method checks if that exists
in the database and raises a `MissingRecordError` if it doesn't exist. It also checks if a duplicate catalogue
category is found within the parent catalogue category.
category is found within the parent catalogue category and raises a `DuplicateRecordError` if it is.
:param catalogue_category: The catalogue category to be created.
:return: The created catalogue category.
Expand All @@ -49,20 +50,22 @@ def create(self, catalogue_category: CatalogueCategoryIn) -> CatalogueCategoryOu
"""
parent_id = str(catalogue_category.parent_id) if catalogue_category.parent_id else None
if parent_id and not self.get(parent_id):
raise MissingRecordError(f"No catalogue category found with ID: {parent_id}")
raise MissingRecordError(f"No parent catalogue category found with ID: {parent_id}")

if self._is_duplicate_catalogue_category(parent_id, catalogue_category.code):
raise DuplicateRecordError("Duplicate catalogue category found within the parent catalogue category")

logger.info("Inserting the new catalogue category into the database")
result = self._collection.insert_one(catalogue_category.dict())
result = self._catalogue_categories_collection.insert_one(catalogue_category.dict())
catalogue_category = self.get(str(result.inserted_id))
return catalogue_category

def delete(self, catalogue_category_id: str) -> None:
"""
Delete a catalogue category by its ID from a MongoDB database. The method checks if the catalogue category has
children elements and raises a `ChildrenElementsExistError` if it does.
Delete a catalogue category by its ID from a MongoDB database.
The method checks if the catalogue category has children elements and raises a `ChildrenElementsExistError`
if it does.
:param catalogue_category_id: The ID of the catalogue category to delete.
:raises ChildrenElementsExistError: If the catalogue category has children elements.
Expand All @@ -75,7 +78,7 @@ def delete(self, catalogue_category_id: str) -> None:
)

logger.info("Deleting catalogue category with ID: %s from the database", catalogue_category_id)
result = self._collection.delete_one({"_id": catalogue_category_id})
result = self._catalogue_categories_collection.delete_one({"_id": catalogue_category_id})
if result.deleted_count == 0:
raise MissingRecordError(f"No catalogue category found with ID: {str(catalogue_category_id)}")

Expand All @@ -88,11 +91,51 @@ def get(self, catalogue_category_id: str) -> Optional[CatalogueCategoryOut]:
"""
catalogue_category_id = CustomObjectId(catalogue_category_id)
logger.info("Retrieving catalogue category with ID: %s from the database", catalogue_category_id)
catalogue_category = self._collection.find_one({"_id": catalogue_category_id})
catalogue_category = self._catalogue_categories_collection.find_one({"_id": catalogue_category_id})
if catalogue_category:
return CatalogueCategoryOut(**catalogue_category)
return None

def update(self, catalogue_category_id: str, catalogue_category: CatalogueCategoryIn):
"""
Update a catalogue category by its ID in a MongoDB database.
The method checks if the catalogue category has children elements and raises a `ChildrenElementsExistError`
if it does. If a parent catalogue category is specified by `parent_id`, the method checks if that exists in the
database and raises a `MissingRecordError` if it doesn't exist. It also checks if a duplicate catalogue category
is found within the parent catalogue category and raises a `DuplicateRecordError` if it is.
:param catalogue_category_id: The ID of the catalogue category to update.
:param catalogue_category: The catalogue category containing the update data.
:return: The updated catalogue category.
:raises ChildrenElementsExistError: If the catalogue category has children elements.
:raises MissingRecordError: If the parent catalogue category specified by `parent_id` doesn't exist.
:raises DuplicateRecordError: If a duplicate catalogue category is found within the parent catalogue category.
"""
catalogue_category_id = CustomObjectId(catalogue_category_id)
if self._has_children_elements(str(catalogue_category_id)):
raise ChildrenElementsExistError(
f"Catalogue category with ID {str(catalogue_category_id)} has children elements and cannot be updated"
)

parent_id = str(catalogue_category.parent_id) if catalogue_category.parent_id else None
if parent_id and not self.get(parent_id):
raise MissingRecordError(f"No parent catalogue category found with ID: {parent_id}")

stored_catalogue_category = self.get(str(catalogue_category_id))
if (
catalogue_category.name != stored_catalogue_category.name
or parent_id != stored_catalogue_category.parent_id
) and self._is_duplicate_catalogue_category(parent_id, catalogue_category.code):
raise DuplicateRecordError("Duplicate catalogue category found within the parent catalogue category")

logger.info("Updating catalogue category with ID: %s in the database", catalogue_category_id)
self._catalogue_categories_collection.update_one(
{"_id": catalogue_category_id}, {"$set": catalogue_category.dict()}
)
catalogue_category = self.get(str(catalogue_category_id))
return catalogue_category

def list(self, path: Optional[str], parent_path: Optional[str]) -> List[CatalogueCategoryOut]:
"""
Retrieve catalogue categories from a MongoDB database based on the provided filters.
Expand All @@ -115,7 +158,7 @@ def list(self, path: Optional[str], parent_path: Optional[str]) -> List[Catalogu
logger.info("%s matching the provided filter(s)", message)
logger.debug("Provided filter(s): %s", query)

catalogue_categories = self._collection.find(query)
catalogue_categories = self._catalogue_categories_collection.find(query)
return [CatalogueCategoryOut(**catalogue_category) for catalogue_category in catalogue_categories]

def _is_duplicate_catalogue_category(self, parent_id: Optional[str], code: str) -> bool:
Expand All @@ -126,22 +169,29 @@ def _is_duplicate_catalogue_category(self, parent_id: Optional[str], code: str)
:param code: The code of the catalogue category to check for duplicates.
:return: `True` if a duplicate catalogue category code is found, `False` otherwise.
"""
logger.info("Checking if catalogue category with code '%s' already exists within the category", code)
logger.info("Checking if catalogue category with code '%s' already exists within the parent category", code)
if parent_id:
parent_id = CustomObjectId(parent_id)

count = self._collection.count_documents({"parent_id": parent_id, "code": code})
count = self._catalogue_categories_collection.count_documents({"parent_id": parent_id, "code": code})
return count > 0

def _has_children_elements(self, catalogue_category_id: str) -> bool:
"""
Check if a catalogue category has children elements based on its ID.
Children elements in this case means whether or not a catalogue category has children catalogue categories or
children catalogue items.
:param catalogue_category_id: The ID of the catalogue category to check.
:return: True if the catalogue category has children elements, False otherwise.
"""
catalogue_category_id = CustomObjectId(catalogue_category_id)
logger.info("Checking if catalogue category with ID '%s' has children elements", catalogue_category_id)
# Check if it has catalogue categories
query = {"parent_id": catalogue_category_id}
count = self._collection.count_documents(query)
count = self._catalogue_categories_collection.count_documents(query)
# Check if it has catalogue items
query = {"catalogue_category_id": catalogue_category_id}
count = count + self._catalogue_items_collection.count_documents(query)
return count > 0
Loading

0 comments on commit 3a1d6cb

Please sign in to comment.