[PRMP-1287] GetDocumentRef lambda

.github/workflows/base-lambdas-reusable-deploy-all.yml

@@ -389,6 +389,21 @@ jobs:
     secrets:
       AWS_ASSUME_ROLE: ${{ secrets.AWS_ASSUME_ROLE }}
 
+  deploy_get_document_reference_lambda:
+    name: Deploy get nrl document reference lambda
+    uses: ./.github/workflows/base-lambdas-reusable-deploy.yml
+    with:
+      environment: ${{ inputs.environment}}
+      python_version: ${{ inputs.python_version }}
+      build_branch: ${{ inputs.build_branch}}
+      sandbox: ${{ inputs.sandbox }}
+      lambda_handler_name: nrl_get_document_reference_handler
+      lambda_aws_name: GetDocumentReference
+      lambda_layer_names: 'core_lambda_layer'
+    secrets:
+      AWS_ASSUME_ROLE: ${{ secrets.AWS_ASSUME_ROLE }}
+
+
   deploy_edge_presign_lambda:
     name: Deploy edge presign cloudfront lambda
     uses: ./.github/workflows/base-lambdas-edge-deploy.yml

lambdas/enums/fhir/fhir_issue_type.py

@@ -0,0 +1,14 @@
+from enum import Enum
+
+
+class FhirIssueCoding(Enum):
+    INVALID = ("invalid", "Invalid")
+    FORBIDDEN = ("forbidden", "Forbidden")
+    NOT_FOUND = ("not-found", "Not Found")
+    EXCEPTION = ("exception", "Exception")
+
+    def code(self):
+        return self.value[0]
+
+    def display(self):
+        return self.value[1]

lambdas/enums/lambda_error.py

@@ -1,26 +1,29 @@
 from enum import Enum
 from typing import Optional
 
+from enums.fhir.fhir_issue_type import FhirIssueCoding
 from utils.error_response import ErrorResponse
 from utils.request_context import request_context
 
 
 class LambdaError(Enum):
-    def create_error_body(self, params: Optional[dict] = None) -> str:
+    def create_error_response(self, params: Optional[dict] = None) -> 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
         )
-        return error_response.create()
+        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) -> str:
+        return self.create_error_response(params).create()
+
     """
     Errors for SearchPatientException
     """
@@ -390,7 +393,29 @@ def to_str(self) -> str:
         "err_code": "US_5001",
         "message": "Dynamo client error",
     }
-
+    """
+       Errors for fhir get document reference lambda 
+    """
+    DocumentReferenceNotFound = {
+        "err_code": "NRL_DR_4041",
+        "message": "Document reference not found",
+        "fhir_coding": FhirIssueCoding.NOT_FOUND,
+    }
+    DocumentReferenceGeneralError = {
+        "err_code": "NRL_DR_4031",
+        "message": "An error occurred while fetching the document",
+        "fhir_coding": FhirIssueCoding.EXCEPTION,
+    }
+    DocumentReferenceUnauthorised = {
+        "err_code": "NRL_DR_4011",
+        "message": "User is unauthorised to view record",
+        "fhir_coding": FhirIssueCoding.FORBIDDEN,
+    }
+    DocumentReferenceInvalidRequest = {
+        "err_code": "NRL_DR_4001",
+        "message": "Invalid request",
+        "fhir_coding": FhirIssueCoding.INVALID,
+    }
     """
         Edge Lambda Errors
     """
@@ -475,6 +500,7 @@ def to_str(self) -> str:
         "message": "Client error",
         "err_code": "AB_XXXX",
         "interaction_id": "88888888-4444-4444-4444-121212121212",
+        "fhir_coding": FhirIssueCoding.FORBIDDEN,
     }
 
     """

lambdas/handlers/manage_nrl_pointer_handler.py

@@ -1,7 +1,7 @@
 import json
 
 from enums.nrl_sqs_upload import NrlActionTypes
-from models.nrl_fhir_document_reference import FhirDocumentReference
+from models.fhir.R4.nrl_fhir_document_reference import DocumentReferenceInfo
 from models.nrl_sqs_message import NrlSqsMessage
 from services.base.nhs_oauth_service import NhsOauthService
 from services.base.ssm_service import SSMService
@@ -49,16 +49,13 @@ def lambda_handler(event, context):
             )
             match nrl_message.action:
                 case NrlActionTypes.CREATE:
-                    document = (
-                        FhirDocumentReference(
-                            **nrl_verified_message,
-                            custodian=nrl_api_service.end_user_ods_code,
-                        )
-                        .build_fhir_dict()
-                        .json()
+                    document = DocumentReferenceInfo(
+                        **nrl_verified_message,
+                        custodian=nrl_api_service.end_user_ods_code,
+                    ).create_fhir_document_reference_object()
+                    nrl_api_service.create_new_pointer(
+                        document.model_dump(exclude_none=True)
                     )
-
-                    nrl_api_service.create_new_pointer(json.loads(document))
                 case NrlActionTypes.DELETE:
                     nrl_api_service.delete_pointer(
                         nrl_message.nhs_number, nrl_message.snomed_code_doc_type

lambdas/handlers/nrl_get_document_reference_handler.py

@@ -0,0 +1,51 @@
+from enums.lambda_error import LambdaError
+from services.dynamic_configuration_service import DynamicConfigurationService
+from services.nrl_get_document_reference_service import NRLGetDocumentReferenceService
+from utils.audit_logging_setup import LoggingService
+from utils.decorators.ensure_env_var import ensure_environment_variables
+from utils.decorators.handle_lambda_exceptions import handle_lambda_exceptions
+from utils.decorators.set_audit_arg import set_request_context_for_logging
+from utils.lambda_exceptions import NRLGetDocumentReferenceException
+from utils.lambda_response import ApiGatewayResponse
+
+logger = LoggingService(__name__)
+
+
+@handle_lambda_exceptions
+@set_request_context_for_logging
+@ensure_environment_variables(
+    names=[
+        "APPCONFIG_APPLICATION",
+        "APPCONFIG_CONFIGURATION",
+        "APPCONFIG_ENVIRONMENT",
+        "LLOYD_GEORGE_DYNAMODB_NAME",
+        "PRESIGNED_ASSUME_ROLE",
+        "CLOUDFRONT_URL",
+    ]
+)
+def lambda_handler(event, context):
+    document_id = event.get("pathParameters", {}).get("id", None)
+    bearer_token = event.get("headers", {}).get("Authorization", None)
+    configuration_service = DynamicConfigurationService()
+    configuration_service.set_auth_ssm_prefix()
+    try:
+        if not document_id or not bearer_token:
+            raise NRLGetDocumentReferenceException(
+                400, LambdaError.DocumentReferenceInvalidRequest
+            )
+        get_document_service = NRLGetDocumentReferenceService()
+        document_ref = get_document_service.handle_get_document_reference_request(
+            document_id, bearer_token
+        )
+
+        return ApiGatewayResponse(
+            status_code=200, body=document_ref, methods="GET"
+        ).create_api_gateway_response()
+    except NRLGetDocumentReferenceException as e:
+        return ApiGatewayResponse(
+            status_code=e.status_code,
+            body=e.error.create_error_response().create_error_fhir_response(
+                e.error.value.get("fhir_coding")
+            ),
+            methods="GET",
+        ).create_api_gateway_response()

lambdas/models/fhir/R4/base_models.py

@@ -0,0 +1,39 @@
+from typing import List, Optional
+
+from pydantic import BaseModel
+
+
+class Coding(BaseModel):
+    system: Optional[str] = "http://hl7.org/fhir/issue-type"
+    code: Optional[str] = None
+    display: Optional[str] = None
+
+
+class CodeableConcept(BaseModel):
+    coding: Optional[List[Coding]] = None
+
+
+class Extension(BaseModel):
+    valueCodeableConcept: Optional[CodeableConcept] = None
+    url: Optional[str] = None
+
+
+class Period(BaseModel):
+    id: Optional[str] = None
+    start: Optional[str] = None
+    end: Optional[str] = None
+
+
+class Identifier(BaseModel):
+    use: Optional[str] = None
+    type: Optional[CodeableConcept] = None
+    system: Optional[str] = None
+    value: Optional[str] = None
+    period: Optional[Period] = None
+
+
+class Reference(BaseModel):
+    reference: Optional[str] = None
+    type: Optional[str] = None
+    identifier: Optional[Identifier] = None
+    display: Optional[str] = None

lambdas/models/fhir/R4/nrl_fhir_document_reference.py

@@ -0,0 +1,157 @@
+from typing import List, Literal, Optional
+
+from enums.snomed_codes import SnomedCode, SnomedCodes
+from models.fhir.R4.base_models import (
+    CodeableConcept,
+    Coding,
+    Extension,
+    Period,
+    Reference,
+)
+from pydantic import BaseModel, ConfigDict
+from pydantic.alias_generators import to_camel
+
+
+class NRLFormatCode(Coding):
+    system: Literal["https://fhir.nhs.uk/England/CodeSystem/England-NRLFormatCode"] = (
+        "https://fhir.nhs.uk/England/CodeSystem/England-NRLFormatCode"
+    )
+    code: Literal["urn:nhs-ic:record-contact", "urn:nhs-ic:unstructured"] = (
+        "urn:nhs-ic:unstructured"
+    )
+    display: Literal["Contact details (HTTP Unsecured)", "Unstructured Document"] = (
+        "Unstructured document"
+    )
+
+
+class Attachment(BaseModel):
+    contentType: str = "application/pdf"
+    language: str = "en-GB"
+    url: Optional[str] = None
+    size: Optional[int] = None
+    hash: Optional[str] = None
+    title: Optional[str] = None
+    creation: Optional[str] = None
+
+
+class ContentStabilityExtensionCoding(Coding):
+    system: Literal[
+        "https://fhir.nhs.uk/England/CodeSystem/England-NRLContentStability"
+    ] = "https://fhir.nhs.uk/England/CodeSystem/England-NRLContentStability"
+    code: Literal["static", "dynamic"] = "static"
+    display: Literal["Static", "Dynamic"] = "Static"
+
+
+class ContentStabilityExtensionValueCodeableConcept(CodeableConcept):
+    coding: List[ContentStabilityExtensionCoding] = [ContentStabilityExtensionCoding()]
+
+
+class ContentStabilityExtension(Extension):
+    url: Literal[
+        "https://fhir.nhs.uk/England/StructureDefinition/Extension-England-ContentStability"
+    ] = "https://fhir.nhs.uk/England/StructureDefinition/Extension-England-ContentStability"
+    valueCodeableConcept: ContentStabilityExtensionValueCodeableConcept = (
+        ContentStabilityExtensionValueCodeableConcept()
+    )
+
+
+class DocumentReferenceContent(BaseModel):
+    attachment: Attachment
+    format: NRLFormatCode = NRLFormatCode()
+    extension: List[ContentStabilityExtension] = [ContentStabilityExtension()]
+
+
+class DocumentReferenceContext(BaseModel):
+    encounter: Optional[List[Reference]] = None
+    event: Optional[List[CodeableConcept]] = None
+    period: Optional[Period] = None
+    facilityType: Optional[CodeableConcept] = None
+    practiceSetting: CodeableConcept
+    sourcePatientInfo: Optional[Reference] = None
+    related: Optional[List[Reference]] = None
+
+
+class DocumentReference(BaseModel):
+    resourceType: Literal["DocumentReference"] = "DocumentReference"
+    status: Literal["current"] = "current"
+    type: Optional[CodeableConcept] = None
+    category: Optional[List[CodeableConcept]] = None
+    subject: Optional[Reference] = None
+    date: Optional[str] = None
+    author: Optional[List[Reference]] = None
+    authenticator: Optional[Reference] = None
+    custodian: Optional[Reference] = None
+    content: List[DocumentReferenceContent]
+    context: Optional[DocumentReferenceContext] = None
+
+
+class DocumentReferenceInfo(BaseModel):
+    model_config = ConfigDict(alias_generator=to_camel)
+    nhs_number: str
+    custodian: str
+    snomed_code_doc_type: SnomedCode = SnomedCodes.LLOYD_GEORGE.value
+    snomed_code_category: SnomedCode = SnomedCodes.CARE_PLAN.value
+    snomed_code_practice_setting: SnomedCode = (
+        SnomedCodes.GENERAL_MEDICAL_PRACTICE.value
+    )
+    attachment: Optional[Attachment] = Attachment()
+
+    def create_fhir_document_reference_object(self):
+        fhir_base_url = "https://fhir.nhs.uk/Id"
+        snomed_url = "http://snomed.info/sct"
+
+        fhir_document_ref = DocumentReference(
+            subject={
+                "identifier": {
+                    "system": fhir_base_url + "/nhs-number",
+                    "value": self.nhs_number,
+                }
+            },
+            custodian={
+                "identifier": {
+                    "system": fhir_base_url + "/ods-organization-code",
+                    "value": self.custodian,
+                },
+            },
+            type={
+                "coding": [
+                    {
+                        "system": snomed_url,
+                        "code": self.snomed_code_doc_type.code,
+                        "display": self.snomed_code_doc_type.display_name,
+                    }
+                ]
+            },
+            content=[{"attachment": self.attachment}],
+            category=[
+                {
+                    "coding": [
+                        {
+                            "system": snomed_url,
+                            "code": self.snomed_code_category.code,
+                            "display": self.snomed_code_category.display_name,
+                        }
+                    ]
+                }
+            ],
+            author=[
+                {
+                    "identifier": {
+                        "system": fhir_base_url + "/ods-organization-code",
+                        "value": self.custodian,
+                    }
+                }
+            ],
+            context={
+                "practiceSetting": {
+                    "coding": [
+                        {
+                            "system": snomed_url,
+                            "code": self.snomed_code_practice_setting.code,
+                            "display": self.snomed_code_practice_setting.display_name,
+                        }
+                    ]
+                }
+            },
+        )
+        return fhir_document_ref