[PRMP-1476] Update User Restriction ODS Code upon MNS Notification (Active Records Only)

lambdas/handlers/mns_notification_handler.py

@@ -1,8 +1,9 @@
 import json
 
+from pydantic import ValidationError
+
 from enums.mns_notification_types import MNSNotificationTypes
 from models.sqs.mns_sqs_message import MNSSQSMessage
-from pydantic import ValidationError
 from services.process_mns_message_service import MNSNotificationService
 from utils.audit_logging_setup import LoggingService
 from utils.decorators.ensure_env_var import ensure_environment_variables
@@ -21,7 +22,8 @@
         "LLOYD_GEORGE_DYNAMODB_NAME",
         "DOCUMENT_REVIEW_DYNAMODB_NAME",
         "MNS_NOTIFICATION_QUEUE_URL",
-    ]
+        "RESTRICTIONS_TABLE_NAME",
+    ],
 )
 @override_error_check
 def lambda_handler(event, context):
@@ -38,7 +40,7 @@ def lambda_handler(event, context):
 
             request_context.patient_nhs_no = mns_message.subject.nhs_number
             logger.info(
-                f"Processing SQS message for nhs number: {mns_message.subject.nhs_number}"
+                f"Processing SQS message for nhs number: {mns_message.subject.nhs_number}",
             )
 
             if mns_message.type in MNSNotificationTypes.__members__.values():

lambdas/models/user_restrictions/user_restrictions.py

@@ -10,11 +10,12 @@ class UserRestrictionsFields(StrEnum):
     ID = "ID"
     CREATOR = "CreatorSmartcard"
     RESTRICTED_USER = "RestrictedSmartcard"
-    REMOVED_BY = "RemoverSmartCard"
-    CUSTODIAN = "Custodian"
+    REMOVED_BY = "RemoverSmartcard"
     NHS_NUMBER = "NhsNumber"
+    CUSTODIAN = "Custodian"
     IS_ACTIVE = "IsActive"
     LAST_UPDATED = "LastUpdated"
+    CREATED = "Created"
 
 
 class UserRestrictionIndexes(StrEnum):

lambdas/services/process_mns_message_service.py

@@ -1,16 +1,20 @@
 import os
 
 from botocore.exceptions import ClientError
+
 from enums.death_notification_status import DeathNotificationStatus
 from enums.mns_notification_types import MNSNotificationTypes
 from enums.patient_ods_inactive_status import PatientOdsInactiveStatus
 from models.document_reference import DocumentReference
 from models.document_review import DocumentUploadReviewReference
 from models.sqs.mns_sqs_message import MNSSQSMessage
+from models.user_restrictions.user_restrictions import UserRestriction
 from services.base.sqs_service import SQSService
 from services.document_reference_service import DocumentReferenceService
 from services.document_upload_review_service import DocumentUploadReviewService
-from services.feature_flags_service import FeatureFlagService
+from services.user_restrictions.user_restriction_dynamo_service import (
+    UserRestrictionDynamoService,
+)
 from utils.audit_logging_setup import LoggingService
 from utils.exceptions import PdsErrorException
 from utils.utilities import get_pds_service
@@ -25,7 +29,7 @@ def __init__(self):
         self.pds_service = get_pds_service()
         self.sqs_service = SQSService()
         self.queue = os.getenv("MNS_NOTIFICATION_QUEUE_URL")
-        self.feature_flag_service = FeatureFlagService()
+        self.restrictions_dynamo_service = UserRestrictionDynamoService()
 
     def handle_mns_notification(self, message: MNSSQSMessage):
         try:
@@ -36,35 +40,31 @@ def handle_mns_notification(self, message: MNSSQSMessage):
                 case MNSNotificationTypes.DEATH_NOTIFICATION:
                     logger.info("Handling death status notification.")
                     self.handle_death_notification(message)
-
-        except PdsErrorException as e:
-            logger.info("An error occurred when calling PDS")
-            logger.info(
-                f"Unable to process message: {message.id}, of type: {message.type}"
-            )
-            logger.info(f"{e}")
-            raise e
-
-        except ClientError as e:
-            logger.info(
-                f"Unable to process message: {message.id}, of type: {message.type}"
+        except (PdsErrorException, ClientError) as e:
+            logger.error(
+                f"Unable to process message: {message.id}, of type: {message.type}",
             )
-            logger.info(f"{e}")
-            raise e
+            logger.error(str(e))
+            raise
 
     def handle_gp_change_notification(self, message: MNSSQSMessage) -> None:
-        lg_documents, review_documents = self.get_all_patient_documents(
-            message.subject.nhs_number
+        nhs_number = message.subject.nhs_number
+        lg_documents, review_documents, restrictions = self._fetch_patient_data(
+            nhs_number,
         )
 
-        if not lg_documents and not review_documents:
+        if not (lg_documents or review_documents or restrictions):
             return
 
-        updated_ods_code = self.get_updated_gp_ods(message.subject.nhs_number)
-        self.update_all_patient_documents(
-            lg_documents, review_documents, updated_ods_code
+        updated_ods_code = self.get_updated_gp_ods(nhs_number)
+        self._apply_ods_update(
+            nhs_number,
+            lg_documents,
+            review_documents,
+            restrictions,
+            updated_ods_code,
         )
-        logger.info("Update complete for change of GP")
+        logger.info("Update complete for change of GP.")
 
     def handle_death_notification(self, message: MNSSQSMessage) -> None:
         death_notification_type = message.data["deathNotificationStatus"]
@@ -73,52 +73,91 @@ def handle_death_notification(self, message: MNSSQSMessage) -> None:
         match death_notification_type:
             case DeathNotificationStatus.INFORMAL:
                 logger.info(
-                    "Patient is deceased - INFORMAL, moving on to the next message."
+                    "Patient is deceased - INFORMAL, moving on to the next message.",
                 )
 
             case DeathNotificationStatus.REMOVED:
-                lg_documents, review_documents = self.get_all_patient_documents(
-                    nhs_number
+                lg_documents, review_documents, restrictions = self._fetch_patient_data(
+                    nhs_number,
                 )
 
-                if lg_documents or review_documents:
-                    updated_ods_code = self.get_updated_gp_ods(nhs_number)
-                    self.update_all_patient_documents(
-                        lg_documents, review_documents, updated_ods_code
-                    )
-                    logger.info("Update complete for death notification change.")
+                if not (lg_documents or review_documents or restrictions):
+                    return
+
+                updated_ods_code = self.get_updated_gp_ods(nhs_number)
+                self._apply_ods_update(
+                    nhs_number,
+                    lg_documents,
+                    review_documents,
+                    restrictions,
+                    updated_ods_code,
+                )
+                logger.info("Update complete for death notification change.")
 
             case DeathNotificationStatus.FORMAL:
-                lg_documents, review_documents = self.get_all_patient_documents(
-                    nhs_number
+                lg_documents, review_documents, restrictions = self._fetch_patient_data(
+                    nhs_number,
+                )
+                self._apply_ods_update(
+                    nhs_number,
+                    lg_documents,
+                    review_documents,
+                    restrictions,
+                    PatientOdsInactiveStatus.DECEASED,
+                )
+                logger.info(
+                    f"Update complete, patient marked {PatientOdsInactiveStatus.DECEASED}.",
                 )
 
-                if lg_documents or review_documents:
-                    self.update_all_patient_documents(
-                        lg_documents,
-                        review_documents,
-                        PatientOdsInactiveStatus.DECEASED,
-                    )
-                    logger.info(
-                        f"Update complete, patient marked {PatientOdsInactiveStatus.DECEASED}."
-                    )
+    def _fetch_patient_data(
+        self,
+        nhs_number: str,
+    ) -> tuple[
+        list[DocumentReference],
+        list[DocumentUploadReviewReference],
+        list[UserRestriction],
+    ]:
+        lg_documents, review_documents = self.get_all_patient_documents(nhs_number)
+        restrictions = (
+            self.restrictions_dynamo_service.query_restrictions_by_nhs_number(
+                nhs_number=nhs_number,
+            )
+        )
+        return lg_documents, review_documents, restrictions
+
+    def _apply_ods_update(
+        self,
+        nhs_number: str,
+        lg_documents: list[DocumentReference],
+        review_documents: list[DocumentUploadReviewReference],
+        restrictions: list[UserRestriction],
+        ods_code: str,
+    ) -> None:
+        if lg_documents or review_documents:
+            self.update_all_patient_documents(lg_documents, review_documents, ods_code)
+        if restrictions:
+            self.update_restrictions(
+                nhs_number=nhs_number,
+                custodian=ods_code,
+                restrictions=restrictions,
+            )
 
     def get_updated_gp_ods(self, nhs_number: str) -> str:
         patient_details = self.pds_service.fetch_patient_details(nhs_number)
         return patient_details.general_practice_ods
 
     def get_all_patient_documents(
-        self, nhs_number: str
+        self,
+        nhs_number: str,
     ) -> tuple[list[DocumentReference], list[DocumentUploadReviewReference]]:
-        """Fetch patient documents from both LG and document review tables."""
         lg_documents = (
             self.lg_document_service.fetch_documents_from_table_with_nhs_number(
-                nhs_number
+                nhs_number,
             )
         )
         review_documents = (
             self.document_review_service.fetch_documents_from_table_with_nhs_number(
-                nhs_number
+                nhs_number,
             )
         )
 
@@ -133,9 +172,26 @@ def update_all_patient_documents(
         """Update documents in both tables if they exist."""
         if lg_documents:
             self.lg_document_service.update_patient_ods_code(
-                lg_documents, updated_ods_code
+                lg_documents,
+                updated_ods_code,
             )
         if review_documents:
             self.document_review_service.update_document_review_custodian(
-                review_documents, updated_ods_code
+                review_documents,
+                updated_ods_code,
+            )
+
+    def update_restrictions(
+        self,
+        nhs_number: str,
+        custodian: str,
+        restrictions: list[UserRestriction],
+    ) -> None:
+        for restriction in restrictions:
+            logger.info(f"Updating restriction {restriction.id}")
+            self.restrictions_dynamo_service.update_restriction_custodian(
+                restriction_id=restriction.id,
+                updated_custodian=custodian,
             )
+
+        logger.info(f"All restrictions for patient {nhs_number} updated.")

lambdas/services/user_restrictions/user_restriction_dynamo_service.py

@@ -17,6 +17,7 @@
 from utils.dynamo_utils import build_mixed_condition_expression
 from utils.exceptions import (
     UserRestrictionConditionCheckFailedException,
+    UserRestrictionDynamoDBException,
     UserRestrictionValidationException,
 )
 
@@ -77,31 +78,33 @@ def update_restriction_inactive(
         removed_by: str,
         patient_id: str,
     ):
-        try:
-            logger.info("Updating user restriction inactive.")
-            current_time = int(datetime.now(timezone.utc).timestamp())
+        logger.info("Updating user restriction inactive.")
+        current_time = int(datetime.now(timezone.utc).timestamp())
 
-            updated_fields = {
-                UserRestrictionsFields.REMOVED_BY.value: removed_by,
-                UserRestrictionsFields.LAST_UPDATED.value: current_time,
-                UserRestrictionsFields.IS_ACTIVE.value: False,
-            }
+        updated_fields = {
+            UserRestrictionsFields.REMOVED_BY.value: removed_by,
+            UserRestrictionsFields.LAST_UPDATED.value: current_time,
+            UserRestrictionsFields.IS_ACTIVE.value: False,
+        }
 
+        try:
             self.dynamo_service.update_item(
                 table_name=self.table_name,
                 key_pair={UserRestrictionsFields.ID.value: restriction_id},
                 updated_fields=updated_fields,
-                condition_expression=f"{UserRestrictionsFields.IS_ACTIVE.value} = :true "
-                f"AND {UserRestrictionsFields.RESTRICTED_USER.value} <> :user_id "
-                f"AND {UserRestrictionsFields.NHS_NUMBER.value} = :patient_id",
+                condition_expression=(
+                    f"{UserRestrictionsFields.IS_ACTIVE} = :true"
+                    f" AND {UserRestrictionsFields.RESTRICTED_USER} <> :user_id"
+                    f" AND {UserRestrictionsFields.NHS_NUMBER} = :patient_id"
+                ),
                 expression_attribute_values={
                     ":true": True,
                     ":user_id": removed_by,
                     ":patient_id": patient_id,
                 },
             )
-
         except ClientError as e:
+            logger.error(e)
             if (
                 e.response["Error"]["Code"]
                 == DynamoClientErrors.CONDITION_CHECK_FAILURE
@@ -111,7 +114,62 @@ def update_restriction_inactive(
                 f"Unexpected DynamoDB error in update_restriction_inactive: "
                 f"{e.response['Error']['Code']} - {e}",
             )
-            raise e
+            raise UserRestrictionDynamoDBException(
+                "An issue occurred while updating user restriction inactive",
+            )
+
+    def query_restrictions_by_nhs_number(
+        self,
+        nhs_number: str,
+    ) -> list[UserRestriction]:
+        try:
+            logger.info("Building IsActive filter for DynamoDB query.")
+            filter_builder = DynamoQueryFilterBuilder()
+            filter_builder.add_condition(
+                UserRestrictionsFields.IS_ACTIVE,
+                AttributeOperator.EQUAL,
+                True,
+            )
+            active_filter_expression = filter_builder.build()
+
+            logger.info("Querying Restrictions by NHS Number.")
+            items = self.dynamo_service.query_table(
+                table_name=self.table_name,
+                index_name=UserRestrictionIndexes.NHS_NUMBER_INDEX,
+                search_key=UserRestrictionsFields.NHS_NUMBER,
+                search_condition=nhs_number,
+                query_filter=active_filter_expression,
+            )
+
+            return self._validate_restrictions(items)
+        except ClientError as e:
+            logger.error(e)
+            raise UserRestrictionDynamoDBException(
+                "An issue occurred while querying restrictions",
+            )
+
+    def update_restriction_custodian(self, restriction_id: str, updated_custodian: str):
+        logger.info(f"Updating custodian for restriction: {restriction_id}")
+        current_time = int(datetime.now(timezone.utc).timestamp())
+
+        updated_fields = {
+            UserRestrictionsFields.LAST_UPDATED.value: current_time,
+            UserRestrictionsFields.CUSTODIAN.value: updated_custodian,
+        }
+
+        try:
+            self.dynamo_service.update_item(
+                table_name=self.table_name,
+                key_pair={UserRestrictionsFields.ID.value: restriction_id},
+                updated_fields=updated_fields,
+            )
+        except ClientError as e:
+            logger.error(
+                f"DynamoDB ClientError when updating custodian for restriction {restriction_id}: {e}",
+            )
+            raise UserRestrictionDynamoDBException(
+                f"An issue occurred while updating restriction custodian for restriction {restriction_id}",
+            ) from e
 
     @staticmethod
     def _build_query_filter(