Refactor document service to use 'table_name' type hint and improve variable naming

Author: NogaNHS

lambdas/services/document_service.py

@@ -54,19 +54,19 @@ def fetch_available_document_references_by_type(
     def fetch_documents_from_table_with_nhs_number(
         self,
         nhs_number: str,
-        table: str = None,
+        table: str | None = None,
         query_filter: Attr | ConditionBase = None,
         model_class: type[BaseModel] = None,
     ) -> list:
         """Fetch documents by NHS number from specified or configured table."""
-        table_to_use = table or self.table_name
+        table_name = table or self.table_name
 
         documents = self.fetch_documents_from_table(
             index_name="NhsNumberIndex",
             search_key="NhsNumber",
             search_condition=nhs_number,
             query_filter=query_filter,
-            table_name=table_to_use,
+            table_name=table_name,
             model_class=model_class,
         )
         return documents
@@ -77,24 +77,24 @@ def fetch_documents_from_table(
         search_key: str,
         index_name: str | None = None,
         query_filter: Attr | ConditionBase = None,
-        table_name: str = None,
+        table_name: str | None = None,
         model_class: type[BaseModel] = None,
     ) -> list:
         """Fetch documents from specified or configured table using model_class."""
         documents = []
-        table_to_use = table_name or self.table_name
-        model_to_use = model_class or self.model_class
+        table_name = table_name or self.table_name
+        model_class = model_class or self.model_class
 
         response = self.dynamo_service.query_table(
-            table_name=table_to_use,
+            table_name=table_name,
             index_name=index_name,
             search_key=search_key,
             search_condition=search_condition,
             query_filter=query_filter,
         )
         for item in response:
             try:
-                document = model_to_use.model_validate(item)
+                document = model_class.model_validate(item)
                 documents.append(document)
             except ValidationError as e:
                 logger.error(f"Validation error on document: {item}")
@@ -103,17 +103,17 @@ def fetch_documents_from_table(
         return documents
 
     def get_nhs_numbers_based_on_ods_code(
-        self, ods_code: str, table_name: str = None
+        self, ods_code: str, table_name: str | None = None
     ) -> list[str]:
         """Get unique NHS numbers for patients with given ODS code."""
-        table_to_use = table_name or self.table_name
+        table_name = table_name or self.table_name
 
         documents = self.fetch_documents_from_table(
             index_name="OdsCodeIndex",
             search_key=DocumentReferenceMetadataFields.CURRENT_GP_ODS.value,
             search_condition=ods_code,
             query_filter=NotDeleted,
-            table_name=table_to_use,
+            table_name=table_name,
         )
         nhs_numbers = list({document.nhs_number for document in documents})
         return nhs_numbers
@@ -169,15 +169,15 @@ def delete_document_object(self, bucket: str, key: str):
 
     def update_document(
         self,
-        table_name: str = None,
+        table_name: str | None = None,
         document: BaseModel = None,
-        update_fields_name: set[str] = None,
+        update_fields_name: set[str] | None = None,
     ):
         """Update document in specified or configured table."""
-        table_to_use = table_name or self.table_name
+        table_name = table_name or self.table_name
 
         self.dynamo_service.update_item(
-            table_name=table_to_use,
+            table_name=table_name,
             key_pair={DocumentReferenceMetadataFields.ID.value: document.id},
             updated_fields=document.model_dump(
                 exclude_none=True, by_alias=True, include=update_fields_name
@@ -188,14 +188,14 @@ def hard_delete_metadata_records(
         self, table_name: str, document_references: list[BaseModel]
     ):
         """Permanently delete metadata from specified or configured table."""
-        table_to_use = table_name or self.table_name
+        table_name = table_name or self.table_name
 
-        logger.info(f"Deleting items in table: {table_to_use} (HARD DELETE)")
+        logger.info(f"Deleting items in table: {table_name} (HARD DELETE)")
         primary_key_name = DocumentReferenceMetadataFields.ID.value
         for reference in document_references:
             primary_key_value = reference.id
             deletion_key = {primary_key_name: primary_key_value}
-            self.dynamo_service.delete_item(table_to_use, deletion_key)
+            self.dynamo_service.delete_item(table_name, deletion_key)
 
     @staticmethod
     def is_upload_in_process(record: DocumentReference) -> bool:
@@ -212,12 +212,12 @@ def get_batch_document_references_by_id(
     ) -> list:
         table_name = doc_type.get_dynamodb_table_name()
 
-        table_to_use = table_name or self.table_name
-        model_to_use = self.model_class
+        table_name = table_name or self.table_name
+        model_class = self.model_class
 
         response = self.dynamo_service.batch_get_items(
-            table_name=table_to_use, key_list=document_ids
+            table_name=table_name, key_list=document_ids
         )
 
-        found_docs = [model_to_use.model_validate(item) for item in response]
+        found_docs = [model_class.model_validate(item) for item in response]
         return found_docs

lambdas/tests/unit/services/test_document_service.py

@@ -341,13 +341,13 @@ def test_get_nhs_numbers_based_on_ods_code(mock_service, mocker):
     )
 
     result = mock_service.get_nhs_numbers_based_on_ods_code(
-        ods_code, "test_lg_dynamoDB_table"
+        ods_code, MOCK_LG_TABLE_NAME
     )
 
     assert result == [expected_nhs_number]
 
     mock_fetch.assert_called_once_with(
-        table_name="test_lg_dynamoDB_table",
+        table_name=MOCK_LG_TABLE_NAME,
         index_name="OdsCodeIndex",
         search_key=DocumentReferenceMetadataFields.CURRENT_GP_ODS.value,
         search_condition=ods_code,