paginate table query

Author: nhsdevws

.github/workflows/sonarcloud.yml

@@ -24,45 +24,6 @@ jobs:
       - name: Install poetry
         run: pip install poetry==2.1.4
 
-      - name: Debug PR Info
-        if: github.event_name == 'pull_request'
-        run: |
-          echo "PR Number: ${{ github.event.pull_request.number }}"
-          echo "PR Head SHA: ${{ github.event.pull_request.head.sha }}"
-          echo "PR Base SHA: ${{ github.event.pull_request.base.sha }}"
-          echo "Current SHA: ${{ github.sha }}"
-
-
-      - name: Debug Trigger Info
-        run: |
-          echo "=== WORKFLOW TRIGGER DEBUG ==="
-          echo "Event Name: ${{ github.event_name }}"
-          echo "Current SHA: ${{ github.sha }}"
-
-          if [ "${{ github.event_name }}" = "pull_request" ]; then
-            echo "=== PULL REQUEST DETAILS ==="
-            echo "PR Number: ${{ github.event.pull_request.number }}"
-            echo "PR Head SHA: ${{ github.event.pull_request.head.sha }}"
-            echo "PR Base SHA: ${{ github.event.pull_request.base.sha }}"
-            echo "PR Head Ref: ${{ github.event.pull_request.head.ref }}"
-            echo "PR Base Ref: ${{ github.event.pull_request.base.ref }}"
-            echo "✅ Using PR HEAD commit for analysis"
-          elif [ "${{ github.event_name }}" = "push" ]; then
-            echo "=== PUSH DETAILS ==="
-            echo "Branch: ${{ github.ref_name }}"
-            echo "Pushed SHA: ${{ github.sha }}"
-            echo "Before SHA: ${{ github.event.before }}"
-            echo "After SHA: ${{ github.event.after }}"
-            echo "✅ Using pushed commit for analysis"
-          else
-            echo "=== OTHER EVENT ==="
-            echo "Unknown event type: ${{ github.event_name }}"
-          fi
-
-
-      - name: Install poetry
-        run: pip install poetry==2.1.2
-
       - uses: actions/setup-python@v5
         with:
           python-version: 3.11

.gitignore

@@ -30,4 +30,3 @@ openapi.json
 
 devtools/volume/
 **/.coverage
-backend/src/1.json

.tool-versions

@@ -1,4 +1,3 @@
-
 poetry 2.1.2
 nodejs 23.11.0
 python 3.11.12

backend/poetry.lock

@@ -3,6 +3,7 @@
 import boto3
 import time
 import simplejson as json
+from clients import logger
 from dataclasses import dataclass
 import botocore.exceptions
 from boto3.dynamodb.conditions import Key, Attr
@@ -34,7 +35,7 @@ def _query_identifier(table, index, pk, identifier, is_present):
                 return queryresponse
 
             if retries > 6:
-                print(f"{identifier}: Crossed {retries} retries")
+                logger.info(f"{identifier}: Crossed {retries} retries")
 
             retries += 1
             # Delay time in milliseconds

backend/src/fhir_batch_repository.py

@@ -392,29 +392,20 @@ def delete_immunization(self, aws_event):
             return self.create_response(403, unauthorized.to_operation_outcome())
 
     def search_immunizations(self, aws_event: APIGatewayProxyEventV1) -> dict:
-        logger.info("SAW: search_immunizations.")
-        
         if response := self.authorize_request(aws_event):
             return response
-        logger.info("SAW: Authorised request")
+
         try:
-            logger.info("SAW: Processing search parameters")
             search_params = process_search_params(process_params(aws_event))
         except ParameterException as e:
             return self._create_bad_request(e.message)
         if search_params is None:
             raise Exception("Failed to parse parameters.")
 
-        logger.info("SAW: fhir_controller. search_params: %s", search_params)
         # Check vaxx type permissions- start
         try:
-            logger.info("SAW: fhir_controller. search_immunizations...5")
-            logger.info("SAW: aws_event=%s", aws_event)
             if aws_event.get("headers"):
-                logger.info("SAW: fhir_controller. search_immunizations...6")
                 supplier_system = self._identify_supplier_system(aws_event)
-                logger.info("SAW: Supplier system identified: %s", supplier_system)
-                logger.info("SAW: Get supplier permissions for: %s", supplier_system)
                 imms_vax_type_perms = get_supplier_permissions(supplier_system)
                 if len(imms_vax_type_perms) == 0:
                     raise UnauthorizedVaxError()
@@ -437,7 +428,7 @@ def search_immunizations(self, aws_event: APIGatewayProxyEventV1) -> dict:
         except UnauthorizedVaxError as unauthorized:
             return self.create_response(403, unauthorized.to_operation_outcome())
         # Check vaxx type permissions on the existing record - end
-        logger.info("SAW: Searching immunizations...")
+
         result = self.fhir_service.search_immunizations(
             search_params.patient_identifier,
             vax_type_perm,
@@ -635,7 +626,7 @@ def check_vaccine_type_permissions(self, aws_event):
             if len(supplier_system) == 0:
                 raise UnauthorizedSystemError()
             imms_vax_type_perms = get_supplier_permissions(supplier_system)
-            print(f" update imms = {imms_vax_type_perms}")
+            logger.info(f" update imms = {imms_vax_type_perms}")
             if len(imms_vax_type_perms) == 0:
                 raise UnauthorizedVaxError()
             # Return the values needed for later use
@@ -650,10 +641,8 @@ def check_vaccine_type_permissions(self, aws_event):
 
     @staticmethod
     def create_response(status_code, body=None, headers=None):
-        logger.info("SAW: Creating response with status code: %d", status_code)
         if body:
             if isinstance(body, dict):
-                logger.info("SAW: return body : %s", body)
                 body = json.dumps(body)
             if headers:
                 headers["Content-Type"] = "application/fhir+json"
@@ -668,10 +657,7 @@ def create_response(status_code, body=None, headers=None):
 
     @staticmethod
     def _identify_supplier_system(aws_event):
-        logger.info("SAW: fhir_controller. _identify_supplier_system...1")
-        headers = aws_event.get("headers", {})
-        logger.info("SAW: headers=%s", headers)
-        supplier_system = headers.get("SupplierSystem")
+        supplier_system = aws_event["headers"]["SupplierSystem"]
         if not supplier_system:
             raise UnauthorizedError("SupplierSystem header is missing")
         return supplier_system

backend/src/fhir_controller.py

@@ -1,3 +1,5 @@
+from urllib import response
+from responses import logger
 import simplejson as json
 import os
 import time
@@ -394,70 +396,49 @@ def delete_immunization(
 
     def find_immunizations(self, patient_identifier: str, vaccine_types: list):
         """it should find all of the specified patient's Immunization events for all of the specified vaccine_types"""
-        
-        # ✅ Add debug logging
-        import logging
-        logger = logging.getLogger(__name__)
-        
-        logger.info("SAW fi...1: find_immunizations called with patient_identifier: '%s', vaccine_types: %s", 
-                    patient_identifier, vaccine_types)
-        
-        # Create the patient PK and log it
-        patient_pk = _make_patient_pk(patient_identifier)
-        logger.info("SAW fi...2: patient_pk created: '%s'", patient_pk)
-
-        condition = Key("PatientPK").eq(patient_pk)
+        condition = Key("PatientPK").eq(_make_patient_pk(patient_identifier))
         is_not_deleted = Attr("DeletedAt").not_exists() | Attr("DeletedAt").eq("reinstated")
 
-        logger.info("SAW fi...2.1: is_not_deleted condition created: %s", is_not_deleted)
+        raw_items = self.get_all_items(condition, is_not_deleted)
 
-        logger.info("SAW fi...3: executing DynamoDB query on PatientGSI index")
-
-        response = self.table.query(
-            IndexName="PatientGSI",
-            KeyConditionExpression=condition,
-            FilterExpression=is_not_deleted,
-        )
-        
-        # ✅ Log the raw DynamoDB response
-        logger.info("SAW fi...4: DynamoDB query response - Count: %s, ScannedCount: %s", 
-                    response.get("Count", 0), response.get("ScannedCount", 0))
-        
-        if "Items" in response:
-            raw_items = response["Items"]
-            logger.info("SAW fi...5: total items returned from DynamoDB: %d", len(raw_items))
-            
-            # Log first few items for debugging
-            if raw_items:
-                logger.info("SAW fi...6: sample raw item keys: %s", list(raw_items[0].keys()))
-                logger.info("SAW fi...7: first few PatientSK values: %s",
-                        [item.get("PatientSK", "MISSING") for item in raw_items[:3]])
-            
+        if raw_items:    
             # Filter the response to contain only the requested vaccine types
             items = [x for x in raw_items if x["PatientSK"].split("#")[0] in vaccine_types]
 
-            logger.info("SAW fi...8: after vaccine_types filtering (%s): %d items", vaccine_types, len(items))
-
-            if items:
-                # Log the vaccine types found
-                found_vaccine_types = [item["PatientSK"].split("#")[0] for item in items]
-                logger.info("SAW fi...9: found vaccine types: %s", found_vaccine_types)
-            else:
-                # Debug why no items matched
-                all_vaccine_types = [item["PatientSK"].split("#")[0] for item in raw_items]
-                logger.warning("SAW fi...10: no items matched vaccine_types filter!")
-                logger.warning("SAW fi...11: requested vaccine_types: %s", vaccine_types)
-                logger.warning("SAW fi...12: available vaccine_types in data: %s", list(set(all_vaccine_types)))
-
             # Return a list of the FHIR immunization resource JSON items
             final_resources = [json.loads(item["Resource"]) for item in items]
-            logger.info("SAW fi...13: returning %d FHIR resources", len(final_resources))
 
             return final_resources
         else:
-            logger.error("SAW fi...14: No 'Items' key in DynamoDB response!")
-            logger.error("SAW fi...15: Response keys: %s", list(response.keys()))
-            raise UnhandledResponseError(message=f"Unhandled error. Query failed", response=response)
+            logger.warning("no items matched patient_identifier filter!")
+            return []
+
+    def get_all_items(self, condition, is_not_deleted):
+        """Query DynamoDB and paginate through all results."""
+        all_items = []
+        last_evaluated_key = None
+
+        while True:
+            query_args = {
+                "IndexName": "PatientGSI",
+                "KeyConditionExpression": condition,
+                "FilterExpression": is_not_deleted,
+            }
+            if last_evaluated_key:
+                query_args["ExclusiveStartKey"] = last_evaluated_key
+
+            response = self.table.query(**query_args)
+            if "Items" not in response:
+                raise UnhandledResponseError(message="No Items in DynamoDB response", response=response)
+  
+            items = response.get("Items", [])
+            all_items.extend(items)
+
+            last_evaluated_key = response.get("LastEvaluatedKey")
+            if not last_evaluated_key:
+                break
+
+        return all_items
 
     @staticmethod
     def _handle_dynamo_response(response):

backend/src/fhir_repository.py

@@ -288,59 +288,40 @@ def search_immunizations(
         Finds all instances of Immunization(s) for a specified patient which are for the specified vaccine type(s).
         Bundles the resources with the relevant patient resource and returns the bundle.
         """
-        logger.info("SAW: search_immunizations...1 nhs_number: %s, vaccine_types: %s, date_from: %s, date_to: %s", 
-                     nhs_number, vaccine_types, date_from, date_to)
         # TODO: is disease type a mandatory field? (I assumed it is)
         #  i.e. Should we provide a search option for getting Patient's entire imms history?
-
-        logger.info("SAW: search_immunizations...2 check nhs_number_mod11_check for nhs_number: %s", nhs_number)
         if not nhs_number_mod11_check(nhs_number):
             return create_diagnostics()
 
-        logger.info("SAW: search_immunizations...3 obtain resources for nhs_number: %s, vaccine_types: %s, date_from: %s, date_to: %s",
-                     nhs_number, vaccine_types, date_from, date_to)
         # Obtain all resources which are for the requested nhs number and vaccine type(s) and within the date range
         resources = [
             r
             for r in self.immunization_repo.find_immunizations(nhs_number, vaccine_types)
             if self.is_valid_date_from(r, date_from) and self.is_valid_date_to(r, date_to)
         ]
-        logger.info("SAW: search_immunizations...4 no of resources found: %d", len(resources))
+
         # Create the patient URN for the fullUrl field.
         # NOTE: This UUID is assigned when a SEARCH request is received and used only for referencing the patient
         # resource from immunisation resources within the bundle. The fullUrl value we are using is a urn (hence the
         # FHIR key name of "fullUrl" is somewhat misleading) which cannot be used to locate any externally stored
         # patient resource. This is as agreed with VDS team for backwards compatibility with Immunisation History API.
         patient_full_url = f"urn:uuid:{str(uuid4())}"
 
-        logger.info("SAW: search_immunizations...5 call get_contained_patient")
         imms_patient_record = get_contained_patient(resources[-1]) if resources else None
 
-        logger.info("SAW: search_immunizations...6 filter resources for search")
-        # log pretty resources for debug
-        logger.info("SAW: search_immunizations...7 resources: %s", [
-            Immunization.parse_obj(r).json(indent=2) for r in resources
-        ])
-# Filter and amend the immunization resources for the SEARCH response
+        # Filter and amend the immunization resources for the SEARCH response
         resources_filtered_for_search = [Filter.search(imms, patient_full_url) for imms in resources]
-        logger.info("SAW: search_immunizations...8 no of items in filtered resources: %d", len(resources_filtered_for_search))
+
         # Add bundle entries for each of the immunization resources
         entries = [
             BundleEntry(
                 resource=Immunization.parse_obj(imms),
                 search=BundleEntrySearch(mode="match"),
                 fullUrl=f"https://api.service.nhs.uk/immunisation-fhir-api/Immunization/{imms['id']}",
             )
-            # get imms and log for debug
             for imms in resources_filtered_for_search
         ]
 
-
-        logger.info("SAW: search_immunizations...9 entries created for bundle: %d", len(entries))
-
-        for entry in entries:
-            logger.debug("SAW: search_immunizations...10 entry: %s", entry.resource.json(indent=2))
-
         # Add patient resource if there is at least one immunization resource
         if len(resources) > 0:
             entries.append(

backend/src/fhir_service.py

@@ -20,7 +20,7 @@ def function_info(func):
     @wraps(func)
     def wrapper(*args, **kwargs):
         event = args[0] if args else {}
-        print(f"Event: {event}")
+        logger.info(f"Event: {event}")
         headers = event.get("headers", {})
         correlation_id = headers.get("X-Correlation-ID", "X-Correlation-ID not passed")
         request_id = headers.get("X-Request-ID", "X-Request-ID not passed")
@@ -40,7 +40,7 @@ def wrapper(*args, **kwargs):
         start = time.time()
         try:
             result = func(*args, **kwargs)
-            print(f"Result:{result}")
+            logger.info(f"Result:{result}")
             end = time.time()
             log_data["time_taken"] = f"{round(end - start, 5)}s"
             status = "500"
@@ -56,7 +56,7 @@ def wrapper(*args, **kwargs):
                         record = result_headers["Location"]
                 if result.get("body"):
                     ops_outcome = json.loads(result["body"])
-                    print(f"ops_outcome: {ops_outcome}")
+                    logger.info(f"ops_outcome: {ops_outcome}")
                     if ops_outcome.get("issue"):
                         outcome_body = ops_outcome["issue"][0]
                         status_code = outcome_body["code"]