VED-372: Use configurable disease mappings in batch process. (#593)

* VED-372: Use configurable disease mappings in batch process.

* VED-372: Call Redis once per batch instead of once per row.

* VED-372: Connect ECS task to Redis.

* VED-372: Fix lint errors.

* VED-372: Add missing Redis patch.

* VED-372: Convert env var value to a string.

* VED-372: Fix import.

* VED-372: Query for correct key.

Author: mfjarvis

recordprocessor/poetry.lock

@@ -25,6 +25,7 @@ structlog               = "^24.1.0"
 pandas                  = "^2.3.0"
 freezegun               = "^1.5.2"
 coverage                = "^7.9.0"
+redis = "^6.2.0"
 
 [build-system]
 requires      = ["poetry-core ~= 1.5.0"]

recordprocessor/pyproject.toml

@@ -4,6 +4,7 @@
 import os
 import time
 from process_row import process_row
+from mappings import map_target_disease
 from send_to_kinesis import send_to_kinesis
 from clients import logger
 from file_level_validation import file_level_validation
@@ -29,26 +30,28 @@ def process_csv_to_fhir(incoming_message_body: dict) -> None:
     created_at_formatted_string = interim_message_body.get("created_at_formatted_string")
     csv_reader = interim_message_body.get("csv_dict_reader")
 
+    target_disease = map_target_disease(vaccine)
+
     row_count = 0  # Initialize a counter for rows
     for row in csv_reader:
         row_count += 1
         row_id = f"{file_id}^{row_count}"
         logger.info("MESSAGE ID : %s", row_id)
 
         # Process the row to obtain the details needed for the message_body and ack file
-        details_from_processing = process_row(vaccine, allowed_operations, row)
+        details_from_processing = process_row(target_disease, allowed_operations, row)
 
         # Create the message body for sending
         outgoing_message_body = {
             "row_id": row_id,
             "file_key": file_key,
             "supplier": supplier,
-            "vax_type": vaccine.value,
+            "vax_type": vaccine,
             "created_at_formatted_string": created_at_formatted_string,
             **details_from_processing,
         }
 
-        send_to_kinesis(supplier, outgoing_message_body, vaccine.value)
+        send_to_kinesis(supplier, outgoing_message_body, vaccine)
 
         logger.info("Total rows processed: %s", row_count)
 

recordprocessor/src/batch_processing.py

@@ -1,10 +1,15 @@
 """Initialise s3 and kinesis clients"""
 
 import logging
+import os
+
+import redis
 from boto3 import client as boto3_client, resource as boto3_resource
 from botocore.config import Config
 
 REGION_NAME = "eu-west-2"
+REDIS_HOST = os.getenv("REDIS_HOST", "")
+REDIS_PORT = os.getenv("REDIS_PORT", 6379)
 
 s3_client = boto3_client("s3", region_name=REGION_NAME)
 kinesis_client = boto3_client(
@@ -14,6 +19,7 @@
 firehose_client = boto3_client("firehose", region_name=REGION_NAME)
 dynamodb_client = boto3_client("dynamodb", region_name=REGION_NAME)
 lambda_client = boto3_client("lambda", region_name=REGION_NAME)
+redis_client = redis.StrictRedis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=True)
 
 dynamodb_resource = boto3_resource("dynamodb", region_name=REGION_NAME)
 

recordprocessor/src/clients.py

@@ -2,7 +2,6 @@
 
 from typing import List, Callable, Dict
 from utils_for_fhir_conversion import _is_not_empty, Generate, Add, Convert
-from mappings import map_target_disease, Vaccine
 from constants import Urls
 
 
@@ -196,13 +195,14 @@ def _decorate_performer(imms: dict, row: Dict[str, str]) -> None:
 ]
 
 
-def convert_to_fhir_imms_resource(row: dict, vaccine: Vaccine) -> dict:
+def convert_to_fhir_imms_resource(row: dict, target_disease: list) -> 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 = {"resourceType": "Immunization", "status": "completed"}
-
-    # Add the targetDisease element based on the vaccine type
-    imms_resource["protocolApplied"] = [{"targetDisease": map_target_disease(vaccine)}]
+    imms_resource = {
+        "resourceType": "Immunization",
+        "status": "completed",
+        "protocolApplied": [{"targetDisease": target_disease}]
+    }
 
     # Apply all decorators to add the relevant fields to the imms_resource
     for decorator in all_decorators:

recordprocessor/src/convert_to_fhir_imms_resource.py

@@ -6,7 +6,6 @@
 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 mappings import Vaccine
 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
@@ -75,9 +74,7 @@ def file_level_validation(incoming_message_body: dict) -> dict:
     """
     try:
         message_id = incoming_message_body.get("message_id")
-        vaccine: Vaccine = next(  # Convert vaccine_type to Vaccine enum
-            vaccine for vaccine in Vaccine if vaccine.value == incoming_message_body.get("vaccine_type").upper()
-        )
+        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")
@@ -89,7 +86,7 @@ def file_level_validation(incoming_message_body: dict) -> dict:
         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.value, permission, csv_data)
+        allowed_operations_set = validate_action_flag_permissions(supplier, vaccine, permission, csv_data)
 
         make_and_upload_ack_file(message_id, file_key, True, True, created_at_formatted_string)
 

recordprocessor/src/file_level_validation.py

@@ -1,70 +1,20 @@
 """Mappings for converting vaccine type into target disease FHIR element"""
-
-from enum import Enum
-from typing import Dict, List
+import json
 from constants import Urls
+from clients import redis_client
 
 
-class Vaccine(Enum):
-    """Disease Codes"""
-
-    COVID_19: str = "COVID19"
-    FLU: str = "FLU"
-    MMR: str = "MMR"
-    RSV: str = "RSV"
-
-
-class Disease(Enum):
-    """Disease Codes"""
-
-    COVID_19: str = "COVID19"
-    FLU: str = "FLU"
-    MEASLES: str = "MEASLES"
-    MUMPS: str = "MUMPS"
-    RUBELLA: str = "RUBELLA"
-    RSV: str = "RSV"
-
-
-class DiseaseCode(Enum):
-    """Disease Codes"""
-
-    COVID_19: str = "840539006"
-    FLU: str = "6142004"
-    MEASLES: str = "14189004"
-    MUMPS: str = "36989005"
-    RUBELLA: str = "36653000"
-    RSV: str = "55735004"
-
-
-class DiseaseDisplayTerm(Enum):
-    """Disease display terms which correspond to disease codes"""
-
-    COVID_19: str = "Disease caused by severe acute respiratory syndrome coronavirus 2"
-    FLU: str = "Influenza"
-    MEASLES: str = "Measles"
-    MUMPS: str = "Mumps"
-    RUBELLA: str = "Rubella"
-    RSV: str = "Respiratory syncytial virus infection (disorder)"
-
-
-VACCINE_DISEASE_MAPPING: Dict[Vaccine, List[Disease]] = {
-    Vaccine.COVID_19: [Disease.COVID_19],
-    Vaccine.FLU: [Disease.FLU],
-    Vaccine.MMR: [Disease.MEASLES, Disease.MUMPS, Disease.RUBELLA],
-    Vaccine.RSV: [Disease.RSV],
-}
-
-
-def map_target_disease(vaccine: Vaccine) -> list:
+def map_target_disease(vaccine: str) -> list:
     """Returns the target disease element for the given vaccine type using the vaccine_disease_mapping"""
-    diseases = VACCINE_DISEASE_MAPPING.get(vaccine, [])
+    diseases_str = redis_client.hget("vacc_to_diseases", vaccine)
+    diseases = json.loads(diseases_str) if diseases_str else []
     return [
         {
             "coding": [
                 {
                     "system": Urls.SNOMED,
-                    "code": DiseaseCode[disease.name].value,
-                    "display": DiseaseDisplayTerm[disease.name].value,
+                    "code": disease["code"],
+                    "display": disease["term"],
                 }
             ]
         }

recordprocessor/src/mappings.py

@@ -2,12 +2,11 @@
 
 from convert_to_fhir_imms_resource import convert_to_fhir_imms_resource
 from constants import Diagnostics
-from mappings import Vaccine
 from clients import logger
 from utils_for_recordprocessor import create_diagnostics_dictionary
 
 
-def process_row(vaccine: Vaccine, allowed_operations: set, row: dict) -> dict:
+def process_row(target_disease: list, allowed_operations: set, row: dict) -> dict:
     """
     Processes a row of the file and returns a dictionary containing the fhir_json, action_flag, imms_id, local_id
     (where applicable), version(where applicable) and any diagnostics.
@@ -51,7 +50,7 @@ def process_row(vaccine: Vaccine, allowed_operations: set, row: dict) -> dict:
 
     # Handle success
     return {
-        "fhir_json": convert_to_fhir_imms_resource(row, vaccine),
+        "fhir_json": convert_to_fhir_imms_resource(row, target_disease),
         "operation_requested": operation_requested,
         "local_id": local_id,
     }

recordprocessor/src/process_row.py

@@ -1,18 +1,15 @@
 """Tests for convert_to_fhir_imms_resource"""
-
 import unittest
 from unittest.mock import patch
 
 from tests.utils_for_recordprocessor_tests.values_for_recordprocessor_tests import (
     MockFhirImmsResources,
     MockFieldDictionaries,
+    TargetDiseaseElements
 )
 from tests.utils_for_recordprocessor_tests.mock_environment_variables import MOCK_ENVIRONMENT_DICT
 
 with patch("os.environ", MOCK_ENVIRONMENT_DICT):
-    # Do not attempt 'from src.mappings import Vaccine' as this imports a different instance of Vaccine
-    # and tests will break
-    from mappings import Vaccine
     from convert_to_fhir_imms_resource import convert_to_fhir_imms_resource
 
 
@@ -24,6 +21,7 @@ def test_convert_to_fhir_imms_resource(self):
         Test that convert_to_fhir_imms_resource gives the expected output. These tests check that the entire
         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),
@@ -41,4 +39,5 @@ def test_convert_to_fhir_imms_resource(self):
 
         for test_name, input_values, expected_output in cases:
             with self.subTest(test_name):
-                self.assertEqual(convert_to_fhir_imms_resource(input_values, Vaccine.RSV), expected_output)
+                output = convert_to_fhir_imms_resource(input_values, TargetDiseaseElements.RSV)
+                self.assertEqual(output, expected_output)

recordprocessor/tests/test_convert_to_fhir_imms_resource.py

@@ -1,37 +1,46 @@
 """Tests for map_target_disease"""
-
+import json
 import unittest
 from unittest.mock import patch
-from typing import List
-from tests.utils_for_recordprocessor_tests.values_for_recordprocessor_tests import TargetDiseaseElements
 from tests.utils_for_recordprocessor_tests.mock_environment_variables import MOCK_ENVIRONMENT_DICT
 
 with patch("os.environ", MOCK_ENVIRONMENT_DICT):
-    from mappings import map_target_disease, Vaccine
+    from mappings import map_target_disease
 
 
+@patch("mappings.redis_client")
 class TestMapTargetDisease(unittest.TestCase):
     """
     Test that map_target_disease returns the correct target disease element for valid vaccine types, or [] for
     invalid vaccine types.
     """
 
-    def test_map_target_disease_valid(self):
+    def test_map_target_disease_valid(self, mock_redis_client):
         """Tests map_target_disease returns the disease coding information when using valid vaccine types"""
-        # NOTE: TEST CASES SHOULD INCLUDE ALL VACCINE TYPES WHICH ARE VALID FOR THIS PRODUCT.
-        # A NEW VACCINE TYPE SHOULD BE ADDED TO THIS TEST EVERY TIME THERE IS A VACCINE TYPE UPLIFT
-        # (note that this will require adding the vaccine type to the TargetDiseaseElements class).
-        # Target disease elements are intentionally hardcoded as a way of ensuring that the correct code and display
-        # values are being used, and that the element is being built up correctly.
-        vaccines: List[Vaccine] = [Vaccine.RSV, Vaccine.COVID_19, Vaccine.FLU, Vaccine.MMR]
-        for vaccine in vaccines:
-            with self.subTest(vaccine=vaccine):
-                self.assertEqual(map_target_disease(vaccine), getattr(TargetDiseaseElements, vaccine.value))
-
-    def test_map_target_disease_invalid(self):
+        mock_redis_client.hget.return_value = json.dumps([{
+            "code": "55735004",
+            "term": "Respiratory syncytial virus infection (disorder)"
+        }])
+
+        self.assertEqual(map_target_disease("RSV"), [{
+            "coding": [{
+                "system": "http://snomed.info/sct",
+                "code": "55735004",
+                "display": "Respiratory syncytial virus infection (disorder)"
+            }]
+        }])
+
+        mock_redis_client.hget.assert_called_with("vacc_to_diseases", "RSV")
+
+    def test_map_target_disease_invalid(self, mock_redis_client):
         """Tests map_target_disease does not return the disease coding information when using invalid vaccine types."""
+
+        mock_redis_client.hget.return_value = None
+
         self.assertEqual(map_target_disease("invalid_vaccine"), [])
 
+        mock_redis_client.hget.assert_called_with("vacc_to_diseases", "invalid_vaccine")
+
 
 if __name__ == "__main__":
     unittest.main()