[PRMP-585] added support for multiple files in sqs message

Author: lillie-dae

lambdas/handlers/review_processor_handler.py

@@ -1,28 +1,25 @@
 import json
+
+from pydantic import ValidationError
 from lambdas.models.sqs.review_message_body import ReviewMessageBody
 from lambdas.services.review_processor_service import ReviewProcessorService
 from utils.audit_logging_setup import LoggingService
 from utils.decorators.ensure_env_var import ensure_environment_variables
-from utils.decorators.handle_lambda_exceptions import handle_lambda_exceptions
 from utils.decorators.override_error_check import override_error_check
 from utils.decorators.set_audit_arg import set_request_context_for_logging
-from utils.decorators.validate_sqs_message_event import validate_sqs_event
-from utils.lambda_response import ApiGatewayResponse
 
 logger = LoggingService(__name__)
 
 
 @set_request_context_for_logging
-@override_error_check
 @ensure_environment_variables(
     names=[
         "DOCUMENT_REVIEW_DYNAMODB_NAME",
         "STAGING_STORE_BUCKET_NAME",
         "PENDING_REVIEW_BUCKET_NAME",
     ]
 )
-@handle_lambda_exceptions
-@validate_sqs_event
+@override_error_check
 def lambda_handler(event, context):
     """
     This handler consumes SQS messages from the document review queue, creates DynamoDB
@@ -34,7 +31,7 @@ def lambda_handler(event, context):
         _context: Lambda context
 
     Returns:
-        ApiGatewayResponse with processing status
+        None
     """
     logger.info("Starting review processor Lambda")
 
@@ -47,23 +44,22 @@ def lambda_handler(event, context):
     for sqs_message in sqs_messages:
         try:
             sqs_message_body = json.loads(sqs_message["body"])
-            message = ReviewMessageBody.model_validate(sqs_message_body)
+            message: ReviewMessageBody = ReviewMessageBody.model_validate(sqs_message_body)
 
             review_service.process_review_message(message)
             processed_count += 1
+        except ValidationError as error:
+            logger.error("Malformed review message")
+            logger.error(error)
+
         except Exception as e:
             logger.error(
                 f"Failed to process review message: {str(e)}",
                 {"Result": "Review processing failed"},
             )
             failed_count += 1
+        logger.info("Continuing to next message.")
 
     logger.info(
         f"Review processor completed: {processed_count} processed, {failed_count} failed"
     )
-
-    return ApiGatewayResponse(
-        status_code=200,
-        body=f"Processed {processed_count} messages",
-        methods="GET",
-    ).create_api_gateway_response()

lambdas/models/sqs/review_message_body.py

@@ -1,12 +1,17 @@
 from pydantic import BaseModel
 
 
-class ReviewMessageBody(BaseModel):
-    """Model for SQS message body from the document review queue."""
+class ReviewMessageFile(BaseModel):
+    """Model for individual file in SQS message body from the document review queue."""
 
     file_name: str
     file_path: str
     """Location in the staging bucket"""
+
+class ReviewMessageBody(BaseModel):
+    """Model for SQS message body from the document review queue."""
+
+    files: list[ReviewMessageFile]
     nhs_number: str
     failure_reason: str
     upload_date: str

lambdas/services/review_processor_service.py

@@ -1,5 +1,6 @@
 import os
 from datetime import datetime, timezone
+import uuid
 
 from enums.review_status import ReviewStatus
 from models.document_review import DocumentReviewFileDetails, DocumentsUploadReview
@@ -43,13 +44,17 @@ def process_review_message(self, review_message: ReviewMessageBody) -> None:
 
         request_context.patient_nhs_no = review_message.nhs_number
 
-        logger.info(f"Processing review for NHS: {review_message.nhs_number}, File: {review_message.file_name}")
+        logger.info(f"Processing review for NHS: {review_message.nhs_number} with {len(review_message.files)} files")
 
-        self._verify_file_exists_in_staging(review_message.file_path)
-        document_upload_review = self._create_review_record(review_message)
+        for file in review_message.files:
+           logger.info(f"Processing review file: {file.file_name}")
+           self._verify_file_exists_in_staging(file.file_path)
 
-        new_file_key = self._move_file_to_review_bucket(review_message, document_upload_review.id)
-        self._update_review_record_with_file_location(document_upload_review.id, new_file_key)
+        review_id = uuid.uuid4().hex
+        files = self._move_files_to_review_bucket(review_message, review_id)
+        document_upload_review = self._build_review_record(review_message, review_id, files)
+
+        self._create_review_record(document_upload_review)
 
         logger.info(
             f"Successfully processed review for {review_message.nhs_number}",
@@ -80,89 +85,64 @@ def _verify_file_exists_in_staging(self, file_path: str) -> None:
             logger.error(f"Error checking file in staging bucket: {str(e)}")
             raise
 
-    def _create_review_record(self, message_data: ReviewMessageBody) -> DocumentsUploadReview:
-        """
-        Create a new review record in DynamoDB.
-
-        Args:
-            message_data: Validated review queue message data
-
-        Returns:
-            Created DocumentsUploadReview object
-
-        Raises:
-            ClientError: If DynamoDB create operation fails
-        """
-        try:
-            files = [DocumentReviewFileDetails(
-                file_name=message_data.file_name,
-                file_location=message_data.file_path
-            )]
-
-            document_review = DocumentsUploadReview(
-                nhs_number=message_data.nhs_number,
-                upload_date=int(datetime.fromisoformat(message_data.upload_date).replace(tzinfo=timezone.utc).timestamp()),
-                review_status=ReviewStatus.PENDING_REVIEW,
-                review_reason=message_data.failure_reason,
-                author=message_data.uploader_ods,
-                custodian=message_data.current_gp,
-                files=files,
-            )
-
-            self.dynamo_service.create_item(
-                table_name=self.review_table_name,
-                item=document_review.model_dump(by_alias=True, exclude_none=True),
-            )
-
-            logger.info(
-                f"Created review record in DynamoDB with ID: {document_review.id}",
-                {"Result": "DynamoDB record created"},
-            )
-
-            return document_review
-
-        except Exception as e:
-            logger.error(f"Failed to create DynamoDB record: {str(e)}")
-            raise
+    def _build_review_record(
+            self, message_data: ReviewMessageBody, review_id: str, files: list[DocumentReviewFileDetails]
+    ) -> DocumentsUploadReview:
+        return DocumentsUploadReview(
+            id=review_id,
+            nhs_number=message_data.nhs_number,
+            review_status=ReviewStatus.PENDING_REVIEW,
+            review_reason=message_data.failure_reason,
+            author=message_data.uploader_ods,
+            custodian=message_data.current_gp,
+            files=files,
+            upload_date=int(datetime.now(tz=timezone.utc).timestamp())
+        )
 
-    def _move_file_to_review_bucket(self, message_data: ReviewMessageBody, review_record_id: str) -> str:
+    def _move_files_to_review_bucket(
+        self, message_data: ReviewMessageBody, review_record_id: str
+    ) -> list[DocumentReviewFileDetails]:
         """
         Move file from staging to review bucket.
 
         Args:
             message_data: Review queue message data
-            review_record_id: ID of the review record (used in destination path)
+            review_record_id: ID of the review record being created
 
         Returns:
-            New file key in review bucket
-
-        Raises:
-            ClientError: If S3 copy or delete operations fail
+            List of DocumentReviewFileDetails objects for the moved files
         """
+        moved_files = []
         try:
-            new_file_key = f"{message_data.nhs_number}/{review_record_id}/{message_data.file_name}"
+            for file in message_data.files:
+                new_file_key = f"{message_data.nhs_number}/{review_record_id}/{file.file_name}"
 
-            logger.info(f"Copying file from ({message_data.file_path}) in staging to review bucket: {new_file_key}")
+                logger.info(f"Copying file from ({file.file_path}) in staging to review bucket: {new_file_key}")
 
-            self.s3_service.copy_across_bucket(
-                source_bucket=self.staging_bucket_name,
-                source_file_key=message_data.file_path,
-                dest_bucket=self.review_bucket_name,
-                dest_file_key=new_file_key,
-            )
+                self.s3_service.copy_across_bucket(
+                    source_bucket=self.staging_bucket_name,
+                    source_file_key=file.file_path,
+                    dest_bucket=self.review_bucket_name,
+                    dest_file_key=new_file_key,
+                )
 
-            logger.info("File successfully copied to review bucket")
-            logger.info(f"Deleting file from staging bucket: {message_data.file_path}")
+                logger.info("File successfully copied to review bucket")
+                logger.info(f"Deleting file from staging bucket: {file.file_path}")
 
-            self._delete_from_staging(message_data.file_path)
-            logger.info(f"Successfully moved file to: {new_file_key}")
+                self._delete_from_staging(file.file_path)
+                logger.info(f"Successfully moved file to: {new_file_key}")
 
-            return new_file_key
+                moved_files.append(DocumentReviewFileDetails(
+                    file_name=file.file_name,
+                    file_location=new_file_key
+                ))
 
         except Exception as e:
             logger.error(f"Failed to move file: {str(e)}")
             raise
 
+        return moved_files
+
     def _delete_from_staging(self, file_key: str) -> None:
         try:
             self.s3_service.delete_object(s3_bucket_name=self.staging_bucket_name, file_key=file_key)
@@ -173,17 +153,15 @@ def _delete_from_staging(self, file_key: str) -> None:
             logger.error(f"Error deleting file from staging: {str(e)}")
             raise
 
-    def _update_review_record_with_file_location(self, review_record_id: str, review_bucket_path: str) -> None:
+    def _create_review_record(self, review_record: DocumentsUploadReview) -> None:
         try:
-            self.dynamo_service.update_item(
+            self.dynamo_service.create_item(
                 table_name=self.review_table_name,
-                key_pair={"ID": review_record_id},
-                updated_fields={"ReviewBucketPath": review_bucket_path},
+                item=review_record
             )
 
-            logger.info(f"Updated review record {review_record_id} with file location: {review_bucket_path}")
+            logger.info(f"Created review record {review_record.id}")
 
         except Exception as e:
-            logger.error(f"Failed to update review record with file location: {str(e)}")
-            logger.warning("Review record created but file location not updated in DynamoDB")
+            logger.error(f"Failed to create review record with id: {review_record.id} -- {str(e)}")
             raise