add upserts for EA files

Author: JamesW1-NHS

lambdas/filenameprocessor/src/audit_table.py

@@ -15,6 +15,7 @@ def upsert_audit_table(
     queue_name: str,
     file_status: str,
     error_details: Optional[str] = None,
+    condition_expression: str = ""
 ) -> None:
     """
     Updates the audit table with the file details
@@ -36,7 +37,7 @@ def upsert_audit_table(
         dynamodb_client.put_item(
             TableName=AUDIT_TABLE_NAME,
             Item=audit_item,
-            ConditionExpression="attribute_not_exists(message_id)",  # Prevents accidental overwrites
+            ConditionExpression=condition_expression,
         )
         logger.info(
             "%s file, with message id %s, successfully added to audit table",

lambdas/filenameprocessor/src/file_name_processor.py

@@ -7,10 +7,16 @@
 """
 
 import argparse
+from botocore.exceptions import ClientError
 from uuid import uuid4
 
 from audit_table import upsert_audit_table
-from common.aws_s3_utils import move_file, move_file_outside_bucket
+from common.aws_s3_utils import (
+    copy_file_outside_bucket,
+    delete_file,
+    is_file_in_bucket,
+    move_file,
+)
 from common.clients import STREAM_NAME, get_s3_client, logger
 from common.log_decorator import logging_decorator
 from common.models.errors import UnhandledAuditTableError
@@ -174,6 +180,7 @@ def handle_batch_file(file_key, bucket_name, message_id, created_at_formatted_st
             expiry_timestamp,
             queue_name,
             FileStatus.QUEUED,
+            condition_expression="attribute_not_exists(message_id)",  # Prevents accidental overwrites
         )
         make_and_send_sqs_message(
             file_key,
@@ -242,9 +249,17 @@ def handle_extended_attributes_file(
     Processes a single record for extended attributes file.
     Returns a dictionary containing information to be included in the logs.
     """
+    
+    # here: the sequence of events should be
+    # 1. upsert 'processing'
+    # 2. move the file to the dest bucket
+    # 3. check the file is present in the dest bucket
+    # 4. if it is, delete it from the src bucket, upsert 'processed'
+    # 5. if it isn't, move it to the archive/ folder, upsert 'failed'
+    # NB for this to work we have to retool upsert so it accepts overwrites, i.e. ignore the ConditionExpression
+
     try:
         extended_attribute_identifier = validate_extended_attributes_file_key(file_key)
-        move_file_outside_bucket(bucket_name, file_key, DPS_DESTINATION_BUCKET_NAME, f"archive/{file_key}")
         queue_name = extended_attribute_identifier
 
         upsert_audit_table(
@@ -255,14 +270,30 @@ def handle_extended_attributes_file(
             queue_name,
             FileStatus.PROCESSING,
         )
+
+        copy_file_outside_bucket(bucket_name, file_key, DPS_DESTINATION_BUCKET_NAME, f"archive/{file_key}")
+        is_file_in_bucket(DPS_DESTINATION_BUCKET_NAME, file_key)
+        delete_file(DPS_DESTINATION_BUCKET_NAME, file_key)
+
+        upsert_audit_table(
+            message_id,
+            file_key,
+            created_at_formatted_string,
+            expiry_timestamp,
+            queue_name,
+            FileStatus.PROCESSED,
+        )
+
         return {
             "statusCode": 200,
             "message": "Extended Attributes file successfully processed",
             "file_key": file_key,
             "message_id": message_id,
             "queue_name": queue_name,
         }
+
     except (  # pylint: disable=broad-exception-caught
+        ClientError,
         VaccineTypePermissionsError,
         InvalidFileKeyError,
         UnhandledAuditTableError,
@@ -275,13 +306,16 @@ def handle_extended_attributes_file(
         extended_attribute_identifier = validate_extended_attributes_file_key(file_key)
         queue_name = extended_attribute_identifier
 
+        # Move file to archive
+        move_file(bucket_name, file_key, f"archive/{file_key}")
+
         upsert_audit_table(
             message_id,
             file_key,
             created_at_formatted_string,
             expiry_timestamp,
             extended_attribute_identifier,
-            file_status,
+            queue_name,
             error_details=str(error),
         )
 

lambdas/filenameprocessor/tests/test_audit_table.py

@@ -73,6 +73,7 @@ def test_upsert_audit_table_with_duplicate_message_id_raises_exception(self):
             queue_name=ravs_rsv_test_file.queue_name,
             file_status=FileStatus.PROCESSED,
             expiry_timestamp=ravs_rsv_test_file.expires_at,
+            condition_expression="attribute_not_exists(message_id)",
         )
 
         assert_audit_table_entry(ravs_rsv_test_file, FileStatus.PROCESSED)
@@ -85,4 +86,5 @@ def test_upsert_audit_table_with_duplicate_message_id_raises_exception(self):
                 queue_name=ravs_rsv_test_file.queue_name,
                 file_status=FileStatus.PROCESSED,
                 expiry_timestamp=ravs_rsv_test_file.expires_at,
+                condition_expression="attribute_not_exists(message_id)",
             )

lambdas/filenameprocessor/tests/test_lambda_handler.py

@@ -269,14 +269,13 @@ def test_lambda_handler_extended_attributes_success(self):
                 return_value=test_cases[0].ods_code + "_COVID",
             ),
             patch(
-                "file_name_processor.move_file_outside_bucket",
+                "file_name_processor.copy_file_outside_bucket",
                 side_effect=lambda src_bucket, key, dst_bucket, dst_key: (
                     s3_client.put_object(
                         Bucket=BucketNames.DESTINATION,
                         Key=dst_key,
                         Body=s3_client.get_object(Bucket=src_bucket, Key=key)["Body"].read(),
                     ),
-                    s3_client.delete_object(Bucket=src_bucket, Key=key),
                 ),
             ),
         ):

lambdas/shared/src/common/aws_s3_utils.py

@@ -5,7 +5,6 @@
 from common.clients import get_s3_client, logger
 
 EXPECTED_BUCKET_OWNER_ACCOUNT = os.getenv("ACCOUNT_ID")
-DESTINATION_BUCKET_NAME = os.getenv("DESTINATION_BUCKET_NAME")
 
 
 def move_file(bucket_name: str, source_file_key: str, destination_file_key: str) -> None:
@@ -20,7 +19,7 @@ def move_file(bucket_name: str, source_file_key: str, destination_file_key: str)
     logger.info("File moved from %s to %s", source_file_key, destination_file_key)
 
 
-def move_file_outside_bucket(source_bucket: str, source_key: str, destination_bucket: str, destination_key: str) -> None:
+def copy_file_outside_bucket(source_bucket: str, source_key: str, destination_bucket: str, destination_key: str) -> None:
     s3_client = get_s3_client()
     s3_client.copy_object(
         CopySource={"Bucket": source_bucket, "Key": source_key},
@@ -29,8 +28,16 @@ def move_file_outside_bucket(source_bucket: str, source_key: str, destination_bu
         ExpectedBucketOwner=EXPECTED_BUCKET_OWNER_ACCOUNT,
         ExpectedSourceBucketOwner=EXPECTED_BUCKET_OWNER_ACCOUNT,
     )
+
+def delete_file(source_bucket: str, source_key: str) -> None:
+    s3_client = get_s3_client()
     s3_client.delete_object(
         Bucket=source_bucket,
         Key=source_key,
         ExpectedBucketOwner=EXPECTED_BUCKET_OWNER_ACCOUNT,
     )
+
+def is_file_in_bucket(bucket_name: str, file_key: str) -> None:
+    s3_client = get_s3_client()
+    s3_client.head_object(Bucket=bucket_name, Key=file_key)
+

lambdas/shared/tests/test_common/test_s3_utils.py

@@ -75,7 +75,7 @@ def test_move_file_within_bucket(self):
         self.mock_logger_info.assert_called_with("File moved from %s to %s", file_key, dest_key)
 
     def test_move_file_outside_bucket_copies_then_deletes(self):
-        """File should be copied to destination bucket under destination_key and removed from source bucket."""
+        """File should be copied to destination bucket under destination_key and deleted from source bucket"""
         source_key = "RSV_Vaccinations_v5_X8E5B_20000101T00000001.csv"
         destination_key = f"archive/{source_key}"
 
@@ -88,8 +88,8 @@ def test_move_file_outside_bucket_copies_then_deletes(self):
         with self.assertRaises(self.s3.exceptions.NoSuchKey):
             self.s3.get_object(Bucket=self.destination_bucket, Key=destination_key)
 
-        # Execute move across buckets
-        aws_s3_utils.move_file_outside_bucket(
+        # Execute copy across buckets
+        aws_s3_utils.copy_file_outside_bucket(
             source_bucket=self.source_bucket,
             source_key=source_key,
             destination_bucket=self.destination_bucket,
@@ -100,6 +100,16 @@ def test_move_file_outside_bucket_copies_then_deletes(self):
         dest_obj = self.s3.get_object(Bucket=self.destination_bucket, Key=destination_key)
         self.assertEqual(dest_obj["Body"].read(), body_content)
 
+        # Assert source object was not deleted
+        src_obj = self.s3.get_object(Bucket=self.source_bucket, Key=source_key)
+        self.assertEqual(src_obj["Body"].read(), body_content)
+
+        # Execute delete file
+        aws_s3_utils.delete_file(
+            source_bucket=self.source_bucket,
+            source_key=source_key,
+        )
+
         # Assert source object was deleted
         with self.assertRaises(self.s3.exceptions.NoSuchKey):
             self.s3.get_object(Bucket=self.source_bucket, Key=source_key)