[PRMP-589] Create SearchDocumentReview lambda logic

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

@@ -682,6 +682,19 @@ jobs:
       lambda_layer_names: "core_lambda_layer"
     secrets:
       AWS_ASSUME_ROLE: ${{ secrets.AWS_ASSUME_ROLE }}
+
+  deploy_search_document_review_lambda:
+    name: Deploy Search Document Review
+    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: search_document_review_handler
+      lambda_aws_name: SearchDocumentReview
+    secrets:
+      AWS_ASSUME_ROLE: ${{ secrets.AWS_ASSUME_ROLE }}
 
   deploy_get_document_reference_by_id_lambda:
     name: Deploy get_document_reference_lambda

lambdas/enums/document_review_accepted_querystring_parameters.py

@@ -0,0 +1,9 @@
+from enum import StrEnum
+
+
+class DocumentReviewQuerystringParameters(StrEnum):
+    LIMIT = "limit"
+    NEXT_PAGE_TOKEN = "nextPageToken"
+    UPLOADER = "uploader"
+    NHS_NUMBER = "nhsNumber"
+

lambdas/enums/lambda_error.py

@@ -60,30 +60,29 @@ def create_error_body(self, params: Optional[dict] = None, **kwargs) -> str:
         "fhir_coding": UKCoreSpineError.VALIDATION_ERROR,
     }
 
-
     """
          Errors for /DocumentReference
     """
     DocRefNoBody = {
         "err_code": "DR_4001",
         "message": "Missing event body",
         "fhir_coding": FhirIssueCoding.REQUIRED,
-        }
+    }
     DocRefPayload = {
         "err_code": "DR_4002",
         "message": "Invalid json in body",
         "fhir_coding": FhirIssueCoding.INVALID,
-        }
+    }
     DocRefProps = {
         "err_code": "DR_4003",
         "message": "Request body missing some properties",
-        "fhircoding": FhirIssueCoding.REQUIRED
+        "fhircoding": FhirIssueCoding.REQUIRED,
     }
     DocRefInvalidFiles = {
         "err_code": "DR_4004",
         "message": "Invalid files or id",
-        "fhir_coding": FhirIssueCoding.INVALID
-        }
+        "fhir_coding": FhirIssueCoding.INVALID,
+    }
     DocRefNoParse = {
         "err_code": "DR_4005",
         "message": "Failed to parse document upload request data",
@@ -117,7 +116,7 @@ def create_error_body(self, params: Optional[dict] = None, **kwargs) -> str:
     DocRefUploadInternalError = {
         "err_code": "DR_5002",
         "message": "An error occurred when creating pre-signed url for document reference",
-        "fhir_coding": FhirIssueCoding.EXCEPTION
+        "fhir_coding": FhirIssueCoding.EXCEPTION,
     }
     DocRefPatientSearchInvalid = {
         "err_code": "DR_5003",
@@ -136,12 +135,12 @@ def create_error_body(self, params: Optional[dict] = None, **kwargs) -> str:
     UpdateDocNHSNumberMismatch = {
         "err_code": "UDR_5005",
         "message": "NHS number did not match",
-        "fhir_coding": FhirIssueCoding.INVARIANT
+        "fhir_coding": FhirIssueCoding.INVARIANT,
     }
     UpdateDocNotLatestVersion = {
         "err_code": "UDR_5006",
         "message": "Document is not the latest version",
-        "fhir_coding": FhirIssueCoding.INVARIANT
+        "fhir_coding": FhirIssueCoding.INVARIANT,
     }
 
     """
@@ -630,3 +629,26 @@ def create_error_body(self, params: Optional[dict] = None, **kwargs) -> str:
         "message": "An internal server error occurred",
         "fhir_coding": FhirIssueCoding.EXCEPTION,
     }
+
+    """
+        Errors for SearchDocumentReviewReference exceptions
+    """
+    DocumentReviewDB = {
+        "err_code": "SDR_5001",
+        "message": RETRIEVE_DOCUMENTS,
+    }
+
+    DocumentReviewValidation = {
+        "err_code": "SDR_5002",
+        "message": "Review document model error",
+    }
+
+    SearchDocumentReviewMissingODS = {
+        "err_code": "SDR_4001",
+        "message": "Missing ODS code in request context",
+    }
+
+    SearchDocumentInvalidQuerystring = {
+        "err_code": "SDR_4002",
+        "message": "Invalid query string passed",
+    }

lambdas/handlers/search_document_review_handler.py

@@ -0,0 +1,118 @@
+import json
+
+from enums.document_review_accepted_querystring_parameters import (
+    DocumentReviewQuerystringParameters,
+)
+from enums.feature_flags import FeatureFlags
+from enums.lambda_error import LambdaError
+from services.feature_flags_service import FeatureFlagService
+from services.search_document_review_service import SearchDocumentReviewService
+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.override_error_check import override_error_check
+from utils.decorators.set_audit_arg import set_request_context_for_logging
+from utils.exceptions import OdsErrorException
+from utils.lambda_exceptions import DocumentReviewException
+from utils.lambda_response import ApiGatewayResponse
+from utils.request_context import request_context
+
+logger = LoggingService(__name__)
+
+
+@set_request_context_for_logging
+@ensure_environment_variables(names=["DOCUMENT_REVIEW_DYNAMODB_NAME"])
+@override_error_check
+@handle_lambda_exceptions
+def lambda_handler(event, context):
+    """
+        Lambda handler for searching documents pending review by custodian.
+        Triggered by GET request to /SearchDocumentReview endpoint.
+    Args:
+        event: API Gateway event containing query optional query string parameters
+               QueryStringParameters: limit - Limit for DynamoDB query, defaulted to 50 if not provided,
+                                      nhsNumber - Patient NHS number, used for filtering DynamoDB results,
+                                      uploader - Author ODS code, used for filtering DynamoDB results,
+                                      nextPageToken - Encoded exclusive start key used to query DynamoDB for next page of results.
+        context: Lambda context
+
+    Returns:
+        API Gateway response containing Document Review references, number of reference returned, and next page token if present.
+        401 - No ODS code or auth token provided in request.
+        500 - Document Review Error, Internal Server Error.
+
+    """
+    try:
+        feature_flag_service = FeatureFlagService()
+        upload_lambda_enabled_flag_object = (
+            feature_flag_service.get_feature_flags_by_flag(
+                FeatureFlags.UPLOAD_DOCUMENT_ITERATION_3_ENABLED
+            )
+        )
+
+        if not upload_lambda_enabled_flag_object[
+            FeatureFlags.UPLOAD_DOCUMENT_ITERATION_3_ENABLED
+        ]:
+            logger.info("Feature flag not enabled, event will not be processed")
+            raise DocumentReviewException(403, LambdaError.FeatureFlagDisabled)
+
+        ods_code = get_ods_code_from_request_context()
+
+        params = parse_querystring_parameters(event)
+
+        search_document_reference_service = SearchDocumentReviewService()
+
+        references, next_page_token = search_document_reference_service.process_request(
+            params=params, ods_code=ods_code
+        )
+
+        response = {"documentReviewReferences": references, "count": len(references)}
+
+        if next_page_token:
+            response["nextPageToken"] = next_page_token
+
+        return ApiGatewayResponse(
+            status_code=200,
+            body=json.dumps(response),
+            methods="GET",
+        ).create_api_gateway_response()
+
+    except OdsErrorException as e:
+        logger.error(e)
+        return ApiGatewayResponse(
+            status_code=401,
+            body=LambdaError.SearchDocumentReviewMissingODS.create_error_body(),
+            methods="GET",
+        ).create_api_gateway_response()
+
+
+def get_ods_code_from_request_context():
+    logger.info("Getting ODS code from request context")
+    try:
+        ods_code = request_context.authorization.get("selected_organisation", {}).get(
+            "org_ods_code"
+        )
+        if not ods_code:
+            raise OdsErrorException()
+
+        return ods_code
+
+    except AttributeError as e:
+        logger.error(e)
+        raise DocumentReviewException(401, LambdaError.SearchDocumentReviewMissingODS)
+
+
+def parse_querystring_parameters(event):
+    logger.info("Parsing query string parameters.")
+    params = event.get("queryStringParameters", {})
+
+    extracted_params = {}
+
+    if not params:
+        return extracted_params
+
+    for param in DocumentReviewQuerystringParameters:
+        if param in params:
+            extracted_params[param.value] = params.get(param)
+
+    return extracted_params

lambdas/services/authoriser_service.py

@@ -117,6 +117,9 @@ def deny_access_policy(self, path, user_role, nhs_number: str = None):
             case "/Feedback":
                 deny_resource = False
 
+            case "/DocumentReview":
+                deny_resource = False
+
             case "/DocumentStatus":
                 deny_resource = (
                     not patient_access_is_allowed or is_user_gp_clinical or is_user_pcse

lambdas/services/base/dynamo_service.py

@@ -36,15 +36,33 @@ def get_table(self, table_name: str):
             logger.error(str(e), {"Result": "Unable to connect to DB"})
             raise e
 
-    def query_table(
+    def query_table_single(
         self,
-        table_name,
-        search_key,
+        table_name: str,
+        search_key: str,
         search_condition: str,
         index_name: str | None = None,
-        requested_fields: list[str] = None,
-        query_filter: Attr | ConditionBase = None,
-    ) -> list[dict]:
+        requested_fields: list[str] | None = None,
+        query_filter: Attr | ConditionBase | None = None,
+        limit: int | None = None,
+        start_key: dict | None = None,
+    ) -> dict:
+        """
+        Execute a single DynamoDB query and return the full response.
+
+        Args:
+            table_name: Name of the DynamoDB table
+            search_key: The partition key name to search on
+            search_condition: The value to match for the search key
+            index_name: Optional GSI/LSI name
+            requested_fields: Optional list of fields to project
+            query_filter: Optional filter expression
+            limit: Optional limit on number of items to return
+            start_key: Optional exclusive start key for pagination
+
+        Returns:
+            Full DynamoDB query response including Items, LastEvaluatedKey, etc.
+        """
         try:
             table = self.get_table(table_name)
 
@@ -54,30 +72,75 @@ def query_table(
 
             if index_name:
                 query_params["IndexName"] = index_name
+
             if requested_fields:
                 projection_expression = ",".join(requested_fields)
                 query_params["ProjectionExpression"] = projection_expression
+
             if query_filter:
                 query_params["FilterExpression"] = query_filter
-            items = []
-            while True:
-                results = table.query(**query_params)
 
-                if results is None or "Items" not in results:
-                    logger.error(f"Unusable results in DynamoDB: {results!r}")
-                    raise DynamoServiceException("Unrecognised response from DynamoDB")
+            if start_key:
+                query_params["ExclusiveStartKey"] = start_key
 
-                items += results["Items"]
+            if limit:
+                query_params["Limit"] = limit
 
-                if "LastEvaluatedKey" in results:
-                    query_params["ExclusiveStartKey"] = results["LastEvaluatedKey"]
-                else:
-                    break
-            return items
+            return table.query(**query_params)
         except ClientError as e:
             logger.error(str(e), {"Result": f"Unable to query table: {table_name}"})
             raise e
 
+    def query_table(
+        self,
+        table_name: str,
+        search_key: str,
+        search_condition: str,
+        index_name: str | None = None,
+        requested_fields: list[str] | None = None,
+        query_filter: Attr | ConditionBase | None = None,
+    ) -> list[dict]:
+        """
+        Execute a DynamoDB query and automatically paginate through all results.
+
+        Args:
+            table_name: Name of the DynamoDB table
+            search_key: The partition key name to search on
+            search_condition: The value to match for the search key
+            index_name: Optional GSI/LSI name
+            requested_fields: Optional list of fields to project
+            query_filter: Optional filter expression
+
+        Returns:
+            List of all items from paginated query results
+        """
+        items = []
+        start_key = None
+
+        while True:
+            results = self.query_table_single(
+                table_name=table_name,
+                search_key=search_key,
+                search_condition=search_condition,
+                index_name=index_name,
+                requested_fields=requested_fields,
+                query_filter=query_filter,
+                start_key=start_key,
+            )
+
+            if results is None or "Items" not in results:
+                logger.error(f"Unusable results in DynamoDB: {results!r}")
+                raise DynamoServiceException("Unrecognised response from DynamoDB")
+
+            items += results["Items"]
+
+            if "LastEvaluatedKey" in results:
+                start_key = results["LastEvaluatedKey"]
+            else:
+                break
+
+        return items
+
     def create_item(self, table_name, item):
         try:
             table = self.get_table(table_name)