From 544bc63e926ab545df1a16346362d7404627d35a Mon Sep 17 00:00:00 2001 From: Joel Davies Date: Thu, 9 Jan 2025 10:52:07 +0000 Subject: [PATCH 1/2] Fix missing optional types on ClientSessions #441 --- .../repositories/catalogue_category.py | 31 +++++++++++++------ .../repositories/catalogue_item.py | 20 ++++++------ .../repositories/item.py | 14 ++++----- .../repositories/manufacturer.py | 14 ++++----- .../repositories/setting.py | 6 ++-- .../repositories/system.py | 20 +++++++----- .../repositories/unit.py | 14 +++++---- .../repositories/usage_status.py | 18 ++++++----- 8 files changed, 81 insertions(+), 56 deletions(-) diff --git a/inventory_management_system_api/repositories/catalogue_category.py b/inventory_management_system_api/repositories/catalogue_category.py index 0a745188..61399d7d 100644 --- a/inventory_management_system_api/repositories/catalogue_category.py +++ b/inventory_management_system_api/repositories/catalogue_category.py @@ -44,7 +44,9 @@ def __init__(self, database: DatabaseDep) -> None: self._catalogue_categories_collection: Collection = self._database.catalogue_categories self._catalogue_items_collection: Collection = self._database.catalogue_items - def create(self, catalogue_category: CatalogueCategoryIn, session: ClientSession = None) -> CatalogueCategoryOut: + def create( + self, catalogue_category: CatalogueCategoryIn, session: Optional[ClientSession] = None + ) -> CatalogueCategoryOut: """ Create a new catalogue category in a MongoDB database. @@ -72,7 +74,9 @@ def create(self, catalogue_category: CatalogueCategoryIn, session: ClientSession catalogue_category = self.get(str(result.inserted_id), session=session) return catalogue_category - def get(self, catalogue_category_id: str, session: ClientSession = None) -> Optional[CatalogueCategoryOut]: + def get( + self, catalogue_category_id: str, session: Optional[ClientSession] = None + ) -> Optional[CatalogueCategoryOut]: """ Retrieve a catalogue category by its ID from a MongoDB database. @@ -89,7 +93,9 @@ def get(self, catalogue_category_id: str, session: ClientSession = None) -> Opti return CatalogueCategoryOut(**catalogue_category) return None - def get_breadcrumbs(self, catalogue_category_id: str, session: ClientSession = None) -> BreadcrumbsGetSchema: + def get_breadcrumbs( + self, catalogue_category_id: str, session: Optional[ClientSession] = None + ) -> BreadcrumbsGetSchema: """ Retrieve the breadcrumbs for a specific catalogue category @@ -111,7 +117,7 @@ def get_breadcrumbs(self, catalogue_category_id: str, session: ClientSession = N collection_name="catalogue_categories", ) - def list(self, parent_id: Optional[str], session: ClientSession = None) -> List[CatalogueCategoryOut]: + def list(self, parent_id: Optional[str], session: Optional[ClientSession] = None) -> List[CatalogueCategoryOut]: """ Retrieve catalogue categories from a MongoDB database based on the provided filters. @@ -126,7 +132,10 @@ def list(self, parent_id: Optional[str], session: ClientSession = None) -> List[ return [CatalogueCategoryOut(**catalogue_category) for catalogue_category in catalogue_categories] def update( - self, catalogue_category_id: str, catalogue_category: CatalogueCategoryIn, session: ClientSession = None + self, + catalogue_category_id: str, + catalogue_category: CatalogueCategoryIn, + session: Optional[ClientSession] = None, ) -> CatalogueCategoryOut: """ Update a catalogue category by its ID in a MongoDB database. @@ -183,7 +192,7 @@ def update( catalogue_category = self.get(str(catalogue_category_id), session=session) return catalogue_category - def delete(self, catalogue_category_id: str, session: ClientSession = None) -> None: + def delete(self, catalogue_category_id: str, session: Optional[ClientSession] = None) -> None: """ Delete a catalogue category by its ID from a MongoDB database. @@ -211,7 +220,7 @@ def _is_duplicate_catalogue_category( parent_id: Optional[str], code: str, catalogue_category_id: CustomObjectId = None, - session: ClientSession = None, + session: Optional[ClientSession] = None, ) -> bool: """ Check if a catalogue category with the same code already exists within the parent category. @@ -233,7 +242,9 @@ def _is_duplicate_catalogue_category( return catalogue_category is not None - def has_child_elements(self, catalogue_category_id: CustomObjectId, session: ClientSession = None) -> bool: + def has_child_elements( + self, catalogue_category_id: CustomObjectId, session: Optional[ClientSession] = None + ) -> bool: """ Check if a catalogue category has child elements based on its ID. @@ -259,7 +270,7 @@ def create_property( self, catalogue_category_id: str, property_in: CatalogueCategoryPropertyIn, - session: ClientSession = None, + session: Optional[ClientSession] = None, ) -> CatalogueCategoryPropertyOut: """ Create a new a property within a catalogue category given its ID in a MongoDB database @@ -293,7 +304,7 @@ def update_property( catalogue_category_id: str, property_id: str, property_in: CatalogueCategoryPropertyIn, - session: ClientSession = None, + session: Optional[ClientSession] = None, ) -> CatalogueCategoryPropertyOut: """ Updates a property given its ID and the ID of the catalogue category it's in diff --git a/inventory_management_system_api/repositories/catalogue_item.py b/inventory_management_system_api/repositories/catalogue_item.py index 2fa32b51..f2993ccc 100644 --- a/inventory_management_system_api/repositories/catalogue_item.py +++ b/inventory_management_system_api/repositories/catalogue_item.py @@ -33,7 +33,7 @@ def __init__(self, database: DatabaseDep) -> None: self._catalogue_items_collection: Collection = self._database.catalogue_items self._items_collection: Collection = self._database.items - def create(self, catalogue_item: CatalogueItemIn, session: ClientSession = None) -> CatalogueItemOut: + def create(self, catalogue_item: CatalogueItemIn, session: Optional[ClientSession] = None) -> CatalogueItemOut: """ Create a new catalogue item in a MongoDB database. @@ -46,7 +46,7 @@ def create(self, catalogue_item: CatalogueItemIn, session: ClientSession = None) catalogue_item = self.get(str(result.inserted_id), session=session) return catalogue_item - def get(self, catalogue_item_id: str, session: ClientSession = None) -> Optional[CatalogueItemOut]: + def get(self, catalogue_item_id: str, session: Optional[ClientSession] = None) -> Optional[CatalogueItemOut]: """ Retrieve a catalogue item by its ID from a MongoDB database. @@ -61,7 +61,9 @@ def get(self, catalogue_item_id: str, session: ClientSession = None) -> Optional return CatalogueItemOut(**catalogue_item) return None - def list(self, catalogue_category_id: Optional[str], session: ClientSession = None) -> List[CatalogueItemOut]: + def list( + self, catalogue_category_id: Optional[str], session: Optional[ClientSession] = None + ) -> List[CatalogueItemOut]: """ Retrieve all catalogue items from a MongoDB database. @@ -85,7 +87,7 @@ def list(self, catalogue_category_id: Optional[str], session: ClientSession = No return [CatalogueItemOut(**catalogue_item) for catalogue_item in catalogue_items] def update( - self, catalogue_item_id: str, catalogue_item: CatalogueItemIn, session: ClientSession = None + self, catalogue_item_id: str, catalogue_item: CatalogueItemIn, session: Optional[ClientSession] = None ) -> CatalogueItemOut: """ Update a catalogue item by its ID in a MongoDB database. @@ -104,7 +106,7 @@ def update( catalogue_item = self.get(str(catalogue_item_id), session=session) return catalogue_item - def delete(self, catalogue_item_id: str, session: ClientSession = None) -> None: + def delete(self, catalogue_item_id: str, session: Optional[ClientSession] = None) -> None: """ Delete a catalogue item by its ID from a MongoDB database. @@ -123,7 +125,7 @@ def delete(self, catalogue_item_id: str, session: ClientSession = None) -> None: if result.deleted_count == 0: raise MissingRecordError(f"No catalogue item found with ID: {str(catalogue_item_id)}") - def has_child_elements(self, catalogue_item_id: CustomObjectId, session: ClientSession = None) -> bool: + def has_child_elements(self, catalogue_item_id: CustomObjectId, session: Optional[ClientSession] = None) -> bool: """ Check if a catalogue item has child elements based on its ID. @@ -137,7 +139,7 @@ def has_child_elements(self, catalogue_item_id: CustomObjectId, session: ClientS item = self._items_collection.find_one({"catalogue_item_id": catalogue_item_id}, session=session) return item is not None - def list_ids(self, catalogue_category_id: str, session: ClientSession = None) -> List[ObjectId]: + def list_ids(self, catalogue_category_id: str, session: Optional[ClientSession] = None) -> List[ObjectId]: """ Retrieve a list of all catalogue item ids with a specific catalogue_category_id from a MongoDB database. Performs a projection to only include _id. (Required for mass updates of properties @@ -161,7 +163,7 @@ def list_ids(self, catalogue_category_id: str, session: ClientSession = None) -> ).distinct("_id") def insert_property_to_all_matching( - self, catalogue_category_id: str, property_in: PropertyIn, session: ClientSession = None + self, catalogue_category_id: str, property_in: PropertyIn, session: Optional[ClientSession] = None ): """ Inserts a property into every catalogue item with a given catalogue_category_id via an update_many query @@ -186,7 +188,7 @@ def insert_property_to_all_matching( ) def update_names_of_all_properties_with_id( - self, property_id: str, new_property_name: str, session: ClientSession = None + self, property_id: str, new_property_name: str, session: Optional[ClientSession] = None ) -> None: """ Updates the name of a property in every catalogue item it is present in diff --git a/inventory_management_system_api/repositories/item.py b/inventory_management_system_api/repositories/item.py index c61ef2e0..6dd4d0b6 100644 --- a/inventory_management_system_api/repositories/item.py +++ b/inventory_management_system_api/repositories/item.py @@ -34,7 +34,7 @@ def __init__(self, database: DatabaseDep) -> None: self._items_collection: Collection = self._database.items self._systems_collection: Collection = self._database.systems - def create(self, item: ItemIn, session: ClientSession = None) -> ItemOut: + def create(self, item: ItemIn, session: Optional[ClientSession] = None) -> ItemOut: """ Create a new item in a MongoDB database. @@ -51,7 +51,7 @@ def create(self, item: ItemIn, session: ClientSession = None) -> ItemOut: item = self.get(str(result.inserted_id), session=session) return item - def get(self, item_id: str, session: ClientSession = None) -> Optional[ItemOut]: + def get(self, item_id: str, session: Optional[ClientSession] = None) -> Optional[ItemOut]: """ Retrieve an item by its ID from a MongoDB database. @@ -67,7 +67,7 @@ def get(self, item_id: str, session: ClientSession = None) -> Optional[ItemOut]: return None def list( - self, system_id: Optional[str], catalogue_item_id: Optional[str], session: ClientSession = None + self, system_id: Optional[str], catalogue_item_id: Optional[str], session: Optional[ClientSession] = None ) -> List[ItemOut]: """ Get all items from the MongoDB database @@ -98,7 +98,7 @@ def list( items = self._items_collection.find(query, session=session) return [ItemOut(**item) for item in items] - def update(self, item_id: str, item: ItemIn, session: ClientSession = None) -> ItemOut: + def update(self, item_id: str, item: ItemIn, session: Optional[ClientSession] = None) -> ItemOut: """ Update an item by its ID in a MongoDB database. @@ -113,7 +113,7 @@ def update(self, item_id: str, item: ItemIn, session: ClientSession = None) -> I item = self.get(str(item_id), session=session) return item - def delete(self, item_id: str, session: ClientSession = None) -> None: + def delete(self, item_id: str, session: Optional[ClientSession] = None) -> None: """ Delete an item by its ID from a MongoDB database. @@ -128,7 +128,7 @@ def delete(self, item_id: str, session: ClientSession = None) -> None: raise MissingRecordError(f"No item found with ID: {str(item_id)}") def insert_property_to_all_in( - self, catalogue_item_ids: List[ObjectId], property_in: PropertyIn, session: ClientSession = None + self, catalogue_item_ids: List[ObjectId], property_in: PropertyIn, session: Optional[ClientSession] = None ): """ Inserts a property into every item with one of the given catalogue_item_id's using an update_many query @@ -155,7 +155,7 @@ def insert_property_to_all_in( # pylint:disable=duplicate-code def update_names_of_all_properties_with_id( - self, property_id: str, new_property_name: str, session: ClientSession = None + self, property_id: str, new_property_name: str, session: Optional[ClientSession] = None ) -> None: """ Updates the name of a property in every item it is present in diff --git a/inventory_management_system_api/repositories/manufacturer.py b/inventory_management_system_api/repositories/manufacturer.py index 2555f41f..363bcf92 100644 --- a/inventory_management_system_api/repositories/manufacturer.py +++ b/inventory_management_system_api/repositories/manufacturer.py @@ -33,7 +33,7 @@ def __init__(self, database: DatabaseDep) -> None: self._manufacturers_collection: Collection = self._database.manufacturers self._catalogue_items_collection: Collection = self._database.catalogue_items - def create(self, manufacturer: ManufacturerIn, session: ClientSession = None) -> ManufacturerOut: + def create(self, manufacturer: ManufacturerIn, session: Optional[ClientSession] = None) -> ManufacturerOut: """ Create a new manufacturer in a MongoDB database. @@ -52,7 +52,7 @@ def create(self, manufacturer: ManufacturerIn, session: ClientSession = None) -> return manufacturer - def get(self, manufacturer_id: str, session: ClientSession = None) -> Optional[ManufacturerOut]: + def get(self, manufacturer_id: str, session: Optional[ClientSession] = None) -> Optional[ManufacturerOut]: """ Retrieve a manufacturer by its ID from a MondoDB database. @@ -67,7 +67,7 @@ def get(self, manufacturer_id: str, session: ClientSession = None) -> Optional[M return ManufacturerOut(**manufacturer) return None - def list(self, session: ClientSession = None) -> List[ManufacturerOut]: + def list(self, session: Optional[ClientSession] = None) -> List[ManufacturerOut]: """ Retrieve all manufacturers from a MongoDB database. @@ -79,7 +79,7 @@ def list(self, session: ClientSession = None) -> List[ManufacturerOut]: return [ManufacturerOut(**manufacturer) for manufacturer in manufacturers] def update( - self, manufacturer_id: str, manufacturer: ManufacturerIn, session: ClientSession = None + self, manufacturer_id: str, manufacturer: ManufacturerIn, session: Optional[ClientSession] = None ) -> ManufacturerOut: """ Update manufacturer by its ID in a MongoDB database. @@ -105,7 +105,7 @@ def update( manufacturer = self.get(str(manufacturer_id), session=session) return manufacturer - def delete(self, manufacturer_id: str, session: ClientSession = None) -> None: + def delete(self, manufacturer_id: str, session: Optional[ClientSession] = None) -> None: """ Delete a manufacturer by its ID from a MongoDB database. @@ -127,7 +127,7 @@ def delete(self, manufacturer_id: str, session: ClientSession = None) -> None: raise MissingRecordError(f"No manufacturer found with ID: {str(manufacturer_id)}") def _is_duplicate_manufacturer( - self, code: str, manufacturer_id: CustomObjectId = None, session: ClientSession = None + self, code: str, manufacturer_id: CustomObjectId = None, session: Optional[ClientSession] = None ) -> bool: """ Check if a manufacturer with the same code already exists. @@ -143,7 +143,7 @@ def _is_duplicate_manufacturer( ) return manufacturer is not None - def _is_manufacturer_in_catalogue_item(self, manufacturer_id: str, session: ClientSession = None) -> bool: + def _is_manufacturer_in_catalogue_item(self, manufacturer_id: str, session: Optional[ClientSession] = None) -> bool: """ Check if a manufacturer is part of a catalogue item based on its ID. diff --git a/inventory_management_system_api/repositories/setting.py b/inventory_management_system_api/repositories/setting.py index 29bb072e..2daa723c 100644 --- a/inventory_management_system_api/repositories/setting.py +++ b/inventory_management_system_api/repositories/setting.py @@ -65,7 +65,7 @@ def __init__(self, database: DatabaseDep) -> None: self._settings_collection: Collection = self._database.settings def upsert( - self, setting: SettingInBaseT, out_model_type: Type[SettingOutBaseT], session: ClientSession = None + self, setting: SettingInBaseT, out_model_type: Type[SettingOutBaseT], session: Optional[ClientSession] = None ) -> SettingOutBaseT: """ Update or insert a setting in a MongoDB database depending on whether it already exists. @@ -83,7 +83,9 @@ def upsert( return self.get(out_model_type=out_model_type, session=session) - def get(self, out_model_type: Type[SettingOutBaseT], session: ClientSession = None) -> Optional[SettingOutBaseT]: + def get( + self, out_model_type: Type[SettingOutBaseT], session: Optional[ClientSession] = None + ) -> Optional[SettingOutBaseT]: """ Retrieve a setting from a MongoDB database. diff --git a/inventory_management_system_api/repositories/system.py b/inventory_management_system_api/repositories/system.py index f15d68ba..ecc1ff09 100644 --- a/inventory_management_system_api/repositories/system.py +++ b/inventory_management_system_api/repositories/system.py @@ -38,7 +38,7 @@ def __init__(self, database: DatabaseDep) -> None: self._systems_collection: Collection = self._database.systems self._items_collection: Collection = self._database.items - def create(self, system: SystemIn, session: ClientSession = None) -> SystemOut: + def create(self, system: SystemIn, session: Optional[ClientSession] = None) -> SystemOut: """ Create a new system in a MongoDB database @@ -64,7 +64,7 @@ def create(self, system: SystemIn, session: ClientSession = None) -> SystemOut: system = self.get(str(result.inserted_id), session=session) return system - def get(self, system_id: str, session: ClientSession = None) -> Optional[SystemOut]: + def get(self, system_id: str, session: Optional[ClientSession] = None) -> Optional[SystemOut]: """ Retrieve a system by its ID from a MongoDB database @@ -79,7 +79,7 @@ def get(self, system_id: str, session: ClientSession = None) -> Optional[SystemO return SystemOut(**system) return None - def get_breadcrumbs(self, system_id: str, session: ClientSession = None) -> BreadcrumbsGetSchema: + def get_breadcrumbs(self, system_id: str, session: Optional[ClientSession] = None) -> BreadcrumbsGetSchema: """ Retrieve the breadcrumbs for a specific system @@ -99,7 +99,7 @@ def get_breadcrumbs(self, system_id: str, session: ClientSession = None) -> Brea collection_name="systems", ) - def list(self, parent_id: Optional[str], session: ClientSession = None) -> list[SystemOut]: + def list(self, parent_id: Optional[str], session: Optional[ClientSession] = None) -> list[SystemOut]: """ Retrieve systems from a MongoDB database based on the provided filters @@ -112,7 +112,7 @@ def list(self, parent_id: Optional[str], session: ClientSession = None) -> list[ systems = self._systems_collection.find(query, session=session) return [SystemOut(**system) for system in systems] - def update(self, system_id: str, system: SystemIn, session: ClientSession = None) -> SystemOut: + def update(self, system_id: str, system: SystemIn, session: Optional[ClientSession] = None) -> SystemOut: """Update a system by its ID in a MongoDB database :param system_id: ID of the system to update @@ -155,7 +155,7 @@ def update(self, system_id: str, system: SystemIn, session: ClientSession = None return self.get(str(system_id), session=session) - def delete(self, system_id: str, session: ClientSession = None) -> None: + def delete(self, system_id: str, session: Optional[ClientSession] = None) -> None: """ Delete a system by its ID from a MongoDB database @@ -176,7 +176,11 @@ def delete(self, system_id: str, session: ClientSession = None) -> None: raise MissingRecordError(f"No system found with ID: {str(system_id)}") def _is_duplicate_system( - self, parent_id: Optional[str], code: str, system_id: CustomObjectId = None, session: ClientSession = None + self, + parent_id: Optional[str], + code: str, + system_id: CustomObjectId = None, + session: Optional[ClientSession] = None, ) -> bool: """ Check if a system with the same code already exists within the parent system @@ -196,7 +200,7 @@ def _is_duplicate_system( ) return system is not None - def _has_child_elements(self, system_id: CustomObjectId, session: ClientSession = None) -> bool: + def _has_child_elements(self, system_id: CustomObjectId, session: Optional[ClientSession] = None) -> bool: """ Check if a system has any child system's or any Item's based on its ID diff --git a/inventory_management_system_api/repositories/unit.py b/inventory_management_system_api/repositories/unit.py index 2dc8d997..7f884d20 100644 --- a/inventory_management_system_api/repositories/unit.py +++ b/inventory_management_system_api/repositories/unit.py @@ -34,7 +34,7 @@ def __init__(self, database: DatabaseDep) -> None: self._units_collection: Collection = self._database.units self._catalogue_categories_collection: Collection = self._database.catalogue_categories - def create(self, unit: UnitIn, session: ClientSession = None) -> UnitOut: + def create(self, unit: UnitIn, session: Optional[ClientSession] = None) -> UnitOut: """ Create a new Unit in a MongoDB database @@ -54,7 +54,7 @@ def create(self, unit: UnitIn, session: ClientSession = None) -> UnitOut: return unit - def list(self, session: ClientSession = None) -> list[UnitOut]: + def list(self, session: Optional[ClientSession] = None) -> list[UnitOut]: """ Retrieve Units from a MongoDB database @@ -64,7 +64,7 @@ def list(self, session: ClientSession = None) -> list[UnitOut]: units = self._units_collection.find(session=session) return [UnitOut(**unit) for unit in units] - def get(self, unit_id: str, session: ClientSession = None) -> Optional[UnitOut]: + def get(self, unit_id: str, session: Optional[ClientSession] = None) -> Optional[UnitOut]: """ Retrieve a Unit by its ID from a MongoDB database. @@ -79,7 +79,7 @@ def get(self, unit_id: str, session: ClientSession = None) -> Optional[UnitOut]: return UnitOut(**unit) return None - def delete(self, unit_id: str, session: ClientSession = None) -> None: + def delete(self, unit_id: str, session: Optional[ClientSession] = None) -> None: """ Delete a unit by its ID from a MongoDB database. @@ -99,7 +99,9 @@ def delete(self, unit_id: str, session: ClientSession = None) -> None: if result.deleted_count == 0: raise MissingRecordError(f"No unit found with ID: {str(unit_id)}") - def _is_duplicate_unit(self, code: str, unit_id: CustomObjectId = None, session: ClientSession = None) -> bool: + def _is_duplicate_unit( + self, code: str, unit_id: CustomObjectId = None, session: Optional[ClientSession] = None + ) -> bool: """ Check if a Unit with the same value already exists in the Units collection @@ -112,7 +114,7 @@ def _is_duplicate_unit(self, code: str, unit_id: CustomObjectId = None, session: unit = self._units_collection.find_one({"code": code, "_id": {"$ne": unit_id}}, session=session) return unit is not None - def _is_unit_in_catalogue_category(self, unit_id: str, session: ClientSession = None) -> bool: + def _is_unit_in_catalogue_category(self, unit_id: str, session: Optional[ClientSession] = None) -> bool: """ Checks if any catalogue categories in the database have a specific unit ID diff --git a/inventory_management_system_api/repositories/usage_status.py b/inventory_management_system_api/repositories/usage_status.py index e83b44d1..640ca986 100644 --- a/inventory_management_system_api/repositories/usage_status.py +++ b/inventory_management_system_api/repositories/usage_status.py @@ -38,7 +38,7 @@ def __init__(self, database: DatabaseDep) -> None: self._items_collection: Collection = self._database.items self._settings_collection: Collection = self._database.settings - def create(self, usage_status: UsageStatusIn, session: ClientSession = None) -> UsageStatusOut: + def create(self, usage_status: UsageStatusIn, session: Optional[ClientSession] = None) -> UsageStatusOut: """ Create a new usage status in MongoDB database @@ -58,7 +58,7 @@ def create(self, usage_status: UsageStatusIn, session: ClientSession = None) -> return usage_status - def list(self, session: ClientSession = None) -> list[UsageStatusOut]: + def list(self, session: Optional[ClientSession] = None) -> list[UsageStatusOut]: """ Retrieve Usage statuses from a MongoDB database @@ -68,7 +68,7 @@ def list(self, session: ClientSession = None) -> list[UsageStatusOut]: usage_statuses = self._usage_statuses_collection.find(session=session) return [UsageStatusOut(**usage_status) for usage_status in usage_statuses] - def get(self, usage_status_id: str, session: ClientSession = None) -> Optional[UsageStatusOut]: + def get(self, usage_status_id: str, session: Optional[ClientSession] = None) -> Optional[UsageStatusOut]: """ Retrieve a usage status by its ID from a MongoDB database. @@ -83,7 +83,7 @@ def get(self, usage_status_id: str, session: ClientSession = None) -> Optional[U return UsageStatusOut(**usage_status) return None - def delete(self, usage_status_id: str, session: ClientSession = None) -> None: + def delete(self, usage_status_id: str, session: Optional[ClientSession] = None) -> None: """ Delete a usage status by its ID from a MongoDB database. @@ -106,7 +106,7 @@ def delete(self, usage_status_id: str, session: ClientSession = None) -> None: raise MissingRecordError(f"No usage status found with ID: {str(usage_status_id)}") def _is_duplicate_usage_status( - self, code: str, usage_status_id: CustomObjectId = None, session: ClientSession = None + self, code: str, usage_status_id: CustomObjectId = None, session: Optional[ClientSession] = None ) -> bool: """ Check if usage status with the same name already exists in the usage statuses collection @@ -122,7 +122,9 @@ def _is_duplicate_usage_status( ) return usage_status is not None - def _is_usage_status_in_item(self, usage_status_id: CustomObjectId, session: ClientSession = None) -> bool: + def _is_usage_status_in_item( + self, usage_status_id: CustomObjectId, session: Optional[ClientSession] = None + ) -> bool: """Checks to see if any of the items in the database have a specific usage status ID. :param usage_status_id: The ID of the usage status that is looked for. @@ -131,7 +133,9 @@ def _is_usage_status_in_item(self, usage_status_id: CustomObjectId, session: Cli """ return self._items_collection.find_one({"usage_status_id": usage_status_id}, session=session) is not None - def _is_usage_status_in_setting(self, usage_status_id: CustomObjectId, session: ClientSession = None) -> bool: + def _is_usage_status_in_setting( + self, usage_status_id: CustomObjectId, session: Optional[ClientSession] = None + ) -> bool: """Checks to see if any of the settings in the database refer to a specific usage status ID. :param usage_status_id: The ID of the usage status that is looked for. From 468cc656708fa70163d130f1304615ae9e6c6c99 Mon Sep 17 00:00:00 2001 From: Joel Davies Date: Tue, 14 Jan 2025 14:11:13 +0000 Subject: [PATCH 2/2] Resolve some further missing optional types #441 --- .../repositories/catalogue_category.py | 2 +- inventory_management_system_api/repositories/manufacturer.py | 2 +- inventory_management_system_api/repositories/system.py | 2 +- inventory_management_system_api/repositories/usage_status.py | 2 +- 4 files changed, 4 insertions(+), 4 deletions(-) diff --git a/inventory_management_system_api/repositories/catalogue_category.py b/inventory_management_system_api/repositories/catalogue_category.py index 61399d7d..edcc72a6 100644 --- a/inventory_management_system_api/repositories/catalogue_category.py +++ b/inventory_management_system_api/repositories/catalogue_category.py @@ -219,7 +219,7 @@ def _is_duplicate_catalogue_category( self, parent_id: Optional[str], code: str, - catalogue_category_id: CustomObjectId = None, + catalogue_category_id: Optional[CustomObjectId] = None, session: Optional[ClientSession] = None, ) -> bool: """ diff --git a/inventory_management_system_api/repositories/manufacturer.py b/inventory_management_system_api/repositories/manufacturer.py index 363bcf92..c7a85d38 100644 --- a/inventory_management_system_api/repositories/manufacturer.py +++ b/inventory_management_system_api/repositories/manufacturer.py @@ -127,7 +127,7 @@ def delete(self, manufacturer_id: str, session: Optional[ClientSession] = None) raise MissingRecordError(f"No manufacturer found with ID: {str(manufacturer_id)}") def _is_duplicate_manufacturer( - self, code: str, manufacturer_id: CustomObjectId = None, session: Optional[ClientSession] = None + self, code: str, manufacturer_id: Optional[CustomObjectId] = None, session: Optional[ClientSession] = None ) -> bool: """ Check if a manufacturer with the same code already exists. diff --git a/inventory_management_system_api/repositories/system.py b/inventory_management_system_api/repositories/system.py index ecc1ff09..27f7009d 100644 --- a/inventory_management_system_api/repositories/system.py +++ b/inventory_management_system_api/repositories/system.py @@ -179,7 +179,7 @@ def _is_duplicate_system( self, parent_id: Optional[str], code: str, - system_id: CustomObjectId = None, + system_id: Optional[CustomObjectId] = None, session: Optional[ClientSession] = None, ) -> bool: """ diff --git a/inventory_management_system_api/repositories/usage_status.py b/inventory_management_system_api/repositories/usage_status.py index 640ca986..cb53457c 100644 --- a/inventory_management_system_api/repositories/usage_status.py +++ b/inventory_management_system_api/repositories/usage_status.py @@ -106,7 +106,7 @@ def delete(self, usage_status_id: str, session: Optional[ClientSession] = None) raise MissingRecordError(f"No usage status found with ID: {str(usage_status_id)}") def _is_duplicate_usage_status( - self, code: str, usage_status_id: CustomObjectId = None, session: Optional[ClientSession] = None + self, code: str, usage_status_id: Optional[CustomObjectId] = None, session: Optional[ClientSession] = None ) -> bool: """ Check if usage status with the same name already exists in the usage statuses collection