Merge branch 'master' into VED-753-CloudWatch-PII

Author: JamesW1-NHS

ack_backend/poetry.lock

@@ -10,7 +10,7 @@ packages = [
 
 [tool.poetry.dependencies]
 python = "~3.11"
-boto3 = "~1.40.22"
+boto3 = "~1.40.28"
 mypy-boto3-dynamodb = "^1.40.20"
 freezegun      = "^1.5.2"
 moto           = "^4"

ack_backend/pyproject.toml

@@ -9,11 +9,11 @@ packages    = [{include = "src"}]
 [tool.poetry.dependencies]
 python                  = "~3.11"
 "fhir.resources"        = "~7.0.2"
-boto3                   = "~1.40.22"
-boto3-stubs-lite        = {extras = ["dynamodb"], version = "~1.40.22"}
+boto3                   = "~1.40.28"
+boto3-stubs-lite        = {extras = ["dynamodb"], version = "~1.40.28"}
 aws-lambda-typing       = "~2.20.0"
 redis                   = "^4.6.0"
-moto                    = "^5.1.11"
+moto                    = "^5.1.12"
 requests                = "~2.32.5"
 responses               = "~0.25.7"
 pydantic                = "~1.10.13"

backend/poetry.lock

@@ -10,8 +10,8 @@ packages    = [{include = "src"}]
 python            = "~3.11"
 coverage          = "^7.10.6"
 aws-lambda-typing = "~2.20.0"
-boto3             = "~1.40.22"
-moto              = "^5.1.11"
+boto3             = "~1.40.28"
+moto              = "^5.1.12"
 
 [build-system]
 requires      = ["poetry-core ~= 1.5.0"]

backend/pyproject.toml

@@ -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/poetry.lock

@@ -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}")