VED-000: Refactor Api_clients Authentication (#1278)

Author: amarauzoma

lambdas/id_sync/src/pds_details.py

@@ -2,37 +2,6 @@
 Operations related to PDS (Patient Demographic Service)
 """
 
-import tempfile
-
-from common.api_clients.authentication import AppRestrictedAuth, Service
-from common.api_clients.pds_service import PdsService
-from common.cache import Cache
-from common.clients import get_secrets_manager_client, logger
-from exceptions.id_sync_exception import IdSyncException
-from os_vars import get_pds_env
-
-pds_env = get_pds_env()
-safe_tmp_dir = tempfile.mkdtemp(dir="/tmp")
-
-
-# Get Patient details from external service PDS using NHS number from MNS notification
-def pds_get_patient_details(nhs_number: str) -> dict:
-    try:
-        cache = Cache(directory=safe_tmp_dir)
-        authenticator = AppRestrictedAuth(
-            service=Service.PDS,
-            secret_manager_client=get_secrets_manager_client(),
-            environment=pds_env,
-            cache=cache,
-        )
-        pds_service = PdsService(authenticator, pds_env)
-        patient = pds_service.get_patient_details(nhs_number)
-        return patient
-    except Exception as e:
-        msg = "Error retrieving patient details from PDS"
-        logger.exception(msg)
-        raise IdSyncException(message=msg) from e
-
 
 def get_nhs_number_from_pds_resource(pds_resource: dict) -> str:
     """Simple helper to get the NHS Number from a PDS Resource. No handling as this is a mandatory field in the PDS

lambdas/mns_publisher/poetry.lock

@@ -19,6 +19,7 @@ boto3 = "~1.42.37"
 mypy-boto3-dynamodb = "^1.42.33"
 moto = "~5.1.20"
 cache = "^1.0.3"
+aws-lambda-powertools = {version = "3.24.0"}
 
 [build-system]
 requires = ["poetry-core >= 1.5.0"]

lambdas/mns_publisher/pyproject.toml

@@ -1,5 +1,6 @@
 # Static constants for the MNS notification creation process
 SPEC_VERSION = "1.0"
-IMMUNISATION_TYPE = "imms-vaccinations-1"
+IMMUNISATION_EVENT_SOURCE = "uk.nhs.vaccinations-data-flow-management"
+IMMUNISATION_EVENT_TYPE = "imms-vaccination-record-change-1"
 
 DYNAMO_DB_TYPE_DESCRIPTORS = ("S", "N", "BOOL", "M", "L")

lambdas/mns_publisher/src/constants.py

@@ -10,7 +10,7 @@
 from common.api_clients.get_pds_details import pds_get_patient_details
 from common.clients import logger
 from common.get_service_url import get_service_url
-from constants import DYNAMO_DB_TYPE_DESCRIPTORS, IMMUNISATION_TYPE, SPEC_VERSION
+from constants import DYNAMO_DB_TYPE_DESCRIPTORS, IMMUNISATION_EVENT_SOURCE, IMMUNISATION_EVENT_TYPE, SPEC_VERSION
 
 IMMUNIZATION_ENV = os.getenv("IMMUNIZATION_ENV")
 IMMUNIZATION_BASE_PATH = os.getenv("IMMUNIZATION_BASE_PATH")
@@ -43,8 +43,8 @@ def create_mns_notification(sqs_event: SQSMessage) -> MnsNotificationPayload:
     return {
         "specversion": SPEC_VERSION,
         "id": str(uuid.uuid4()),
-        "source": immunisation_url,
-        "type": IMMUNISATION_TYPE,
+        "source": IMMUNISATION_EVENT_SOURCE,
+        "type": IMMUNISATION_EVENT_TYPE,
         "time": date_and_time,
         "subject": nhs_number,
         "dataref": f"{immunisation_url}/Immunization/{imms_id}",

lambdas/mns_publisher/src/create_notification.py

@@ -0,0 +1,21 @@
+"""
+Centralised observability for MNS publisher Lambda.
+
+log_uncaught_exceptions=True ensures unexpected exceptions are captured as
+structured JSON logs at the Lambda boundary.
+"""
+
+from __future__ import annotations
+
+import os
+
+from aws_lambda_powertools import Logger
+
+_SERVICE_NAME = "mns-immunisation-publisher."
+
+logger: Logger = Logger(
+    service=_SERVICE_NAME,
+    level=os.environ.get("LOG_LEVEL", "INFO"),
+    log_uncaught_exceptions=True,
+    location=os.environ.get("POWERTOOLS_LOGGER_LOG_CALLABLE_LOCATION", "false").lower() == "true",
+)

lambdas/mns_publisher/src/observability.py

@@ -1,33 +1,63 @@
 import json
 import os
-from typing import Tuple
 
+from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord
 from aws_lambda_typing.events.sqs import SQSMessage
 
 from common.api_clients.mns_service import MnsService
 from common.api_clients.mns_setup import get_mns_service
 from common.api_clients.mock_mns_service import MockMnsService
-from common.clients import logger
 from create_notification import create_mns_notification
+from observability import logger
 
 mns_env = os.getenv("MNS_ENV", "int")
-MNS_TEST_QUEUE_URL = os.getenv("MNS_TEST_QUEUE_URL")
+_mns_service: MnsService | MockMnsService | None = None
+SqsRecord = SQSRecord | SQSMessage
 
 
-def process_records(records: list[SQSMessage]) -> dict[str, list]:
+def _get_message_id(record: SqsRecord) -> str:
+    if isinstance(record, SQSRecord):
+        return record.message_id
+
+    return record.get("messageId", "unknown")
+
+
+def _get_body(record: SqsRecord) -> dict | str:
+    if isinstance(record, SQSRecord):
+        return record.body
+
+    return record.get("body", {})
+
+
+def _as_sqs_message(record: SqsRecord) -> SQSMessage:
+    if isinstance(record, SQSRecord):
+        return record.raw_event
+
+    return record
+
+
+def _get_runtime_mns_service() -> MnsService | MockMnsService:
+    global _mns_service
+    if _mns_service is None:
+        _mns_service = get_mns_service(mns_env=mns_env)
+
+    return _mns_service
+
+
+def process_records(records: list[SqsRecord]) -> dict[str, list]:
     """
     Process multiple SQS records.
     Args: records: List of SQS records to process
     Returns: List of failed item identifiers for partial batch failure
     """
     batch_item_failures = []
-    mns_service = get_mns_service(mns_env=mns_env)
+    mns_service = _get_runtime_mns_service()
 
     for record in records:
         try:
             process_record(record, mns_service)
         except Exception:
-            message_id = record.get("messageId", "unknown")
+            message_id = _get_message_id(record)
             batch_item_failures.append({"itemIdentifier": message_id})
             logger.exception("Failed to process record", extra={"message_id": message_id})
 
@@ -39,7 +69,7 @@ def process_records(records: list[SQSMessage]) -> dict[str, list]:
     return {"batchItemFailures": batch_item_failures}
 
 
-def process_record(record: SQSMessage, mns_service: MnsService | MockMnsService) -> None:
+def process_record(record: SqsRecord, mns_service: MnsService | MockMnsService) -> None:
     """
     Process a single SQS record.
     Args:
@@ -50,34 +80,36 @@ def process_record(record: SQSMessage, mns_service: MnsService | MockMnsService)
     message_id, immunisation_id = extract_trace_ids(record)
     notification_id = None
 
-    mns_notification_payload = create_mns_notification(record)
+    mns_notification_payload = create_mns_notification(_as_sqs_message(record))
     notification_id = mns_notification_payload.get("id")
 
     action_flag = mns_notification_payload.get("filtering", {}).get("action")
     logger.info(
         "Processing message",
-        extra={
-            "notification_id": notification_id,
-            "message_id": message_id,
-            "immunisation_id": immunisation_id,
-            "action_flag": action_flag,
-        },
+        notification_id=notification_id,
+        message_id=message_id,
+        immunisation_id=immunisation_id,
+        action_flag=action_flag,
     )
 
     mns_service.publish_notification(mns_notification_payload)
-    logger.info("Successfully created MNS notification", extra={"mns_notification_id": notification_id})
+
+    logger.info(
+        "Successfully created MNS notification",
+        mns_notification_id=notification_id,
+    )
 
 
-def extract_trace_ids(record: SQSMessage) -> Tuple[str, str | None]:
+def extract_trace_ids(record: SqsRecord) -> tuple[str, str | None]:
     """
     Extract identifiers for tracing from SQS record.
     Returns: Tuple of (message_id, immunisation_id)
     """
-    sqs_message_id = record.get("messageId", "unknown")
+    sqs_message_id = _get_message_id(record)
     immunisation_id = None
 
     try:
-        sqs_event_body = record.get("body", {})
+        sqs_event_body = _get_body(record)
         if isinstance(sqs_event_body, str):
             sqs_event_body = json.loads(sqs_event_body)
 

lambdas/mns_publisher/src/process_records.py

@@ -3,7 +3,7 @@
 import unittest
 from unittest.mock import MagicMock, patch
 
-from constants import IMMUNISATION_TYPE, SPEC_VERSION
+from constants import IMMUNISATION_EVENT_SOURCE, IMMUNISATION_EVENT_TYPE, SPEC_VERSION
 from create_notification import (
     _unwrap_dynamodb_value,
     calculate_age_at_vaccination,
@@ -81,8 +81,8 @@ def test_success_create_mns_notification_complete_payload(self, mock_uuid, mock_
         result = create_mns_notification(self.sample_sqs_event)
 
         self.assertEqual(result["specversion"], SPEC_VERSION)
-        self.assertEqual(result["type"], IMMUNISATION_TYPE)
-        self.assertEqual(result["source"], self.expected_immunisation_url)
+        self.assertEqual(result["type"], IMMUNISATION_EVENT_TYPE)
+        self.assertEqual(result["source"], IMMUNISATION_EVENT_SOURCE)
         self.assertEqual(result["subject"], "9481152782")
 
         expected_dataref = f"{self.expected_immunisation_url}/Immunization/d058014c-b0fd-4471-8db9-3316175eb825"

lambdas/mns_publisher/tests/test_create_notification.py

@@ -125,7 +125,7 @@ def setUpClass(cls):
         cls.sample_sqs_record = load_sample_sqs_event()
 
     @patch("process_records.logger")
-    @patch("process_records.get_mns_service")
+    @patch("process_records._get_runtime_mns_service")
     @patch("process_records.process_record")
     def test_process_records_all_success(self, mock_process_record, mock_get_mns, mock_logger):
         """Test processing multiple records with all successes."""
@@ -145,7 +145,7 @@ def test_process_records_all_success(self, mock_process_record, mock_get_mns, mo
         mock_logger.info.assert_called_with("Successfully processed all 2 messages")
 
     @patch("process_records.logger")
-    @patch("process_records.get_mns_service")
+    @patch("process_records._get_runtime_mns_service")
     @patch("process_records.process_record")
     def test_process_records_partial_failure(self, mock_process_record, mock_get_mns, mock_logger):
         """Test processing with some failures."""
@@ -167,7 +167,7 @@ def test_process_records_partial_failure(self, mock_process_record, mock_get_mns
         mock_logger.warning.assert_called_with("Batch completed with 1 failures")
 
     @patch("process_records.logger")
-    @patch("process_records.get_mns_service")
+    @patch("process_records._get_runtime_mns_service")
     @patch("process_records.process_record")
     def test_process_records_empty_list(self, mock_process_record, mock_get_mns, mock_logger):
         """Test processing empty record list."""
@@ -181,7 +181,7 @@ def test_process_records_empty_list(self, mock_process_record, mock_get_mns, moc
         mock_logger.info.assert_called_with("Successfully processed all 0 messages")
 
     @patch("process_records.logger")
-    @patch("process_records.get_mns_service")
+    @patch("process_records._get_runtime_mns_service")
     @patch("process_records.process_record")
     def test_process_records_mns_service_created_once(self, mock_process_record, mock_get_mns, mock_logger):
         """Test that MNS service is created only once for batch."""
@@ -300,7 +300,7 @@ def test_successful_notification_creation_with_gp(self, mock_logger, mock_get_to
 
     @responses.activate
     @patch("common.api_clients.authentication.AppRestrictedAuth.get_access_token")
-    @patch("process_records.get_mns_service")
+    @patch("process_records._get_runtime_mns_service")
     @patch("process_records.logger")
     def test_pds_failure(self, mock_logger, mock_get_mns, mock_get_token):
         """