Update empty identifier search bundle url

Also moved some common code that was only used in the api service layer to the relevant place.

Author: edhall-nhs

lambdas/backend/src/service/fhir_service.py

@@ -4,7 +4,7 @@
 import os
 import urllib.parse
 import uuid
-from typing import Optional, cast
+from typing import Any, Optional, cast
 from uuid import uuid4
 
 from fhir.resources.R4B.bundle import (
@@ -36,7 +36,6 @@
 from common.models.utils.generic_utils import (
     get_contained_patient,
     get_occurrence_datetime,
-    make_search_bundle,
 )
 from common.models.utils.validation_utils import (
     get_vaccine_type,
@@ -103,7 +102,7 @@ def get_immunization_by_identifier(
         resource, resource_metadata = self.immunization_repo.get_immunization_by_identifier(identifier)
 
         if not resource:
-            return make_search_bundle(resource, None, elements, identifier, base_url)
+            return self.make_empty_identifier_search_bundle(base_url)
 
         vaccination_type = get_vaccine_type(resource)
 
@@ -113,7 +112,9 @@ def get_immunization_by_identifier(
         patient_full_url = f"urn:uuid:{str(uuid4())}"
         filtered_resource = Filter.search(resource, patient_full_url)
 
-        return make_search_bundle(filtered_resource, resource_metadata.resource_version, elements, identifier, base_url)
+        return self.make_identifier_search_bundle(
+            filtered_resource, resource_metadata.resource_version, elements, identifier, base_url
+        )
 
     def get_immunization_and_version_by_id(self, imms_id: str, supplier_system: str) -> tuple[Immunization, str]:
         """
@@ -336,6 +337,58 @@ def is_valid_date_to(self, immunization: dict, date_to: Optional[datetime.date])
 
         return occurrence_datetime.date() <= date_to
 
+    def make_identifier_search_bundle(
+        self,
+        resource: Optional[dict],
+        version_id: Optional[int],
+        elements: Optional[set[str]],
+        identifier: Identifier,
+        base_url: str,
+    ) -> FhirBundle:
+        searched_url = f"{base_url}?identifier={identifier.system}|{identifier.value}" + (
+            f"&_elements={','.join(sorted(elements))}" if elements else ""
+        )
+
+        if not resource:
+            return self.make_empty_identifier_search_bundle(searched_url)
+
+        meta = {"versionId": version_id}
+
+        # Full Immunization payload to be returned if only the identifier parameter was provided and truncated when
+        # _elements is used
+        if elements is not None:
+            resource_for_bundle: dict[str, Any] = {"resourceType": "Immunization"}
+            if "id" in elements:
+                resource_for_bundle["id"] = resource["id"]
+            if "meta" in elements:
+                resource_for_bundle["meta"] = meta
+
+        else:
+            resource_for_bundle = copy.deepcopy(resource)
+            resource_for_bundle["meta"] = meta
+
+        entry = BundleEntry.construct(
+            fullUrl=f"{base_url}/{resource['id']}",
+            resource=(
+                Immunization.construct(**resource_for_bundle)
+                if elements
+                else Immunization.parse_obj(resource_for_bundle)
+            ),
+            search=BundleEntrySearch.construct(mode="match") if not elements else None,
+        )
+
+        return FhirBundle(
+            type="searchset",
+            link=[BundleLink(relation="self", url=searched_url)],
+            entry=[entry],
+            total=1,
+        )
+
+    @staticmethod
+    def make_empty_identifier_search_bundle(base_url: str) -> FhirBundle:
+        no_results_url = f"{base_url}?identifier=None"
+        return FhirBundle(entry=[], link=[BundleLink(relation="self", url=no_results_url)], type="searchset", total=0)
+
     @staticmethod
     def process_patient_for_bundle(patient: dict):
         """

lambdas/backend/tests/service/test_fhir_service.py

@@ -280,8 +280,7 @@ def test_get_immunization_by_identifier_returns_empty_search_when_not_found(self
             result.link[0],
             BundleLink.construct(
                 relation="self",
-                url="https://internal-dev.api.service.nhs.uk/immunisation-fhir-api/FHIR/R4/Immunization?identifier=some-"
-                "system|some-value",
+                url="https://internal-dev.api.service.nhs.uk/immunisation-fhir-api/FHIR/R4/Immunization?identifier=None",
             ),
         )
         self.assertEqual(result.entry, [])

lambdas/shared/src/common/models/utils/generic_utils.py

@@ -1,21 +1,10 @@
 """Generic utilities"""
 
 import base64
-import copy
 import datetime
 import urllib.parse
-from typing import Any, Literal, Optional
-
-from fhir.resources.R4B.bundle import (
-    Bundle as FhirBundle,
-)
-from fhir.resources.R4B.bundle import (
-    BundleEntry,
-    BundleEntrySearch,
-    BundleLink,
-)
-from fhir.resources.R4B.identifier import Identifier
-from fhir.resources.R4B.immunization import Immunization
+from typing import Literal, Optional
+
 from stdnum.verhoeff import validate
 
 from common.models.constants import Constants
@@ -155,56 +144,6 @@ def create_diagnostics_error(value):
     return exp_error
 
 
-def make_empty_search_bundle(searched_url: str) -> FhirBundle:
-    return FhirBundle(entry=[], link=[BundleLink(relation="self", url=searched_url)], type="searchset", total=0)
-
-
-# A lot of stuff in here is not very generic. Consider moving to relevant lambdas
-def make_search_bundle(
-    resource: Optional[dict],
-    version_id: Optional[int],
-    elements: Optional[set[str]],
-    identifier: Identifier,
-    base_url: str,
-) -> FhirBundle:
-    searched_url = f"{base_url}?identifier={identifier.system}|{identifier.value}" + (
-        f"&_elements={','.join(sorted(elements))}" if elements else ""
-    )
-
-    if not resource:
-        return make_empty_search_bundle(searched_url)
-
-    meta = {"versionId": version_id}
-
-    # Full Immunization payload to be returned if only the identifier parameter was provided and truncated when
-    # _elements is used
-    if elements is not None:
-        resource_for_bundle: dict[str, Any] = {"resourceType": "Immunization"}
-        if "id" in elements:
-            resource_for_bundle["id"] = resource["id"]
-        if "meta" in elements:
-            resource_for_bundle["meta"] = meta
-
-    else:
-        resource_for_bundle = copy.deepcopy(resource)
-        resource_for_bundle["meta"] = meta
-
-    entry = BundleEntry.construct(
-        fullUrl=f"{base_url}/{resource['id']}",
-        resource=(
-            Immunization.construct(**resource_for_bundle) if elements else Immunization.parse_obj(resource_for_bundle)
-        ),
-        search=BundleEntrySearch.construct(mode="match") if not elements else None,
-    )
-
-    return FhirBundle(
-        type="searchset",
-        link=[BundleLink(relation="self", url=searched_url)],
-        entry=[entry],
-        total=1,
-    )
-
-
 def check_keys_in_sources(event, not_required_keys):
     # Decode and parse the body, assuming it is JSON and base64-encoded
     def decode_and_parse_body(encoded_body):