implementation

Author: nhsdevws

lambdas/redis_sync/src/record_processor.py

@@ -1,14 +1,15 @@
 from redis_cacher import RedisCacher
 from common.clients import logger
 from common.s3_event import S3EventRecord
+from common.service_return import ServiceReturn
 '''
     Record Processor
     This module processes individual S3 records from an event.
     It is used to upload data to Redis ElastiCache.
 '''
 
 
-def process_record(record: S3EventRecord) -> dict:
+def process_record(record: S3EventRecord) -> ServiceReturn:
     try:
         logger.info("Processing S3 r bucket: %s, key: %s",
                     record.get_bucket_name(), record.get_object_key())
@@ -20,17 +21,21 @@ def process_record(record: S3EventRecord) -> dict:
         }
 
         try:
-            result = RedisCacher.upload(bucket_name, file_key)
-            result.update(base_log_data)
-            return result
+            service_result = RedisCacher.upload(bucket_name, file_key)
+            if service_result.is_success:
+                result = service_result.value
+                result.update(base_log_data)
+                return ServiceReturn(value=result)
+            else:
+                return ServiceReturn(
+                    status=500,
+                    message=f"Failed to upload to cache for filename '{file_key}': {service_result.message}")
 
         except Exception as error:  # pylint: disable=broad-except
             logger.exception("Error uploading to cache for filename '%s'", file_key)
             error_data = {"status": "error", "message": str(error)}
             error_data.update(base_log_data)
-            return error_data
+            return ServiceReturn(value=error_data)
 
     except Exception:  # pylint: disable=broad-except
-        msg = "Error processing record"
-        logger.exception(msg)
-        return {"status": "error", "message": msg}
+        return ServiceReturn(value={"status": "error", "message": "Error processing record"})

lambdas/redis_sync/src/redis_cacher.py

@@ -12,7 +12,7 @@ class RedisCacher:
     """Class to handle interactions with ElastiCache (Redis) for configuration files."""
 
     @staticmethod
-    def upload(bucket_name: str, file_key: str) -> dict:
+    def upload(bucket_name: str, file_key: str) -> ServiceReturn:
         try:
             logger.info("Upload from s3 to Redis cache. file '%s'. bucket '%s'", file_key, bucket_name)
 
@@ -46,8 +46,8 @@ def upload(bucket_name: str, file_key: str) -> dict:
                     redis_client.hdel(key, *fields_to_delete)
                     logger.info("Deleted mapping fields for %s: %s", key, fields_to_delete)
 
-            return {"status": "success", "message": f"File {file_key} uploaded to Redis cache."}
+            return ServiceReturn(value={"status": "success", "message": f"File {file_key} uploaded to Redis cache."})
         except Exception:
             msg = f"Error uploading file '{file_key}' to Redis cache"
             logger.exception(msg)
-            return {"status": "error", "message": msg}
+            return ServiceReturn(status=500, message=msg)

lambdas/redis_sync/src/redis_sync.py

@@ -4,6 +4,7 @@
 from common.log_decorator import logging_decorator
 from common.redis_client import get_redis_client
 from common.s3_event import S3Event
+from common.service_return import ServiceReturn
 '''
     Event Processor
     The Business Logic for the Redis Sync Lambda Function.
@@ -15,18 +16,20 @@ def _process_all_records(s3_records: list) -> dict:
     error_count = 0
     file_keys = []
     for record in s3_records:
-        record_result = process_record(record)
-        file_keys.append(record_result["file_key"])
-        if record_result["status"] == "error":
+        service_result = process_record(record)
+        file_keys.append(service_result.value.get("file_key"))
+        if service_result.status == 500:
             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",
-                "file_keys": file_keys}
+        return ServiceReturn(value={"status": "error", "message": f"Processed {record_count} records with {error_count} errors",
+                                    "file_keys": file_keys})
     else:
         logger.info("Successfully processed all %d records", record_count)
-        return {"status": "success", "message": f"Successfully processed {record_count} records",
-                "file_keys": file_keys}
+        return ServiceReturn(
+            value={"status": "success",
+                   "message": f"Successfully processed {record_count} records",
+                   "file_keys": file_keys})
 
 
 @logging_decorator(prefix="redis_sync", stream_name=STREAM_NAME)
@@ -44,7 +47,12 @@ def handler(event, _):
                 logger.info(no_records)
                 return {"status": "success", "message": no_records}
             else:
-                return _process_all_records(s3_records)
+                service_result = _process_all_records(s3_records)
+                if service_result.is_success:
+                    return service_result.value
+                else:
+                    return {"status": "error", "message": service_result.value.get("message"),
+                            "file_keys": service_result.value.get("file_keys", [])}
         else:
             logger.info(no_records)
             return {"status": "success", "message": no_records}

lambdas/shared/src/common/s3_reader.py

@@ -12,7 +12,7 @@ class S3Reader:
     """
 
     @staticmethod
-    def read(bucket_name, file_key):
+    def read(bucket_name, file_key) -> ServiceReturn:
         try:
             s3_file = s3_client.get_object(Bucket=bucket_name, Key=file_key)
             return ServiceReturn(value=s3_file["Body"].read().decode("utf-8"))