Refactored record processor

Author: dlzhry2nhs

recordprocessor/src/audit_table.py

@@ -1,20 +1,35 @@
 """Add the filename to the audit table and check for duplicates."""
+from typing import Optional
 
 from clients import dynamodb_client, logger
 from errors import UnhandledAuditTableError
 from constants import AUDIT_TABLE_NAME, AuditTableKeys
 
 
-def update_audit_table_status(file_key: str, message_id: str, status: str) -> None:
+def update_audit_table_status(
+    file_key: str,
+    message_id: str,
+    status: str,
+    error_details: Optional[str] = None
+) -> None:
     """Updates the status in the audit table to the requested value"""
+    update_expression = f"SET #{AuditTableKeys.STATUS} = :status"
+    expression_attr_names = {f"#{AuditTableKeys.STATUS}": "status"}
+    expression_attr_values = {":status": {"S": status}}
+
+    if error_details is not None:
+        update_expression = update_expression + f", #{AuditTableKeys.ERROR_DETAILS} = :error_details"
+        expression_attr_names[f"#{AuditTableKeys.ERROR_DETAILS}"] = "error_details"
+        expression_attr_values[":error_details"] = {"S": error_details}
+
     try:
         # Update the status in the audit table to "Processed"
         dynamodb_client.update_item(
             TableName=AUDIT_TABLE_NAME,
             Key={AuditTableKeys.MESSAGE_ID: {"S": message_id}},
-            UpdateExpression="SET #status = :status",
-            ExpressionAttributeNames={"#status": "status"},
-            ExpressionAttributeValues={":status": {"S": status}},
+            UpdateExpression=update_expression,
+            ExpressionAttributeNames=expression_attr_names,
+            ExpressionAttributeValues=expression_attr_values,
             ConditionExpression="attribute_exists(message_id)",
         )
 

recordprocessor/src/batch_processor.py

@@ -3,6 +3,7 @@
 import json
 import os
 import time
+from json import JSONDecodeError
 
 from constants import FileStatus
 from process_row import process_row
@@ -66,9 +67,23 @@ def main(event: str) -> None:
     logger.info("task started")
     start = time.time()
     try:
-        process_csv_to_fhir(incoming_message_body=json.loads(event))
+        incoming_message_body = json.loads(event)
+    except JSONDecodeError as error:
+        logger.error("Error decoding incoming message: %s", error)
+        return
+
+    try:
+        process_csv_to_fhir(incoming_message_body=incoming_message_body)
     except Exception as error:  # pylint: disable=broad-exception-caught
         logger.error("Error processing message: %s", error)
+        message_id = incoming_message_body.get("message_id")
+        file_key = incoming_message_body.get("file_key")
+
+        # If an unexpected error occurs, attempt to mark the event as failed. If the event is so malformed that this
+        # also fails, we will still get the error alert and the event will remain in processing meaning the supplier +
+        # vacc type queue is blocked until we resolve the issue
+        update_audit_table_status(file_key, message_id, FileStatus.FAILED, error_details=str(error))
+
     end = time.time()
     logger.info("Total time for completion: %ss", round(end - start, 5))
 

recordprocessor/src/constants.py

@@ -55,7 +55,14 @@ class FileStatus:
     PROCESSING = "Processing"
     PREPROCESSED = "Preprocessed"  # All entries in file converted to FHIR and forwarded to Kinesis
     PROCESSED = "Processed"  # All entries processed and ack file created
-    DUPLICATE = "Duplicate"
+    NOT_PROCESSED = "Not processed"
+    FAILED = "Failed"
+
+
+class FileNotProcessedReason(StrEnum):
+    """Reasons why a file was not processed"""
+    UNAUTHORISED = "Unauthorised"
+    INVALID_FILE_HEADERS = "Invalid file headers"
 
 
 class AuditTableKeys:
@@ -66,6 +73,7 @@ class AuditTableKeys:
     QUEUE_NAME = "queue_name"
     STATUS = "status"
     TIMESTAMP = "timestamp"
+    ERROR_DETAILS = "error_details"
 
 
 class Diagnostics:

recordprocessor/src/file_level_validation.py

@@ -8,7 +8,8 @@
 from errors import InvalidHeaders, NoOperationPermissions
 from logging_decorator import file_level_validation_logging_decorator
 from audit_table import update_audit_table_status
-from constants import SOURCE_BUCKET_NAME, EXPECTED_CSV_HEADERS, permission_to_operation_map, FileStatus, Permission
+from constants import SOURCE_BUCKET_NAME, EXPECTED_CSV_HEADERS, permission_to_operation_map, FileStatus, Permission, \
+    FileNotProcessedReason
 
 
 def validate_content_headers(csv_content_reader) -> None:
@@ -17,6 +18,18 @@ def validate_content_headers(csv_content_reader) -> None:
         raise InvalidHeaders("File headers are invalid.")
 
 
+def get_file_status_for_error(error: Exception) -> str:
+    """Returns the appropriate file status based on the error that occurred"""
+    if isinstance(error, NoOperationPermissions):
+        return f"{FileStatus.NOT_PROCESSED} - {FileNotProcessedReason.UNAUTHORISED}"
+
+    # TODO - discuss with team. Do we want client errors to leave pipeline unblocked. Or block and investigate?
+    elif isinstance(error, InvalidHeaders):
+        return f"{FileStatus.NOT_PROCESSED} - {FileNotProcessedReason.INVALID_FILE_HEADERS}"
+
+    return FileStatus.FAILED
+
+
 def get_permitted_operations(
     supplier: str, vaccine_type: str, allowed_permissions_list: list
 ) -> set:
@@ -107,12 +120,13 @@ def file_level_validation(incoming_message_body: dict) -> dict:
         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 = get_file_status_for_error(error)
 
         try:
             move_file(SOURCE_BUCKET_NAME, file_key, f"archive/{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, FileStatus.PROCESSED)
+        update_audit_table_status(file_key, message_id, file_status, error_details=str(error))
         raise

recordprocessor/tests/test_audit_table.py

@@ -65,6 +65,22 @@ def test_update_audit_table_status(self):
 
         self.assertIn(expected_table_entry, table_items)
 
+    def test_update_audit_table_status_including_error_details(self):
+        """Checks audit table correctly updates a record including some error details"""
+        add_entry_to_table(MockFileDetails.rsv_ravs, file_status=FileStatus.QUEUED)
+        ravs_rsv_test_file = FileDetails("RSV", "RAVS", "X26")
+
+        update_audit_table_status(ravs_rsv_test_file.file_key, ravs_rsv_test_file.message_id, FileStatus.FAILED,
+                                  error_details="Test error details")
+
+        table_items = dynamodb_client.scan(TableName=AUDIT_TABLE_NAME).get("Items", [])
+        self.assertEqual(1, len(table_items))
+        self.assertEqual({
+            **MockFileDetails.rsv_ravs.audit_table_entry,
+            "status": {"S": FileStatus.FAILED},
+            "error_details": {"S": "Test error details"}
+        }, table_items[0])
+
     def test_update_audit_table_status_throws_exception_with_invalid_id(self):
         emis_flu_test_file_2 = FileDetails("FLU", "EMIS", "YGM41")
 

recordprocessor/tests/test_recordprocessor_main.py

@@ -3,14 +3,17 @@
 import unittest
 import json
 from decimal import Decimal
+from json import JSONDecodeError
 from unittest.mock import patch
 from datetime import datetime, timedelta, timezone
-from moto import mock_s3, mock_kinesis, mock_firehose
+from moto import mock_s3, mock_kinesis, mock_firehose, mock_dynamodb
 from boto3 import client as boto3_client
 
 from tests.utils_for_recordprocessor_tests.utils_for_recordprocessor_tests import (
     GenericSetUp,
     GenericTearDown,
+    add_entry_to_table,
+    assert_audit_table_entry
 )
 from tests.utils_for_recordprocessor_tests.values_for_recordprocessor_tests import (
     MockFileDetails,
@@ -25,36 +28,41 @@
 from tests.utils_for_recordprocessor_tests.mock_environment_variables import MOCK_ENVIRONMENT_DICT, BucketNames, Kinesis
 
 with patch("os.environ", MOCK_ENVIRONMENT_DICT):
-    from constants import Diagnostics
+    from constants import Diagnostics, FileStatus, FileNotProcessedReason, AUDIT_TABLE_NAME, AuditTableKeys
     from batch_processor import main
 
 s3_client = boto3_client("s3", region_name=REGION_NAME)
 kinesis_client = boto3_client("kinesis", region_name=REGION_NAME)
 firehose_client = boto3_client("firehose", region_name=REGION_NAME)
+dynamo_db_client = boto3_client("dynamodb", region_name=REGION_NAME)
 yesterday = datetime.now(timezone.utc) - timedelta(days=1)
 mock_rsv_emis_file = MockFileDetails.rsv_emis
 
 
 @patch.dict("os.environ", MOCK_ENVIRONMENT_DICT)
+@mock_dynamodb
 @mock_s3
 @mock_kinesis
 @mock_firehose
 class TestRecordProcessor(unittest.TestCase):
     """Tests for main function for RecordProcessor"""
 
     def setUp(self) -> None:
-        GenericSetUp(s3_client, firehose_client, kinesis_client)
+        GenericSetUp(s3_client, firehose_client, kinesis_client, dynamo_db_client)
 
         redis_patcher = patch("mappings.redis_client")
+        batch_processor_logger_patcher = patch("batch_processor.logger")
         self.addCleanup(redis_patcher.stop)
+        self.mock_batch_processor_logger = batch_processor_logger_patcher.start()
         mock_redis_client = redis_patcher.start()
         mock_redis_client.hget.return_value = json.dumps([{
             "code": "55735004",
             "term": "Respiratory syncytial virus infection (disorder)"
         }])
 
     def tearDown(self) -> None:
-        GenericTearDown(s3_client, firehose_client, kinesis_client)
+        GenericTearDown(s3_client, firehose_client, kinesis_client, dynamo_db_client)
+        patch.stopall()
 
     @staticmethod
     def upload_source_files(source_file_content):  # pylint: disable=dangerous-default-value
@@ -148,9 +156,11 @@ def test_e2e_full_permissions(self):
         Tests that file containing CREATE, UPDATE and DELETE is successfully processed when the supplier has
         full permissions.
         """
+        test_file = mock_rsv_emis_file
         self.upload_source_files(ValidMockFileContent.with_new_and_update_and_delete)
+        add_entry_to_table(test_file, FileStatus.PROCESSING)
 
-        main(mock_rsv_emis_file.event_full_permissions)
+        main(test_file.event_full_permissions)
 
         # Assertion case tuples are stuctured as
         # (test_name, index, expected_kinesis_data_ignoring_fhir_json,expect_success)
@@ -176,15 +186,18 @@ def test_e2e_full_permissions(self):
         ]
         self.make_inf_ack_assertions(file_details=mock_rsv_emis_file, passed_validation=True)
         self.make_kinesis_assertions(assertion_cases)
+        assert_audit_table_entry(test_file, FileStatus.PREPROCESSED)
 
     def test_e2e_partial_permissions(self):
         """
         Tests that file containing CREATE, UPDATE and DELETE is successfully processed when the supplier only has CREATE
         permissions.
         """
+        test_file = mock_rsv_emis_file
+        add_entry_to_table(test_file, FileStatus.PROCESSING)
         self.upload_source_files(ValidMockFileContent.with_new_and_update_and_delete)
 
-        main(mock_rsv_emis_file.event_create_permissions_only)
+        main(test_file.event_create_permissions_only)
 
         # Assertion case tuples are stuctured as
         # (test_name, index, expected_kinesis_data_ignoring_fhir_json,expect_success)
@@ -226,15 +239,18 @@ def test_e2e_partial_permissions(self):
         ]
         self.make_inf_ack_assertions(file_details=mock_rsv_emis_file, passed_validation=True)
         self.make_kinesis_assertions(assertion_cases)
+        assert_audit_table_entry(test_file, FileStatus.PREPROCESSED)
 
     def test_e2e_no_required_permissions(self):
         """
         Tests that file containing UPDATE and DELETE is successfully processed when the supplier has CREATE permissions
         only.
         """
+        test_file = mock_rsv_emis_file
+        add_entry_to_table(test_file, FileStatus.PROCESSING)
         self.upload_source_files(ValidMockFileContent.with_update_and_delete)
 
-        main(mock_rsv_emis_file.event_create_permissions_only)
+        main(test_file.event_create_permissions_only)
 
         kinesis_records = kinesis_client.get_records(ShardIterator=self.get_shard_iterator(), Limit=10)["Records"]
         self.assertEqual(len(kinesis_records), 2)
@@ -244,27 +260,39 @@ def test_e2e_no_required_permissions(self):
             self.assertIn("diagnostics", data_dict)
             self.assertNotIn("fhir_json", data_dict)
         self.make_inf_ack_assertions(file_details=mock_rsv_emis_file, passed_validation=True)
+        assert_audit_table_entry(test_file, FileStatus.PREPROCESSED)
 
     def test_e2e_no_permissions(self):
         """
         Tests that file containing UPDATE and DELETE is successfully processed when the supplier has no permissions.
         """
+        test_file = mock_rsv_emis_file
+        add_entry_to_table(test_file, FileStatus.PROCESSING)
         self.upload_source_files(ValidMockFileContent.with_update_and_delete)
 
-        main(mock_rsv_emis_file.event_no_permissions)
+        main(test_file.event_no_permissions)
 
         kinesis_records = kinesis_client.get_records(ShardIterator=self.get_shard_iterator(), Limit=10)["Records"]
+        table_entry = dynamo_db_client.get_item(
+            TableName=AUDIT_TABLE_NAME, Key={AuditTableKeys.MESSAGE_ID: {"S": test_file.message_id}}
+        ).get("Item")
         self.assertEqual(len(kinesis_records), 0)
         self.make_inf_ack_assertions(file_details=mock_rsv_emis_file, passed_validation=False)
+        self.assertDictEqual(table_entry, {
+            **test_file.audit_table_entry,
+            "status": {"S": f"{FileStatus.NOT_PROCESSED} - {FileNotProcessedReason.UNAUTHORISED}"},
+            "error_details": {"S": "EMIS does not have permissions to perform any of the requested actions."}
+        })
 
     def test_e2e_invalid_action_flags(self):
         """Tests that file is successfully processed when the ACTION_FLAG field is empty or invalid."""
-
+        test_file = mock_rsv_emis_file
+        add_entry_to_table(test_file, FileStatus.PROCESSING)
         self.upload_source_files(
             ValidMockFileContent.with_update_and_delete.replace("update", "").replace("delete", "INVALID")
         )
 
-        main(mock_rsv_emis_file.event_full_permissions)
+        main(test_file.event_full_permissions)
 
         expected_kinesis_data = {
             "diagnostics": {
@@ -288,14 +316,16 @@ def test_e2e_invalid_action_flags(self):
     def test_e2e_differing_amounts_of_data(self):
         """Tests that file containing rows with differing amounts of data present is processed as expected"""
         # Create file content with different amounts of data present in each row
+        test_file = mock_rsv_emis_file
+        add_entry_to_table(test_file, FileStatus.PROCESSING)
         headers = "|".join(MockFieldDictionaries.all_fields.keys())
         all_fields_values = "|".join(f'"{v}"' for v in MockFieldDictionaries.all_fields.values())
         mandatory_fields_only_values = "|".join(f'"{v}"' for v in MockFieldDictionaries.mandatory_fields_only.values())
         critical_fields_only_values = "|".join(f'"{v}"' for v in MockFieldDictionaries.critical_fields_only.values())
         file_content = f"{headers}\n{all_fields_values}\n{mandatory_fields_only_values}\n{critical_fields_only_values}"
         self.upload_source_files(file_content)
 
-        main(mock_rsv_emis_file.event_full_permissions)
+        main(test_file.event_full_permissions)
 
         all_fields_row_expected_kinesis_data = {
             "operation_requested": "UPDATE",
@@ -329,6 +359,8 @@ def test_e2e_kinesis_failed(self):
         Tests that, for a file with valid content and supplier with full permissions, when the kinesis send fails, the
         ack file is created and documents an error.
         """
+        test_file = mock_rsv_emis_file
+        add_entry_to_table(test_file, FileStatus.PROCESSING)
         self.upload_source_files(ValidMockFileContent.with_new_and_update)
         # Delete the kinesis stream, to cause kinesis send to fail
         kinesis_client.delete_stream(StreamName=Kinesis.STREAM_NAME, EnforceConsumerDeletion=True)
@@ -340,11 +372,14 @@ def test_e2e_kinesis_failed(self):
         ):  # noqa: E999
             mock_time.time.side_effect = [1672531200, 1672531200.123456]
             mock_datetime.now.return_value = datetime(2024, 1, 1, 12, 0, 0)
-            main(mock_rsv_emis_file.event_full_permissions)
+            main(test_file.event_full_permissions)
 
         # Since the failure occured at row level, not file level, the ack file should still be created
         # and firehose logs should indicate a successful file level validation
-        self.make_inf_ack_assertions(file_details=mock_rsv_emis_file, passed_validation=True)
+        table_entry = dynamo_db_client.get_item(
+            TableName=AUDIT_TABLE_NAME, Key={AuditTableKeys.MESSAGE_ID: {"S": test_file.message_id}}
+        ).get("Item")
+        self.make_inf_ack_assertions(file_details=test_file, passed_validation=True)
         expected_log_data = {
             "function_name": "record_processor_file_level_validation",
             "date_time": "2024-01-01 12:00:00",
@@ -357,6 +392,25 @@ def test_e2e_kinesis_failed(self):
             "message": "Successfully sent for record processing",
         }
         mock_send_log_to_firehose.assert_called_with(expected_log_data)
+        self.assertDictEqual(table_entry, {
+            **test_file.audit_table_entry,
+            "status": {"S": FileStatus.FAILED},
+            "error_details": {"S": "An error occurred (ResourceNotFoundException) when calling the PutRecord operation"
+                                   ": Stream imms-batch-internal-dev-processingdata-stream under account 123456789012"
+                                   " not found."}
+        })
+
+    def test_e2e_error_is_logged_if_invalid_json_provided(self):
+        """This scenario should not happen. If it does, it means our batch processing system config is broken and we
+        have received malformed content from SQS -> EventBridge. In this case we log the error so we will be alerted.
+        However, we cannot do anything with the AuditDB record as we cannot retrieve information from the event"""
+        malformed_event = '{"test": {}'
+        main(malformed_event)
+
+        logged_message = self.mock_batch_processor_logger.error.call_args[0][0]
+        exception = self.mock_batch_processor_logger.error.call_args[0][1]
+        self.assertEqual(logged_message, "Error decoding incoming message: %s")
+        self.assertIsInstance(exception, JSONDecodeError)
 
 
 if __name__ == "__main__":