[PRMT-435] Stitching manual trigger doesn't add all records to queue (#671)

* [PRMT-435]- Add batching to manual trigger when adding entires to the queue

* [PRMT-435]- added tests and better logging

* [PRMT-435]- fixed small issue and added logging

* [PRMT-435]- commented test for now

* [PRMT-435]- commented test for now

* [PRMT-435]- added more logging

* [PRMT-435]- added tests, and fixed logic

* [PRMT-435]- fixed comments

* [PRMT-435]- relocated batch method, and made it less specific

* [PRMT-435]- accepted changed

* [PRMT-435]- fixed test and merged

* [PRMT-435]- added extra logging

Author: PedroSoaresNHS

lambdas/handlers/pdf_stitching_handler.py

@@ -44,10 +44,8 @@ def lambda_handler(event, context):
 @validate_sqs_event
 def handle_sqs_request(event, pdf_stitching_service):
     request_context.app_interaction = LoggingAppInteraction.STITCH_RECORD.value
-
     logger.info("Received PDF Stitching SQS message event")
-    event_message_records = event.get("Records")
-
+    event_message_records = event.get("Records", [])
     for message in event_message_records:
         try:
             request_context.patient_nhs_no = ""
@@ -56,7 +54,6 @@ def handle_sqs_request(event, pdf_stitching_service):
             request_context.patient_nhs_no = stitching_message.nhs_number
             pdf_stitching_service.process_message(stitching_message=stitching_message)
         except (JSONDecodeError, ValidationError) as e:
-            logger.error("Malformed PDF Stitching SQS message")
             logger.error(
                 f"Failed to parse PDF stitching from SQS message: {str(e)}",
                 {"Results": "Failed to stitch PDF"},

lambdas/services/base/sqs_service.py

@@ -1,3 +1,5 @@
+import uuid
+
 import boto3
 from botocore.client import Config as BotoConfig
 
@@ -16,6 +18,23 @@ def send_message_fifo(self, queue_url: str, message_body: str, group_id: str):
     def send_message_standard(self, queue_url: str, message_body: str):
         self.client.send_message(QueueUrl=queue_url, MessageBody=message_body)
 
+    def send_message_batch_standard(self, queue_url: str, messages: list[str], delay=0):
+        entries = []
+        for i, body in enumerate(messages):
+            entries.append(
+                {
+                    "Id": str(uuid.uuid4()),
+                    "MessageBody": body,
+                    "DelaySeconds": delay,
+                }
+            )
+
+        response = self.client.send_message_batch(
+            QueueUrl=queue_url,
+            Entries=entries,
+        )
+        return response
+
     def send_message_with_attr(
         self, queue_url: str, message_body: str, attributes: dict
     ):

lambdas/services/document_service.py

@@ -59,23 +59,31 @@ def fetch_documents_from_table(
         query_filter: Attr | ConditionBase = None,
     ) -> list[DocumentReference]:
         documents = []
+        exclusive_start_key = None
+
+        while True:
+            response = self.dynamo_service.query_table_by_index(
+                table_name=table,
+                index_name=index_name,
+                search_key=search_key,
+                search_condition=search_condition,
+                requested_fields=DocumentReferenceMetadataFields.list(),
+                query_filter=query_filter,
+                exclusive_start_key=exclusive_start_key,
+            )
 
-        response = self.dynamo_service.query_table_by_index(
-            table_name=table,
-            index_name=index_name,
-            search_key=search_key,
-            search_condition=search_condition,
-            query_filter=query_filter,
-        )
-
-        for item in response["Items"]:
-            try:
-                document = DocumentReference.model_validate(item)
-                documents.append(document)
-            except ValidationError as e:
-                logger.error(f"Validation error on document: {item}")
-                logger.error(f"{e}")
-                continue
+            for item in response["Items"]:
+                try:
+                    document = DocumentReference.model_validate(item)
+                    documents.append(document)
+                except ValidationError as e:
+                    logger.error(f"Validation error on document: {item}")
+                    logger.error(f"{e}")
+                    continue
+            if "LastEvaluatedKey" in response:
+                exclusive_start_key = response["LastEvaluatedKey"]
+            else:
+                break
         return documents
 
     def get_nhs_numbers_based_on_ods_code(self, ods_code: str) -> list[str]:

lambdas/services/mock_pds_service.py

@@ -27,19 +27,27 @@ def pds_request(self, nhs_number: str, *args, **kwargs) -> Response:
 
         try:
             for file in all_mock_files:
-                with open(file) as f:
-                    mock_pds_results.append(json.load(f))
+                with open(file) as mock_file:
+                    mock_pds_results.append(json.load(mock_file))
 
         except FileNotFoundError:
             raise PdsErrorException("Error when requesting patient from PDS")
 
         pds_patient: dict = {}
         response = Response()
         if self.always_pass_mock:
-            pds_patient_index = 3
-            pds_patient = mock_pds_results[pds_patient_index]
+            mock_file_name = "pds_patient_9000000068_M85143_gp.json"
+            file_path = os.path.join(
+                parent_dir_of_this_file, "services", "mock_data", mock_file_name
+            )
+            try:
+                with open(file_path) as open_file:
+                    pds_patient = json.load(open_file)
+            except FileNotFoundError:
+                raise PdsErrorException(f"Mock file '{mock_file_name}' not found")
             pds_patient["id"] = nhs_number
             pds_patient["identifier"][0]["value"] = nhs_number
+            print(f"pds_patient={pds_patient}")
         else:
             for result in mock_pds_results:
                 mock_patient_nhs_number = result.get("id")