21 fail

Author: nhsdevws

delta_backend/src/delta.py

@@ -7,9 +7,10 @@
 import logging
 from botocore.exceptions import ClientError
 from log_firehose import FirehoseLogger
-from Converter import Converter
-from helpers.delta_data import DeltaData
-from helpers.mappings import OperationName, EventName
+from helpers.db_processor import DbProcessor
+from helpers.mappings import OperationName, EventName, ActionFlag
+from helpers.record_processor import RecordProcessor
+from helpers.sqs_utils import send_message
 
 failure_queue_url = os.environ["AWS_SQS_QUEUE_URL"]
 delta_table_name = os.environ["DELTA_TABLE_NAME"]
@@ -18,22 +19,24 @@
 logger = logging.getLogger()
 logger.setLevel("INFO")
 firehose_logger = FirehoseLogger()
-delta_data = DeltaData(delta_table_name, delta_source)
 
 
-def send_message(record):
-    # Create a message
-    message_body = record
-    # Use boto3 to interact with SQS
-    sqs_client = boto3.client("sqs")
-    try:
-        # Send the record to the queue
-        print(f"Sending record to DLQ: {message_body}")
-        sqs_client.send_message(QueueUrl=failure_queue_url, MessageBody=json.dumps(message_body))
-        logger.info("Record saved successfully to the DLQ")
-    except ClientError as e:
-        logger.error(f"Error sending record to DLQ: {e}")
+# def send_message(record):
+#     # Create a message
+#     message_body = record
+#     # Use boto3 to interact with SQS
+#     sqs_client = boto3.client("sqs")
+#     try:
+#         # Send the record to the queue
+#         print(f"Sending record to DLQ: {message_body}")
+#         sqs_client.send_message(QueueUrl=failure_queue_url, MessageBody=json.dumps(message_body))
+#         logger.info("Record saved successfully to the DLQ")
+#     except ClientError as e:
+#         logger.error(f"Error sending record to DLQ: {e}")
 
+def get_vaccine_type(patientsk) -> str:
+    parsed = [str.strip(str.lower(s)) for s in patientsk.split("#")]
+    return parsed[0]
 
 def handler(event, context):
     logger.info("Starting Delta Handler")
@@ -45,81 +48,17 @@ def handler(event, context):
     try:
         dynamodb = boto3.resource("dynamodb", region_name="eu-west-2")
         delta_table = dynamodb.Table(delta_table_name)
-
-        # Converting ApproximateCreationDateTime directly to string will give Unix timestamp
-        # I am converting it to isofformat for filtering purpose. This can be changed accordingly
-
+        db_processor = DbProcessor(delta_table, delta_source, logger)
+        record_processor = RecordProcessor(delta_table,
+                                           delta_source,
+                                           log_data,
+                                           db_processor,
+                                           firehose_logger,
+                                           firehose_log,
+                                           logger)
         for record in event["Records"]:
-            start = time.time()
-            log_data["date_time"] = str(datetime.now())
-            intrusion_check = False
-            approximate_creation_time = datetime.utcfromtimestamp(record["dynamodb"]["ApproximateCreationDateTime"])
-            expiry_time = approximate_creation_time + timedelta(days=30)
-            expiry_time_epoch = int(expiry_time.timestamp())
-            error_records = []
-            response = str()
-            imms_id = str()
-            operation = str()
-            if record["eventName"] != EventName.DELETE_PHYSICAL:
-                new_image = record["dynamodb"]["NewImage"]
-                imms_id = new_image["PK"]["S"].split("#")[1]
-                supplier_system = new_image["SupplierSystem"]["S"]
-                if supplier_system not in ("DPSFULL", "DPSREDUCED"):
-                    response, error_records = delta_data.write_to_db(new_image, imms_id, approximate_creation_time, expiry_time_epoch)
-                else:
-                    operation_outcome["statusCode"] = "200"
-                    operation_outcome["statusDesc"] = "Record from DPS skipped"
-                    log_data["operation_outcome"] = operation_outcome
-                    firehose_log["event"] = log_data
-                    firehose_logger.send_log(firehose_log)
-                    logger.info(f"Record from DPS skipped for {imms_id}")
-                    return {"statusCode": 200, "body": f"Record from DPS skipped for {imms_id}"}
-            else:
-                operation = OperationName.DELETE_PHYSICAL
-                new_image = record["dynamodb"]["Keys"]
-                logger.info(f"Record to delta:{new_image}")
-                imms_id = new_image["PK"]["S"].split("#")[1]
-                response = delta_table.put_item(
-                    Item={
-                        "PK": str(uuid.uuid4()),
-                        "ImmsID": imms_id,
-                        "Operation": OperationName.DELETE_PHYSICAL,
-                        "VaccineType": "default",
-                        "SupplierSystem": "default",
-                        "DateTimeStamp": approximate_creation_time.isoformat(),
-                        "Source": delta_source,
-                        "Imms": "",
-                        "ExpiresAt": expiry_time_epoch,
-                    }
-                )
-            end = time.time()
-            log_data["time_taken"] = f"{round(end - start, 5)}s"
-            operation_outcome = {"record": imms_id, "operation_type": operation}
-            if response["ResponseMetadata"]["HTTPStatusCode"] == 200:
-                if error_records:
-                    log = f"Partial success: successfully synced into delta, but issues found within record {imms_id}"
-                    operation_outcome["statusCode"] = "207"
-                    operation_outcome["statusDesc"] = (
-                        f"Partial success: successfully synced into delta, but issues found within record {json.dumps(error_records)}"
-                    )
-                else:
-                    log = f"Record Successfully created for {imms_id}"
-                    operation_outcome["statusCode"] = "200"
-                    operation_outcome["statusDesc"] = "Successfully synched into delta"
-                log_data["operation_outcome"] = operation_outcome
-                firehose_log["event"] = log_data
-                firehose_logger.send_log(firehose_log)
-                logger.info(log)
-                return {"statusCode": 200, "body": "Records processed successfully"}
-            else:
-                log = f"Record NOT created for {imms_id}"
-                operation_outcome["statusCode"] = "500"
-                operation_outcome["statusDesc"] = "Exception"
-                log_data["operation_outcome"] = operation_outcome
-                firehose_log["event"] = log_data
-                firehose_logger.send_log(firehose_log)
-                logger.info(log)
-                return {"statusCode": 500, "body": "Records not processed successfully"}
+            record_processor.process_record(record)
+
 
     except Exception as e:
         operation_outcome["statusCode"] = "500"
@@ -138,3 +77,4 @@ def handler(event, context):
             "statusCode": 500,
             "body": "Records not processed",
         }
+

delta_backend/src/helpers/db_processor.py

@@ -0,0 +1,62 @@
+import json
+import uuid
+from Converter import Converter
+
+
+class DbProcessor:
+    def __init__(self, delta_table, delta_source, logger):
+        self.delta_table = delta_table
+        self.delta_source = delta_source
+        self.logger = logger
+
+    def get_data(self):
+        return self.data     
+    
+    def remove(self, imms_id, operation,
+                    approximate_creation_time, expiry_time_epoch):
+            response = self.delta_table.put_item(
+                Item={
+                    "PK": str(uuid.uuid4()),
+                    "ImmsID": imms_id,
+                    "Operation": operation,
+                    "VaccineType": "default",
+                    "SupplierSystem": "default",
+                    "DateTimeStamp": approximate_creation_time.isoformat(),
+                    "Source": self.delta_source,
+                    "Imms": "",
+                    "ExpiresAt": expiry_time_epoch,
+                }
+            )
+            return response
+        
+    def write(self, new_image, imms_id, operation, vaccine_type, supplier_system,
+                    approximate_creation_time, expiry_time_epoch):
+        try:
+            # vaccine_type = self.get_vaccine_type(new_image["PatientSK"]["S"])
+            # supplier_system = new_image["SupplierSystem"]["S"]
+            # if supplier_system not in ("DPSFULL", "DPSREDUCED"):
+            #     operation = new_image["Operation"]["S"]
+            #     if operation == OperationName.CREATE:
+            #         operation = ActionFlag.CREATE
+            resource_json = json.loads(new_image["Resource"]["S"])
+            FHIRConverter = Converter(json.dumps(resource_json))
+            flat_json = FHIRConverter.runConversion(resource_json)  # Get the flat JSON
+            error_records = FHIRConverter.getErrorRecords()
+            flat_json[0]["ACTION_FLAG"] = operation
+            response = self.delta_table.put_item(
+                Item={
+                    "PK": str(uuid.uuid4()),
+                    "ImmsID": imms_id,
+                    "Operation": operation,
+                    "VaccineType": vaccine_type,
+                    "SupplierSystem": supplier_system,
+                    "DateTimeStamp": approximate_creation_time.isoformat(),
+                    "Source": self.delta_source,
+                    "Imms": str(flat_json),
+                    "ExpiresAt": expiry_time_epoch,
+                }
+            )
+            return response, operation, error_records
+        except Exception as e:
+            print(f"Error writing to DB: {e}")
+            return None

delta_backend/src/helpers/delta_data.py

@@ -16,7 +16,11 @@ class OperationName:
     DELETE_LOGICAL = "DELETE"
     DELETE_PHYSICAL = "REMOVE"
 
-ActionFlag = OperationName
+class ActionFlag:
+    CREATE = "CREATE"
+    UPDATE = "UPDATE"
+    DELETE_LOGICAL = "DELETE"
+    DELETE_PHYSICAL = "REMOVE"
 
 class VaccineTypes:
     """Vaccine types"""

delta_backend/src/helpers/mappings.py

@@ -0,0 +1,87 @@
+import unittest
+from helpers.record_processor import RecordProcessor
+
+
+class TestGetOpOutcome(unittest.TestCase):
+    def test_get_op_outcome_basic(self):
+        # Arrange
+        status_code = 200
+        status_desc = "Operation successful"
+
+        # Act
+        outcome = RecordProcessor.get_op_outcome(status_code, status_desc)
+
+        # Assert
+        self.assertEqual(outcome["statusCode"], "200")
+        self.assertEqual(outcome["statusDesc"], "Operation successful")
+        self.assertNotIn("diagnostics", outcome)
+        self.assertNotIn("record", outcome)
+        self.assertNotIn("operation_type", outcome)
+
+    def test_get_op_outcome_with_diagnostics(self):
+        # Arrange
+        status_code = 500
+        status_desc = "Operation failed"
+        diagnostics = "An error occurred during processing"
+
+        # Act
+        outcome = RecordProcessor.get_op_outcome(status_code, status_desc, diagnostics=diagnostics)
+
+        # Assert
+        self.assertEqual(outcome["statusCode"], "500")
+        self.assertEqual(outcome["statusDesc"], "Operation failed")
+        self.assertEqual(outcome["diagnostics"], "An error occurred during processing")
+        self.assertNotIn("record", outcome)
+        self.assertNotIn("operation_type", outcome)
+
+    def test_get_op_outcome_with_record(self):
+        # Arrange
+        status_code = 200
+        status_desc = "Record processed successfully"
+        record = "12345"
+
+        # Act
+        outcome = RecordProcessor.get_op_outcome(status_code, status_desc, record=record)
+
+        # Assert
+        self.assertEqual(outcome["statusCode"], "200")
+        self.assertEqual(outcome["statusDesc"], "Record processed successfully")
+        self.assertEqual(outcome["record"], "12345")
+        self.assertNotIn("diagnostics", outcome)
+        self.assertNotIn("operation_type", outcome)
+
+    def test_get_op_outcome_with_operation_type(self):
+        # Arrange
+        status_code = 200
+        status_desc = "Operation completed"
+        operation_type = "CREATE"
+
+        # Act
+        outcome = RecordProcessor.get_op_outcome(status_code, status_desc, operation_type=operation_type)
+
+        # Assert
+        self.assertEqual(outcome["statusCode"], "200")
+        self.assertEqual(outcome["statusDesc"], "Operation completed")
+        self.assertEqual(outcome["operation_type"], "CREATE")
+        self.assertNotIn("diagnostics", outcome)
+        self.assertNotIn("record", outcome)
+
+    def test_get_op_outcome_with_all_fields(self):
+        # Arrange
+        status_code = 207
+        status_desc = "Partial success"
+        diagnostics = "Some records failed"
+        record = "67890"
+        operation_type = "UPDATE"
+
+        # Act
+        outcome = RecordProcessor.get_op_outcome(
+            status_code, status_desc, diagnostics=diagnostics, record=record, operation_type=operation_type
+        )
+
+        # Assert
+        self.assertEqual(outcome["statusCode"], "207")
+        self.assertEqual(outcome["statusDesc"], "Partial success")
+        self.assertEqual(outcome["diagnostics"], "Some records failed")
+        self.assertEqual(outcome["record"], "67890")
+        self.assertEqual(outcome["operation_type"], "UPDATE")

delta_backend/src/helpers/processor_logging.py

@@ -0,0 +1,23 @@
+def get_op_outcome( status_code: int, status_desc: str, diagnostics: str = None, record: str = None, operation_type: str = None) -> dict:
+    """
+    Constructs the operation_outcome dictionary.
+
+    Args:
+        status_code (int): status code e.g., "200", "500"
+        status_desc (str): Operation's outcome
+        diagnostics (str, optional): Additional diagnostic information.
+        record (str, optional): record identifier (e.g., imms_id).
+        operation_type (str, optional): The type of operation performed.
+
+
+    Returns:
+        dict: The constructed operation_outcome dictionary.
+    """
+    outcome = { "statusCode": str(status_code), "statusDesc": status_desc}
+    if diagnostics:
+        outcome["diagnostics"] = diagnostics
+    if record:
+        outcome["record"] = record
+    if operation_type:
+        outcome["operation_type"] = operation_type
+    return outcome