[PRMP-1512] - Add message to sqs-stitching queue during bulk upload (#554)

Author: MohammadIqbalAD-NHS

lambdas/models/pdf_stitching_sqs_message.py

@@ -0,0 +1,7 @@
+from enums.snomed_codes import SnomedCode, SnomedCodes
+from pydantic import BaseModel
+
+
+class PdfStitchingSqsMessage(BaseModel):
+    nhs_number: str
+    snomed_code_doc_type: SnomedCode = SnomedCodes.LLOYD_GEORGE.value

lambdas/repositories/bulk_upload/bulk_upload_sqs_repository.py

@@ -1,7 +1,7 @@
 import os
 import uuid
 
-from models.nrl_sqs_message import NrlSqsMessage
+from models.pdf_stitching_sqs_message import PdfStitchingSqsMessage
 from models.staging_metadata import StagingMetadata
 from services.base.sqs_service import SQSService
 from utils.audit_logging_setup import LoggingService
@@ -42,11 +42,10 @@ def put_sqs_message_back_to_queue(self, sqs_message: dict):
             group_id=f"back_to_queue_bulk_upload_{uuid.uuid4()}",
         )
 
-    def send_message_to_nrl_fifo(
-        self, queue_url: str, message: NrlSqsMessage, group_id: str
+    def send_message_to_pdf_stitching_queue(
+        self, queue_url: str, message: PdfStitchingSqsMessage
     ):
-        self.sqs_repository.send_message_fifo(
+        self.sqs_repository.send_message_standard(
             queue_url=queue_url,
             message_body=message.model_dump_json(),
-            group_id=group_id,
         )

lambdas/services/bulk_upload_service.py

@@ -4,14 +4,12 @@
 
 import pydantic
 from botocore.exceptions import ClientError
-from enums.nrl_sqs_upload import NrlActionTypes
 from enums.patient_ods_inactive_status import PatientOdsInactiveStatus
 from enums.snomed_codes import SnomedCodes
 from enums.upload_status import UploadStatus
 from enums.virus_scan_result import VirusScanResult
-from models.fhir.R4.nrl_fhir_document_reference import Attachment
 from models.nhs_document_reference import NHSDocumentReference
-from models.nrl_sqs_message import NrlSqsMessage
+from models.pdf_stitching_sqs_message import PdfStitchingSqsMessage
 from models.staging_metadata import MetadataFile, StagingMetadata
 from repositories.bulk_upload.bulk_upload_dynamo_repository import (
     BulkUploadDynamoRepository,
@@ -57,7 +55,7 @@ def __init__(self, strict_mode):
         self.pdf_content_type = "application/pdf"
         self.unhandled_messages = []
         self.file_path_cache = {}
-        self.nrl_queue_url = os.environ["NRL_SQS_URL"]
+        self.pdf_stitching_queue_url = os.environ["PDF_STITCHING_SQS_URL"]
 
     def process_message_queue(self, records: list):
         for index, message in enumerate(records, start=1):
@@ -251,9 +249,7 @@ def handle_sqs_message(self, message: dict):
         )
 
         try:
-            last_document_processed = self.create_lg_records_and_copy_files(
-                staging_metadata, patient_ods_code
-            )
+            self.create_lg_records_and_copy_files(staging_metadata, patient_ods_code)
             logger.info(
                 f"Successfully uploaded the Lloyd George records for patient: {staging_metadata.nhs_number}",
                 {"Result": "Successful upload"},
@@ -290,28 +286,18 @@ def handle_sqs_message(self, message: dict):
             accepted_reason,
             patient_ods_code,
         )
-        if len(file_names) == 1:
-            document_api_endpoint = (
-                os.environ.get("APIM_API_URL", "")
-                + "/DocumentReference/"
-                + SnomedCodes.LLOYD_GEORGE.value.code
-                + "~"
-                + last_document_processed.id
-            )
-            doc_details = Attachment(
-                url=document_api_endpoint,
-                content_type="application/pdf",
-            )
-            nrl_sqs_message = NrlSqsMessage(
-                nhs_number=staging_metadata.nhs_number,
-                action=NrlActionTypes.CREATE,
-                attachment=doc_details,
-            )
-            self.sqs_repository.send_message_to_nrl_fifo(
-                queue_url=self.nrl_queue_url,
-                message=nrl_sqs_message,
-                group_id=f"nrl_sqs_{uuid.uuid4()}",
-            )
+
+        pdf_stitching_sqs_message = PdfStitchingSqsMessage(
+            nhs_number=staging_metadata.nhs_number,
+            snomed_code_doc_type=SnomedCodes.LLOYD_GEORGE.value,
+        )
+        self.sqs_repository.send_message_to_pdf_stitching_queue(
+            queue_url=self.pdf_stitching_queue_url,
+            message=pdf_stitching_sqs_message,
+        )
+        logger.info(
+            f"Message sent to stitching queue for patient {staging_metadata.nhs_number}"
+        )
 
     def resolve_source_file_path(self, staging_metadata: StagingMetadata):
         sample_file_path = staging_metadata.files[0].file_path
@@ -357,7 +343,6 @@ def create_lg_records_and_copy_files(
         self, staging_metadata: StagingMetadata, current_gp_ods: str
     ):
         nhs_number = staging_metadata.nhs_number
-        document_reference = None
         for file_metadata in staging_metadata.files:
             document_reference = self.convert_to_document_reference(
                 file_metadata, nhs_number, current_gp_ods
@@ -371,8 +356,6 @@ def create_lg_records_and_copy_files(
             )
             document_reference.set_uploaded_to_true()
             self.dynamo_repository.create_record_in_lg_dynamo_table(document_reference)
-        # returning last document ref until stitching as default is implemented
-        return document_reference
 
     def rollback_transaction(self):
         try:

lambdas/tests/unit/conftest.py

@@ -29,6 +29,7 @@
 MOCK_ZIP_OUTPUT_BUCKET_ENV_NAME = "ZIPPED_STORE_BUCKET_NAME"
 MOCK_ZIP_TRACE_TABLE_ENV_NAME = "ZIPPED_STORE_DYNAMODB_NAME"
 MOCK_METADATA_NRL_SQS_URL_ENV_NAME = "NRL_SQS_URL"
+MOCK_PDF_STITCHING_SQS_URL_ENV_NAME = "PDF_STITCHING_SQS_URL"
 
 MOCK_LG_STAGING_STORE_BUCKET_ENV_NAME = "STAGING_STORE_BUCKET_NAME"
 MOCK_LG_METADATA_SQS_QUEUE_ENV_NAME = "METADATA_SQS_QUEUE_URL"
@@ -113,6 +114,10 @@
 NRL_SQS_URL = "https://sqs.us-east-1.amazonaws.com/177715257436/MyQueue"
 APIM_API_URL = "https://apim.api.service.uk"
 
+PDF_STITCHING_SQS_URL = (
+    "https://sqs.us-east-1.amazonaws.com/977715257439/MyPdfStitchingQueue"
+)
+
 
 @pytest.fixture
 def set_env(monkeypatch):
@@ -131,6 +136,7 @@ def set_env(monkeypatch):
     monkeypatch.setenv(MOCK_LG_INVALID_SQS_QUEUE_ENV_NAME, MOCK_LG_INVALID_SQS_QUEUE)
     monkeypatch.setenv(MOCK_AUTH_STATE_TABLE_NAME_ENV_NAME, AUTH_STATE_TABLE_NAME)
     monkeypatch.setenv(MOCK_METADATA_NRL_SQS_URL_ENV_NAME, NRL_SQS_URL)
+    monkeypatch.setenv(MOCK_PDF_STITCHING_SQS_URL_ENV_NAME, PDF_STITCHING_SQS_URL)
     monkeypatch.setenv(MOCK_AUTH_SESSION_TABLE_NAME_ENV_NAME, AUTH_SESSION_TABLE_NAME)
     monkeypatch.setenv(MOCK_OIDC_CALLBACK_URL_ENV_NAME, OIDC_CALLBACK_URL)
     monkeypatch.setenv(MOCK_OIDC_CLIENT_ID_ENV_NAME, OIDC_CLIENT_ID)

lambdas/tests/unit/helpers/data/bulk_upload/test_data.py

@@ -1,9 +1,11 @@
 import os
 
+from enums.snomed_codes import SnomedCodes
 from enums.virus_scan_result import VirusScanResult
 from freezegun import freeze_time
 from models.nhs_document_reference import NHSDocumentReference
 from models.nrl_sqs_message import NrlSqsMessage
+from models.pdf_stitching_sqs_message import PdfStitchingSqsMessage
 from models.staging_metadata import MetadataFile, StagingMetadata
 from tests.unit.conftest import MOCK_LG_BUCKET, TEST_CURRENT_GP_ODS, TEST_UUID
 
@@ -155,6 +157,17 @@ def build_test_nrl_sqs_fifo_message(nhs_number: str, action: str) -> NrlSqsMessa
     return nrl_sqs_message
 
 
+def build_test_pdf_stitching_sqs_message(
+    nhs_number: str, snomed_code_doc_type
+) -> PdfStitchingSqsMessage:
+    message_body = {
+        "nhs_number": nhs_number,
+        "snomed_code_doc_type": snomed_code_doc_type,
+    }
+    pdf_stitching_sqs_message = PdfStitchingSqsMessage(**message_body)
+    return pdf_stitching_sqs_message
+
+
 @freeze_time("2024-01-01 12:00:00")
 def build_test_document_reference(file_name: str, nhs_number: str = "9000000009"):
     doc_ref = NHSDocumentReference(
@@ -181,6 +194,10 @@ def build_test_document_reference(file_name: str, nhs_number: str = "9000000009"
 TEST_NRL_SQS_MESSAGE = build_test_nrl_sqs_fifo_message(
     TEST_NHS_NUMBER_FOR_BULK_UPLOAD, NrlActionTypes.CREATE
 )
+TEST_SNOMED_CODE_FOR_PDF_STITCHING = SnomedCodes.LLOYD_GEORGE.value
+TEST_PDF_STITCHING_SQS_MESSAGE = build_test_pdf_stitching_sqs_message(
+    TEST_NHS_NUMBER_FOR_BULK_UPLOAD, TEST_SNOMED_CODE_FOR_PDF_STITCHING
+)
 TEST_STAGING_METADATA_WITH_INVALID_FILENAME = build_test_staging_metadata(
     [*make_valid_lg_file_names(2), "invalid_file_name.txt"]
 )

lambdas/tests/unit/repositories/bulk_upload/test_bulk_upload_sqs_repository.py

@@ -2,14 +2,16 @@
 
 import pytest
 from repositories.bulk_upload.bulk_upload_sqs_repository import BulkUploadSqsRepository
-from tests.unit.conftest import MOCK_LG_METADATA_SQS_QUEUE, NRL_SQS_URL
+from tests.unit.conftest import MOCK_LG_METADATA_SQS_QUEUE, PDF_STITCHING_SQS_URL
 from tests.unit.helpers.data.bulk_upload.test_data import (
-    TEST_GROUP_ID,
     TEST_NHS_NUMBER_FOR_BULK_UPLOAD,
-    TEST_NRL_SQS_MESSAGE,
+    TEST_PDF_STITCHING_SQS_MESSAGE,
     TEST_SQS_MESSAGE,
     TEST_STAGING_METADATA,
 )
+from utils.audit_logging_setup import LoggingService
+
+logger = LoggingService(__name__)
 
 
 @pytest.fixture
@@ -47,15 +49,13 @@ def test_put_sqs_message_back_to_queue(set_env, repo_under_test, mock_uuid):
     )
 
 
-def test_send_message_to_nrl_sqs_fifo(set_env, repo_under_test):
-    repo_under_test.send_message_to_nrl_fifo(
-        NRL_SQS_URL,
-        TEST_NRL_SQS_MESSAGE,
-        TEST_GROUP_ID,
+def test_send_message_to_pdf_stitching_queue(set_env, repo_under_test):
+    repo_under_test.send_message_to_pdf_stitching_queue(
+        PDF_STITCHING_SQS_URL,
+        TEST_PDF_STITCHING_SQS_MESSAGE,
     )
-    message_body = TEST_NRL_SQS_MESSAGE
-    repo_under_test.sqs_repository.send_message_fifo.assert_called_with(
-        queue_url=NRL_SQS_URL,
+    message_body = TEST_PDF_STITCHING_SQS_MESSAGE
+    repo_under_test.sqs_repository.send_message_standard.assert_called_with(
+        queue_url=PDF_STITCHING_SQS_URL,
         message_body=message_body.model_dump_json(),
-        group_id="123",
     )

lambdas/tests/unit/services/test_bulk_upload_service.py

@@ -3,23 +3,17 @@
 
 import pytest
 from botocore.exceptions import ClientError
-from enums.nrl_sqs_upload import NrlActionTypes
 from enums.patient_ods_inactive_status import PatientOdsInactiveStatus
-from enums.snomed_codes import SnomedCodes
 from enums.upload_status import UploadStatus
 from enums.virus_scan_result import SCAN_RESULT_TAG_KEY, VirusScanResult
 from freezegun import freeze_time
-from models.fhir.R4.nrl_fhir_document_reference import Attachment
-from models.nrl_sqs_message import NrlSqsMessage
 from models.pds_models import Patient
 from repositories.bulk_upload.bulk_upload_s3_repository import BulkUploadS3Repository
 from repositories.bulk_upload.bulk_upload_sqs_repository import BulkUploadSqsRepository
 from services.bulk_upload_service import BulkUploadService
 from tests.unit.conftest import (
-    APIM_API_URL,
     MOCK_LG_BUCKET,
     MOCK_STAGING_STORE_BUCKET,
-    NRL_SQS_URL,
     TEST_CURRENT_GP_ODS,
 )
 from tests.unit.helpers.data.bulk_upload.test_data import (
@@ -236,7 +230,7 @@ def test_handle_sqs_message_happy_path(
     mock_pds_validation_strict.assert_called()
     mock_report_upload_complete.assert_called()
     mock_remove_ingested_file_from_source_bucket.assert_called()
-    repo_under_test.sqs_repository.send_message_to_nrl_fifo.assert_not_called()
+    repo_under_test.sqs_repository.send_message_to_pdf_stitching_queue.assert_called()
 
 
 def test_handle_sqs_message_happy_path_single_file(
@@ -250,14 +244,6 @@ def test_handle_sqs_message_happy_path_single_file(
     mock_ods_validation,
 ):
     TEST_STAGING_METADATA.retries = 0
-    mock_nrl_attachment = Attachment(
-        url=f"{APIM_API_URL}/DocumentReference/{SnomedCodes.LLOYD_GEORGE.value.code}~{TEST_DOCUMENT_REFERENCE.id}",
-    )
-    mock_nrl_message = NrlSqsMessage(
-        nhs_number=TEST_STAGING_METADATA.nhs_number,
-        action=NrlActionTypes.CREATE,
-        attachment=mock_nrl_attachment,
-    )
     mock_create_lg_records_and_copy_files = mocker.patch.object(
         BulkUploadService, "create_lg_records_and_copy_files"
     )
@@ -277,11 +263,7 @@ def test_handle_sqs_message_happy_path_single_file(
     )
     mock_report_upload_complete.assert_called()
     mock_remove_ingested_file_from_source_bucket.assert_called()
-    repo_under_test.sqs_repository.send_message_to_nrl_fifo.assert_called_with(
-        queue_url=NRL_SQS_URL,
-        message=mock_nrl_message,
-        group_id=f"nrl_sqs_{mock_uuid}",
-    )
+    repo_under_test.sqs_repository.send_message_to_pdf_stitching_queue.assert_called()
 
 
 def set_up_mocks_for_non_ascii_files(
@@ -394,7 +376,7 @@ def test_handle_sqs_message_calls_report_upload_failure_when_patient_record_alre
     mock_report_upload_failure.assert_called_with(
         TEST_STAGING_METADATA, UploadStatus.FAILED, str(mocked_error), ""
     )
-    repo_under_test.sqs_repository.send_message_to_nrl_fifo.assert_not_called()
+    repo_under_test.sqs_repository.send_message_to_pdf_stitching_queue.assert_not_called()
 
 
 def test_handle_sqs_message_calls_report_upload_failure_when_lg_file_name_invalid(
@@ -431,7 +413,7 @@ def test_handle_sqs_message_calls_report_upload_failure_when_lg_file_name_invali
         str(mocked_error),
         "",
     )
-    repo_under_test.sqs_repository.send_message_to_nrl_fifo.assert_not_called()
+    repo_under_test.sqs_repository.send_message_to_pdf_stitching_queue.assert_not_called()
 
 
 def test_handle_sqs_message_report_failure_when_document_is_infected(
@@ -469,7 +451,7 @@ def test_handle_sqs_message_report_failure_when_document_is_infected(
     )
     mock_create_lg_records_and_copy_files.assert_not_called()
     mock_remove_ingested_file_from_source_bucket.assert_not_called()
-    repo_under_test.sqs_repository.send_message_to_nrl_fifo.assert_not_called()
+    repo_under_test.sqs_repository.send_message_to_pdf_stitching_queue.assert_not_called()
 
 
 def test_handle_sqs_message_report_failure_when_document_not_exist(
@@ -499,7 +481,7 @@ def test_handle_sqs_message_report_failure_when_document_not_exist(
         "One or more of the files is not accessible from staging bucket",
         "Y12345",
     )
-    repo_under_test.sqs_repository.send_message_to_nrl_fifo.assert_not_called()
+    repo_under_test.sqs_repository.send_message_to_pdf_stitching_queue.assert_not_called()
 
 
 def test_handle_sqs_message_calls_report_upload_successful_when_patient_is_formally_deceased(
@@ -685,7 +667,7 @@ def test_handle_sqs_message_put_staging_metadata_back_to_queue_when_virus_scan_r
     mock_report_upload_failure.assert_not_called()
     mock_create_lg_records_and_copy_files.assert_not_called()
     mock_remove_ingested_file_from_source_bucket.assert_not_called()
-    repo_under_test.sqs_repository.send_message_to_nrl_fifo.assert_not_called()
+    repo_under_test.sqs_repository.send_message_to_pdf_stitching_queue.assert_not_called()
 
 
 def test_handle_sqs_message_rollback_transaction_when_validation_pass_but_file_transfer_failed_halfway(
@@ -1039,7 +1021,6 @@ def test_handle_sqs_message_lenient_mode_happy_path(
     mock_pds_validation_strict.assert_not_called()
     mock_report_upload_complete.assert_called()
     mock_remove_ingested_file_from_source_bucket.assert_called()
-    service.sqs_repository.send_message_to_nrl_fifo.assert_not_called()
 
 
 def test_concatenate_acceptance_reason(repo_under_test):