parent 0719a0d

author Akol125 <akinola.olutola1@nhs.net> 1741732167 +0000
committer Robert Novac <robert.novac@airelogic.com> 1742314388 +0000

READ request- drop obfuscation for s-flag patient

Author: Akol125

backend/.envrc.default

@@ -1,3 +1,3 @@
-layout pyenv 3.10.12
+layout pyenv 3.10.16
 
-dotenv
+dotenv

backend/.gitignore

@@ -1,2 +1,7 @@
 build
 .venv
+.direnv
+.env
+.envrc
+terraform.tfstate
+.python-version

backend/src/fhir_service.py

@@ -76,7 +76,7 @@ def get_immunization_by_identifier(
             base_url = f"{get_service_url()}/Immunization"
             response = form_json(imms_resp, element, identifier, base_url)
             return response
-
+    
     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,
@@ -85,26 +85,19 @@ def get_immunization_by_id(self, imms_id: str, imms_vax_type_perms: str) -> Opti
         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", {}))
+        # Returns the Immunisation full resource with no obfuscation
+        resource = imms_resp.get("Resource", {})
+        imms_filtered_for_read = Filter.read(resource) if resource else {}
 
-        # 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")
 
         return {
             "Version": imms_resp.get("Version", ""),
-            "Resource": Immunization.parse_obj(imms_filtered_for_read_and_s_flag),
+            "Resource": Immunization.parse_obj(imms_filtered_for_read),
         }
 
     def get_immunization_by_id_all(self, imms_id: str, imms: dict) -> 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.
         """
         imms["id"] = imms_id

backend/src/filter.py

@@ -94,7 +94,6 @@ def add_use_to_identifier(imms: dict) -> dict:
 
 class Filter:
     """Functions for filtering a FHIR Immunization Resource"""
-
     @staticmethod
     def read(imms: dict) -> dict:
         """Apply filtering for READ request"""

backend/src/get_imms_handler.py

@@ -16,7 +16,7 @@ def get_imms_handler(event, context):
 
 def get_immunization_by_id(event, controller: FhirController):
     try:
-        return controller.get_immunization_by_id(event)
+        return controller.get_immunization_by_id(event)  
     except Exception:  # pylint: disable = broad-exception-caught
         exp_error = create_operation_outcome(
             resource_id=str(uuid.uuid4()),

backend/tests/test_fhir_service.py

@@ -157,7 +157,7 @@ def setUp(self):
         self.pds_service = create_autospec(PdsService)
         self.validator = create_autospec(ImmunizationValidator)
         self.fhir_service = FhirService(self.imms_repo, self.pds_service, self.validator)
-
+    
     def test_get_immunization_by_id(self):
         """it should find an Immunization by id"""
         imms_id = "an-id"
@@ -204,23 +204,7 @@ def test_get_immunization_by_id_patient_not_restricted(self):
 
         # Then
         self.assertEqual(actual_output["Resource"], expected_output)
-
-    def test_get_immunization_by_id_patient_restricted(self):
-        """it should return a filtered Immunization when patient is restricted"""
-        imms_id = "restricted_id"
-        immunization_data = load_json_data("completed_covid19_immunization_event.json")
-        filtered_immunization = load_json_data("completed_covid19_immunization_event_filtered_for_s_flag_and_read.json")
-        self.imms_repo.get_immunization_by_id.return_value = {"Resource": immunization_data}
-        patient_data = {"meta": {"security": [{"code": "R"}]}}
-        self.fhir_service.pds_service.get_patient_details.return_value = patient_data
-
-        # When
-        resp_imms = self.fhir_service.get_immunization_by_id(imms_id, "COVID19:read")
-        act_res = resp_imms["Resource"]
-        filtered_immunization_res = Immunization.parse_obj(filtered_immunization)
-        # Then
-        self.assertEqual(act_res, filtered_immunization_res)
-
+       
     def test_pre_validation_failed(self):
         """it should throw exception if Immunization is not valid"""
         imms_id = "an-id"

devtools/.terraform.lock.hcl

@@ -0,0 +1,78 @@
+import boto3
+
+# Get the DynamoDB table name
+TABLE_NAME = "imms-internal-dev-imms-events"
+
+if not TABLE_NAME:
+    raise ValueError("DYNAMODB_TABLE_NAME environment variable is not set")
+
+# Initialize DynamoDB client
+dynamodb = boto3.resource("dynamodb")
+table = dynamodb.Table(TABLE_NAME)
+
+def get_primary_keys():
+    """Retrieve the primary key schema of the table."""
+    response = table.key_schema
+    return [key["AttributeName"] for key in response]
+
+<<<<<<< HEAD
+=======
+def get_total_count():
+    """Get the total count of items in the table, handling pagination."""
+    total_count = 0
+    last_evaluated_key = None
+    
+    while True:
+        if last_evaluated_key:
+            response = table.scan(Select='COUNT', ExclusiveStartKey=last_evaluated_key)
+        else:
+            response = table.scan(Select='COUNT')
+        
+        total_count += response.get("Count", 0)
+        last_evaluated_key = response.get("LastEvaluatedKey")
+        
+        if not last_evaluated_key:
+            break
+    
+    return total_count
+
+>>>>>>> 6caf6a5f (added)
+def clear_dynamodb():
+    """Deletes all items from the DynamoDB table, handling pagination."""
+    print(f"Clearing DynamoDB table: {TABLE_NAME}")
+    
+    primary_keys = get_primary_keys()
+    if not primary_keys:
+        raise ValueError("Unable to retrieve primary key schema")
+    
+<<<<<<< HEAD
+=======
+    total_count = get_total_count()
+    print(f"Total items before deletion: {total_count}")
+    
+>>>>>>> 6caf6a5f (added)
+    deleted_count = 0
+    
+    while True:
+        scan = table.scan()
+        items = scan.get("Items", [])
+        
+        if not items:
+            break
+
+        with table.batch_writer() as batch:
+            for item in items:
+                key = {pk: item[pk] for pk in primary_keys}
+                batch.delete_item(Key=key)
+                deleted_count += 1
+
+        print(f"Deleted {len(items)} items...")
+
+    print(f"Total {deleted_count} items deleted from DynamoDB")
+
+if __name__ == "__main__":
+<<<<<<< HEAD
+    clear_dynamodb()
+=======
+    clear_dynamodb()
+>>>>>>> 6caf6a5f (added)