Refactor get by ID journey

Author: dlzhry2nhs

backend/src/controller/__init__.py

@@ -0,0 +1,28 @@
+"""Utility module for interacting with the AWS API Gateway event provided to controllers"""
+from typing import Optional
+
+from aws_lambda_typing.events import APIGatewayProxyEventV1
+
+from controller.constants import SUPPLIER_SYSTEM_HEADER_NAME
+from models.errors import UnauthorizedError
+from utils import dict_utils
+
+
+def get_path_parameter(event: APIGatewayProxyEventV1, param_name: str) -> str:
+    return dict_utils.get_field(
+        dict(event),
+        "pathParameters",
+        param_name,
+        default=""
+    )
+
+
+def get_supplier_system_header(event: APIGatewayProxyEventV1) -> str:
+    """Retrieves the supplier system header from the API Gateway event"""
+    supplier_system: Optional[str] = dict_utils.get_field(dict(event), "headers", SUPPLIER_SYSTEM_HEADER_NAME)
+
+    if supplier_system is None:
+        # SupplierSystem header must be provided for looking up permissions
+        raise UnauthorizedError()
+
+    return supplier_system

backend/src/controller/aws_apig_event_utils.py

@@ -0,0 +1,24 @@
+"""Utility module providing helper functions for dealing with response formats for AWS API Gateway"""
+import json
+from typing import Optional
+
+
+def create_response(
+    status_code: int,
+    body: Optional[dict | str] = None,
+    headers: Optional[dict] = None
+):
+    """Creates response body as per Lambda -> API Gateway proxy integration"""
+    if body is not None:
+        if isinstance(body, dict):
+            body = json.dumps(body)
+        if headers:
+            headers["Content-Type"] = "application/fhir+json"
+        else:
+            headers = {"Content-Type": "application/fhir+json"}
+
+    return {
+        "statusCode": status_code,
+        "headers": headers if headers else {},
+        **({"body": body} if body else {}),
+    }

backend/src/controller/aws_apig_response_utils.py

@@ -0,0 +1,5 @@
+"""FHIR Controller constants"""
+
+
+SUPPLIER_SYSTEM_HEADER_NAME = "SupplierSystem"
+E_TAG_HEADER_NAME = "E-Tag"

backend/src/controller/constants.py

@@ -0,0 +1,42 @@
+"""Module for the global FHIR API exception handler"""
+import functools
+import uuid
+from typing import Callable, Type
+
+from clients import logger
+from constants import GENERIC_SERVER_ERROR_DIAGNOSTICS_MESSAGE
+from controller.aws_apig_response_utils import create_response
+from models.errors import UnauthorizedVaxError, UnauthorizedError, ResourceNotFoundError, create_operation_outcome, \
+    Severity, Code
+
+
+_CUSTOM_EXCEPTION_TO_STATUS_MAP: dict[Type[Exception], int] = {
+    UnauthorizedError: 403,
+    UnauthorizedVaxError: 403,
+    ResourceNotFoundError: 404
+}
+
+
+def fhir_api_exception_handler(function: Callable) -> Callable:
+    """Decorator to handle any expected FHIR API exceptions or unexpected exception and provide a valid response to
+    the client"""
+
+    @functools.wraps(function)
+    def wrapper(*args, **kwargs):
+        try:
+            return function(*args, **kwargs)
+        except tuple(_CUSTOM_EXCEPTION_TO_STATUS_MAP) as exc:
+            status_code = _CUSTOM_EXCEPTION_TO_STATUS_MAP[type(exc)]
+            return create_response(status_code=status_code, body=exc.to_operation_outcome())
+        except Exception:  # pylint: disable = broad-exception-caught
+            logger.exception("Unhandled exception")
+            server_error = create_operation_outcome(
+                resource_id=str(uuid.uuid4()),
+                severity=Severity.error,
+                code=Code.server_error,
+                diagnostics=GENERIC_SERVER_ERROR_DIAGNOSTICS_MESSAGE,
+            )
+            return create_response(500, server_error)
+
+    return wrapper
+

backend/src/controller/fhir_api_exception_handler.py

@@ -3,8 +3,8 @@
 import pprint
 import uuid
 
-
-from fhir_controller import FhirController, make_controller
+from controller.aws_apig_response_utils import create_response
+from controller.fhir_controller import FhirController, make_controller
 from local_lambda import load_string
 from models.errors import Severity, Code, create_operation_outcome
 from log_structure import function_info
@@ -29,7 +29,7 @@ def create_immunization(event, controller: FhirController):
             code=Code.server_error,
             diagnostics=GENERIC_SERVER_ERROR_DIAGNOSTICS_MESSAGE,
         )
-        return FhirController.create_response(500, exp_error)
+        return create_response(500, exp_error)
 
 
 if __name__ == "__main__":

backend/src/fhir_batch_controller.py …/src/controller/fhir_batch_controller.pybackend/src/fhir_batch_controller.py renamed to backend/src/controller/fhir_batch_controller.py backend/src/fhir_batch_controller.py renamed to backend/src/controller/fhir_batch_controller.py

@@ -3,8 +3,8 @@
 import pprint
 import uuid
 
-
-from fhir_controller import FhirController, make_controller
+from controller.aws_apig_response_utils import create_response
+from controller.fhir_controller import FhirController, make_controller
 from models.errors import Severity, Code, create_operation_outcome
 from log_structure import function_info
 from constants import GENERIC_SERVER_ERROR_DIAGNOSTICS_MESSAGE
@@ -28,7 +28,7 @@ def delete_immunization(event, controller: FhirController):
             code=Code.server_error,
             diagnostics=GENERIC_SERVER_ERROR_DIAGNOSTICS_MESSAGE,
         )
-        return FhirController.create_response(500, exp_error)
+        return create_response(500, exp_error)
 
 
 if __name__ == "__main__":

backend/src/fhir_controller.py backend/src/controller/fhir_controller.pybackend/src/fhir_controller.py renamed to backend/src/controller/fhir_controller.py

@@ -102,19 +102,17 @@ def get_immunization_by_identifier(self, identifier_pk: str) -> tuple[Optional[d
         else:
             return None, None
 
-    def get_immunization_by_id(self, imms_id: str) -> Optional[dict]:
+    def get_immunization_by_id(self, imms_id: str) -> tuple[Optional[dict], Optional[str]]:
         response = self.table.get_item(Key={"PK": _make_immunization_pk(imms_id)})
         item = response.get("Item")
 
         if not item:
-            return None
+            return None, None
+
         if item.get("DeletedAt") and item["DeletedAt"] != "reinstated":
-            return None
+            return None, None
 
-        return {
-            "Resource": json.loads(item["Resource"]),
-            "Version": item["Version"]
-        }
+        return json.loads(item.get("Resource", {})), str(item.get("Version", ""))
 
     def get_immunization_by_id_all(self, imms_id: str, imms: dict) -> Optional[dict]:
         response = self.table.get_item(Key={"PK": _make_immunization_pk(imms_id)})