decorator

Author: nhsdevws

recordprocessor/src/clients.py

@@ -1,10 +1,12 @@
 """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 +23,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):
@@ -70,3 +71,6 @@ def map_target_disease(vaccine: Vaccine) -> list:
         }
         for disease in diseases
     ]
+
+
+VACCINE_DISEASE_MAPPING = redis_cacher.get(RedisCacheKeys.DISEASE_MAPPING_FILE_KEY)

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 {}

recordprocessor/tests/test_redis_cacher.py

@@ -0,0 +1,61 @@
+import unittest
+import json
+from unittest.mock import patch
+from redis_cacher import RedisCacher
+from constants import RedisCacheKeys
+
+
+class TestRedisCacher(unittest.TestCase):
+
+    def setUp(self):
+        # mock s3_reader and transform_map
+        self.s3_reader_patcher = patch("redis_cacher.S3Reader")
+        self.mock_s3_reader = self.s3_reader_patcher.start()
+        self.transform_map_patcher = patch("redis_cacher.transform_map")
+        self.mock_transform_map = self.transform_map_patcher.start()
+        self.redis_client_patcher = patch("redis_cacher.redis_client")
+        self.mock_redis_client = self.redis_client_patcher.start()
+
+    def tearDown(self):
+        self.s3_reader_patcher.stop()
+        self.transform_map_patcher.stop()
+        self.redis_client_patcher.stop()
+
+    def test_upload(self):
+        mock_data = '{"a": "b"}'
+        mock_transformed_data = '{"b": "c"}'
+        self.mock_s3_reader.read = unittest.mock.Mock()
+        self.mock_s3_reader.read.return_value = mock_data
+        self.mock_transform_map.return_value = mock_transformed_data
+
+        bucket_name = "bucket"
+        file_key = "file-key"
+        result = RedisCacher.upload(bucket_name, file_key)
+
+        self.mock_s3_reader.read.assert_called_once_with(bucket_name, file_key)
+        self.mock_transform_map.assert_called_once_with(mock_data, file_key)
+        self.mock_redis_client.set.assert_called_once_with(file_key, mock_transformed_data)
+        self.assertTrue(result)
+
+    def test_get_cached_config_json(self):
+        """Test getting cached config JSON from Redis."""
+        cache_key = "some-key"
+        cached_data = '{"some-data-key": "some-data-value"}'
+        self.mock_redis_client.get.return_value = cached_data
+        result = RedisCacher.get_cached_config_json(cache_key)
+        self.assertEqual(result, json.loads(cached_data))
+        self.mock_redis_client.get.assert_called_once_with(cache_key)
+
+    def test_get_cached_permissions_config_json(self):
+        cached_permissions = '{"permissions": "perm_config"}'
+        self.mock_redis_client.get.return_value = cached_permissions
+        result = RedisCacher.get_cached_permissions_config_json()
+        self.assertEqual(result, json.loads(cached_permissions))
+        self.mock_redis_client.get.assert_called_once_with(RedisCacheKeys.PERMISSIONS_CONFIG_FILE_KEY)
+
+    def test_get_cached_disease_mapping_json(self):
+        cached_disease_mapping = '{"disease": "disease-mapping"}'
+        self.mock_redis_client.get.return_value = cached_disease_mapping
+        result = RedisCacher.get_cached_disease_mapping_json()
+        self.assertEqual(result, json.loads(cached_disease_mapping))
+        self.mock_redis_client.get.assert_called_once_with(RedisCacheKeys.DISEASE_MAPPING_FILE_KEY)

redis_sync/src/clients.py

@@ -13,5 +13,6 @@
 REDIS_PORT = os.getenv("REDIS_PORT", 6379)
 
 s3_client = boto3_client("s3", region_name=REGION_NAME)
+firehose_client = boto3_client("firehose", region_name=REGION_NAME)
 logger.info(f"Connecting to Redis at {REDIS_HOST}:{REDIS_PORT}")
 redis_client = redis.StrictRedis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=True)

redis_sync/src/event_processor.py

@@ -10,13 +10,21 @@
 def event_processor(event, context):
 
     try:
-
         logger.info("Processing S3 event with %d records", len(event.get('Records', [])))
         s3Event = S3Event(event)
+        record_count = len(s3Event.get_s3_records())
+        error_count = 0
         for record in s3Event.get_s3_records():
-            record_processor(record)
-        return True
+            record_result = record_processor(record)
+            if record_result["status"] == "error":
+                error_count += 1
+        if error_count > 0:
+            logger.error("Processed %d records with %d errors", record_count, error_count)
+            return {"status": "error", "message": f"Processed {record_count} records with {error_count} errors"}
+        else:
+            logger.info("Successfully processed all %d records", record_count)
+            return {"status": "success", "message": f"Successfully processed {record_count} records"}
 
     except Exception:
         logger.exception("Error processing S3 event")
-        return False
+        return {"status": "error", "message": "Error processing S3 event"}

redis_sync/src/logging_decorator.py

@@ -0,0 +1,58 @@
+"""This module contains the logging decorator for sending the appropriate logs to Cloudwatch and Firehose."""
+
+import json
+import os
+import time
+from datetime import datetime
+from functools import wraps
+from clients import firehose_client, logger
+
+STREAM_NAME = os.getenv("SPLUNK_FIREHOSE_NAME", "immunisation-fhir-api-internal-dev-splunk-firehose")
+
+
+def send_log_to_firehose(log_data: dict) -> None:
+    """Sends the log_message to Firehose"""
+    try:
+        record = {"Data": json.dumps({"event": log_data}).encode("utf-8")}
+        response = firehose_client.put_record(DeliveryStreamName=STREAM_NAME, Record=record)
+        logger.info("Log sent to Firehose: %s", response)  # TODO: Should we be logging full response?
+    except Exception as error:  # pylint:disable = broad-exception-caught
+        logger.exception("Error sending log to Firehose: %s", error)
+
+
+def generate_and_send_logs(
+    start_time, base_log_data: dict, additional_log_data: dict, is_error_log: bool = False
+) -> None:
+    """Generates log data which includes the base_log_data, additional_log_data, and time taken (calculated using the
+    current time and given start_time) and sends them to Cloudwatch and Firehose."""
+    log_data = {**base_log_data, "time_taken": f"{round(time.time() - start_time, 5)}s", **additional_log_data}
+    log_function = logger.error if is_error_log else logger.info
+    log_function(json.dumps(log_data))
+    send_log_to_firehose(log_data)
+
+
+def logging_decorator(func):
+    """
+    Sends the appropriate logs to Cloudwatch and Firehose based on the function result.
+    NOTE: The function must return a dictionary as its only return value. The dictionary is expected to contain
+    all of the required additional details for logging.
+    NOTE: Logs will include the result of the function call or, in the case of an Exception being raised,
+    a status code of 500 and the error message.
+    """
+
+    @wraps(func)
+    def wrapper(*args, **kwargs):
+        base_log_data = {"function_name": f"filename_processor_{func.__name__}", "date_time": str(datetime.now())}
+        start_time = time.time()
+
+        try:
+            result = func(*args, **kwargs)
+            generate_and_send_logs(start_time, base_log_data, additional_log_data=result)
+            return result
+
+        except Exception as e:
+            additional_log_data = {"statusCode": 500, "error": str(e)}
+            generate_and_send_logs(start_time, base_log_data, additional_log_data, is_error_log=True)
+            raise
+
+    return wrapper

redis_sync/src/record_processor.py

@@ -1,13 +1,18 @@
 from clients import logger
 from s3_event import S3EventRecord
 from redis_cacher import RedisCacher
+from logging_decorator import logging_decorator
 '''
     Record Processor
     This module processes individual S3 records from an event.
     It is used to upload data to Redis ElastiCache.
 '''
 
 
+# NOTE: logging_decorator is applied to handle_record function, rather than lambda_handler, because
+# the logging_decorator is for an individual record, whereas the lambda_handler could potentially be handling
+# multiple records.
+@logging_decorator
 def record_processor(record: S3EventRecord) -> bool:
 
     try:
@@ -21,9 +26,9 @@ def record_processor(record: S3EventRecord) -> bool:
             return RedisCacher.upload(bucket_name, file_key)
 
         except Exception as error:  # pylint: disable=broad-except
-            logger.error("Error uploading to cache for file '%s': %s", file_key, error)
-            return False
+            logger.exception("Error uploading to cache for file '%s'", file_key)
+            return {"status": "error", "message": str(error)}
 
     except Exception:  # pylint: disable=broad-except
         logger.exception("Error obtaining file_key")
-        return False
+        return {"status": "error", "message": "Error obtaining file_key"}

redis_sync/src/redis_cacher.py

@@ -20,7 +20,7 @@ def upload(bucket_name: str, file_key: str) -> None:
 
         # Use the file_key as the Redis key and file content as the value
         redis_client.set(file_key, trx_data)
-        return True
+        return {"status": "success", "message": f"File {file_key} uploaded to Redis cache."}
 
     @staticmethod
     def get_cached_config_json(file_type) -> dict: