Add Expected Lifetime to Catalogue Items

inventory_management_system_api/migrations/scripts/expected_lifetime_migration.py

@@ -0,0 +1,188 @@
+"""
+Module providing a migration for the optional expected_lifetime field under catalogue items
+"""
+
+import logging
+from typing import Any, Collection, List, Optional
+
+from pydantic import BaseModel, ConfigDict, Field, HttpUrl, ValidationError, field_serializer, field_validator
+from pymongo.client_session import ClientSession
+from pymongo.database import Database
+
+from inventory_management_system_api.migrations.migration import BaseMigration
+from inventory_management_system_api.models.catalogue_item import PropertyIn, PropertyOut
+from inventory_management_system_api.models.custom_object_id_data_types import CustomObjectIdField, StringObjectIdField
+from inventory_management_system_api.models.mixins import CreatedModifiedTimeInMixin, CreatedModifiedTimeOutMixin
+
+logger = logging.getLogger()
+
+
+# pylint: disable=duplicate-code
+class NewCatalogueItemBase(BaseModel):
+    """
+    Base database model for a catalogue item.
+    """
+
+    catalogue_category_id: CustomObjectIdField
+    manufacturer_id: CustomObjectIdField
+    name: str
+    description: Optional[str] = None
+    cost_gbp: float
+    cost_to_rework_gbp: Optional[float] = None
+    days_to_replace: float
+    days_to_rework: Optional[float] = None
+    drawing_number: Optional[str] = None
+    drawing_link: Optional[HttpUrl] = None
+    expected_lifetime: Optional[float] = None
+    item_model_number: Optional[str] = None
+    is_obsolete: bool
+    obsolete_reason: Optional[str] = None
+    obsolete_replacement_catalogue_item_id: Optional[CustomObjectIdField] = None
+    notes: Optional[str] = None
+    properties: List[PropertyIn] = []
+
+    @field_validator("properties", mode="before")
+    @classmethod
+    def validate_properties(cls, properties: Any) -> Any:
+        """
+        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 specific to this catalogue item as defined in the corresponding
+            catalogue category.
+        :return: The list of properties specific to this catalogue item or an empty list.
+        """
+        if properties is None:
+            properties = []
+        return properties
+
+    @field_serializer("drawing_link")
+    def serialize_url(self, url: HttpUrl):
+        """
+        Convert `url` to string when the model is dumped.
+        :param url: The `HttpUrl` object.
+        :return: The URL as a string.
+        """
+        return url if url is None else str(url)
+
+
+class NewCatalogueItemIn(CreatedModifiedTimeInMixin, NewCatalogueItemBase):
+    """
+    Input database model for a catalogue item.
+    """
+
+
+class OldCatalogueItemBase(BaseModel):
+    """
+    Base database model for a catalogue item.
+    """
+
+    catalogue_category_id: CustomObjectIdField
+    manufacturer_id: CustomObjectIdField
+    name: str
+    description: Optional[str] = None
+    cost_gbp: float
+    cost_to_rework_gbp: Optional[float] = None
+    days_to_replace: float
+    days_to_rework: Optional[float] = None
+    drawing_number: Optional[str] = None
+    drawing_link: Optional[HttpUrl] = None
+    item_model_number: Optional[str] = None
+    is_obsolete: bool
+    obsolete_reason: Optional[str] = None
+    obsolete_replacement_catalogue_item_id: Optional[CustomObjectIdField] = None
+    notes: Optional[str] = None
+    properties: List[PropertyIn] = []
+
+    @field_validator("properties", mode="before")
+    @classmethod
+    def validate_properties(cls, properties: Any) -> Any:
+        """
+        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 specific to this catalogue item as defined in the corresponding
+            catalogue category.
+        :return: The list of properties specific to this catalogue item or an empty list.
+        """
+        if properties is None:
+            properties = []
+        return properties
+
+    @field_serializer("drawing_link")
+    def serialize_url(self, url: HttpUrl):
+        """
+        Convert `url` to string when the model is dumped.
+        :param url: The `HttpUrl` object.
+        :return: The URL as a string.
+        """
+        return url if url is None else str(url)
+
+
+class OldCatalogueItemOut(CreatedModifiedTimeOutMixin, OldCatalogueItemBase):
+    """
+    Output database model for a catalogue item.
+    """
+
+    id: StringObjectIdField = Field(alias="_id")
+    catalogue_category_id: StringObjectIdField
+    manufacturer_id: StringObjectIdField
+    obsolete_replacement_catalogue_item_id: Optional[StringObjectIdField] = None
+    properties: List[PropertyOut] = []
+
+    model_config = ConfigDict(populate_by_name=True, arbitrary_types_allowed=True)
+
+
+# pylint: enable=duplicate-code
+
+
+class Migration(BaseMigration):
+    """Migration for Catalogue Items' Optional Expected Lifetime"""
+
+    description = "Migration for Catalogue Items' Optional Expected Lifetime"
+
+    def __init__(self, database: Database):
+
+        self._catalogue_items_collection: Collection = database.catalogue_items
+
+    def forward(self, session: ClientSession):
+        """This function should actually perform the migration
+
+        All database functions should be given the session in order to ensure all updates are done within a transaction
+        """
+        catalogue_items = self._catalogue_items_collection.find({}, session=session)
+
+        logger.info("expected_lifetime forward migration")
+        for catalogue_item in catalogue_items:
+            try:
+                old_catalogue_item = OldCatalogueItemOut(**catalogue_item)
+
+                new_catalogue_item = NewCatalogueItemIn(**old_catalogue_item.model_dump())
+
+                update_data = {
+                    **new_catalogue_item.model_dump(),
+                    "modified_time": old_catalogue_item.modified_time,
+                }
+
+                self._catalogue_items_collection.replace_one(
+                    {"_id": catalogue_item["_id"]},
+                    update_data,
+                    session=session,
+                )
+
+            except ValidationError as ve:
+                logger.error("Validation failed for item with id %s: %s", catalogue_item["_id"], ve)
+
+                continue
+
+    def backward(self, session: ClientSession):
+        """This function should reverse the migration
+
+        All database functions should be given the session in order to ensure all updates are done within a transaction
+        """
+
+        logger.info("expected_lifetime backward migration")
+        result = self._catalogue_items_collection.update_many(
+            {}, {"$unset": {"expected_lifetime": ""}}, session=session
+        )
+        return result

inventory_management_system_api/models/catalogue_item.py

@@ -51,6 +51,7 @@ class CatalogueItemBase(BaseModel):
     days_to_rework: Optional[float] = None
     drawing_number: Optional[str] = None
     drawing_link: Optional[HttpUrl] = None
+    expected_lifetime: Optional[float] = None
     item_model_number: Optional[str] = None
     is_obsolete: bool
     obsolete_reason: Optional[str] = None

inventory_management_system_api/schemas/catalogue_item.py

@@ -45,6 +45,7 @@ class CatalogueItemPostSchema(BaseModel):
     days_to_rework: Optional[float] = Field(default=None, description="The number of days to rework the catalogue item")
     drawing_number: Optional[str] = Field(default=None, description="The drawing number of the catalogue item")
     drawing_link: Optional[HttpUrl] = Field(default=None, description="The link to the drawing of the catalogue item")
+    expected_lifetime: Optional[float] = Field(default=None, description="The expected lifetime of the catalogue item")
     item_model_number: Optional[str] = Field(default=None, description="The model number of the catalogue item")
     is_obsolete: bool = Field(description="Whether the catalogue item is obsolete or not")
     obsolete_reason: Optional[str] = Field(

scripts/generate_mock_data.py

@@ -350,6 +350,7 @@ def generate_random_catalogue_item(
         "days_to_rework": optional_catalogue_item_field(lambda: fake.random.randint(0, 100)),
         "drawing_number": optional_catalogue_item_field(lambda: str(fake.random.randint(1000, 10000))),
         "drawing_link": optional_catalogue_item_field(fake.image_url),
+        "expected_lifetime": optional_catalogue_item_field(lambda: fake.random.randint(1000, 10000)),
         "item_model_number": optional_catalogue_item_field(fake.isbn13),
         "is_obsolete": bool(obsolete_replacement_catalogue_item_id),
         "obsolete_replacement_catalogue_item_id": obsolete_replacement_catalogue_item_id,

test/mock_data.py

@@ -441,6 +441,7 @@
     "days_to_rework": None,
     "drawing_number": None,
     "drawing_link": None,
+    "expected_lifetime": None,
     "item_model_number": None,
     "obsolete_reason": None,
     "obsolete_replacement_catalogue_item_id": None,
@@ -457,6 +458,7 @@
     "cost_to_rework_gbp": 9001,
     "days_to_rework": 3,
     "drawing_number": "12345-1",
+    "expected_lifetime": 3002,
     "drawing_link": "http://example.com/",
     "item_model_number": "123456-1",
     "is_obsolete": False,