Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Refactor manufacturer tests #313

Merged
merged 9 commits into from
Jul 24, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
100 changes: 48 additions & 52 deletions inventory_management_system_api/repositories/manufacturer.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,79 +21,74 @@


class ManufacturerRepo:
"""Repository for managing manufacturer in MongoDb database"""
"""Repository for managing manufacturers in a MongoDb database."""

def __init__(self, database: DatabaseDep) -> None:
"""Initialize the `ManufacturerRepo` with MongoDB database instance
"""
Initialise the `ManufacturerRepo` with a MongoDB database instance.

:param database: The database to use.
"""

self._database = database
self._manufacturers_collection: Collection = self._database.manufacturers
self._catalogue_item_collection: Collection = self._database.catalogue_items
self._catalogue_items_collection: Collection = self._database.catalogue_items

def create(self, manufacturer: ManufacturerIn, session: ClientSession = None) -> ManufacturerOut:
"""
Create a new manufacturer in MongoDB database
Create a new manufacturer in a MongoDB database.

:param manufacturer: The manufacturer to be created
:param session: PyMongo ClientSession to use for database operations
:return: The created manufacturer
:raises DuplicateRecordError: If a duplicate manufacturer is found within collection
:param manufacturer: The manufacturer to be created.
:param session: PyMongo ClientSession to use for database operations.
:raises DuplicateRecordError: If a duplicate manufacturer is found.
:return: The created manufacturer.
"""

if self._is_duplicate_manufacturer(manufacturer.code, session=session):
raise DuplicateRecordError("Duplicate manufacturer found")

logger.info("Inserting new manufacturer into database")
logger.info("Inserting the new manufacturer into database")

result = self._manufacturers_collection.insert_one(manufacturer.model_dump(), session=session)
manufacturer = self.get(str(result.inserted_id), session=session)

return manufacturer

def get(self, manufacturer_id: str, session: ClientSession = None) -> Optional[ManufacturerOut]:
"""Retrieve a manufacturer from database by its id

:param manufacturer_id: The ID of the manufacturer
:param session: PyMongo ClientSession to use for database operations
:return: The retrieved manufacturer, or `None` if not found
"""
Retrieve a manufacturer by its ID from a MondoDB database.

:param manufacturer_id: The ID of the manufacturer to retrieve.
:param session: PyMongo ClientSession to use for database operations.
:return: The retrieved manufacturer, or `None` if not found.
"""
manufacturer_id = CustomObjectId(manufacturer_id)

logger.info("Retrieving manufacturer with ID %s from database", manufacturer_id)
logger.info("Retrieving manufacturer with ID: %s from database", manufacturer_id)
manufacturer = self._manufacturers_collection.find_one({"_id": manufacturer_id}, session=session)
if manufacturer:
return ManufacturerOut(**manufacturer)
return None

def list(self, session: ClientSession = None) -> List[ManufacturerOut]:
"""Get all manufacturers from database

:param session: PyMongo ClientSession to use for database operations
:return: List of manufacturers, or empty list if no manufacturers
"""
Retrieve all manufacturers from a MongoDB database.

logger.info("Getting all manufacturers from database")

:param session: PyMongo ClientSession to use for database operations.
:return: List of manufacturers, or empty list if no manufacturers.
"""
logger.info("Getting all manufacturers from the database")
manufacturers = self._manufacturers_collection.find(session=session)

return [ManufacturerOut(**manufacturer) for manufacturer in manufacturers]

def update(
self, manufacturer_id: str, manufacturer: ManufacturerIn, session: ClientSession = None
) -> ManufacturerOut:
"""Update manufacturer by its ID in database

:param: manufacturer_id: The id of the manufacturer to be updated
:param: manufacturer: The manufacturer with the update data
:param session: PyMongo ClientSession to use for database operations

:raises: DuplicateRecordError: if changed manufacturer name is a duplicate name
"""
Update manufacturer by its ID in a MongoDB database.

:returns: the updated manufacturer
:param manufacturer_id: The ID of the manufacturer to update.
:param manufacturer: The manufacturer containing the update data.
:param session: PyMongo ClientSession to use for database operations.
:raises DuplicateRecordError: If a duplicate manufacturer is found.
:return: The updated manufacturer.
"""
manufacturer_id = CustomObjectId(manufacturer_id)

Expand All @@ -102,7 +97,7 @@ def update(
if self._is_duplicate_manufacturer(manufacturer.code, manufacturer_id, session=session):
raise DuplicateRecordError("Duplicate manufacturer found")

logger.info("Updating manufacturer with ID %s", manufacturer_id)
logger.info("Updating manufacturer with ID: %s", manufacturer_id)
self._manufacturers_collection.update_one(
{"_id": manufacturer_id}, {"$set": manufacturer.model_dump()}, session=session
)
Expand All @@ -112,21 +107,21 @@ def update(

def delete(self, manufacturer_id: str, session: ClientSession = None) -> None:
"""
Delete a manufacturer by its ID from MongoDB database.
Checks if manufacturer is a part of an item, and does not delete if it is
Delete a manufacturer by its ID from a MongoDB database.

The method checks if the manufacturer is part of a catalogue item, and raises a `PartOfCatalogueItemError` if it
is.

:param manufacturer_id: The ID of the manufacturer to delete
:param session: PyMongo ClientSession to use for database operations
:raises PartOfCatalogueItemError: if manufacturer is a part of a catalogue item
:raises MissingRecordError: if supplied manufacturer ID does not exist in the database
:param manufacturer_id: The ID of the manufacturer to delete.
:param session: PyMongo ClientSession to use for database operations.
:raises PartOfCatalogueItemError: If the manufacturer is part of a catalogue item.
:raises MissingRecordError: If the manufacturer doesn't exist.
"""
manufacturer_id = CustomObjectId(manufacturer_id)
if self._is_manufacturer_in_catalogue_item(str(manufacturer_id), session=session):
raise PartOfCatalogueItemError(
f"The manufacturer with id {str(manufacturer_id)} is a part of a Catalogue Item"
)
raise PartOfCatalogueItemError(f"Manufacturer with ID '{str(manufacturer_id)}' is part of a catalogue item")

logger.info("Deleting manufacturer with ID %s from the database", manufacturer_id)
logger.info("Deleting manufacturer with ID: %s from the database", manufacturer_id)
result = self._manufacturers_collection.delete_one({"_id": manufacturer_id}, session=session)
if result.deleted_count == 0:
raise MissingRecordError(f"No manufacturer found with ID: {str(manufacturer_id)}")
Expand All @@ -135,12 +130,12 @@ def _is_duplicate_manufacturer(
self, code: str, manufacturer_id: CustomObjectId = None, session: ClientSession = None
) -> bool:
"""
Check if manufacturer with the same name already exists in the manufacturer collection
Check if a manufacturer with the same code already exists.

:param code: The code of the manufacturer to check for duplicates.
:param manufacturer_id: The ID of the manufacturer to check if the duplicate manufacturer found is itself.
:param session: PyMongo ClientSession to use for database operations
:return `True` if duplicate manufacturer, `False` otherwise
:param session: PyMongo ClientSession to use for database operations.
:return: `True` if a duplicate manufacturer is found, `False` otherwise.
"""
logger.info("Checking if manufacturer with code '%s' already exists", code)
manufacturer = self._manufacturers_collection.find_one(
Expand All @@ -149,13 +144,14 @@ def _is_duplicate_manufacturer(
return manufacturer is not None

def _is_manufacturer_in_catalogue_item(self, manufacturer_id: str, session: ClientSession = None) -> bool:
"""Checks to see if any of the documents in the database have a specific manufacturer id
"""
Check if a manufacturer is part of a catalogue item based on its ID.

:param manufacturer_id: The ID of the manufacturer that is looked for
:param session: PyMongo ClientSession to use for database operations
:return: Returns True if 1 or more documents have the manufacturer ID, false if none do
:param manufacturer_id: The ID of the manufacturer to check.
:param session: PyMongo ClientSession to use for database operations.
:return: `True` if the manufacturer is part of a catalogue item, `False` otherwise.
"""
manufacturer_id = CustomObjectId(manufacturer_id)
return (
self._catalogue_item_collection.find_one({"manufacturer_id": manufacturer_id}, session=session) is not None
self._catalogue_items_collection.find_one({"manufacturer_id": manufacturer_id}, session=session) is not None
)
42 changes: 19 additions & 23 deletions inventory_management_system_api/routers/v1/manufacturer.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
"""
Module for providing an API router which defines routes for managing manufacturer using the
`Manufacturer` service.
Module for providing an API router which defines routes for managing manufacturer using the `ManufacturerService`
service.
"""

import logging
Expand Down Expand Up @@ -30,8 +30,8 @@

@router.post(
path="",
summary="Create new manufacturer",
response_description="The new manufacturer",
summary="Create a new manufacturer",
response_description="The created manufacturer",
status_code=status.HTTP_201_CREATED,
)
def create_manufacturer(
Expand All @@ -40,26 +40,23 @@ def create_manufacturer(
# pylint: disable=missing-function-docstring
logger.info("Creating a new manufacturer")
logger.debug("Manufacturer data is %s", manufacturer)

try:
manufacturer = manufacturer_service.create(manufacturer)
return ManufacturerSchema(**manufacturer.model_dump())

except DuplicateRecordError as exc:
message = "A manufacturer with the same name has been found"
message = "A manufacturer with the same name already exists"
logger.exception(message)
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=message) from exc


@router.get(
path="",
summary="Get all manufacturers",
summary="Get manufacturers",
response_description="List of manufacturers",
)
def get_all_manufacturers(manufacturer_service: ManufacturerServiceDep) -> List[ManufacturerSchema]:
def get_manufacturers(manufacturer_service: ManufacturerServiceDep) -> List[ManufacturerSchema]:
# pylint: disable=missing-function-docstring
logger.info("Getting manufacturers")

manufacturers = manufacturer_service.list()
return [ManufacturerSchema(**manufacturer.model_dump()) for manufacturer in manufacturers]

Expand All @@ -69,52 +66,51 @@ def get_all_manufacturers(manufacturer_service: ManufacturerServiceDep) -> List[
summary="Get a manufacturer by ID",
response_description="Single manufacturer",
)
def get_one_manufacturer(
def get_manufacturer(
manufacturer_id: Annotated[str, Path(description="The ID of the manufacturer to be retrieved")],
manufacturer_service: ManufacturerServiceDep,
) -> ManufacturerSchema:
# pylint: disable=missing-function-docstring
logger.info("Getting manufacturer with ID %s", manufacturer_id)
logger.info("Getting manufacturer with ID: %s", manufacturer_id)
message = "Manufacturer not found"
try:
manufacturer = manufacturer_service.get(manufacturer_id)
if not manufacturer:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=message)
return ManufacturerSchema(**manufacturer.model_dump())
except InvalidObjectIdError as exc:
logger.exception("The ID is not a valid object value")
logger.exception("The ID is not a valid ObjectId value")
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=message) from exc

return ManufacturerSchema(**manufacturer.model_dump())


@router.patch(
path="/{manufacturer_id}",
summary="Update a manufacturer by its ID",
summary="Update a manufacturer partially by ID",
response_description="Manufacturer updated successfully",
)
def edit_manufacturer(
def partial_update_manufacturer(
manufacturer: ManufacturerPatchSchema,
manufacturer_id: Annotated[str, Path(description="The ID of the manufacturer that is to be updated")],
manufacturer_service: ManufacturerServiceDep,
) -> ManufacturerSchema:
# pylint: disable=missing-function-docstring
logger.info("Updating manufacturer with ID %s", manufacturer_id)
logger.info("Partially updating manufacturer with ID: %s", manufacturer_id)
try:
updated_manufacturer = manufacturer_service.update(manufacturer_id, manufacturer)
return ManufacturerSchema(**updated_manufacturer.model_dump())
except (MissingRecordError, InvalidObjectIdError) as exc:
message = "The specified manufacturer does not exist"
message = "Manufacturer not found"
logger.exception(message)
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=message) from exc
except DuplicateRecordError as exc:
message = "A manufacturer with the same name has been found"
message = "A manufacturer with the same name already exists"
logger.exception(message)
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=message) from exc


@router.delete(
path="/{manufacturer_id}",
summary="Delete a manufacturer by its ID",
summary="Delete a manufacturer by ID",
response_description="Manufacturer deleted successfully",
status_code=status.HTTP_204_NO_CONTENT,
)
Expand All @@ -127,10 +123,10 @@ def delete_manufacturer(
try:
manufacturer_service.delete(manufacturer_id)
except (MissingRecordError, InvalidObjectIdError) as exc:
message = "The specified manufacturer does not exist"
message = "Manufacturer not found"
logger.exception(message)
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=message) from exc
except PartOfCatalogueItemError as exc:
message = "The specified manufacturer is a part of a Catalogue Item"
message = "The specified manufacturer is a part of a catalogue item"
logger.exception(message)
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=message) from exc
Loading