init

Author: JamesW1-NHS

ack_backend/src/ack_processor.py

@@ -40,12 +40,11 @@ def lambda_handler(event, context):
             message_id = (incoming_message_body[0].get("row_id", "")).split("^")[0]
             vaccine_type = incoming_message_body[0].get("vaccine_type")
             supplier = incoming_message_body[0].get("supplier")
-            supplier_queue = f"{supplier}_{vaccine_type}"
             created_at_formatted_string = incoming_message_body[0].get("created_at_formatted_string")
 
         for message in incoming_message_body:
             ack_data_rows.append(convert_message_to_ack_row(message, created_at_formatted_string))
 
-    update_ack_file(file_key, message_id, supplier_queue, created_at_formatted_string, ack_data_rows)
+    update_ack_file(file_key, message_id, supplier, vaccine_type, created_at_formatted_string, ack_data_rows)
 
     return {"statusCode": 200, "body": json.dumps("Lambda function executed successfully!")}

ack_backend/src/logging_decorators.py

@@ -77,6 +77,32 @@ def wrapper(message, created_at_formatted_string):
     return wrapper
 
 
+def upload_ack_file_logging_decorator(func):
+    """This decorator logs when record processing is complete."""
+
+    @wraps(func)
+    def wrapper(*args, **kwargs):
+
+        base_log_data = {"function_name": f"ack_processor_{func.__name__}", "date_time": str(datetime.now())}
+        start_time = time.time()
+
+        try:
+            result = func(*args, **kwargs)
+            if result is not None:
+                message_for_logs = "Record processing complete"
+                base_log_data.update(result)
+                additional_log_data = {"status": "success", "statusCode": 200, "message": message_for_logs}
+                generate_and_send_logs(start_time, base_log_data, additional_log_data)
+            return result
+
+        except Exception as error:
+            additional_log_data = {"status": "fail", "statusCode": 500, "diagnostics": str(error)}
+            generate_and_send_logs(start_time, base_log_data, additional_log_data, is_error_log=True)
+            raise
+
+    return wrapper
+
+
 def ack_lambda_handler_logging_decorator(func):
     """This decorator logs the execution info for the ack lambda handler."""
 

ack_backend/src/update_ack_file.py

@@ -2,12 +2,13 @@
 
 import json
 from io import StringIO, BytesIO
-from typing import Union
+from typing import Union, Optional
 from botocore.exceptions import ClientError
 from constants import ACK_HEADERS, SOURCE_BUCKET_NAME, ACK_BUCKET_NAME, FILE_NAME_PROC_LAMBDA_NAME
 from audit_table import change_audit_table_status_to_processed, get_next_queued_file_details
 from clients import s3_client, logger, lambda_client
 from utils_for_ack_lambda import get_row_count
+from logging_decorators import upload_ack_file_logging_decorator
 
 
 def create_ack_data(
@@ -65,15 +66,17 @@ def obtain_current_ack_content(temp_ack_file_key: str) -> StringIO:
     return accumulated_csv_content
 
 
+@upload_ack_file_logging_decorator
 def upload_ack_file(
     temp_ack_file_key: str,
     message_id: str,
-    supplier_queue: str,
+    supplier: str,
+    vaccine_type: str,
     accumulated_csv_content: StringIO,
     ack_data_rows: list,
     archive_ack_file_key: str,
     file_key: str,
-) -> None:
+) -> Optional[dict]:
     """Adds the data row to the uploaded ack file"""
     for row in ack_data_rows:
         data_row_str = [str(item) for item in row.values()]
@@ -91,17 +94,30 @@ def upload_ack_file(
 
         # Update the audit table and invoke the filename lambda with next file in the queue (if one exists)
         change_audit_table_status_to_processed(file_key, message_id)
+        supplier_queue = f"{supplier}_{vaccine_type}"
         next_queued_file_details = get_next_queued_file_details(supplier_queue)
         if next_queued_file_details:
             invoke_filename_lambda(next_queued_file_details["filename"], next_queued_file_details["message_id"])
 
+        # Ingestion of this file is complete
+        result = {
+            "message_id": message_id,
+            "file_key": file_key,
+            "supplier": supplier,
+            "vaccine_type": vaccine_type,
+            "row_count": row_count_source,
+        }
+    else:
+        result = None
     logger.info("Ack file updated to %s: %s", ACK_BUCKET_NAME, archive_ack_file_key)
+    return result
 
 
 def update_ack_file(
     file_key: str,
     message_id: str,
-    supplier_queue: str,
+    supplier: str,
+    vaccine_type: str,
     created_at_formatted_string: str,
     ack_data_rows: list,
 ) -> None:
@@ -113,7 +129,8 @@ def update_ack_file(
     upload_ack_file(
         temp_ack_file_key,
         message_id,
-        supplier_queue,
+        supplier,
+        vaccine_type,
         accumulated_csv_content,
         ack_data_rows,
         archive_ack_file_key,