VED-470: Stream records in record processor instead of loading everything into memory.

Author: mfjarvis

recordprocessor/src/batch_processing.py

@@ -20,7 +20,7 @@ def process_csv_to_fhir(incoming_message_body: dict) -> None:
         interim_message_body = file_level_validation(incoming_message_body=incoming_message_body)
     except (InvalidHeaders, NoOperationPermissions, Exception):  # pylint: disable=broad-exception-caught
         # If the file is invalid, processing should cease immediately
-        return None
+        return
 
     file_id = interim_message_body.get("message_id")
     vaccine = interim_message_body.get("vaccine")

recordprocessor/src/constants.py

@@ -87,8 +87,8 @@ class Urls:
     VACCINATION_PROCEDURE = "https://fhir.hl7.org.uk/StructureDefinition/Extension-UKCore-VaccinationProcedure"
 
 
-class ActionFlag(StrEnum):
-    CREATE = "NEW"
+class Operation(StrEnum):
+    CREATE = "CREATE"
     UPDATE = "UPDATE"
     DELETE = "DELETE"
 
@@ -99,8 +99,8 @@ class Permission(StrEnum):
     DELETE = "D"
 
 
-permission_to_action_flag_map = {
-    Permission.CREATE: ActionFlag.CREATE,
-    Permission.UPDATE: ActionFlag.UPDATE,
-    Permission.DELETE: ActionFlag.DELETE
-    }
+permission_to_operation_map = {
+    Permission.CREATE: Operation.CREATE,
+    Permission.UPDATE: Operation.UPDATE,
+    Permission.DELETE: Operation.DELETE
+}

recordprocessor/src/file_level_validation.py

@@ -2,15 +2,13 @@
 Functions for completing file-level validation
 (validating headers and ensuring that the supplier has permission to perform at least one of the requested operations)
 """
-
-from unique_permission import get_unique_action_flags_from_s3
 from clients import logger, s3_client
 from make_and_upload_ack_file import make_and_upload_ack_file
 from utils_for_recordprocessor import get_csv_content_dict_reader, invoke_filename_lambda
 from errors import InvalidHeaders, NoOperationPermissions
 from logging_decorator import file_level_validation_logging_decorator
 from audit_table import change_audit_table_status_to_processed, get_next_queued_file_details
-from constants import SOURCE_BUCKET_NAME, EXPECTED_CSV_HEADERS, permission_to_action_flag_map, ActionFlag, Permission
+from constants import SOURCE_BUCKET_NAME, EXPECTED_CSV_HEADERS, permission_to_operation_map, Permission
 
 
 def validate_content_headers(csv_content_reader) -> None:
@@ -19,25 +17,9 @@ def validate_content_headers(csv_content_reader) -> None:
         raise InvalidHeaders("File headers are invalid.")
 
 
-def validate_action_flag_permissions(
-    supplier: str, vaccine_type: str, allowed_permissions_list: list, csv_data: str
+def get_permitted_operations(
+    supplier: str, vaccine_type: str, allowed_permissions_list: list
 ) -> set:
-    """
-    Validates that the supplier has permission to perform at least one of the requested operations for the given
-    vaccine type and returns the set of allowed operations for that vaccine type.
-    Raises a NoPermissionsError if the supplier does not have permission to perform any of the requested operations.
-    """
-
-    # Get unique ACTION_FLAG values from the S3 file
-    required_action_flags = get_unique_action_flags_from_s3(csv_data)
-
-    valid_action_flag_values = {flag.value for flag in ActionFlag}
-    required_action_flags = required_action_flags & valid_action_flag_values  # intersection
-
-    if not required_action_flags:
-        logger.warning("No valid ACTION_FLAGs found in file. Skipping permission validation.")
-        return set()
-
     # Check if supplier has permission for the subject vaccine type and extract permissions
     permission_strs_for_vaccine_type = {
         permission_str
@@ -54,24 +36,17 @@ def validate_action_flag_permissions(
     }
 
     # Map Permission key to action flag
-    permitted_action_flags_for_vaccine_type = {
-        permission_to_action_flag_map[permission].value
+    permitted_operations_for_vaccine_type = {
+        permission_to_operation_map[permission].value
         for permission in permissions_for_vaccine_type
     }
 
-    if not required_action_flags.intersection(permitted_action_flags_for_vaccine_type):
+    if not permitted_operations_for_vaccine_type:
         raise NoOperationPermissions(
             f"{supplier} does not have permissions to perform any of the requested actions."
         )
 
-    logger.info(
-         "%s permissions %s match one of the requested permissions required to %s",
-         supplier,
-         allowed_permissions_list,
-         permitted_action_flags_for_vaccine_type,
-     )
-
-    return {permission.name for permission in permissions_for_vaccine_type}
+    return permitted_operations_for_vaccine_type
 
 
 def move_file(bucket_name: str, source_file_key: str, destination_file_key: str) -> None:
@@ -103,12 +78,12 @@ def file_level_validation(incoming_message_body: dict) -> dict:
         created_at_formatted_string = incoming_message_body.get("created_at_formatted_string")
 
         # Fetch the data
-        csv_reader, csv_data = get_csv_content_dict_reader(file_key)
+        csv_reader = get_csv_content_dict_reader(file_key)
 
         validate_content_headers(csv_reader)
 
         # Validate has permission to perform at least one of the requested actions
-        allowed_operations_set = validate_action_flag_permissions(supplier, vaccine, permission, csv_data)
+        allowed_operations_set = get_permitted_operations(supplier, vaccine, permission)
 
         make_and_upload_ack_file(message_id, file_key, True, True, created_at_formatted_string)
 

recordprocessor/src/unique_permission.py

@@ -3,7 +3,7 @@
 import os
 import json
 from csv import DictReader
-from io import StringIO
+from io import StringIO, TextIOWrapper
 from clients import s3_client, lambda_client, logger
 from constants import SOURCE_BUCKET_NAME, FILE_NAME_PROC_LAMBDA_NAME
 
@@ -15,12 +15,12 @@ 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) -> (DictReader, str):
+def get_csv_content_dict_reader(file_key: str) -> DictReader:
     """Returns the requested file contents from the source bucket in the form of a DictReader"""
     response = s3_client.get_object(Bucket=os.getenv("SOURCE_BUCKET_NAME"), Key=file_key)
-    # TODO - this reads everything into memory! Look into streaming instead
-    csv_data = response["Body"].read().decode("utf-8")
-    return DictReader(StringIO(csv_data), delimiter="|"), csv_data
+    binary_io = response["Body"]
+    text_io = TextIOWrapper(binary_io, encoding="utf-8")
+    return DictReader(text_io, delimiter="|")
 
 
 def create_diagnostics_dictionary(error_type, status_code, error_message) -> dict:

recordprocessor/src/utils_for_recordprocessor.py

@@ -11,7 +11,7 @@
 
 with patch("os.environ", MOCK_ENVIRONMENT_DICT):
     from errors import NoOperationPermissions, InvalidHeaders
-    from file_level_validation import validate_content_headers, validate_action_flag_permissions
+    from file_level_validation import validate_content_headers, get_permitted_operations
 
 
 test_file = MockFileDetails.rsv_emis
@@ -43,80 +43,56 @@ def test_validate_content_headers(self):
                 with self.assertRaises(InvalidHeaders):
                     validate_content_headers(test_data)
 
-    def test_validate_action_flag_permissions(self):
-        """
-        Tests that validate_action_flag_permissions returns True if supplier has permissions to perform at least one
-        of the requested CRUD operations for the given vaccine type, and False otherwise
-        """
+    def test_get_permitted_operations(self):
         # Set up test file content. Note that ValidFileContent has action flags in lower case
-        valid_file_content = ValidMockFileContent.with_new_and_update
-        valid_content_new_and_update_lowercase = valid_file_content
-        valid_content_new_and_update_uppercase = valid_file_content.replace("new", "NEW").replace("update", "UPDATE")
-        valid_content_new_and_update_mixedcase = valid_file_content.replace("new", "New").replace("update", "uPdAte")
-        valid_content_new_and_delete_lowercase = valid_file_content.replace("update", "delete")
-        valid_content_update_and_delete_lowercase = valid_file_content.replace("new", "delete").replace(
-            "update", "UPDATE"
-        )
 
         # Case: Supplier has permissions to perform at least one of the requested operations
         # Test case tuples are stuctured as (vaccine_type, vaccine_permissions, file_content, expected_output)
         test_cases = [
             # FLU, full permissions, lowercase action flags
-            ("FLU", ["FLU.CRUD"], valid_content_new_and_update_lowercase, {"CREATE", "UPDATE", "DELETE"}),
+            ("FLU", ["FLU.CRUD"], {"CREATE", "UPDATE", "DELETE"}),
             # FLU, partial permissions, uppercase action flags
-            ("FLU", ["FLU.C"], valid_content_new_and_update_uppercase, {"CREATE"}),
+            ("FLU", ["FLU.C"], {"CREATE"}),
             # FLU, full permissions, mixed case action flags
-            ("FLU", ["FLU.CRUD"], valid_content_new_and_update_mixedcase, {"CREATE", "UPDATE", "DELETE"}),
+            ("FLU", ["FLU.CRUD"], {"CREATE", "UPDATE", "DELETE"}),
             # FLU, partial permissions (create)
-            ("FLU", ["FLU.D", "FLU.C"], valid_content_new_and_update_lowercase, {"CREATE", "DELETE"}),
+            ("FLU", ["FLU.D", "FLU.C"], {"CREATE", "DELETE"}),
             # FLU, partial permissions (update)
-            ("FLU", ["FLU.U"], valid_content_new_and_update_lowercase, {"UPDATE"}),
+            ("FLU", ["FLU.U"], {"UPDATE"}),
             # FLU, partial permissions (delete)
-            ("FLU", ["FLU.D"], valid_content_new_and_delete_lowercase, {"DELETE"}),
+            ("FLU", ["FLU.D"], {"DELETE"}),
             # COVID19, full permissions
-            ("COVID19", ["COVID19.CRUD"], valid_content_new_and_delete_lowercase, {"CREATE", "UPDATE", "DELETE"}),
+            ("COVID19", ["COVID19.CRUD"], {"CREATE", "UPDATE", "DELETE"}),
             # COVID19, partial permissions
-            ("COVID19", ["COVID19.U"], valid_content_update_and_delete_lowercase, {"UPDATE"}),
+            ("COVID19", ["COVID19.U"], {"UPDATE"}),
             # RSV, full permissions
-            ("RSV", ["RSV.CRUD"], valid_content_new_and_delete_lowercase, {"CREATE", "UPDATE", "DELETE"}),
+            ("RSV", ["RSV.CRUD"], {"CREATE", "UPDATE", "DELETE"}),
             # RSV, partial permissions
-            ("RSV", ["RSV.U"], valid_content_update_and_delete_lowercase, {"UPDATE"}),
+            ("RSV", ["RSV.U"], {"UPDATE"}),
             # RSV, full permissions, mixed case action flags
-            ("RSV", ["RSV.CRUD"], valid_content_new_and_update_mixedcase, {"CREATE", "UPDATE", "DELETE"}),
+            ("RSV", ["RSV.CRUD"], {"CREATE", "UPDATE", "DELETE"}),
         ]
 
-        for vaccine_type, vaccine_permissions, file_content, expected_output in test_cases:
+        for vaccine_type, vaccine_permissions, expected_output in test_cases:
             with self.subTest(f"Vaccine_type {vaccine_type} - permissions {vaccine_permissions}"):
                 self.assertEqual(
-                    validate_action_flag_permissions("TEST_SUPPLIER", vaccine_type, vaccine_permissions, file_content),
+                    get_permitted_operations("TEST_SUPPLIER", vaccine_type, vaccine_permissions),
                     expected_output,
                 )
 
         # Case: Supplier has no permissions to perform any of the requested operations
         # Test case tuples are stuctured as (vaccine_type, vaccine_permissions, file_content)
         invalid_cases = [
-            # FLU, no permissions
-            ("FLU", ["FLU.U", "COVID19.CRUDS"], valid_content_new_and_delete_lowercase),
             # COVID19, no permissions
-            ("COVID19", ["FLU.C", "FLU.U"], valid_content_update_and_delete_lowercase),
-            # RSV, no permissions
-            ("RSV", ["FLU.C", "FLU.U"], valid_content_update_and_delete_lowercase),
+            ("COVID19", ["FLU.CRUDS", "RSV.CUD"]),
+            # RSV, no valid permissions
+            ("RSV", ["FLU.C", "RSV.XYZ"]),
         ]
 
-        for vaccine_type, vaccine_permissions, file_content in invalid_cases:
+        for vaccine_type, vaccine_permissions in invalid_cases:
             with self.subTest():
                 with self.assertRaises(NoOperationPermissions):
-                    validate_action_flag_permissions("TEST_SUPPLIER", vaccine_type, vaccine_permissions, file_content)
-
-        no_flag_cases = [
-            ("FLU", ["FLU.C"], valid_file_content.replace("new", "").replace("update", "")),
-            ("COVID19", ["COVID19.CRUD"], valid_file_content.replace("new", "INVALID").replace("update", "")),
-            ]
-
-        for vaccine_type, permissions, file_content in no_flag_cases:
-            with self.subTest(f"{vaccine_type} with invalid or missing ACTION_FLAGs"):
-                result = validate_action_flag_permissions("TEST_SUPPLIER", vaccine_type, permissions, file_content)
-                self.assertEqual(result, set())
+                    get_permitted_operations("TEST_SUPPLIER", vaccine_type, vaccine_permissions)
 
 
 if __name__ == "__main__":

recordprocessor/tests/test_file_level_validation.py

@@ -171,7 +171,7 @@ def test_splunk_logger_handled_failure(self):
             # CASE: No operation permissions
             (
                 ValidMockFileContent.with_new_and_update,
-                MOCK_FILE_DETAILS.event_delete_permissions_only_dict,  # No permission for NEW or UPDATE
+                MOCK_FILE_DETAILS.event_no_permissions_dict,  # No permission for NEW or UPDATE
                 NoOperationPermissions,
                 403,
                 f"{MOCK_FILE_DETAILS.supplier} does not have permissions to perform any of the requested actions.",

recordprocessor/tests/test_logging_decorator.py

@@ -81,13 +81,17 @@ def test_process_csv_to_fhir_partial_permissions(self):
         self.assertEqual(mock_send_to_kinesis.call_count, 3)
 
     def test_process_csv_to_fhir_no_permissions(self):
-        """Tests that process_csv_to_fhir does not send a message to kinesis when the supplier has no permissions"""
+        """Tests that process_csv_to_fhir does not send fhir_json to kinesis when the supplier has no permissions"""
         self.upload_source_file(file_key=test_file.file_key, file_content=ValidMockFileContent.with_update_and_delete)
 
         with patch("batch_processing.send_to_kinesis") as mock_send_to_kinesis:
             process_csv_to_fhir(deepcopy(test_file.event_create_permissions_only_dict))
 
-        self.assertEqual(mock_send_to_kinesis.call_count, 0)
+        self.assertEqual(mock_send_to_kinesis.call_count, 2)
+        for (_supplier, message_body, _vaccine), _kwargs in mock_send_to_kinesis.call_args_list:
+            self.assertIn("diagnostics", message_body)
+            self.assertNotIn("fhir_json", message_body)
+
 
     def test_process_csv_to_fhir_invalid_headers(self):
         """Tests that process_csv_to_fhir does not send a message to kinesis when the csv has invalid headers"""

recordprocessor/tests/test_process_csv_to_fhir.py

@@ -227,7 +227,7 @@ 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)
 
-    def test_e2e_no_permissions(self):
+    def test_e2e_no_required_permissions(self):
         """
         Tests that file containing UPDATE and DELETE is successfully processed when the supplier has CREATE permissions
         only.
@@ -236,6 +236,23 @@ def test_e2e_no_permissions(self):
 
         main(mock_rsv_emis_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)
+        for record in kinesis_records:
+            data_bytes = record["Data"]
+            data_dict = json.loads(data_bytes)
+            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)
+
+    def test_e2e_no_permissions(self):
+        """
+        Tests that file containing UPDATE and DELETE is successfully processed when the supplier has no permissions.
+        """
+        self.upload_source_files(ValidMockFileContent.with_update_and_delete)
+
+        main(mock_rsv_emis_file.event_no_permissions)
+
         kinesis_records = kinesis_client.get_records(ShardIterator=self.get_shard_iterator(), Limit=10)["Records"]
         self.assertEqual(len(kinesis_records), 0)
         self.make_inf_ack_assertions(file_details=mock_rsv_emis_file, passed_validation=False)