[NDR-100] Document Reference metadata changes (#654)

* [NDR-100] migrate NhsDocumentReference to use DocumentReference

* [NDR-100] migrate SearchDocumentReference to use DocumentReference

Author: NogaNHS

lambdas/enums/dynamo_filter.py

@@ -9,6 +9,7 @@ class AttributeOperator(Enum):
     LESS_THAN = "lt"
     LESS_THAN_OR_EQUAL = "lte"
     IN = "is_in"
+    NOT_EXISTS = "not_exists"
 
 
 class ConditionOperator(Enum):

lambdas/enums/lambda_error.py

@@ -8,26 +8,22 @@
 
 class LambdaError(Enum):
     def create_error_response(
-        self, params: Optional[dict] = None, **kwargs) -> ErrorResponse:
+        self, params: Optional[dict] = None, **kwargs
+    ) -> ErrorResponse:
         err_code = self.value["err_code"]
         message = self.value["message"]
         if "%" in message and params:
             message = message % params
         interaction_id = getattr(request_context, "request_id", None)
         error_response = ErrorResponse(
-            err_code=err_code,
-            message=message,
-            interaction_id=interaction_id,
-            **kwargs
+            err_code=err_code, message=message, interaction_id=interaction_id, **kwargs
         )
         return error_response
 
     def to_str(self) -> str:
         return f"[{self.value['err_code']}] {self.value['message']}"
 
-    def create_error_body(
-        self, params: Optional[dict] = None, **kwargs
-    ) -> str:
+    def create_error_body(self, params: Optional[dict] = None, **kwargs) -> str:
         return self.create_error_response(params, **kwargs).create()
 
     """

lambdas/enums/snomed_codes.py

@@ -1,4 +1,5 @@
 from enum import Enum
+from typing import Optional
 
 from pydantic import BaseModel
 
@@ -16,3 +17,19 @@ class SnomedCodes(Enum):
     GENERAL_MEDICAL_PRACTICE = SnomedCode(
         code="1060971000000108", display_name="General practice service"
     )
+
+    @classmethod
+    def find_by_code(cls, code: str) -> Optional["SnomedCode"]:
+        """
+        Find a SnomedCodes enum value by its code string.
+
+        Args:
+            code: The SNOMED code string to search for (e.g. "16521000000101")
+
+        Returns:
+            The matching SnomedCodes enum value or None if not found
+        """
+        for snomed_enum in cls:
+            if snomed_enum.value.code == code:
+                return snomed_enum.value
+        return None

lambdas/enums/virus_scan_result.py

@@ -1,7 +1,7 @@
-from enum import Enum
+from enum import StrEnum
 
 
-class VirusScanResult(str, Enum):
+class VirusScanResult(StrEnum):
     CLEAN = "Clean"
     INFECTED = "Infected"
     INFECTED_ALLOWED = "InfectedAllowed"

lambdas/models/access_audit.py

@@ -16,7 +16,7 @@
 
 
 class AccessAuditReason(BaseModel):
-    model_config = ConfigDict(populate_by_name=True, alias_generator=to_pascal)
+    model_config = ConfigDict(validate_by_name=True, alias_generator=to_pascal)
     nhs_number: str = Field(exclude=True)
     request_type: AccessAuditRequestType = Field(exclude=True)
     user_session_id: str

lambdas/models/document_reference.py

@@ -1,12 +1,27 @@
 import pathlib
 from datetime import datetime, timezone
-from typing import Optional
+from typing import Literal, Optional
 
 from enums.metadata_field_names import DocumentReferenceMetadataFields
+from enums.snomed_codes import SnomedCodes
 from enums.supported_document_types import SupportedDocumentTypes
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, model_validator
 from pydantic.alias_generators import to_camel, to_pascal
-from utils.exceptions import InvalidDocumentReferenceException
+
+# Constants
+DATE_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ"
+DEFAULT_CONTENT_TYPE = "application/pdf"
+S3_PREFIX = "s3://"
+THREE_MINUTES_IN_SECONDS = 60 * 3
+
+
+class UploadRequestDocument(BaseModel):
+    model_config = ConfigDict(use_enum_values=True)
+
+    fileName: str
+    contentType: str
+    docType: SupportedDocumentTypes
+    clientId: str
 
 
 class UploadDocumentReference(BaseModel):
@@ -19,43 +34,115 @@ class UploadDocumentReferences(BaseModel):
     files: list[UploadDocumentReference] = Field(...)
 
 
-class SearchDocumentReference(BaseModel):
-    model_config = ConfigDict(
-        alias_generator=to_camel,
-        populate_by_name=True,
-    )
-    id: str
-    created: str
-    file_name: str
-    virus_scanner_result: str
-    file_size: int
-
-
 class DocumentReference(BaseModel):
     model_config = ConfigDict(
+        validate_by_alias=True,
+        validate_by_name=True,
         alias_generator=to_pascal,
         use_enum_values=True,
-        populate_by_name=True,
     )
 
     id: str = Field(..., alias=str(DocumentReferenceMetadataFields.ID.value))
-    content_type: str
-    created: str
-    deleted: str
-    file_location: str
+    author: str = Field(default=None, exclude=True)
+    content_type: str = Field(default=DEFAULT_CONTENT_TYPE)
+    created: str = Field(
+        default_factory=lambda: datetime.now(timezone.utc).strftime(DATE_FORMAT)
+    )
+    document_scan_creation: str = Field(
+        default_factory=lambda: datetime.date(datetime.now()).isoformat(),
+    )
+    current_gp_ods: str = Field(default=None)
+    custodian: str = Field(default=None)
+    deleted: str = Field(default=None)
+    doc_status: Literal[
+        "registered",
+        "partial",
+        "preliminary",
+        "final",
+        "amended",
+        "corrected",
+        "appended",
+        "cancelled",
+        "entered-in-error",
+        "deprecated",
+        "unknown",
+    ] = Field(default="preliminary")
+    doc_type: str = Field(default=None, exclude=True)
+    document_snomed_code_type: Optional[str] = Field(
+        default=SnomedCodes.LLOYD_GEORGE.value.code
+    )
+    file_location: str = ""
     file_name: str
+    file_size: int = Field(default=None)
+    last_updated: int = Field(
+        default_factory=lambda: int(datetime.now(timezone.utc).timestamp()),
+    )
     nhs_number: str
+    s3_bucket_name: str = Field(exclude=True, default=None)
+    s3_file_key: str = Field(exclude=True, default=None)
+    status: Literal["current", "superseded", "entered-in-error"] = Field(
+        default="current"
+    )
+    sub_folder: str = Field(default=None, exclude=True)
     ttl: Optional[int] = Field(
         alias=str(DocumentReferenceMetadataFields.TTL.value), default=None
     )
-    virus_scanner_result: str
-    # Allow current_gp_ods to be nullable so that we can cope with existing records.
-    # After we updated all existing records with this field, consider to set this as non-Optional
-    current_gp_ods: Optional[str] = None
-    uploaded: bool
-    uploading: bool
-    last_updated: int
+    uploaded: bool = Field(default=False)
+    uploading: bool = Field(default=False)
+    version: str = Field(default="1")
+    virus_scanner_result: str = Field(default=None)
+
+    def model_dump_camel_case(self, *args, **kwargs):
+        model_dump_results = self.model_dump(*args, **kwargs)
+        camel_case_model_dump_results = {}
+        for key in model_dump_results:
+            camel_case_model_dump_results[to_camel(key)] = model_dump_results[key]
+        return camel_case_model_dump_results
+
+    @model_validator(mode="before")
+    @classmethod
+    def set_location_properties(cls, data, *args, **kwargs):
+        """Set S3 location properties based on available data."""
+        if "file_location" in data or "FileLocation" in data:
+            file_location = data.get("file_location") or data.get("FileLocation")
+            bucket, key = cls._parse_s3_location(file_location)
+            data["s3_bucket_name"] = bucket
+            data["s3_file_key"] = key
+        elif "s3_bucket_name" in data:
+            data["s3_file_key"] = cls._build_s3_key(data)
+            data["file_location"] = cls._build_s3_location(
+                data["s3_bucket_name"], data["s3_file_key"]
+            )
+        return data
 
+    @staticmethod
+    def _parse_s3_location(file_location: str) -> list[str]:
+        """Parse S3 location into bucket and key components."""
+        location_without_prefix = file_location.replace(S3_PREFIX, "")
+        return location_without_prefix.split("/", 1)
+
+    @staticmethod
+    def _build_s3_key(data: dict) -> str:
+        """Build the S3 key from document data."""
+        key_parts = []
+
+        if "sub_folder" in data:
+            key_parts.append(data["sub_folder"])
+            if "doc_type" in data:
+                key_parts.append(data["doc_type"])
+
+        key_parts.extend([data["nhs_number"], data["id"]])
+        s3_key = "/".join(key_parts)
+
+        return s3_key
+
+    @staticmethod
+    def _build_s3_location(bucket: str, key: str) -> str:
+        """Build a complete S3 location from bucket and key."""
+        normalized_key = key[1:] if key.startswith("/") else key
+        return f"{S3_PREFIX}{bucket}/{normalized_key}"
+
+    # File path handling methods
     def get_file_name_path(self):
         return pathlib.Path(self.file_name)
 
@@ -65,54 +152,21 @@ def get_base_name(self):
     def get_file_extension(self):
         return self.get_file_name_path().suffix
 
-    def get_file_bucket(self):
-        try:
-            file_bucket = self.file_location.replace("s3://", "").split("/")[0]
-            if file_bucket:
-                return file_bucket
-            raise InvalidDocumentReferenceException(
-                "Failed to parse bucket from file location"
-            )
-        except IndexError:
-            raise InvalidDocumentReferenceException(
-                "Failed to parse bucket from file location"
-            )
-
-    def get_file_key(self):
-        try:
-            file_key = self.file_location.replace("s3://", "").split("/", 1)[1]
-            if file_key:
-                return file_key
-            raise InvalidDocumentReferenceException(
-                "Failed to parse object key from file location"
-            )
-        except IndexError:
-            raise InvalidDocumentReferenceException(
-                "Failed to parse object key from file location"
-            )
-
     def create_unique_filename(self, duplicates: int):
         return f"{self.get_base_name()}({duplicates}){self.get_file_extension()}"
 
+    # Status methods
     def last_updated_within_three_minutes(self) -> bool:
-        three_minutes_ago = datetime.now(timezone.utc).timestamp() - 60 * 3
+        three_minutes_ago = (
+            datetime.now(timezone.utc).timestamp() - THREE_MINUTES_IN_SECONDS
+        )
         return self.last_updated >= three_minutes_ago
 
-    def __eq__(self, other):
-        if isinstance(other, DocumentReference):
-            return (
-                self.id == other.id
-                and self.content_type == other.content_type
-                and self.created == other.created
-                and self.deleted == other.deleted
-                and self.file_location == other.file_location
-                and self.file_name == other.file_name
-                and self.nhs_number == other.nhs_number
-                and self.ttl == other.ttl
-                and self.virus_scanner_result == other.virus_scanner_result
-                and self.current_gp_ods == other.current_gp_ods
-                and self.uploaded == other.uploaded
-                and self.uploading == other.uploading
-                and self.last_updated == other.last_updated
-            )
-        return False
+    def set_deleted(self) -> None:
+        self.deleted = datetime.now(timezone.utc).strftime(DATE_FORMAT)
+
+    def set_virus_scanner_result(self, updated_virus_scanner_result) -> None:
+        self.virus_scanner_result = updated_virus_scanner_result
+
+    def set_uploaded_to_true(self):
+        self.uploaded = True

lambdas/models/feedback_model.py

@@ -5,7 +5,7 @@
 
 
 class Feedback(BaseModel):
-    model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)
+    model_config = ConfigDict(validate_by_name=True, alias_generator=to_camel)
 
     feedback_content: str
     experience: str = Field(alias="howSatisfied")

lambdas/models/fhir/R4/fhir_document_reference.py

@@ -1,4 +1,3 @@
-import datetime
 from typing import Any, Dict, List, Literal, Optional
 
 from enums.snomed_codes import SnomedCode, SnomedCodes
@@ -113,7 +112,7 @@ class DocumentReference(BaseModel):
         "deprecated",
         "unknown",
     ] = "final"
-    status: Literal["current"] = "current"
+    status: Literal["current", "superseded"] = "current"
     type: Optional[CodeableConcept] = None
     category: Optional[List[CodeableConcept]] = None
     subject: Optional[Reference] = None
@@ -228,18 +227,23 @@ def create_fhir_document_reference_object(
 
         return DocumentReference(
             id=document.id,
+            docStatus=document.doc_status,
             type=CodeableConcept(
                 coding=self._create_snomed_coding(self.snomed_code_doc_type)
             ),
             subject=Reference(**self._create_identifier("nhs-number", self.nhs_number)),
             content=[DocumentReferenceContent(attachment=self.attachment)],
-            date=datetime.datetime.now().isoformat(),
+            date=document.created,
             author=[
                 Reference(
-                    **self._create_identifier("ods-organization-code", self.custodian)
+                    **self._create_identifier(
+                        "ods-organization-code", document.author or self.custodian
+                    )
                 )
             ],
             custodian=Reference(
-                **self._create_identifier("ods-organization-code", self.custodian)
+                **self._create_identifier(
+                    "ods-organization-code", document.custodian or self.custodian
+                )
             ),
         )