PRMP-643 MNS update with review table

lambdas/enums/document_review_status.py

@@ -0,0 +1,7 @@
+from enum import StrEnum
+
+
+class DocumentReviewStatus(StrEnum):
+    PENDING_REVIEW = "PENDING_REVIEW"
+    APPROVED = "APPROVED"
+    REJECTED = "REJECTED"

lambdas/handlers/mns_notification_handler.py

@@ -19,6 +19,7 @@
         "APPCONFIG_CONFIGURATION",
         "APPCONFIG_ENVIRONMENT",
         "LLOYD_GEORGE_DYNAMODB_NAME",
+        "DOCUMENT_REVIEW_DYNAMODB_NAME",
         "MNS_NOTIFICATION_QUEUE_URL",
     ]
 )

lambdas/models/document_review.py

@@ -0,0 +1,43 @@
+import uuid
+from typing import Optional
+
+from enums.document_review_status import DocumentReviewStatus
+from enums.metadata_field_names import DocumentReferenceMetadataFields
+from enums.snomed_codes import SnomedCodes
+from pydantic import BaseModel, ConfigDict, Field
+from pydantic.alias_generators import to_pascal
+
+
+class DocumentReviewFileDetails(BaseModel):
+    model_config = ConfigDict(
+        validate_by_alias=True,
+        validate_by_name=True,
+        alias_generator=to_pascal,
+    )
+
+    file_name: str
+    file_location: str
+
+
+class DocumentUploadReview(BaseModel):
+    model_config = ConfigDict(
+        validate_by_alias=True,
+        validate_by_name=True,
+        alias_generator=to_pascal,
+        use_enum_values=True,
+    )
+    id: str = Field(default_factory=lambda: str(uuid.uuid4()), alias=str(DocumentReferenceMetadataFields.ID.value))
+    author: str
+    custodian: str
+    review_status: DocumentReviewStatus = Field(default=DocumentReviewStatus.PENDING_REVIEW)
+    review_reason: str
+    review_date: int = Field(default=None)
+    reviewer: str = Field(default=None)
+    upload_date: int
+    files: list[DocumentReviewFileDetails]
+    nhs_number: str
+    ttl: Optional[int] = Field(
+        alias=str(DocumentReferenceMetadataFields.TTL.value), default=None
+    )
+    document_reference_id: str = Field(default=None)
+    document_snomed_code_type: str = Field(default=SnomedCodes.LLOYD_GEORGE.value.code)

lambdas/services/document_service.py

@@ -5,6 +5,7 @@
 from enums.metadata_field_names import DocumentReferenceMetadataFields
 from enums.supported_document_types import SupportedDocumentTypes
 from models.document_reference import DocumentReference
+from models.document_review import DocumentUploadReview
 from pydantic import ValidationError
 from services.base.dynamo_service import DynamoDBService
 from services.base.s3_service import S3Service
@@ -38,14 +39,19 @@ def fetch_available_document_references_by_type(
         )
 
     def fetch_documents_from_table_with_nhs_number(
-        self, nhs_number: str, table: str, query_filter: Attr | ConditionBase = None
-    ) -> list[DocumentReference]:
+        self,
+        nhs_number: str,
+        table: str,
+        query_filter: Attr | ConditionBase = None,
+        model_class=DocumentReference,
+    ) -> list[DocumentReference] | list[DocumentUploadReview]:
         documents = self.fetch_documents_from_table(
             table=table,
             index_name="NhsNumberIndex",
             search_key="NhsNumber",
             search_condition=nhs_number,
             query_filter=query_filter,
+            model_class=model_class,
         )
 
         return documents
@@ -57,7 +63,8 @@ def fetch_documents_from_table(
         search_key: str,
         index_name: str = None,
         query_filter: Attr | ConditionBase = None,
-    ) -> list[DocumentReference]:
+        model_class=DocumentReference,
+    ) -> list[DocumentReference] | list[DocumentUploadReview]:
         documents = []
 
         response = self.dynamo_service.query_table(
@@ -69,7 +76,7 @@ def fetch_documents_from_table(
         )
         for item in response:
             try:
-                document = DocumentReference.model_validate(item)
+                document = model_class.model_validate(item)
                 documents.append(document)
             except ValidationError as e:
                 logger.error(f"Validation error on document: {item}")
@@ -140,7 +147,7 @@ def delete_document_object(self, bucket: str, key: str):
     def update_document(
         self,
         table_name: str,
-        document_reference: DocumentReference,
+        document_reference: DocumentReference | DocumentUploadReview,
         update_fields_name: set[str] = None,
     ):
         self.dynamo_service.update_item(

lambdas/services/process_mns_message_service.py

@@ -6,6 +6,7 @@
 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 DocumentUploadReview
 from models.sqs.mns_sqs_message import MNSSQSMessage
 from services.base.sqs_service import SQSService
 from services.document_service import DocumentService
@@ -20,11 +21,13 @@
 class MNSNotificationService:
     def __init__(self):
         self.document_service = DocumentService()
-        self.table = os.getenv("LLOYD_GEORGE_DYNAMODB_NAME")
+        self.lg_table = os.getenv("LLOYD_GEORGE_DYNAMODB_NAME")
+        self.document_review_table = os.getenv("DOCUMENT_REVIEW_DYNAMODB_NAME")
         self.pds_service = get_pds_service()
         self.sqs_service = SQSService()
         self.queue = os.getenv("MNS_NOTIFICATION_QUEUE_URL")
         self.DOCUMENT_UPDATE_FIELDS = {"current_gp_ods", "custodian", "last_updated"}
+        self.DOCUMENT_REVIEW_UPDATE_FIELDS = {"custodian"}
         self.PCSE_ODS = PCSE_ODS_CODE
 
     def handle_mns_notification(self, message: MNSSQSMessage):
@@ -53,15 +56,17 @@ def handle_mns_notification(self, message: MNSSQSMessage):
             raise e
 
     def handle_gp_change_notification(self, message: MNSSQSMessage) -> None:
-        patient_document_references = self.get_patient_documents(
+        lg_documents, review_documents = self.get_all_patient_documents(
             message.subject.nhs_number
         )
 
-        if not patient_document_references:
+        if not lg_documents and not review_documents:
             return
 
         updated_ods_code = self.get_updated_gp_ods(message.subject.nhs_number)
-        self.update_patient_ods_code(patient_document_references, updated_ods_code)
+        self.update_all_patient_documents(
+            lg_documents, review_documents, updated_ods_code
+        )
         logger.info("Update complete for change of GP")
 
     def handle_death_notification(self, message: MNSSQSMessage) -> None:
@@ -75,58 +80,120 @@ def handle_death_notification(self, message: MNSSQSMessage) -> None:
                 )
 
             case DeathNotificationStatus.REMOVED:
-                patient_document_references = self.get_patient_documents(nhs_number)
-                if patient_document_references:
+                lg_documents, review_documents = self.get_all_patient_documents(
+                    nhs_number
+                )
+
+                if lg_documents or review_documents:
                     updated_ods_code = self.get_updated_gp_ods(nhs_number)
-                    self.update_patient_ods_code(
-                        patient_document_references, updated_ods_code
+                    self.update_all_patient_documents(
+                        lg_documents, review_documents, updated_ods_code
                     )
                     logger.info("Update complete for death notification change.")
 
             case DeathNotificationStatus.FORMAL:
-                patient_document_references = self.get_patient_documents(nhs_number)
-                if patient_document_references:
-                    self.update_patient_ods_code(
-                        patient_document_references, PatientOdsInactiveStatus.DECEASED
+                lg_documents, review_documents = self.get_all_patient_documents(
+                    nhs_number
+                )
+
+                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 update_patient_ods_code(
-        self, patient_documents: list[DocumentReference], updated_ods_code: str
+        self,
+        patient_documents: list[DocumentReference],
+        updated_ods_code: str,
     ) -> None:
         if not patient_documents:
             return
+        updated_custodian = updated_ods_code
+        if updated_ods_code in [
+            PatientOdsInactiveStatus.DECEASED,
+            PatientOdsInactiveStatus.SUSPENDED,
+        ]:
+            updated_custodian = self.PCSE_ODS
+
         for reference in patient_documents:
             logger.info("Updating patient document reference...")
 
             if (
                 reference.current_gp_ods != updated_ods_code
                 or reference.custodian != updated_ods_code
             ):
-                updated_custodian = updated_ods_code
-                if updated_ods_code in [
-                    PatientOdsInactiveStatus.DECEASED,
-                    PatientOdsInactiveStatus.SUSPENDED,
-                ]:
-                    updated_custodian = self.PCSE_ODS
+
                 reference.current_gp_ods = updated_ods_code
                 reference.custodian = updated_custodian
                 reference.last_updated = int(datetime.now().timestamp())
 
                 self.document_service.update_document(
-                    self.table,
+                    self.lg_table,
                     reference,
                     self.DOCUMENT_UPDATE_FIELDS,
                 )
 
+    def update_document_review_custodian(
+        self, patient_documents: list[DocumentUploadReview], updated_ods_code: str
+    ) -> None:
+        if not patient_documents:
+            return
+
+        if updated_ods_code in [
+            PatientOdsInactiveStatus.DECEASED,
+            PatientOdsInactiveStatus.SUSPENDED,
+        ]:
+            updated_ods_code = self.PCSE_ODS
+
+        for review in patient_documents:
+            logger.info("Updating document review custodian...")
+
+            if review.custodian != updated_ods_code:
+                review.custodian = updated_ods_code
+
+                self.document_service.update_document(
+                    self.document_review_table,
+                    review,
+                    self.DOCUMENT_REVIEW_UPDATE_FIELDS,
+                )
+
     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_patient_documents(self, nhs_number: str) -> list[DocumentReference]:
+    def get_patient_documents(
+        self, nhs_number: str, table: str, model_class: type
+    ) -> list[DocumentReference] | list[DocumentUploadReview]:
         """Fetch patient documents and return them if they exist."""
         return self.document_service.fetch_documents_from_table_with_nhs_number(
-            nhs_number, self.table
+            nhs_number, table, model_class=model_class
         )
+
+    def get_all_patient_documents(
+        self, nhs_number: str
+    ) -> tuple[list[DocumentReference], list[DocumentUploadReview]]:
+        """Fetch patient documents from both LG and document review tables."""
+        lg_documents = self.get_patient_documents(
+            nhs_number, self.lg_table, DocumentReference
+        )
+        review_documents = self.get_patient_documents(
+            nhs_number, self.document_review_table, DocumentUploadReview
+        )
+        return lg_documents, review_documents
+
+    def update_all_patient_documents(
+        self,
+        lg_documents: list[DocumentReference],
+        review_documents: list[DocumentUploadReview],
+        updated_ods_code: str,
+    ) -> None:
+        """Update documents in both tables if they exist."""
+        if lg_documents:
+            self.update_patient_ods_code(lg_documents, updated_ods_code)
+        if review_documents:
+            self.update_document_review_custodian(review_documents, updated_ods_code)

lambdas/tests/unit/conftest.py

@@ -225,6 +225,7 @@ def set_env(monkeypatch):
     monkeypatch.setenv("SLACK_BOT_TOKEN", MOCK_SLACK_BOT_TOKEN)
     monkeypatch.setenv("SLACK_CHANNEL_ID", MOCK_ALERTING_SLACK_CHANNEL_ID)
     monkeypatch.setenv("ITOC_TESTING_ODS_CODES", MOCK_ITOC_ODS_CODES)
+    monkeypatch.setenv("DOCUMENT_REVIEW_DYNAMODB_NAME", "test_document_review")
 
 
 EXPECTED_PARSED_PATIENT_BASE_CASE = PatientDetails(