Merge branch 'master' into VED-755-Handling-NHS-Confusions

Author: Akol125

batch_processor_filter/src/batch_audit_repository.py

@@ -12,6 +12,7 @@ class BatchAuditRepository:
         | Key(AuditTableKeys.STATUS).eq(FileStatus.PREPROCESSED)
         | Key(AuditTableKeys.STATUS).eq(FileStatus.PROCESSING)
     )
+    _PROCESSING_AND_FAILED_STATUSES = {FileStatus.PROCESSING, FileStatus.FAILED}
 
     def __init__(self):
         self._batch_audit_table = boto3.resource("dynamodb", region_name=REGION_NAME).Table(AUDIT_TABLE_NAME)
@@ -25,16 +26,20 @@ def is_duplicate_file(self, file_key: str) -> bool:
 
         return len(matching_files) > 0
 
-    def is_event_processing_for_supplier_and_vacc_type(self, supplier: str, vacc_type: str) -> bool:
+    def is_event_processing_or_failed_for_supplier_and_vacc_type(self, supplier: str, vacc_type: str) -> bool:
         queue_name = f"{supplier}_{vacc_type}"
 
-        files_in_processing = self._batch_audit_table.query(
-            IndexName=AUDIT_TABLE_QUEUE_NAME_GSI,
-            KeyConditionExpression=Key(AuditTableKeys.QUEUE_NAME).eq(queue_name) & Key(AuditTableKeys.STATUS)
-            .eq(FileStatus.PROCESSING)
-        ).get("Items", [])
+        for status in self._PROCESSING_AND_FAILED_STATUSES:
+            files_in_queue = self._batch_audit_table.query(
+                IndexName=AUDIT_TABLE_QUEUE_NAME_GSI,
+                KeyConditionExpression=Key(AuditTableKeys.QUEUE_NAME).eq(queue_name) & Key(AuditTableKeys.STATUS)
+                .eq(status)
+            ).get("Items", [])
+
+            if len(files_in_queue) > 0:
+                return True
 
-        return len(files_in_processing) > 0
+        return False
 
     def update_status(self, message_id: str, updated_status: str) -> None:
         self._batch_audit_table.update_item(

batch_processor_filter/src/batch_processor_filter_service.py

@@ -1,19 +1,15 @@
 """Batch processor filter service module"""
 import boto3
 import json
-import logging
 
 from batch_audit_repository import BatchAuditRepository
 from batch_file_created_event import BatchFileCreatedEvent
 from batch_file_repository import BatchFileRepository
-from constants import REGION_NAME, FileStatus, QUEUE_URL
+from constants import REGION_NAME, FileStatus, QUEUE_URL, FileNotProcessedReason
 from exceptions import EventAlreadyProcessingForSupplierAndVaccTypeError
+from logger import logger
 from send_log_to_firehose import send_log_to_firehose
 
-logging.basicConfig(level="INFO")
-logger = logging.getLogger()
-logger.setLevel("INFO")
-
 
 class BatchProcessorFilterService:
     """Batch processor filter service class. Provides the business logic for the Lambda function"""
@@ -38,15 +34,21 @@ def apply_filter(self, batch_file_created_event: BatchFileCreatedEvent) -> None:
 
         if self._is_duplicate_file(filename):
             # Mark as processed and return without error so next event will be picked up from queue
-            logger.info("A duplicate file has already been processed. Filename: %s", filename)
-            self._batch_audit_repository.update_status(message_id, FileStatus.DUPLICATE)
+            logger.error("A duplicate file has already been processed. Filename: %s", filename)
+            self._batch_audit_repository.update_status(
+                message_id,
+                f"{FileStatus.NOT_PROCESSED} - {FileNotProcessedReason.DUPLICATE}"
+            )
             self._batch_file_repo.upload_failure_ack(batch_file_created_event)
             self._batch_file_repo.move_source_file_to_archive(filename)
             return
 
-        if self._batch_audit_repository.is_event_processing_for_supplier_and_vacc_type(supplier, vaccine_type):
+        if self._batch_audit_repository.is_event_processing_or_failed_for_supplier_and_vacc_type(
+            supplier,
+            vaccine_type
+        ):
             # Raise error so event is returned to queue and retried again later
-            logger.info("Batch event already being processed for supplier and vacc type. Filename: %s", filename)
+            logger.info("Batch event already processing for supplier and vacc type. Filename: %s", filename)
             raise EventAlreadyProcessingForSupplierAndVaccTypeError(f"Batch event already processing for supplier: "
                                                                     f"{supplier} and vacc type: {vaccine_type}")
 

batch_processor_filter/src/constants.py

@@ -18,7 +18,13 @@ class FileStatus(StrEnum):
     PROCESSING = "Processing"
     PREPROCESSED = "Preprocessed"
     PROCESSED = "Processed"
-    DUPLICATE = "Not processed - duplicate"
+    NOT_PROCESSED = "Not processed"
+    FAILED = "Failed"
+
+
+class FileNotProcessedReason(StrEnum):
+    """Reasons why a file was not processed"""
+    DUPLICATE = "Duplicate"
 
 
 class AuditTableKeys(StrEnum):
@@ -29,3 +35,4 @@ class AuditTableKeys(StrEnum):
     QUEUE_NAME = "queue_name"
     STATUS = "status"
     TIMESTAMP = "timestamp"
+    ERROR_DETAILS = "error_details"

batch_processor_filter/src/exception_decorator.py

@@ -0,0 +1,24 @@
+"""Module for the batch processor filter Lambda exception wrapper"""
+from functools import wraps
+from typing import Callable
+
+from exceptions import EventAlreadyProcessingForSupplierAndVaccTypeError
+from logger import logger
+
+
+def exception_decorator(func: Callable):
+    """Wrapper for the Lambda Handler. It ensures that any unhandled exceptions are logged for monitoring and alerting
+    purposes."""
+
+    @wraps(func)
+    def exception_wrapper(*args, **kwargs):
+        try:
+            func(*args, **kwargs)
+        except EventAlreadyProcessingForSupplierAndVaccTypeError as exc:
+            # Re-raise so event will be returned to SQS and retried for this expected error
+            raise exc
+        except Exception as exc:  # pylint:disable = broad-exception-caught
+            logger.error("An unhandled exception occurred in the batch processor filter Lambda", exc_info=exc)
+            raise exc
+
+    return exception_wrapper

batch_processor_filter/src/lambda_handler.py

@@ -3,13 +3,15 @@
 
 from batch_file_created_event import BatchFileCreatedEvent
 from batch_processor_filter_service import BatchProcessorFilterService
+from exception_decorator import exception_decorator
 from exceptions import InvalidBatchSizeError
 
 
 service = BatchProcessorFilterService()
 
 
-def lambda_handler(event: events.SQSEvent, _: context):
+@exception_decorator
+def lambda_handler(event: events.SQSEvent, _: context.Context):
     event_records = event.get("Records", [])
 
     # Terraform is configured so the Lambda will get a max batch size of 1. We are using SQS FIFO with the message group

batch_processor_filter/src/logger.py

@@ -0,0 +1,7 @@
+"""Module for the batch processor filter Lambda logger"""
+import logging
+
+
+logging.basicConfig(level="INFO")
+logger = logging.getLogger()
+logger.setLevel("INFO")

batch_processor_filter/tests/test_lambda_handler.py

@@ -1,9 +1,10 @@
 import json
+from json import JSONDecodeError
 
 import boto3
 import copy
 from unittest import TestCase
-from unittest.mock import patch
+from unittest.mock import patch, Mock, ANY
 
 import botocore
 from moto import mock_aws
@@ -77,6 +78,8 @@ def setUp(self):
 
         self.logger_patcher = patch("batch_processor_filter_service.logger")
         self.mock_logger = self.logger_patcher.start()
+        self.exception_decorator_logger_patcher = patch("exception_decorator.logger")
+        self.mock_exception_decorator_logger = self.exception_decorator_logger_patcher.start()
         self.firehose_log_patcher = patch("batch_processor_filter_service.send_log_to_firehose")
         self.mock_firehose_send_log = self.firehose_log_patcher.start()
 
@@ -90,8 +93,7 @@ def tearDown(self):
                 s3_client.delete_object(Bucket=bucket_name, Key=obj["Key"])
             s3_client.delete_bucket(Bucket=bucket_name)
 
-        self.logger_patcher.stop()
-        self.firehose_log_patcher.stop()
+        patch.stopall()
 
     def _assert_source_file_moved(self, filename: str):
         """Check used in the duplicate scenario to validate that the original uploaded file is moved"""
@@ -110,7 +112,7 @@ def _assert_ack_file_created(self, ack_file_key: str):
 
     def test_lambda_handler_raises_error_when_empty_batch_received(self):
         with self.assertRaises(InvalidBatchSizeError) as exc:
-            lambda_handler({"Records": []}, {})
+            lambda_handler({"Records": []}, Mock())
 
         self.assertEqual(str(exc.exception), "Received 0 records, expected 1")
 
@@ -119,10 +121,24 @@ def test_lambda_handler_raises_error_when_more_than_one_record_in_batch(self):
             lambda_handler({"Records": [
                 make_sqs_record(self.default_batch_file_event),
                 make_sqs_record(self.default_batch_file_event),
-            ]}, {})
+            ]}, Mock())
 
         self.assertEqual(str(exc.exception), "Received 2 records, expected 1")
 
+    def test_lambda_handler_decorator_logs_unhandled_exceptions(self):
+        """The exception decorator should log the error when an unhandled exception occurs"""
+        with self.assertRaises(JSONDecodeError):
+            lambda_handler({"Records": [
+                {
+                    "body": "{'malformed}"
+                }
+            ]}, Mock())
+
+        self.mock_exception_decorator_logger.error.assert_called_once_with(
+            "An unhandled exception occurred in the batch processor filter Lambda",
+            exc_info=ANY
+        )
+
     def test_lambda_handler_handles_duplicate_file_scenario(self):
         """Should update the audit table status to duplicate for the incoming record"""
         # Add the duplicate entry that has already been processed
@@ -137,61 +153,69 @@ def test_lambda_handler_handles_duplicate_file_scenario(self):
         # Create the source file in S3
         s3_client.put_object(Bucket=self.mock_source_bucket, Key=test_file_name)
 
-        lambda_handler({"Records": [make_sqs_record(duplicate_file_event)]}, {})
+        lambda_handler({"Records": [make_sqs_record(duplicate_file_event)]}, Mock())
 
         status = get_audit_entry_status_by_id(dynamodb_client, AUDIT_TABLE_NAME, duplicate_file_event["message_id"])
-        self.assertEqual(status, "Not processed - duplicate")
+        self.assertEqual(status, "Not processed - Duplicate")
 
         sqs_messages = sqs_client.receive_message(QueueUrl=self.mock_queue_url)
         self.assertEqual(sqs_messages.get("Messages", []), [])
         self._assert_source_file_moved(test_file_name)
         self._assert_ack_file_created("Menacwy_Vaccinations_v5_TEST_20250820T10210000_InfAck_20250826T14372600.csv")
 
-        self.mock_logger.info.assert_called_once_with(
+        self.mock_logger.error.assert_called_once_with(
             "A duplicate file has already been processed. Filename: %s",
             test_file_name
         )
 
     def test_lambda_handler_raises_error_when_event_already_processing_for_supplier_and_vacc_type(self):
         """Should raise exception so that the event is returned to the originating queue to be retried later"""
-        # Add an audit entry for a batch event that is already processing
-        add_entry_to_mock_table(dynamodb_client, AUDIT_TABLE_NAME, self.default_batch_file_event, FileStatus.PROCESSING)
-
-        test_event: BatchFileCreatedEvent = BatchFileCreatedEvent(
-            message_id="3b60c4f7-ef67-43c7-8f0d-4faee04d7d0e",
-            vaccine_type="MENACWY",  # Same vacc type
-            supplier="TESTSUPPLIER",  # Same supplier
-            permission=["some-permissions"],
-            filename="Menacwy_Vaccinations_v5_TEST_20250826T15003000.csv",  # Different timestamp
-            created_at_formatted_string="20250826T15003000"
-        )
-        # Add the audit record for the incoming event
-        add_entry_to_mock_table(dynamodb_client, AUDIT_TABLE_NAME, test_event, FileStatus.QUEUED)
-
-        with self.assertRaises(EventAlreadyProcessingForSupplierAndVaccTypeError) as exc:
-            lambda_handler({"Records": [make_sqs_record(test_event)]}, {})
-
-        self.assertEqual(
-            str(exc.exception),
-            "Batch event already processing for supplier: TESTSUPPLIER and vacc type: MENACWY"
-        )
-
-        status = get_audit_entry_status_by_id(dynamodb_client, AUDIT_TABLE_NAME, test_event["message_id"])
-        self.assertEqual(status, "Queued")
-
-        sqs_messages = sqs_client.receive_message(QueueUrl=self.mock_queue_url)
-        self.assertEqual(sqs_messages.get("Messages", []), [])
-
-        self.mock_logger.info.assert_called_once_with(
-            "Batch event already being processed for supplier and vacc type. Filename: %s",
-            "Menacwy_Vaccinations_v5_TEST_20250826T15003000.csv"
-        )
+        test_cases = {
+            ("Event is already being processed for supplier + vacc type queue", FileStatus.PROCESSING),
+            ("There is a failed event to be checked in supplier + vacc type queue", FileStatus.FAILED)
+        }
+
+        for msg, file_status in test_cases:
+            self.mock_logger.reset_mock()
+            with self.subTest(msg=msg):
+                # Add an audit entry for a batch event that is already processing or failed
+                add_entry_to_mock_table(dynamodb_client, AUDIT_TABLE_NAME, self.default_batch_file_event, file_status)
+
+                test_event: BatchFileCreatedEvent = BatchFileCreatedEvent(
+                    message_id="3b60c4f7-ef67-43c7-8f0d-4faee04d7d0e",
+                    vaccine_type="MENACWY",  # Same vacc type
+                    supplier="TESTSUPPLIER",  # Same supplier
+                    permission=["some-permissions"],
+                    filename="Menacwy_Vaccinations_v5_TEST_20250826T15003000.csv",  # Different timestamp
+                    created_at_formatted_string="20250826T15003000"
+                )
+                # Add the audit record for the incoming event
+                add_entry_to_mock_table(dynamodb_client, AUDIT_TABLE_NAME, test_event, FileStatus.QUEUED)
+
+                with self.assertRaises(EventAlreadyProcessingForSupplierAndVaccTypeError) as exc:
+                    lambda_handler({"Records": [make_sqs_record(test_event)]}, Mock())
+
+                self.assertEqual(
+                    str(exc.exception),
+                    "Batch event already processing for supplier: TESTSUPPLIER and vacc type: MENACWY"
+                )
+
+                status = get_audit_entry_status_by_id(dynamodb_client, AUDIT_TABLE_NAME, test_event["message_id"])
+                self.assertEqual(status, "Queued")
+
+                sqs_messages = sqs_client.receive_message(QueueUrl=self.mock_queue_url)
+                self.assertEqual(sqs_messages.get("Messages", []), [])
+
+                self.mock_logger.info.assert_called_once_with(
+                    "Batch event already processing for supplier and vacc type. Filename: %s",
+                    "Menacwy_Vaccinations_v5_TEST_20250826T15003000.csv"
+                )
 
     def test_lambda_handler_processes_event_successfully(self):
         """Should update the audit entry status to Processing and forward to SQS"""
         add_entry_to_mock_table(dynamodb_client, AUDIT_TABLE_NAME, self.default_batch_file_event, FileStatus.QUEUED)
 
-        lambda_handler({"Records": [make_sqs_record(self.default_batch_file_event)]}, {})
+        lambda_handler({"Records": [make_sqs_record(self.default_batch_file_event)]}, Mock())
 
         status = get_audit_entry_status_by_id(dynamodb_client, AUDIT_TABLE_NAME,
                                               self.default_batch_file_event["message_id"])
@@ -223,7 +247,7 @@ def test_lambda_handler_processes_event_successfully_when_event_for_same_supplie
         )
         add_entry_to_mock_table(dynamodb_client, AUDIT_TABLE_NAME, test_event, FileStatus.QUEUED)
 
-        lambda_handler({"Records": [make_sqs_record(test_event)]}, {})
+        lambda_handler({"Records": [make_sqs_record(test_event)]}, Mock())
 
         status = get_audit_entry_status_by_id(dynamodb_client, AUDIT_TABLE_NAME, test_event["message_id"])
         self.assertEqual(status, "Processing")

filenameprocessor/src/audit_table.py

@@ -1,4 +1,5 @@
 """Add the filename to the audit table and check for duplicates."""
+from typing import Optional
 from clients import dynamodb_client, logger
 from errors import UnhandledAuditTableError
 from constants import AUDIT_TABLE_NAME, AuditTableKeys
@@ -10,23 +11,29 @@ def upsert_audit_table(
     created_at_formatted_str: str,
     expiry_timestamp: int,
     queue_name: str,
-    file_status: str
+    file_status: str,
+    error_details: Optional[str] = None
 ) -> None:
     """
     Updates the audit table with the file details
     """
+    audit_item = {
+        AuditTableKeys.MESSAGE_ID: {"S": message_id},
+        AuditTableKeys.FILENAME: {"S": file_key},
+        AuditTableKeys.QUEUE_NAME: {"S": queue_name},
+        AuditTableKeys.STATUS: {"S": file_status},
+        AuditTableKeys.TIMESTAMP: {"S": created_at_formatted_str},
+        AuditTableKeys.EXPIRES_AT: {"N": str(expiry_timestamp)}
+    }
+
+    if error_details is not None:
+        audit_item[AuditTableKeys.ERROR_DETAILS] = {"S": error_details}
+
     try:
         # Add to the audit table (regardless of whether it is a duplicate)
         dynamodb_client.put_item(
             TableName=AUDIT_TABLE_NAME,
-            Item={
-                AuditTableKeys.MESSAGE_ID: {"S": message_id},
-                AuditTableKeys.FILENAME: {"S": file_key},
-                AuditTableKeys.QUEUE_NAME: {"S": queue_name},
-                AuditTableKeys.STATUS: {"S": file_status},
-                AuditTableKeys.TIMESTAMP: {"S": created_at_formatted_str},
-                AuditTableKeys.EXPIRES_AT: {"N": str(expiry_timestamp)},
-            },
+            Item=audit_item,
             ConditionExpression="attribute_not_exists(message_id)",  # Prevents accidental overwrites
         )
         logger.info("%s file, with message id %s, successfully added to audit table", file_key, message_id)