PRMP-643 MNS update with review table (#840)

Author: NogaNHS

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/enums/feature_flags.py

@@ -10,3 +10,4 @@ class FeatureFlags(StrEnum):
         "lloydGeorgeValidationStrictModeEnabled"
     )
     UPLOAD_DOCUMENT_ITERATION_2_ENABLED = "uploadDocumentIteration2Enabled"
+    UPLOAD_DOCUMENT_ITERATION_3_ENABLED = "uploadDocumentIteration3Enabled"

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,48 @@
+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 DocumentUploadReviewReference(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_reference_service.py

@@ -0,0 +1,74 @@
+from datetime import datetime
+
+from enums.supported_document_types import SupportedDocumentTypes
+from models.document_reference import DocumentReference
+from services.document_service import DocumentService
+from utils.audit_logging_setup import LoggingService
+from utils.dynamo_utils import filter_uploaded_docs_and_recently_uploading_docs
+from utils.exceptions import FileUploadInProgress, NoAvailableDocument
+
+logger = LoggingService(__name__)
+
+
+class DocumentReferenceService(DocumentService):
+    """Service for handling DocumentReference operations."""
+
+    def __init__(self, doc_type: SupportedDocumentTypes = SupportedDocumentTypes.LG):
+        super().__init__()
+        self.doc_type = doc_type
+
+    @property
+    def table_name(self) -> str:
+        return self.doc_type.get_dynamodb_table_name()
+
+    @property
+    def model_class(self) -> type:
+        return DocumentReference
+
+    @property
+    def s3_bucket(self) -> str:
+        return self.doc_type.get_s3_bucket_name()
+
+    def get_available_lloyd_george_record_for_patient(
+        self, nhs_number: str
+    ) -> list[DocumentReference]:
+        """Get available Lloyd George records for a patient, checking for upload status."""
+        filter_expression = filter_uploaded_docs_and_recently_uploading_docs()
+        available_docs = self.fetch_documents_from_table_with_nhs_number(
+            nhs_number,
+            query_filter=filter_expression,
+        )
+
+        file_in_progress_message = (
+            "The patients Lloyd George record is in the process of being uploaded"
+        )
+        if not available_docs:
+            raise NoAvailableDocument()
+        for document in available_docs:
+            if document.uploading and not document.uploaded:
+                raise FileUploadInProgress(file_in_progress_message)
+        return available_docs
+
+    def update_patient_ods_code(
+        self,
+        patient_documents: list[DocumentReference],
+        updated_ods_code: str,
+    ) -> None:
+        update_field = {"current_gp_ods", "custodian", "last_updated"}
+        if not patient_documents:
+            return
+
+        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
+            ):
+                reference.current_gp_ods = updated_ods_code
+                reference.custodian = updated_ods_code
+                reference.last_updated = int(datetime.now().timestamp())
+
+                self.update_document(
+                    document=reference, update_fields_name=update_field
+                )

lambdas/services/document_service.py

@@ -5,25 +5,39 @@
 from enums.metadata_field_names import DocumentReferenceMetadataFields
 from enums.supported_document_types import SupportedDocumentTypes
 from models.document_reference import DocumentReference
-from pydantic import ValidationError
+from pydantic import BaseModel, ValidationError
 from services.base.dynamo_service import DynamoDBService
 from services.base.s3_service import S3Service
 from utils.audit_logging_setup import LoggingService
 from utils.common_query_filters import NotDeleted
-from utils.dynamo_utils import filter_uploaded_docs_and_recently_uploading_docs
-from utils.exceptions import (
-    DocumentServiceException,
-    FileUploadInProgress,
-    NoAvailableDocument,
-)
+from utils.exceptions import DocumentServiceException
 
 logger = LoggingService(__name__)
 
 
 class DocumentService:
+    """Service for document operations."""
+
     def __init__(self):
         self.s3_service = S3Service()
         self.dynamo_service = DynamoDBService()
+        self._lg_table_name = os.getenv("LLOYD_GEORGE_DYNAMODB_NAME")
+        self._lg_s3_bucket = os.getenv("LLOYD_GEORGE_BUCKET_NAME")
+
+    @property
+    def table_name(self) -> str:
+        """DynamoDB table name. Can be overridden by child classes."""
+        return self._lg_table_name
+
+    @property
+    def s3_bucket(self) -> str:
+        """S3 bucket name. Can be overridden by child classes."""
+        return self._lg_s3_bucket
+
+    @property
+    def model_class(self) -> type[BaseModel]:
+        """Pydantic model class. Can be overridden by child classes."""
+        return DocumentReference
 
     def fetch_available_document_references_by_type(
         self,
@@ -34,56 +48,74 @@ def fetch_available_document_references_by_type(
         table_name = doc_type.get_dynamodb_table_name()
 
         return self.fetch_documents_from_table_with_nhs_number(
-            nhs_number, table_name, query_filter=query_filter
+            nhs_number,
+            table_name,
+            query_filter=query_filter,
         )
 
     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 | None = None,
+        query_filter: Attr | ConditionBase = None,
+        model_class: type[BaseModel] = None,
+    ) -> list:
+        """Fetch documents by NHS number from specified or configured table."""
+        table_name = table or self.table_name
+
         documents = self.fetch_documents_from_table(
-            table=table,
             index_name="NhsNumberIndex",
             search_key="NhsNumber",
             search_condition=nhs_number,
             query_filter=query_filter,
+            table_name=table_name,
+            model_class=model_class,
         )
-
         return documents
 
     def fetch_documents_from_table(
         self,
-        table: str,
         search_condition: str,
         search_key: str,
         index_name: str | None = None,
         query_filter: Attr | ConditionBase = None,
-    ) -> list[DocumentReference]:
+        table_name: str | None = None,
+        model_class: type[BaseModel] = None,
+    ) -> list:
+        """Fetch documents from specified or configured table using model_class."""
         documents = []
+        table_name = table_name or self.table_name
+        model_class = model_class or self.model_class
 
         response = self.dynamo_service.query_table(
-            table_name=table,
+            table_name=table_name,
             index_name=index_name,
             search_key=search_key,
             search_condition=search_condition,
             query_filter=query_filter,
         )
         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}")
                 logger.error(f"{e}")
                 continue
         return documents
 
-    def get_nhs_numbers_based_on_ods_code(self, ods_code: str) -> list[str]:
+    def get_nhs_numbers_based_on_ods_code(
+        self, ods_code: str, table_name: str | None = None
+    ) -> list[str]:
+        """Get unique NHS numbers for patients with given ODS code."""
+        table_name = table_name or self.table_name
+
         documents = self.fetch_documents_from_table(
-            table=os.environ["LLOYD_GEORGE_DYNAMODB_NAME"],
             index_name="OdsCodeIndex",
             search_key=DocumentReferenceMetadataFields.CURRENT_GP_ODS.value,
             search_condition=ods_code,
             query_filter=NotDeleted,
+            table_name=table_name,
         )
         nhs_numbers = list({document.nhs_number for document in documents})
         return nhs_numbers
@@ -139,21 +171,27 @@ def delete_document_object(self, bucket: str, key: str):
 
     def update_document(
         self,
-        table_name: str,
-        document_reference: DocumentReference,
-        update_fields_name: set[str] = None,
+        table_name: str | None = None,
+        document: BaseModel = None,
+        update_fields_name: set[str] | None = None,
     ):
+        """Update document in specified or configured table."""
+        table_name = table_name or self.table_name
+
         self.dynamo_service.update_item(
             table_name=table_name,
-            key_pair={DocumentReferenceMetadataFields.ID.value: document_reference.id},
-            updated_fields=document_reference.model_dump(
+            key_pair={DocumentReferenceMetadataFields.ID.value: document.id},
+            updated_fields=document.model_dump(
                 exclude_none=True, by_alias=True, include=update_fields_name
             ),
         )
 
     def hard_delete_metadata_records(
-        self, table_name: str, document_references: list[DocumentReference]
+        self, table_name: str, document_references: list[BaseModel]
     ):
+        """Permanently delete metadata from specified or configured table."""
+        table_name = table_name or self.table_name
+
         logger.info(f"Deleting items in table: {table_name} (HARD DELETE)")
         primary_key_name = DocumentReferenceMetadataFields.ID.value
         for reference in document_references:
@@ -162,41 +200,26 @@ def hard_delete_metadata_records(
             self.dynamo_service.delete_item(table_name, deletion_key)
 
     @staticmethod
-    def is_upload_in_process(record: DocumentReference):
+    def is_upload_in_process(record: DocumentReference) -> bool:
+        """Check if a document upload is currently in progress."""
         return (
             not record.uploaded
             and record.uploading
             and record.last_updated_within_three_minutes()
             and record.doc_status != "final"
         )
 
-    def get_available_lloyd_george_record_for_patient(
-        self, nhs_number
-    ) -> list[DocumentReference]:
-        filter_expression = filter_uploaded_docs_and_recently_uploading_docs()
-        available_docs = self.fetch_available_document_references_by_type(
-            nhs_number,
-            SupportedDocumentTypes.LG,
-            query_filter=filter_expression,
-        )
-
-        file_in_progress_message = (
-            "The patients Lloyd George record is in the process of being uploaded"
-        )
-        if not available_docs:
-            raise NoAvailableDocument()
-        for document in available_docs:
-            if document.uploading and not document.uploaded:
-                raise FileUploadInProgress(file_in_progress_message)
-        return available_docs
-
     def get_batch_document_references_by_id(
         self, document_ids: list[str], doc_type: SupportedDocumentTypes
-    ) -> list[DocumentReference]:
+    ) -> list:
         table_name = doc_type.get_dynamodb_table_name()
+
+        table_name = table_name or self.table_name
+        model_class = self.model_class
+
         response = self.dynamo_service.batch_get_items(
             table_name=table_name, key_list=document_ids
         )
 
-        found_docs = [DocumentReference.model_validate(item) for item in response]
+        found_docs = [model_class.model_validate(item) for item in response]
         return found_docs