Merge pull request #67 from ral-facilities/add-manufacturers-#63

As an operator/admin, I want to be able to add manufacturers

inventory_management_system_api/main.py

@@ -11,7 +11,7 @@
 
 from inventory_management_system_api.core.config import config
 from inventory_management_system_api.core.logger_setup import setup_logger
-from inventory_management_system_api.routers.v1 import catalogue_category, catalogue_item, system
+from inventory_management_system_api.routers.v1 import catalogue_category, catalogue_item, system, manufacturer
 
 app = FastAPI(title=config.api.title, description=config.api.description)
 
@@ -64,6 +64,7 @@ async def custom_validation_exception_handler(request: Request, exc: RequestVali
 
 app.include_router(catalogue_category.router)
 app.include_router(catalogue_item.router)
+app.include_router(manufacturer.router)
 app.include_router(system.router)
 
 

inventory_management_system_api/models/manufacturer.py

@@ -0,0 +1,24 @@
+"""
+Module for defining the database models for representing manufacturer.
+"""
+
+
+from pydantic import BaseModel, Field
+
+
+from inventory_management_system_api.models.catalogue_category import StringObjectIdField
+
+
+class ManufacturerIn(BaseModel):
+    """Input database model for a manufacturer"""
+
+    name: str
+    code: str
+    url: str
+    address: str
+
+
+class ManufacturerOut(ManufacturerIn):
+    """Output database model for a manufacturer"""
+
+    id: StringObjectIdField = Field(alias="_id")

inventory_management_system_api/repositories/manufacturer.py

@@ -0,0 +1,76 @@
+"""
+Module for providing a repository for managing manufacturers in a MongoDB database.
+"""
+import logging
+from typing import Optional
+
+from fastapi import Depends
+from pymongo.collection import Collection
+from pymongo.database import Database
+
+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.manufacturer import ManufacturerIn, ManufacturerOut
+
+logger = logging.getLogger()
+
+
+class ManufacturerRepo:
+    """Repository for managing manufacturer in MongoDb database"""
+
+    def __init__(self, database: Database = Depends(get_database)) -> None:
+        """Initialize the `ManufacturerRepo` with MongoDB database instance
+
+        :param database: The database to use.
+        """
+
+        self._database = database
+        self._collection: Collection = self._database.manufacturer
+
+    def create(self, manufacturer: ManufacturerIn) -> ManufacturerOut:
+        """
+        Create a new manufacturer in MongoDB database
+
+        :param manufacturer: The manufacturer to be created
+        :return: The created manufacturer
+        :raises DuplicateRecordError: If a duplicate manufacturer is found within collection
+        """
+
+        if self._is_duplicate_manufacturer(manufacturer.code):
+            raise DuplicateRecordError("Duplicate manufacturer found")
+
+        logger.info("Inserting new manufacturer into database")
+
+        result = self._collection.insert_one(manufacturer.dict())
+        manufacturer = self.get(str(result.inserted_id))
+
+        return manufacturer
+
+    def get(self, manufacturer_id: str) -> Optional[ManufacturerOut]:
+        """Retrieve a manufacturer from database by its id
+
+
+        :param manufacturer_id: The ID of the manufacturer
+        :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)
+        manufacturer = self._collection.find_one({"_id": manufacturer_id})
+        if manufacturer:
+            return ManufacturerOut(**manufacturer)
+        return None
+
+    def _is_duplicate_manufacturer(self, code: str) -> bool:
+        """
+        Check if manufacturer with the same url already exists in the manufacturer collection
+
+        :param code: The code of the manufacturer to check for duplicates.
+        :return `True` if duplicate manufacturer, `False` otherwise
+        """
+        logger.info("Checking if manufacturer with code '%s' already exists", code)
+        count = self._collection.count_documents({"code": code})
+        return count > 0

inventory_management_system_api/routers/v1/manufacturer.py

@@ -0,0 +1,40 @@
+"""
+Module for providing an API router which defines routes for managing manufacturer using the
+`Manufacturer` service.
+"""
+import logging
+
+from fastapi import APIRouter, status, Depends, HTTPException
+from inventory_management_system_api.core.exceptions import DuplicateRecordError
+
+from inventory_management_system_api.schemas.manufacturer import ManufacturerPostRequestSchema, ManufacturerSchema
+from inventory_management_system_api.services.manufacturer import ManufacturerService
+
+
+logger = logging.getLogger()
+
+router = APIRouter(prefix="/v1/manufacturer", tags=["manufacturer"])
+
+
+@router.post(
+    path="/",
+    summary="Create new manufacturer",
+    response_description="The new manufacturer",
+    status_code=status.HTTP_201_CREATED,
+)
+def create_manufacturer(
+    manufacturer: ManufacturerPostRequestSchema,
+    manufacturer_service: ManufacturerService = Depends(),
+) -> ManufacturerSchema:
+    # 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.dict())
+
+    except DuplicateRecordError as exc:
+        message = "A manufacturer with the same name has been found"
+        logger.exception(message)
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=message) from exc

inventory_management_system_api/schemas/manufacturer.py

@@ -0,0 +1,20 @@
+"""
+Module for defining the API schema models for representing manufacturers.
+"""
+
+from pydantic import BaseModel, Field
+
+
+class ManufacturerPostRequestSchema(BaseModel):
+    """Schema model for manufactuer creation request"""
+
+    name: str = Field(description="Name of manufacturer")
+    url: str = Field(description="URL of manufacturer")
+    address: str = Field(description="Address of manufacturer")
+
+
+class ManufacturerSchema(ManufacturerPostRequestSchema):
+    """Schema model for manufacturer response"""
+
+    id: str = Field(description="The ID of manufacturer")
+    code: str = Field(description="The code of the manufacturer")

inventory_management_system_api/services/catalogue_category.py

@@ -15,6 +15,7 @@
 )
 from inventory_management_system_api.services import utils
 
+
 logger = logging.getLogger()
 
 

inventory_management_system_api/services/manufacturer.py

@@ -0,0 +1,50 @@
+"""
+Module for providing a service for managing manufacturers using the `ManufacturerRepo` repository.
+"""
+import logging
+import re
+
+from fastapi import Depends
+from inventory_management_system_api.models.manufacturer import ManufacturerIn, ManufacturerOut
+
+from inventory_management_system_api.repositories.manufacturer import ManufacturerRepo
+from inventory_management_system_api.schemas.manufacturer import ManufacturerPostRequestSchema
+
+logger = logging.getLogger()
+
+
+class ManufacturerService:
+    """Service for managing manufacturers"""
+
+    def __init__(self, manufacturer_repository: ManufacturerRepo = Depends(ManufacturerRepo)) -> None:
+        """
+        Initialise the manufacturer service with a ManufacturerRepo
+
+        :param manufacturer_repository: The `ManufacturerRepo` repository to use.
+        """
+
+        self._manufacturer_repository = manufacturer_repository
+
+    def create(self, manufacturer: ManufacturerPostRequestSchema) -> ManufacturerOut:
+        """
+        Create a new manufacturer.
+        :param manufacturer: The manufacturer to be created.
+        :return: The created manufacturer.
+        """
+        code = self._generate_code(manufacturer.name)
+        return self._manufacturer_repository.create(
+            ManufacturerIn(name=manufacturer.name, code=code, url=manufacturer.url, address=manufacturer.address)
+        )
+
+    def _generate_code(self, name: str) -> str:
+        """
+        Generate code for manufacturer based on its name, used to check for duplicate manufacturers
+
+        The code is generated by changing name to lowercase and replacing spaces hypens,
+        and removing trailing/preceding spaces
+
+        :param name: The name of the manufacturer
+        :return: The generated code for the manufacturer
+        """
+        name = name.lower().strip()
+        return re.sub(r"\s", "-", name)

test/e2e/test_manufacturer.py

@@ -0,0 +1,58 @@
+"""
+End-to-End tests for the manufacturer router.
+"""
+import pytest
+
+
+from inventory_management_system_api.core.database import get_database
+
+
+@pytest.fixture(name="cleanup_manufacturer", autouse=True)
+def fixture_cleanup_manufacturer():
+    """
+    Fixture to clean up the manufacturer collection in test database
+    """
+    database = get_database()
+    yield
+    database.manufacturer.delete_many({})
+
+
+def test_create_manufacturer(test_client):
+    """Test creating a manufacturer"""
+    manufacturer_post = {
+        "name": "Manufacturer A",
+        "url": "example.com",
+        "address": "Street A",
+    }
+
+    response = test_client.post("/v1/manufacturer", json=manufacturer_post)
+
+    assert response.status_code == 201
+
+    manufacturer = response.json()
+
+    assert manufacturer["name"] == manufacturer_post["name"]
+    assert manufacturer["url"] == manufacturer_post["url"]
+    assert manufacturer["address"] == manufacturer_post["address"]
+
+
+def test_check_duplicate_url_within_manufacturer(test_client):
+    """Test creating a manufactuer with a duplicate name"""
+
+    manufacturer_post = {
+        "name": "Manufacturer A",
+        "url": "example.com",
+        "address": "Street A",
+    }
+    test_client.post("/v1/manufacturer", json=manufacturer_post)
+
+    manufacturer_post = {
+        "name": "Manufacturer A",
+        "url": "test.com",
+        "address": "Street B",
+    }
+
+    response = test_client.post("/v1/manufacturer", json=manufacturer_post)
+
+    assert response.status_code == 409
+    assert response.json()["detail"] == "A manufacturer with the same name has been found"

test/unit/repositories/conftest.py

@@ -13,6 +13,7 @@
 
 from inventory_management_system_api.repositories.catalogue_category import CatalogueCategoryRepo
 from inventory_management_system_api.repositories.catalogue_item import CatalogueItemRepo
+from inventory_management_system_api.repositories.manufacturer import ManufacturerRepo
 from inventory_management_system_api.repositories.system import SystemRepo
 
 
@@ -27,6 +28,7 @@ def fixture_database_mock() -> Mock:
     database_mock = Mock(Database)
     database_mock.catalogue_categories = Mock(Collection)
     database_mock.catalogue_items = Mock(Collection)
+    database_mock.manufacturer = Mock(Collection)
     database_mock.systems = Mock(Collection)
     return database_mock
 
@@ -53,6 +55,12 @@ def fixture_catalogue_item_repository(database_mock: Mock) -> CatalogueItemRepo:
     return CatalogueItemRepo(database_mock)
 
 
+@pytest.fixture(name="manufacturer_repository")
+def fixture_manufacturer_repository(database_mock: Mock) -> ManufacturerRepo:
+    """
+    Fixture to create ManufacturerRepo instance
+    """
+    return ManufacturerRepo(database_mock)
 @pytest.fixture(name="system_repository")
 def fixture_system_repository(database_mock: Mock) -> SystemRepo:
     """
@@ -65,6 +73,7 @@ def fixture_system_repository(database_mock: Mock) -> SystemRepo:
 
 
 class RepositoryTestHelpers:
+
     """
     A utility class containing common helper methods for the repository tests.