Merge branch 'master' into AMB-2144-mesh-integration

Author: amitkr-22

backend/src/fhir_service.py

@@ -19,11 +19,10 @@
 from models.field_names import FieldNames
 from models.errors import InvalidPatientId, CustomValidationError, UnhandledResponseError
 from models.fhir_immunization import ImmunizationValidator
-from models.utils.generic_utils import nhs_number_mod11_check, get_occurrence_datetime, create_diagnostics, form_json
+from models.utils.generic_utils import nhs_number_mod11_check, get_occurrence_datetime, create_diagnostics, form_json, get_contained_patient
 from models.constants import Constants
 from models.errors import MandatoryError
 from pds_service import PdsService
-from s_flag_handler import handle_s_flag
 from timer import timed
 from filter import Filter
 
@@ -79,27 +78,18 @@ def get_immunization_by_identifier(
 
     def get_immunization_by_id(self, imms_id: str, imms_vax_type_perms: str) -> Optional[dict]:
         """
-        Get an Immunization by its ID. Return None if not found. If the patient doesn't have an NHS number,
+        Get an Immunization by its ID. Return None if it is not found. If the patient doesn't have an NHS number,
         return the Immunization without calling PDS or checking S flag.
         """
         if not (imms_resp := self.immunization_repo.get_immunization_by_id(imms_id, imms_vax_type_perms)):
             return None
 
-        # Remove fields rom the imms resource which are not to be returned for read
-        imms_filtered_for_read = Filter.read(imms_resp.get("Resource", {}))
-
-        # Handle s-flag filtering, where applicable
-        if not (nhs_number := obtain_field_value(imms_filtered_for_read, FieldNames.patient_identifier_value)):
-            imms_filtered_for_read_and_s_flag = imms_filtered_for_read
-        else:
-            if patient := self.pds_service.get_patient_details(nhs_number):
-                imms_filtered_for_read_and_s_flag = handle_s_flag(imms_filtered_for_read, patient)
-            else:
-                raise UnhandledResponseError("unable to validate NHS number with downstream service")
+        # Returns the Immunisation full resource with no obfuscation
+        resource = imms_resp.get("Resource", {})
 
         return {
             "Version": imms_resp.get("Version", ""),
-            "Resource": Immunization.parse_obj(imms_filtered_for_read_and_s_flag),
+            "Resource": Immunization.parse_obj(resource),
         }
 
     def get_immunization_by_id_all(self, imms_id: str, imms: dict) -> Optional[dict]:
@@ -271,15 +261,18 @@ def process_patient_for_bundle(patient: dict):
         """
 
         # Remove unwanted top-level fields
-        fields_to_keep = ["id", "resourceType", "identifier", "birthDate"]
+        fields_to_keep = {"resourceType", "identifier"}
         new_patient = {k: v for k, v in patient.items() if k in fields_to_keep}
 
         # Remove unwanted identifier fields
-        new_identifiers = []
-        for identifier in new_patient["identifier"]:
-            identifier_fields_to_keep = ["system", "value"]
-            new_identifiers.append({k: v for k, v in identifier.items() if k in identifier_fields_to_keep})
-        new_patient["identifier"] = new_identifiers
+        identifier_fields_to_keep = {"system", "value"}
+        new_patient["identifier"] = [
+            {k: v for k, v in identifier.items() if k in identifier_fields_to_keep}
+            for identifier in new_patient.get("identifier", [])
+        ]
+
+        if new_patient["identifier"]:
+            new_patient["id"] = new_patient["identifier"][0].get("value")
 
         return new_patient
 
@@ -323,25 +316,22 @@ def search_immunizations(
             if self.is_valid_date_from(r, date_from) and self.is_valid_date_to(r, date_to)
         ]
 
-        # Check whether the Superseded NHS number present in PDS
-        if pds_patient := self.pds_service.get_patient_details(nhs_number):
-            if pds_patient["identifier"][0]["value"] != nhs_number:
-                return create_diagnostics()
-
         # 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())}"
 
+        imms_patient_record = get_contained_patient(resources[-1]) if resources else None
+
         # Filter and amend the immunization resources for the SEARCH response
-        resources_filtered_for_search = [Filter.search(imms, patient_full_url, pds_patient) for imms in resources]
+        resources_filtered_for_search = [Filter.search(imms, patient_full_url) for imms in resources]
 
         # Add bundle entries for each of the immunization resources
         entries = [
             BundleEntry(
-                resource=Immunization.parse_obj(handle_s_flag(imms, pds_patient)),
+                resource=Immunization.parse_obj(imms),
                 search=BundleEntrySearch(mode="match"),
                 fullUrl=f"https://api.service.nhs.uk/immunisation-fhir-api/Immunization/{imms['id']}",
             )
@@ -352,7 +342,7 @@ def search_immunizations(
         if len(resources) > 0:
             entries.append(
                 BundleEntry(
-                    resource=self.process_patient_for_bundle(pds_patient),
+                    resource=self.process_patient_for_bundle(imms_patient_record),
                     search=BundleEntrySearch(mode="include"),
                     fullUrl=patient_full_url,
                 )

backend/src/filter.py

@@ -1,6 +1,6 @@
 """Functions for filtering a FHIR Immunization Resource"""
 
-from models.utils.generic_utils import is_actor_referencing_contained_resource, get_contained_practitioner
+from models.utils.generic_utils import is_actor_referencing_contained_resource, get_contained_practitioner, get_contained_patient
 from constants import Urls
 
 
@@ -14,7 +14,7 @@ def remove_reference_to_contained_practitioner(imms: dict) -> dict:
 
     # Remove reference to the contained practitioner from imms[performer]
     imms["performer"] = [
-        x for x in imms["performer"] if not is_actor_referencing_contained_resource(x, contained_practitioner["id"])
+         x for x in imms["performer"] if not is_actor_referencing_contained_resource(x, contained_practitioner["id"])
     ]
 
     return imms
@@ -96,25 +96,11 @@ class Filter:
     """Functions for filtering a FHIR Immunization Resource"""
 
     @staticmethod
-    def read(imms: dict) -> dict:
-        """Apply filtering for READ request"""
-        imms.pop("identifier")
-        return imms
-
-    @staticmethod
-    def search(imms: dict, patient_full_url: str, bundle_patient: dict = None) -> dict:
+    def search(imms: dict, patient_full_url: str) -> dict:
         """Apply filtering for an individual FHIR Immunization Resource as part of SEARCH request"""
         imms = remove_reference_to_contained_practitioner(imms)
-        imms.pop("contained")
-        imms["patient"] = create_reference_to_patient_resource(patient_full_url, bundle_patient)
+        imms["patient"] = create_reference_to_patient_resource(patient_full_url, get_contained_patient(imms))
         imms = add_use_to_identifier(imms)
-        return imms
+        imms.pop("contained")
 
-    @staticmethod
-    def s_flag(imms: dict) -> dict:
-        """Apply filtering for patients with 'RESTRICTED' flag"""
-        imms = replace_address_postal_codes(imms)
-        imms = replace_organization_values(imms)
-        if imms.get("location"):
-            imms["location"] = {"identifier": {"system": "urn:iso:std:iso:3166", "value": "GB"}}
         return imms