VED-845: refactor record-processor (#1126)

* refactoring file level validation

Author: motola

lambdas/ack_backend/src/constants.py

@@ -4,6 +4,8 @@
 TEMP_ACK_DIR = "TempAck"
 BATCH_FILE_PROCESSING_DIR = "processing"
 BATCH_FILE_ARCHIVE_DIR = "archive"
+LAMBDA_FUNCTION_NAME_PREFIX = "ack_processor"
+DEFAULT_STREAM_NAME = "immunisation-fhir-api-internal-dev-splunk-firehose"
 
 
 ACK_HEADERS = [

lambdas/ack_backend/src/logging_decorators.py

@@ -6,9 +6,9 @@
 from functools import wraps
 
 from common.log_decorator import generate_and_send_logs
+from constants import DEFAULT_STREAM_NAME, LAMBDA_FUNCTION_NAME_PREFIX
 
-PREFIX = "ack_processor"
-STREAM_NAME = os.getenv("SPLUNK_FIREHOSE_NAME", "immunisation-fhir-api-internal-dev-splunk-firehose")
+STREAM_NAME = os.getenv("SPLUNK_FIREHOSE_NAME", DEFAULT_STREAM_NAME)
 
 
 def convert_message_to_ack_row_logging_decorator(func):
@@ -17,7 +17,7 @@ def convert_message_to_ack_row_logging_decorator(func):
     @wraps(func)
     def wrapper(message, created_at_formatted_string):
         base_log_data = {
-            "function_name": f"{PREFIX}_{func.__name__}",
+            "function_name": f"{LAMBDA_FUNCTION_NAME_PREFIX}_{func.__name__}",
             "date_time": str(datetime.now()),
         }
         start_time = time.time()
@@ -69,41 +69,13 @@ def wrapper(message, created_at_formatted_string):
     return wrapper
 
 
-def complete_batch_file_process_logging_decorator(func):
-    """This decorator logs when record processing is complete."""
-
-    @wraps(func)
-    def wrapper(*args, **kwargs):
-        base_log_data = {
-            "function_name": f"{PREFIX}_{func.__name__}",
-            "date_time": str(datetime.now()),
-        }
-        start_time = time.time()
-
-        # NB this doesn't require a try-catch block as the wrapped function never throws an exception
-        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(STREAM_NAME, start_time, base_log_data, additional_log_data)
-
-        return result
-
-    return wrapper
-
-
 def ack_lambda_handler_logging_decorator(func):
     """This decorator logs the execution info for the ack lambda handler."""
 
     @wraps(func)
     def wrapper(event, context, *args, **kwargs):
         base_log_data = {
-            "function_name": f"{PREFIX}_{func.__name__}",
+            "function_name": f"{LAMBDA_FUNCTION_NAME_PREFIX}_{func.__name__}",
             "date_time": str(datetime.now()),
         }
         start_time = time.time()

lambdas/ack_backend/src/update_ack_file.py

@@ -1,6 +1,8 @@
 """Functions for uploading the data to the ack file"""
 
+import os
 import time
+from datetime import datetime
 from io import BytesIO, StringIO
 
 from botocore.exceptions import ClientError
@@ -12,15 +14,19 @@
 )
 from common.aws_s3_utils import move_file
 from common.clients import get_s3_client, logger
+from common.log_decorator import generate_and_send_logs
 from common.models.batch_constants import ACK_BUCKET_NAME, SOURCE_BUCKET_NAME
 from constants import (
     ACK_HEADERS,
     BATCH_FILE_ARCHIVE_DIR,
     BATCH_FILE_PROCESSING_DIR,
     COMPLETED_ACK_DIR,
+    DEFAULT_STREAM_NAME,
+    LAMBDA_FUNCTION_NAME_PREFIX,
     TEMP_ACK_DIR,
 )
-from logging_decorators import complete_batch_file_process_logging_decorator
+
+STREAM_NAME = os.getenv("SPLUNK_FIREHOSE_NAME", DEFAULT_STREAM_NAME)
 
 
 def create_ack_data(
@@ -58,7 +64,6 @@ def create_ack_data(
     }
 
 
-@complete_batch_file_process_logging_decorator
 def complete_batch_file_process(
     message_id: str,
     supplier: str,
@@ -68,6 +73,8 @@ 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"""
+    start_time = time.time()
+
     ack_filename = f"{file_key.replace('.csv', f'_BusAck_{created_at_formatted_string}.csv')}"
 
     move_file(ACK_BUCKET_NAME, f"{TEMP_ACK_DIR}/{ack_filename}", f"{COMPLETED_ACK_DIR}/{ack_filename}")
@@ -81,7 +88,7 @@ def complete_batch_file_process(
     successful_record_count = total_ack_rows_processed - total_failures
     set_audit_record_success_count_and_end_time(file_key, message_id, successful_record_count, ingestion_end_time)
 
-    return {
+    result = {
         "message_id": message_id,
         "file_key": file_key,
         "supplier": supplier,
@@ -91,6 +98,29 @@ def complete_batch_file_process(
         "failure_count": total_failures,
     }
 
+    log_batch_file_process(
+        start_time=start_time,
+        result=result,
+        function_name=f"{LAMBDA_FUNCTION_NAME_PREFIX}_complete_batch_file_process",
+    )
+
+    return result
+
+
+def log_batch_file_process(start_time: float, result: dict, function_name: str) -> None:
+    """Logs the batch file processing completion to Splunk"""
+    base_log_data = {
+        "function_name": function_name,
+        "date_time": str(datetime.now()),
+        **result,
+    }
+    additional_log_data = {
+        "status": "success",
+        "statusCode": 200,
+        "message": "Record processing complete",
+    }
+    generate_and_send_logs(STREAM_NAME, start_time, base_log_data, additional_log_data)
+
 
 def obtain_current_ack_content(temp_ack_file_key: str) -> StringIO:
     """Returns the current ack file content if the file exists, or else initialises the content with the ack headers."""

lambdas/ack_backend/tests/test_splunk_logging.py

@@ -425,6 +425,7 @@ def test_splunk_update_ack_file_logged(self):
 
         with (  # noqa: E999
             patch("common.log_decorator.send_log_to_firehose") as mock_send_log_to_firehose,  # noqa: E999
+            patch("update_ack_file.datetime") as mock_datetime,
             patch("common.log_decorator.logger") as mock_logger,  # noqa: E999
             patch("update_ack_file.get_record_count_and_failures_by_message_id", return_value=(99, 2)),
             patch(
@@ -435,6 +436,7 @@ def test_splunk_update_ack_file_logged(self):
             ) as mock_set_records_succeeded_count_and_end_time,  # noqa: E999
             patch("ack_processor.increment_records_failed_count"),  # noqa: E999
         ):  # noqa: E999
+            mock_datetime.now.return_value = ValidValues.fixed_datetime
             result = lambda_handler(generate_event(messages, include_eof_message=True), context={})
 
         self.assertEqual(result, EXPECTED_ACK_LAMBDA_RESPONSE_FOR_SUCCESS)

lambdas/recordprocessor/src/batch_processor.py

@@ -62,10 +62,10 @@ def process_csv_to_fhir(incoming_message_body: dict) -> int:
             logger.warning(f"Encoding Error: {err}.")
             new_encoder = "cp1252"
             logger.info(f"Encode error at row {row_count} with {encoder}. Switch to {new_encoder}")
-            encoder = new_encoder
+            encoding = new_encoder
 
             # load alternative encoder
-            csv_reader = get_csv_content_dict_reader(f"{PROCESSING_DIR_NAME}/{file_key}", encoder=encoder)
+            csv_reader = get_csv_content_dict_reader(f"{PROCESSING_DIR_NAME}/{file_key}", encoding=encoding)
             # re-read the file and skip processed rows
             row_count, err = process_rows(
                 file_id,

lambdas/recordprocessor/src/file_level_validation.py

@@ -70,24 +70,21 @@ def file_level_validation(incoming_message_body: dict) -> dict:
     NOTE: If file level validation fails the source file is moved to the archive folder, the audit table is updated
     to reflect the file has been processed and the filename lambda is invoked with the next file in the queue.
     """
+    message_id = None
+    file_key = None
+    created_at_formatted_string = None
+
     try:
         message_id = incoming_message_body.get("message_id")
         vaccine = incoming_message_body.get("vaccine_type").upper()
         supplier = incoming_message_body.get("supplier").upper()
         file_key = incoming_message_body.get("filename")
         permission = incoming_message_body.get("permission")
         created_at_formatted_string = incoming_message_body.get("created_at_formatted_string")
-        encoder = incoming_message_body.get("encoder", "utf-8")
+        encoding = incoming_message_body.get("encoding", "utf-8")
 
         # Fetch the data
-        try:
-            csv_reader = get_csv_content_dict_reader(file_key, encoder=encoder)
-            validate_content_headers(csv_reader)
-        except UnicodeDecodeError as e:
-            logger.warning("Invalid Encoding detected: %s", e)
-            # retry with cp1252 encoding
-            csv_reader = get_csv_content_dict_reader(file_key, encoder="cp1252")
-            validate_content_headers(csv_reader)
+        csv_reader = get_validated_csv_reader(file_key, encoding=encoding)
 
         # Validate has permission to perform at least one of the requested actions
         allowed_operations_set = get_permitted_operations(supplier, vaccine, permission)
@@ -110,24 +107,48 @@ def file_level_validation(incoming_message_body: dict) -> dict:
         }
 
     except Exception as error:
-        logger.error("Error in file_level_validation: %s", error)
-
-        # NOTE: The Exception may occur before the file_id, file_key and created_at_formatted_string are assigned
-        message_id = message_id or "Unable to ascertain message_id"
-        file_key = file_key or "Unable to ascertain file_key"
-        created_at_formatted_string = created_at_formatted_string or "Unable to ascertain created_at_formatted_string"
-        make_and_upload_ack_file(message_id, file_key, False, False, created_at_formatted_string)
-        file_status = (
-            f"{FileStatus.NOT_PROCESSED} - {FileNotProcessedReason.UNAUTHORISED}"
-            if isinstance(error, NoOperationPermissions)
-            else FileStatus.FAILED
+        handle_file_level_validation_exception(
+            error, message_id=message_id, file_key=file_key, created_at_formatted_string=created_at_formatted_string
         )
+        raise
 
-        try:
-            move_file(SOURCE_BUCKET_NAME, file_key, f"{ARCHIVE_DIR_NAME}/{file_key}")
-        except Exception as move_file_error:
-            logger.error("Failed to move file to archive: %s", move_file_error)
 
-        # Update the audit table
-        update_audit_table_status(file_key, message_id, file_status, error_details=str(error))
-        raise
+def get_validated_csv_reader(file_key: str, encoding: str = "utf-8") -> DictReader:
+    """Helper function to get a validated CSV DictReader object."""
+    try:
+        csv_reader = get_csv_content_dict_reader(file_key, encoding=encoding)
+        validate_content_headers(csv_reader)
+        return csv_reader
+    except UnicodeDecodeError as e:
+        logger.warning("Invalid Encoding detected: %s", e)
+        # Retry using cp-1252 encoding if the expected utf-8 fails
+        # This is a known issue with a supplier - see VED-754 for details
+        csv_reader = get_csv_content_dict_reader(file_key, encoding="cp1252")
+
+    validate_content_headers(csv_reader)
+    return csv_reader
+
+
+def handle_file_level_validation_exception(
+    error: Exception, message_id: str | None, file_key: str | None, created_at_formatted_string: str | None
+) -> None:
+    logger.error("Error in file_level_validation: %s", error)
+
+    # NOTE: The Exception may occur before the file_id, file_key and created_at_formatted_string are assigned
+    message_id = message_id or "Unable to ascertain message_id"
+    file_key = file_key or "Unable to ascertain file_key"
+    created_at_formatted_string = created_at_formatted_string or "Unable to ascertain created_at_formatted_string"
+    make_and_upload_ack_file(message_id, file_key, False, False, created_at_formatted_string)
+    file_status = (
+        f"{FileStatus.NOT_PROCESSED} - {FileNotProcessedReason.UNAUTHORISED}"
+        if isinstance(error, NoOperationPermissions)
+        else FileStatus.FAILED
+    )
+
+    try:
+        move_file(SOURCE_BUCKET_NAME, file_key, f"{ARCHIVE_DIR_NAME}/{file_key}")
+    except Exception as move_file_error:
+        logger.error("Failed to move file to archive: %s", move_file_error)
+
+    # Update the audit table
+    update_audit_table_status(file_key, message_id, file_status, error_details=str(error))

lambdas/recordprocessor/src/utils_for_recordprocessor.py

@@ -14,11 +14,11 @@ def get_environment() -> str:
     return _env if _env in ["internal-dev", "int", "ref", "sandbox", "prod"] else "internal-dev"
 
 
-def get_csv_content_dict_reader(file_key: str, encoder="utf-8") -> DictReader:
+def get_csv_content_dict_reader(file_key: str, encoding="utf-8") -> DictReader:
     """Returns the requested file contents from the source bucket in the form of a DictReader"""
     response = get_s3_client().get_object(Bucket=os.getenv("SOURCE_BUCKET_NAME"), Key=file_key)
     binary_io = response["Body"]
-    text_io = TextIOWrapper(binary_io, encoding=encoder, newline="")
+    text_io = TextIOWrapper(binary_io, encoding=encoding, newline="")
     return DictReader(text_io, delimiter="|")