Use Map Cache

Author: nhsdevws

recordprocessor/src/clients.py

@@ -1,10 +1,13 @@
 """Initialise s3 and kinesis clients"""
 
 import logging
+import os
+
 from boto3 import client as boto3_client, resource as boto3_resource
 from botocore.config import Config
+from redis_cacher import RedisCacher
 
-REGION_NAME = "eu-west-2"
+REGION_NAME = os.getenv("AWS_REGION")
 
 s3_client = boto3_client("s3", region_name=REGION_NAME)
 kinesis_client = boto3_client(
@@ -21,3 +24,7 @@
 logging.basicConfig(level="INFO")
 logger = logging.getLogger()
 logger.setLevel("INFO")
+
+REDIS_HOST = os.getenv("REDIS_HOST")
+REDIS_PORT = os.getenv("REDIS_PORT")
+redis_cacher = RedisCacher(REDIS_HOST, REDIS_PORT, logger)

recordprocessor/src/constants.py

@@ -84,3 +84,8 @@ class Urls:
     NHS_NUMBER = "https://fhir.nhs.uk/Id/nhs-number"
     NULL_FLAVOUR_CODES = "http://terminology.hl7.org/CodeSystem/v3-NullFlavor"
     VACCINATION_PROCEDURE = "https://fhir.hl7.org.uk/StructureDefinition/Extension-UKCore-VaccinationProcedure"
+
+
+class RedisCacheKeys:
+    PERMISSIONS_CONFIG_FILE_KEY = "permissions_config.json"
+    DISEASE_MAPPING_FILE_KEY = "disease_mapping.json"

recordprocessor/src/mappings.py

@@ -2,7 +2,8 @@
 
 from enum import Enum
 from typing import Dict, List
-from constants import Urls
+from constants import Urls, RedisCacheKeys
+from redis_cacher import redis_cacher
 
 
 class Vaccine(Enum):
@@ -54,6 +55,8 @@ class DiseaseDisplayTerm(Enum):
     Vaccine.RSV: [Disease.RSV],
 }
 
+VACCINE_DISEASE_MAPPING = redis_cacher.get(RedisCacheKeys.DISEASE_MAPPING_FILE_KEY)
+
 
 def map_target_disease(vaccine: Vaccine) -> list:
     """Returns the target disease element for the given vaccine type using the vaccine_disease_mapping"""

recordprocessor/src/redis_cacher.py

@@ -0,0 +1,28 @@
+import json
+import redis
+
+
+class RedisCacher():
+    """ RedisCacher abstraction class to decouple application code
+    from direct use of Redis client.
+    Also centralised error handling & extensibility.
+    """
+
+    def __init__(self, redis_host, redis_port, logger):
+        try:
+            # Attempt to connect to Redis
+            self.redis_client = redis.StrictRedis(redis_host, redis_port, decode_responses=True)
+            # Check the connection with a PING command
+            if self.redis_client.ping():
+                logger.info("Successfully connected to Redis.")
+            else:
+                logger.error("Failed to connect to Redis.")
+        except Exception as e:
+            logger.exception(f"Connection to Redis failed: {e}")
+
+    def get(self, key: str) -> dict:
+        """Gets the value from Redis cache for the given key."""
+        value = self.redis_client.get(key)
+        if value is not None:
+            return json.loads(value)
+        return {}

redis_sync/src/event_processor.py

@@ -1,6 +1,10 @@
 from clients import logger
 from s3_event import S3Event
 from record_processor import record_processor
+'''
+    Event Processor
+    The Business Logic for the Redis Sync Lambda Function.
+    This module processes S3 events and iterates through each record to process them individually.'''
 
 
 def event_processor(event, context):