[PRMP-189] Update BulkUploadMetadataProcessor lambda to add stored_file_name to the bulk upload queue metadata (#807)

Author: PedroSoaresNHS

lambdas/models/staging_metadata.py

@@ -1,7 +1,4 @@
-from typing import Optional
-
-from pydantic import BaseModel, ConfigDict, Field, ValidationInfo, field_validator
-from pydantic_core import PydanticCustomError
+from pydantic import BaseModel, ConfigDict, Field, field_validator
 
 METADATA_FILENAME = "metadata.csv"
 NHS_NUMBER_FIELD_NAME = "NHS-NO"
@@ -13,44 +10,38 @@ def to_upper_case_with_hyphen(field_name: str) -> str:
     return field_name.upper().replace("_", "-")
 
 
-class MetadataFile(BaseModel):
+class MetadataBase(BaseModel):
     model_config = ConfigDict(
-        alias_generator=to_upper_case_with_hyphen, validate_by_name=True
+        validate_by_name=True,
+        populate_by_name=True,
     )
 
-    file_path: str = Field(alias="FILEPATH")
-    page_count: str = Field(alias="PAGE COUNT")
-    nhs_number: Optional[str] = Field(
-        alias=NHS_NUMBER_FIELD_NAME, exclude=True, default=None
-    )
-    gp_practice_code: str
-    section: str
-    sub_section: Optional[str]
+    file_path: str
+    gp_practice_code: str = Field(min_length=1)
     scan_date: str
-    scan_id: str
-    user_id: str
-    upload: str
 
-    @field_validator("gp_practice_code")
-    @classmethod
-    def ensure_gp_practice_code_non_empty(
-        cls, gp_practice_code: str, info: ValidationInfo
-    ) -> str:
-        if not gp_practice_code:
-            patient_nhs_number = info.data.get("nhs_number", "")
-            raise PydanticCustomError(
-                "MissingGPPracticeCode",
-                "missing GP-PRACTICE-CODE for patient {patient_nhs_number}",
-                {"patient_nhs_number": patient_nhs_number},
-            )
-        return gp_practice_code
-
-
-class StagingMetadata(BaseModel):
-    model_config = ConfigDict(validate_by_name=True)
-
-    nhs_number: str = Field(default=NHS_NUMBER_PLACEHOLDER, alias=NHS_NUMBER_FIELD_NAME)
-    files: list[MetadataFile]
+
+class BulkUploadQueueMetadata(MetadataBase):
+    stored_file_name: str
+
+
+class MetadataFile(MetadataBase):
+    model_config = ConfigDict(
+        alias_generator=to_upper_case_with_hyphen,
+    )
+    nhs_number: str = Field(alias=NHS_NUMBER_FIELD_NAME)
+    file_path: str = Field(alias="FILEPATH")
+    page_count: str = Field(default=None, alias="PAGE COUNT")
+    section: str = None
+    sub_section: str = None
+    scan_id: str = None
+    user_id: str = None
+    upload: str = None
+
+
+class StagingSqsMetadata(BaseModel):
+    nhs_number: str
+    files: list[BulkUploadQueueMetadata]
     retries: int = 0
 
     @field_validator("nhs_number")

lambdas/repositories/bulk_upload/bulk_upload_dynamo_repository.py

@@ -4,7 +4,7 @@
 from enums.upload_status import UploadStatus
 from models.document_reference import DocumentReference
 from models.report.bulk_upload_report import BulkUploadReport
-from models.staging_metadata import StagingMetadata
+from models.staging_metadata import StagingSqsMetadata
 from services.base.dynamo_service import DynamoDBService
 from utils.audit_logging_setup import LoggingService
 
@@ -29,7 +29,7 @@ def create_record_in_lg_dynamo_table(self, document_reference: DocumentReference
 
     def write_report_upload_to_dynamo(
         self,
-        staging_metadata: StagingMetadata,
+        staging_metadata: StagingSqsMetadata,
         upload_status: UploadStatus,
         reason: str = None,
         pds_ods_code: str = "",

lambdas/repositories/bulk_upload/bulk_upload_s3_repository.py

@@ -3,7 +3,7 @@
 from botocore.exceptions import ClientError
 from enums.virus_scan_result import SCAN_RESULT_TAG_KEY, VirusScanResult
 from models.document_reference import DocumentReference
-from models.staging_metadata import StagingMetadata
+from models.staging_metadata import StagingSqsMetadata
 from services.base.s3_service import S3Service
 from utils.audit_logging_setup import LoggingService
 from utils.exceptions import (
@@ -28,7 +28,7 @@ def __init__(self):
         self.dest_bucket_files_in_transaction = []
 
     def check_virus_result(
-        self, staging_metadata: StagingMetadata, file_path_cache: dict
+        self, staging_metadata: StagingSqsMetadata, file_path_cache: dict
     ):
         for file_metadata in staging_metadata.files:
             file_path = file_metadata.file_path

lambdas/repositories/bulk_upload/bulk_upload_sqs_repository.py

@@ -2,7 +2,7 @@
 import uuid
 
 from models.sqs.pdf_stitching_sqs_message import PdfStitchingSqsMessage
-from models.staging_metadata import StagingMetadata
+from models.staging_metadata import StagingSqsMetadata
 from services.base.sqs_service import SQSService
 from utils.audit_logging_setup import LoggingService
 from utils.request_context import request_context
@@ -16,7 +16,7 @@ def __init__(self):
         self.invalid_queue_url = os.environ["INVALID_SQS_QUEUE_URL"]
         self.metadata_queue_url = os.environ["METADATA_SQS_QUEUE_URL"]
 
-    def put_staging_metadata_back_to_queue(self, staging_metadata: StagingMetadata):
+    def put_staging_metadata_back_to_queue(self, staging_metadata: StagingSqsMetadata):
         request_context.patient_nhs_no = staging_metadata.nhs_number
         setattr(staging_metadata, "retries", (staging_metadata.retries + 1))
         _logger.info("Returning message to sqs queue...")

lambdas/services/bulk_upload_metadata_processor_service.py

@@ -3,13 +3,19 @@
 import shutil
 import tempfile
 import uuid
+from collections import defaultdict
 from datetime import datetime
 from typing import Iterable
 
 import pydantic
 from botocore.exceptions import ClientError
 from enums.upload_status import UploadStatus
-from models.staging_metadata import METADATA_FILENAME, MetadataFile, StagingMetadata
+from models.staging_metadata import (
+    METADATA_FILENAME,
+    BulkUploadQueueMetadata,
+    MetadataFile,
+    StagingSqsMetadata,
+)
 from repositories.bulk_upload.bulk_upload_dynamo_repository import (
     BulkUploadDynamoRepository,
 )
@@ -41,7 +47,6 @@ def __init__(self, metadata_formatter_service: MetadataPreprocessorService):
 
         self.temp_download_dir = tempfile.mkdtemp()
 
-        self.corrections = {}
         self.practice_directory = metadata_formatter_service.practice_directory
         self.file_key = (
             f"{metadata_formatter_service.practice_directory}/{METADATA_FILENAME}"
@@ -53,7 +58,7 @@ def __init__(self, metadata_formatter_service: MetadataPreprocessorService):
     def process_metadata(self):
         try:
             metadata_file = self.download_metadata_from_s3()
-            staging_metadata_list = self.csv_to_staging_metadata(metadata_file)
+            staging_metadata_list = self.csv_to_sqs_metadata(metadata_file)
             logger.info("Finished parsing metadata")
 
             self.send_metadata_to_fifo_sqs(staging_metadata_list)
@@ -90,50 +95,59 @@ def download_metadata_from_s3(self) -> str:
         )
         return local_file_path
 
-    def csv_to_staging_metadata(self, csv_file_path: str) -> list[StagingMetadata]:
+    def csv_to_sqs_metadata(self, csv_file_path: str) -> list[StagingSqsMetadata]:
         logger.info("Parsing bulk upload metadata")
-        patients = {}
+        patients: defaultdict[tuple[str, str], list[BulkUploadQueueMetadata]] = (
+            defaultdict(list)
+        )
+
         with open(
             csv_file_path, mode="r", encoding="utf-8-sig", errors="replace"
         ) as csv_file_handler:
             csv_reader: Iterable[dict] = csv.DictReader(csv_file_handler)
             for row in csv_reader:
                 self.process_metadata_row(row, patients)
+
         return [
-            StagingMetadata(
-                nhs_number=key[0],
-                files=value,
+            StagingSqsMetadata(
+                nhs_number=nhs_number,
+                files=files,
             )
-            for (key, value) in patients.items()
+            for (nhs_number, _), files in patients.items()
         ]
 
-    def process_metadata_row(self, row: dict, patients: dict) -> None:
+    def process_metadata_row(
+        self, row: dict, patients: dict[tuple[str, str], list[BulkUploadQueueMetadata]]
+    ) -> None:
         file_metadata = MetadataFile.model_validate(row)
         nhs_number, ods_code = self.extract_patient_info(file_metadata)
-        patient_record_key = (nhs_number, ods_code)
-
-        if patient_record_key not in patients:
-            patients[patient_record_key] = [file_metadata]
-        else:
-            patients[patient_record_key].append(file_metadata)
 
         try:
-            self.validate_correct_filename(file_metadata)
+            correct_file_name = self.validate_and_correct_filename(file_metadata)
         except InvalidFileNameException as error:
-            self.handle_invalid_filename(
-                file_metadata, error, patient_record_key, patients
-            )
-            patients.pop(patient_record_key)
+            self.handle_invalid_filename(file_metadata, error, nhs_number)
+            return
+
+        sqs_metadata = self.convert_to_sqs_metadata(file_metadata, correct_file_name)
+        patients[(nhs_number, ods_code)].append(sqs_metadata)
+
+    @staticmethod
+    def convert_to_sqs_metadata(
+        file: MetadataFile, stored_file_name: str
+    ) -> BulkUploadQueueMetadata:
+        return BulkUploadQueueMetadata(
+            **file.model_dump(), stored_file_name=stored_file_name
+        )
 
     def extract_patient_info(self, file_metadata: MetadataFile) -> tuple[str, str]:
         nhs_number = file_metadata.nhs_number
         ods_code = file_metadata.gp_practice_code
         return nhs_number, ods_code
 
-    def validate_correct_filename(
+    def validate_and_correct_filename(
         self,
         file_metadata: MetadataFile,
-    ) -> None:
+    ) -> str:
         try:
             validate_file_name(file_metadata.file_path.split("/")[-1])
             valid_filepath = file_metadata.file_path
@@ -142,39 +156,40 @@ def validate_correct_filename(
                 file_metadata.file_path
             )
 
-        if valid_filepath:
-            self.corrections[file_metadata.file_path] = valid_filepath
+        return valid_filepath
 
     def handle_invalid_filename(
         self,
         file_metadata: MetadataFile,
         error: InvalidFileNameException,
-        key: tuple[str, str],
-        patients: dict[tuple[str, str], list[MetadataFile]],
+        nhs_number: str,
     ) -> None:
         logger.error(
             f"Failed to process {file_metadata.file_path} due to error: {error}"
         )
-        failed_entry = StagingMetadata(
-            nhs_number=key[0],
-            files=patients[key],
+        failed_file = self.convert_to_sqs_metadata(
+            file_metadata, file_metadata.file_path
+        )
+        failed_entry = StagingSqsMetadata(
+            nhs_number=nhs_number,
+            files=[failed_file],
         )
         self.dynamo_repository.write_report_upload_to_dynamo(
             failed_entry, UploadStatus.FAILED, str(error)
         )
 
     def send_metadata_to_fifo_sqs(
-        self, staging_metadata_list: list[StagingMetadata]
+        self, staging_sqs_metadata_list: list[StagingSqsMetadata]
     ) -> None:
         sqs_group_id = f"bulk_upload_{uuid.uuid4()}"
 
-        for staging_metadata in staging_metadata_list:
-            nhs_number = staging_metadata.nhs_number
+        for staging_sqs_metadata in staging_sqs_metadata_list:
+            nhs_number = staging_sqs_metadata.nhs_number
             logger.info(f"Sending metadata for patientId: {nhs_number}")
 
             self.sqs_service.send_message_with_nhs_number_attr_fifo(
                 queue_url=self.metadata_queue_url,
-                message_body=staging_metadata.model_dump_json(by_alias=True),
+                message_body=staging_sqs_metadata.model_dump_json(by_alias=True),
                 nhs_number=nhs_number,
                 group_id=sqs_group_id,
             )