Minor refactoring

Author: dlzhry2nhs

lambdas/ack_backend/src/ack_processor.py

@@ -44,7 +44,9 @@ def lambda_handler(event, _):
         for message in incoming_message_body:
             ack_data_rows.append(convert_message_to_ack_row(message, created_at_formatted_string))
 
+        # Get the row number of the final record processed
         if i == len(event["Records"]) - 1:
+            # Format of the row id is {batch_message_id}^{row_number}
             total_ack_rows_processed = int(incoming_message_body[-1].get("row_id", "").split("^")[1])
 
     update_ack_file(file_key, created_at_formatted_string, ack_data_rows)

lambdas/ack_backend/src/constants.py

@@ -4,6 +4,11 @@
 
 AUDIT_TABLE_NAME = os.getenv("AUDIT_TABLE_NAME")
 
+COMPLETED_ACK_DIR = "forwardedFile"
+TEMP_ACK_DIR = "TempAck"
+BATCH_FILE_PROCESSING_DIR = "processing"
+BATCH_FILE_ARCHIVE_DIR = "archive"
+
 
 def get_source_bucket_name() -> str:
     """Get the SOURCE_BUCKET_NAME environment from environment variables."""

lambdas/ack_backend/src/logging_decorators.py

@@ -49,7 +49,7 @@ def wrapper(message, created_at_formatted_string):
     return wrapper
 
 
-def upload_ack_file_logging_decorator(func):
+def complete_batch_file_process_logging_decorator(func):
     """This decorator logs when record processing is complete."""
 
     @wraps(func)

lambdas/ack_backend/src/update_ack_file.py

@@ -4,8 +4,9 @@
 from io import StringIO, BytesIO
 from audit_table import change_audit_table_status_to_processed
 from common.clients import get_s3_client, logger
-from constants import ACK_HEADERS, get_source_bucket_name, get_ack_bucket_name
-from logging_decorators import upload_ack_file_logging_decorator
+from constants import ACK_HEADERS, get_source_bucket_name, get_ack_bucket_name, COMPLETED_ACK_DIR, TEMP_ACK_DIR, \
+    BATCH_FILE_PROCESSING_DIR, BATCH_FILE_ARCHIVE_DIR
+from logging_decorators import complete_batch_file_process_logging_decorator
 
 
 def create_ack_data(
@@ -43,7 +44,7 @@ def create_ack_data(
     }
 
 
-@upload_ack_file_logging_decorator
+@complete_batch_file_process_logging_decorator
 def complete_batch_file_process(
     message_id: str,
     supplier: str,
@@ -54,12 +55,18 @@ def complete_batch_file_process(
 ) -> dict:
     """Mark the batch file as processed. This involves moving the ack and original file to destinations and updating
     the audit table status"""
-    ack_bucket_name = get_ack_bucket_name()
-    source_bucket_name = get_source_bucket_name()
     ack_filename = f"{file_key.replace('.csv', f'_BusAck_{created_at_formatted_string}.csv')}"
 
-    move_file(ack_bucket_name, f"TempAck/{ack_filename}", f"forwardedFile/{ack_filename}")
-    move_file(source_bucket_name, f"processing/{file_key}", f"archive/{file_key}")
+    move_file(
+        get_ack_bucket_name(),
+        f"{TEMP_ACK_DIR}/{ack_filename}",
+        f"{COMPLETED_ACK_DIR}/{ack_filename}"
+    )
+    move_file(
+        get_source_bucket_name(),
+        f"{BATCH_FILE_PROCESSING_DIR}/{file_key}",
+        f"{BATCH_FILE_ARCHIVE_DIR}/{file_key}"
+    )
 
     change_audit_table_status_to_processed(file_key, message_id)
 
@@ -99,8 +106,8 @@ def update_ack_file(
 ) -> None:
     """Updates the ack file with the new data row based on the given arguments"""
     ack_filename = f"{file_key.replace('.csv', f'_BusAck_{created_at_formatted_string}.csv')}"
-    temp_ack_file_key = f"TempAck/{ack_filename}"
-    archive_ack_file_key = f"forwardedFile/{ack_filename}"
+    temp_ack_file_key = f"{TEMP_ACK_DIR}/{ack_filename}"
+    archive_ack_file_key = f"{COMPLETED_ACK_DIR}/{ack_filename}"
     accumulated_csv_content = obtain_current_ack_content(temp_ack_file_key)
 
     for row in ack_data_rows:

lambdas/ack_backend/src/utils_for_ack_lambda.py

@@ -13,7 +13,7 @@ def is_ack_processing_complete(batch_event_message_id: str, processed_ack_count:
     record_count = get_record_count_by_message_id(batch_event_message_id)
 
     if not record_count:
-        # Record Count is not set on the audit item until all rows have been preprocessed and sent to Kinesis
+        # Record count is not set on the audit item until all rows have been preprocessed and sent to Kinesis
         return False
 
     _BATCH_EVENT_ID_TO_RECORD_COUNT_MAP[batch_event_message_id] = record_count