wip

Author: nhsdevws

delta_backend/src/delta.py

@@ -15,6 +15,7 @@
 from log_firehose import FirehoseLogger
 
 failure_queue_url = os.environ["AWS_SQS_QUEUE_URL"]
+ieds_table_name = os.environ["IEDS_TABLE_NAME"]
 delta_table_name = os.environ["DELTA_TABLE_NAME"]
 delta_source = os.environ["SOURCE"]
 region_name = "eu-west-2"

lambdas/id_sync/src/clients.py

@@ -0,0 +1,5 @@
+from os_vars import get_ieds_table_name, get_delta_table_name
+
+
+ieds_table_name = get_ieds_table_name()
+delta_table_name = get_delta_table_name()

lambdas/id_sync/src/id_updater.py

@@ -0,0 +1,41 @@
+from boto3.dynamodb.conditions import Key
+from os_vars import get_ieds_table_name
+from common.clients import get_delta_table
+
+ieds_table = None
+
+
+def get_ieds_table():
+    """Get the IEDS table."""
+    global ieds_table
+    if ieds_table is None:
+        ieds_tablename = get_ieds_table_name()
+        ieds_table = get_delta_table(ieds_tablename)
+    return ieds_table
+
+
+def check_record_exist_in_IEDS(id: str) -> bool:
+    """Check if a record exists in the IEDS table for the given ID."""
+    search_patient_pk = f"Patient#{id}"
+
+    # Only fetch 1 record to check existence
+    response = get_ieds_table().query(
+        KeyConditionExpression=Key("PK").eq(search_patient_pk),
+        Limit=1
+    )
+    return response.get("Count", 0) > 0
+
+
+def update_patient_id_in_IEDS(old_id: str, new_id: str) -> dict:
+    """Update the patient ID in the IEDS table."""
+    # check if old_id and new_id are not empty
+    if not old_id or not new_id:
+        return {"status": "error", "message": "Old ID and New ID cannot be empty"}
+    else:
+        # update the table with new id
+        get_ieds_table().update_item(
+            Key={"PK": f"Patient#{old_id}"},
+            UpdateExpression="SET PK = :new_id",
+            ExpressionAttributeValues={":new_id": f"Patient#{new_id}"}
+        )
+        return {"status": "success", "message": f"Updated IEDS, patient ID: {old_id} to {new_id}"}

lambdas/id_sync/src/ieds_db_operations.py

@@ -0,0 +1,26 @@
+'''
+    nhs number Processor
+'''
+from common.clients import logger
+from pds_details import get_pds_patient_details
+from typing import Optional
+from ieds_db_operations import check_record_exist_in_IEDS, update_patient_id_in_IEDS
+
+
+def process_nhs_number(nhs_number: str) -> Optional[str]:
+    # get patient details from PDS
+    patient_details = get_pds_patient_details(nhs_number)
+    if not patient_details:
+        return {"status": "error", "message": f"No records returned for ID: {nhs_number}"}
+
+    patient_details_id = patient_details.get("id")
+
+    # if patient NHS != id, update patient index of vax events to new number
+    if patient_details_id != nhs_number:
+        if check_record_exist_in_IEDS(patient_details_id):
+            logger.info("Updating patient ID from %s to %s", id, patient_details_id)
+            return update_patient_id_in_IEDS(nhs_number, patient_details_id)
+        else:
+            return {"status": "error", "message": f"No records returned for ID: {nhs_number}"}
+    else:
+        return {"status": "success", "message": "No update required"}

lambdas/id_sync/src/nhs_number_processor.py

@@ -0,0 +1,48 @@
+# os_vars.py - Improved approach (✅ Better)
+import os
+from typing import Optional
+
+
+def get_ieds_table_name() -> str:
+    """Get the IEDS table name from environment variables."""
+    return os.environ["IEDS_TABLE_NAME"]
+
+
+def get_delta_table_name() -> str:
+    """Get the Delta table name from environment variables."""
+    return os.environ["DELTA_TABLE_NAME"]
+
+
+def get_pds_env() -> str:
+    """Get the PDS environment from environment variables."""
+    return os.getenv("PDS_ENV", "int")
+
+
+# Optional: Cached versions for performance
+_ieds_table_name: Optional[str] = None
+_delta_table_name: Optional[str] = None
+_pds_env: Optional[str] = None
+
+
+def get_ieds_table_name_cached() -> str:
+    """Get the IEDS table name (cached version)."""
+    global _ieds_table_name
+    if _ieds_table_name is None:
+        _ieds_table_name = os.environ["IEDS_TABLE_NAME"]
+    return _ieds_table_name
+
+
+def get_delta_table_name_cached() -> str:
+    """Get the Delta table name (cached version)."""
+    global _delta_table_name
+    if _delta_table_name is None:
+        _delta_table_name = os.environ["DELTA_TABLE_NAME"]
+    return _delta_table_name
+
+
+def get_pds_env_cached() -> str:
+    """Get the PDS environment (cached version)."""
+    global _pds_env
+    if _pds_env is None:
+        _pds_env = os.getenv("PDS_ENV", "int")
+    return _pds_env

lambdas/id_sync/src/os_vars.py

@@ -3,10 +3,12 @@
 '''
 from common.clients import logger, secrets_manager_client
 from common.cache import Cache
-from clients import pds_env
+from os_vars import get_pds_env
 from common.pds_service import PdsService
 from common.authentication import AppRestrictedAuth, Service
 
+pds_env = get_pds_env()
+
 
 def get_pds_patient_details(nhs_number: str) -> dict:
     try:

lambdas/id_sync/src/pds_details.py

@@ -1,41 +1,21 @@
 '''
     record Processor
 '''
-from common.aws_lambda_sqs_event_record import AwsLambdaSqsEventRecord
 from common.clients import logger
-from pds_details import get_pds_patient_details
 import json
 from typing import Optional
-from to_do_code import check_records_exist, update_patient_index
+from nhs_number_processor import process_nhs_number
 
 
-def process_record(event_record: AwsLambdaSqsEventRecord):
-    record = AwsLambdaSqsEventRecord(event_record) if isinstance(event_record, dict) else event_record
-    logger.info("Processing record: %s", record)
+def process_record(event_record):
 
-    id = get_id(event_record.body)
-
-    if id:
-        # TODO This code is a placeholder for checking if records exist in the database - defaulting to True for now
-        exists = check_records_exist(id)
-
-        if exists:
-            # get patient details from PDS
-            patient_details = get_pds_patient_details(id)
-            if not patient_details:
-                return {"status": "error", "message": f"No records returned for ID: {id}"}
-
-            patient_details_id = patient_details.get("id")
-
-            # if patient NHS != id, update patient index of vax events to new number
-            if patient_details_id != id:
-                return update_patient_index(id, patient_details_id)
-            else:
-                return {"status": "success", "message": "No update required"}
-        else:
-            return {"status": "error", "message": f"No records found for ID: {id}"}
+    logger.info("Processing record: %s", event_record)
+    body = event_record.get('body', {})
+    nhs_number = get_id(body)
+    if nhs_number:
+        return process_nhs_number(nhs_number)
     else:
-        return {"status": "error", "message": "No ID found in event record"}
+        return {"status": "error", "message": "No NHS number found in event record"}
 
 
 def get_id(event_body) -> Optional[str]:

lambdas/id_sync/src/record_processor.py

@@ -1,13 +1,28 @@
 '''
     record Processor
 '''
+# from common.clients import get_delta_table
+from boto3.dynamodb.conditions import Key
+from os_vars import get_ieds_table_name, get_delta_table_name
+from common.clients import get_delta_table
 
+ieds_tablename = get_ieds_table_name()
+delta_tablename = get_delta_table_name()
 
-def check_records_exist(id: str) -> bool:
-    # TODO: Implement logic to check if records exist in the database
-    return True
 
+def check_records_exist(dynamodb_table, id: str) -> bool:
+    # dynamodb query to check that records exist for the given ID
+    search_patient_pk = f"Patient#{id}"
+    response = dynamodb_table.query(
+        KeyConditionExpression=Key("PK").eq(search_patient_pk)
+    )
+    return response.get("Count", 0) > 0
 
-def update_patient_index(old_id: str, new_id: str):
-    # TODO: Implement logic to update patient index in Redis or other data store
-    return {"status": "success", "message": f"Updated patient idx from {old_id} to {new_id}", "TODO": "Implement logic"}
+
+def update_patient_id(old_id: str, new_id: str):
+
+    # check if old_id and new_id are not empty
+    if not old_id or not new_id:
+        return {"status": "error", "message": "Old ID and New ID cannot be empty"}
+    else:
+        return {"status": "success", "message": f"Updated patient {old_id} to {new_id}", "TODO": "Implement logic"}