VED-32 Minimise FHIR JSON in batch delete operation (#735)

Author: dlzhry2nhs

backend/src/forwarding_batch_lambda.py

@@ -1,4 +1,4 @@
-"""Functions for forwarding each row to the Imms API"""
+"""Lambda Handler which streams batch file entries from Kinesis and forwards to the Imms FHIR API"""
 
 import os
 import simplejson as json
@@ -47,12 +47,12 @@ def create_diagnostics_dictionary(error: Exception) -> dict:
 
 
 def forward_request_to_dynamo(
-    message_body: any, table: any, is_present: bool, batchcontroller: ImmunizationBatchController
+    message_body: any, table: any, is_present: bool, batch_controller: ImmunizationBatchController
 ):
     """Forwards the request to the Imms API (where possible) and updates the ack file with the outcome"""
     row_id = message_body.get("row_id")
     logger.info("FORWARDED MESSAGE: ID %s", row_id)
-    return batchcontroller.send_request_to_dynamo(message_body, table, is_present)
+    return batch_controller.send_request_to_dynamo(message_body, table, is_present)
 
 
 def forward_lambda_handler(event, _):
@@ -62,6 +62,7 @@ def forward_lambda_handler(event, _):
     array_of_messages = []
     array_of_identifiers = []
     controller = make_batch_controller()
+
     for record in event["Records"]:
         try:
             kinesis_payload = record["kinesis"]["data"]

backend/tests/test_forwarding_batch_lambda.py

@@ -67,18 +67,18 @@ def tearDown(self):
         patch.stopall()
 
     @staticmethod
-    def generate_fhir_json(include_fhir_json=True, identifier_value=None):
+    def generate_fhir_json(include_fhir_json=True, identifier_value=None, operation_requested="CREATE"):
         """Generates the fhir json for cases where included and None if excluded"""
-        if include_fhir_json:
-            fhir_json = copy.deepcopy(MockFhirImmsResources.all_fields)
+        if not include_fhir_json:
+            return None
 
-            if "identifier" in fhir_json and fhir_json["identifier"]:
-                if identifier_value is not None:
-                    fhir_json["identifier"][0]["value"] = identifier_value
+        fhir_json = copy.deepcopy(MockFhirImmsResources.all_fields) if operation_requested != "DELETE" else (
+            copy.deepcopy(MockFhirImmsResources.delete_operation_fields))
 
-            return fhir_json
+        if fhir_json.get("identifier") and identifier_value is not None:
+            fhir_json["identifier"][0]["value"] = identifier_value
 
-        return None
+        return fhir_json
 
     @staticmethod
     def generate_details_from_processing(
@@ -90,7 +90,11 @@ def generate_details_from_processing(
             "local_id": local_id,
         }
         if include_fhir_json:
-            details["fhir_json"] = TestForwardLambdaHandler.generate_fhir_json(include_fhir_json, identifier_value)
+            details["fhir_json"] = TestForwardLambdaHandler.generate_fhir_json(
+                include_fhir_json,
+                identifier_value,
+                operation_requested
+            )
         return details
 
     @staticmethod

backend/tests/utils/test_utils_for_batch.py

@@ -158,3 +158,32 @@ class MockFhirImmsResources:
             }
         ],
     }
+
+    # VED-32 Object for the delete batch operation will only contain the minimum fieldset
+    delete_operation_fields = {
+        "resourceType": "Immunization",
+        "status": "completed",
+        "protocolApplied": [
+            {
+                "targetDisease": [
+                    {
+                        "coding": [
+                            {
+                                "system": "http://snomed.info/sct",
+                                "code": "398102009",
+                                "display": "Acute poliomyelitis",
+                            }
+                        ]
+                    }
+                ],
+                "doseNumberPositiveInt": 1,
+            }
+        ],
+        "recorded": "2024-09-04",
+        "identifier": [
+            {
+                "value": "RSV_002",
+                "system": "https://www.ravs.england.nhs.uk/"
+            }
+        ]
+    }

recordprocessor/Dockerfile

@@ -24,5 +24,5 @@ COPY src .
 RUN chmod 644 $(find . -type f) && chmod 755 $(find . -type d)
 # Switch to the non-root user for running the container
 USER 1001:1001
-CMD ["python", "batch_processing.py"]
-ENTRYPOINT ["python", "batch_processing.py"]
+CMD ["python", "batch_processor.py"]
+ENTRYPOINT ["python", "batch_processor.py"]

recordprocessor/src/batch_processing.py recordprocessor/src/batch_processor.pyrecordprocessor/src/batch_processing.py renamed to recordprocessor/src/batch_processor.py

@@ -1,4 +1,4 @@
-"""Functions for processing the file on a row-by-row basis"""
+"""Application to convert rows from batch files to FHIR and forward to Kinesis for further downstream processing"""
 
 import json
 import os
@@ -32,7 +32,7 @@ def process_csv_to_fhir(incoming_message_body: dict) -> None:
 
     target_disease = map_target_disease(vaccine)
 
-    row_count = 0  # Initialize a counter for rows
+    row_count = 0
     for row in csv_reader:
         row_count += 1
         row_id = f"{file_id}^{row_count}"

recordprocessor/src/convert_to_fhir_imms_resource.py

@@ -1,8 +1,9 @@
-""" "Decorators to add the relevant fields to the FHIR immunization resource from the batch stream"""
+"""Decorators to add the relevant fields to the FHIR immunization resource from the batch stream"""
 
 from typing import List, Callable, Dict
+
 from utils_for_fhir_conversion import _is_not_empty, Generate, Add, Convert
-from constants import Urls
+from constants import Operation, Urls
 
 
 ImmunizationDecorator = Callable[[Dict, Dict[str, str]], None]
@@ -125,7 +126,7 @@ def _decorate_vaccination(imms: dict, row: Dict[str, str]) -> None:
     Add.custom_item(imms, "doseQuantity", dose_quantity_values, Generate.dictionary(dose_quantity_dict))
 
     # If DOSE_SEQUENCE is empty, default FHIR "doseNumberString" to "Dose sequence not recorded",
-    # otherwise assume the sender's intentiion is to supply a positive integer
+    # otherwise assume the sender's intention is to supply a positive integer
     if _is_not_empty(dose_sequence := row.get("DOSE_SEQUENCE")):
         Add.item(imms["protocolApplied"][0], "doseNumberPositiveInt", dose_sequence, Convert.integer)
     else:
@@ -145,10 +146,10 @@ def _decorate_performer(imms: dict, row: Dict[str, str]) -> None:
         performing_prof_surname := row.get("PERFORMING_PROFESSIONAL_SURNAME"),
         performing_prof_forename := row.get("PERFORMING_PROFESSIONAL_FORENAME"),
     ]
-    peformer_values = organization_values + practitioner_values
+    performer_values = organization_values + practitioner_values
 
     # Add performer if there is at least one non-empty performer value
-    if any(_is_not_empty(value) for value in peformer_values):
+    if any(_is_not_empty(value) for value in performer_values):
         imms["performer"] = []
 
         # Add organization if there is at least one non-empty organization value
@@ -195,7 +196,15 @@ def _decorate_performer(imms: dict, row: Dict[str, str]) -> None:
 ]
 
 
-def convert_to_fhir_imms_resource(row: dict, target_disease: list) -> dict:
+def _get_decorators_for_action_flag(action_flag: Operation) -> List[ImmunizationDecorator]:
+    # VED-32 DELETE action only requires the immunisation decorator
+    if action_flag == Operation.DELETE:
+        return [_decorate_immunization]
+
+    return all_decorators
+
+
+def convert_to_fhir_imms_resource(row: dict, target_disease: list, action_flag: Operation | str) -> dict:
     """Converts a row of data to a FHIR Immunization Resource"""
     # Prepare the imms_resource. Note that all data sent via this service is assumed to be for completed vaccinations.
     imms_resource = {
@@ -204,8 +213,9 @@ def convert_to_fhir_imms_resource(row: dict, target_disease: list) -> dict:
         "protocolApplied": [{"targetDisease": target_disease}]
     }
 
-    # Apply all decorators to add the relevant fields to the imms_resource
-    for decorator in all_decorators:
+    required_decorators = _get_decorators_for_action_flag(action_flag)
+
+    for decorator in required_decorators:
         decorator(imms_resource, row)
 
     return imms_resource

recordprocessor/src/process_row.py

@@ -40,7 +40,7 @@ def process_row(target_disease: list, allowed_operations: set, row: dict) -> dic
         }
 
     # Handle missing UNIQUE_ID or UNIQUE_ID_URI
-    if not (row.get("UNIQUE_ID_URI") and row.get("UNIQUE_ID")):
+    if not (unique_id_uri and unique_id):
         logger.error("Invalid row format: row is missing either UNIQUE_ID or UNIQUE_ID_URI")
         return {
             "diagnostics": create_diagnostics_dictionary("MISSING_UNIQUE_ID", 400, Diagnostics.MISSING_UNIQUE_ID),
@@ -50,7 +50,7 @@ def process_row(target_disease: list, allowed_operations: set, row: dict) -> dic
 
     # Handle success
     return {
-        "fhir_json": convert_to_fhir_imms_resource(row, target_disease),
+        "fhir_json": convert_to_fhir_imms_resource(row, target_disease, action_flag),
         "operation_requested": operation_requested,
         "local_id": local_id,
     }

recordprocessor/tests/test_convert_to_fhir_imms_resource.py

@@ -1,5 +1,6 @@
 """Tests for convert_to_fhir_imms_resource"""
 import unittest
+from typing import Tuple, List
 from unittest.mock import patch
 
 from tests.utils_for_recordprocessor_tests.values_for_recordprocessor_tests import (
@@ -10,7 +11,13 @@
 from tests.utils_for_recordprocessor_tests.mock_environment_variables import MOCK_ENVIRONMENT_DICT
 
 with patch("os.environ", MOCK_ENVIRONMENT_DICT):
-    from convert_to_fhir_imms_resource import convert_to_fhir_imms_resource
+    from convert_to_fhir_imms_resource import (
+        _decorate_immunization,
+        _get_decorators_for_action_flag,
+        all_decorators,
+        convert_to_fhir_imms_resource,
+        ImmunizationDecorator
+    )
 
 
 class TestConvertToFhirImmsResource(unittest.TestCase):
@@ -22,22 +29,46 @@ def test_convert_to_fhir_imms_resource(self):
         outputted FHIR Immunization Resource matches the expected output.
         """
 
-        # Test cases tuples are structure as (test_name, input_values, expected_output)
-        cases = [
-            ("All fields", MockFieldDictionaries.all_fields, MockFhirImmsResources.all_fields),
+        # Test cases tuples are structured as (test_name, input_values, expected_output, action_flag)
+        test_cases = [
+            ("All fields", MockFieldDictionaries.all_fields, MockFhirImmsResources.all_fields, "UPDATE"),
             (
                 "Mandatory fields only",
                 MockFieldDictionaries.mandatory_fields_only,
                 MockFhirImmsResources.mandatory_fields_only,
+                "UPDATE"
             ),
             (
                 "Critical fields only",
                 MockFieldDictionaries.critical_fields_only,
                 MockFhirImmsResources.critical_fields,
+                "NEW"
             ),
+            (
+                "Delete action only converts minimal fields",
+                MockFieldDictionaries.mandatory_fields_delete_action,
+                MockFhirImmsResources.delete_operation_fields,
+                "DELETE"
+            )
         ]
 
-        for test_name, input_values, expected_output in cases:
+        for test_name, input_values, expected_output, action_flag in test_cases:
             with self.subTest(test_name):
-                output = convert_to_fhir_imms_resource(input_values, TargetDiseaseElements.RSV)
+                output = convert_to_fhir_imms_resource(input_values, TargetDiseaseElements.RSV, action_flag)
                 self.assertEqual(output, expected_output)
+
+    def test_get_decorators_for_action_flag(self):
+        """
+        Test that the _test_get_decorators_for_action_flag function returns the correct list of decorators based on the
+        action flag provided.
+        """
+        test_cases: List[Tuple[str, str, List[ImmunizationDecorator]]] = [
+            ("Delete action only returns one decorator", "DELETE", [_decorate_immunization]),
+            ("Update action returns all decorators", "UPDATE", all_decorators),
+            ("Create action returns all decorators", "CREATE", all_decorators)
+        ]
+
+        for test_name, action_flag, expected_decorators in test_cases:
+            with self.subTest(test_name):
+                result = _get_decorators_for_action_flag(action_flag)
+                self.assertEqual(result, expected_decorators)

recordprocessor/tests/test_process_csv_to_fhir.py

@@ -17,7 +17,7 @@
 from tests.utils_for_recordprocessor_tests.mock_environment_variables import MOCK_ENVIRONMENT_DICT, BucketNames
 
 with patch("os.environ", MOCK_ENVIRONMENT_DICT):
-    from batch_processing import process_csv_to_fhir
+    from batch_processor import process_csv_to_fhir
 
 
 s3_client = boto3.client("s3", region_name=REGION_NAME)
@@ -61,7 +61,7 @@ def test_process_csv_to_fhir_full_permissions(self):
             file_key=test_file.file_key, file_content=ValidMockFileContent.with_new_and_update_and_delete
         )
 
-        with patch("batch_processing.send_to_kinesis") as mock_send_to_kinesis:
+        with patch("batch_processor.send_to_kinesis") as mock_send_to_kinesis:
             process_csv_to_fhir(deepcopy(test_file.event_full_permissions_dict))
 
         self.assertEqual(mock_send_to_kinesis.call_count, 3)
@@ -75,7 +75,7 @@ def test_process_csv_to_fhir_partial_permissions(self):
             file_key=test_file.file_key, file_content=ValidMockFileContent.with_new_and_update_and_delete
         )
 
-        with patch("batch_processing.send_to_kinesis") as mock_send_to_kinesis:
+        with patch("batch_processor.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, 3)
@@ -84,7 +84,7 @@ def test_process_csv_to_fhir_no_permissions(self):
         """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:
+        with patch("batch_processor.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, 2)
@@ -99,7 +99,7 @@ def test_process_csv_to_fhir_invalid_headers(self):
             file_content=ValidMockFileContent.with_new_and_update.replace("NHS_NUMBER", "NHS_NUMBERS"),
         )
 
-        with patch("batch_processing.send_to_kinesis") as mock_send_to_kinesis:
+        with patch("batch_processor.send_to_kinesis") as mock_send_to_kinesis:
             process_csv_to_fhir(deepcopy(test_file.event_full_permissions_dict))
 
         self.assertEqual(mock_send_to_kinesis.call_count, 0)

recordprocessor/tests/test_recordprocessor_main.py

@@ -26,7 +26,7 @@
 
 with patch("os.environ", MOCK_ENVIRONMENT_DICT):
     from constants import Diagnostics
-    from batch_processing import main
+    from batch_processor import main
 
 s3_client = boto3_client("s3", region_name=REGION_NAME)
 kinesis_client = boto3_client("kinesis", region_name=REGION_NAME)