refactor fhir_parsers

Author: Akol125

lambdas/shared/src/common/validator/parsers/fhir_parser.py

@@ -1,105 +1,126 @@
-"""
-A parser for navigating and extracting data from FHIR JSON resources.
-
-Identifiy a path to a node using '|' to separate levels and
-"""
-
 import json
 
 
 class FHIRParser:
-    # parser variables
     def __init__(self):
-        self.fhir_resource = {}
+        self.fhir_resources = {}
 
-    # -------------------------------------------
-    # File Management
-    # used for files
-    def parse_fhir_file(self, fhir_file_name):
+    # This opens a file with FHIR resource data
+    def parse_fhir_file(self, fhir_file_name: str) -> None:
         with open(fhir_file_name) as json_file:
-            self.fhir_resource = json.load(json_file)
+            self.fhir_resources = json.load(json_file)
 
-    # used for JSON FHIR Resource data
-    def parse_fhir_data(self, fhir_data):
-        self.fhir_resource = fhir_data
+    # This is used for JSON FHIR Resource data events
+    def parse_fhir_data(self, fhir_data: dict) -> None:
+        self.fhir_resources = fhir_data
 
-    # ------------------------------------------------
-    # Scan and Identify
-    # scan for a key name or a value
-    def _scan_values_for_match(self, parent, match_value):
+    def _locate_fhir_item_in_dict(self, fhir_resource: dict, fhir_field: str) -> bool:
+        """
+        Checks whether a given FHIR field name or value exists in a FHIR dictionary.
+         :param fhir_resource: The FHIR resource as a dictionary.
+         :param fhir_field: The FHIR field name or value to search for.
+         :return: True if the field name or value matches a key item in the resource, False otherwise.
+        """
         try:
-            for key in parent:
-                if parent[key] == match_value:
+            for key in fhir_resource:
+                if fhir_resource[key] == fhir_field:
                     return True
             return False
         except Exception:
             return False
 
-    # locate an index for an item in a list
-    def _locate_list_id(self, parent, locator):
-        field_list = locator.split(":")
+    def _locate_fhir_item_in_list(self, fhir_resource: list, fhir_field: str) -> dict:
+        """
+        Locates and returns the first FHIR item (dictionary) in a list that contains
+        the specified FHIR field name or value.
+        :param fhir_resource: The FHIR resource as a list of dictionaries.
+        :param fhir_field: The FHIR field name or value to search for.
+        :return: The first matching FHIR item (dictionary) if found, otherwise an empty string.
+        """
+        field_list = fhir_field.split(":")
         node_id = 0
         index = 0
         try:
-            while index < len(parent):
-                for key in parent[index]:
-                    if (parent[index][key] == field_list[1]) or (key == field_list[1]):
+            while index < len(fhir_resource):
+                for key in fhir_resource[index]:
+                    if (fhir_resource[index][key] == field_list[1]) or (key == field_list[1]):
                         node_id = index
                         break
                     else:
-                        if self._scan_values_for_match(parent[index][key], field_list[1]):
+                        if self._locate_fhir_item_in_dict(fhir_resource[index][key], field_list[1]):
                             node_id = index
                             break
                 index += 1
         except Exception:
             return ""
-        return parent[node_id]
+        return fhir_resource[node_id]
 
     # identify a node in the FHIR data
-    def _get_node(self, parent, child):
-        # check for indices
+    def _extract_fhir_node_value(self, fhir_resource: dict | list, fhir_field_key: str) -> str:
+        """
+        Safely retrieves a value from a FHIR resource by key or index.
+        :param fhir_resource: The FHIR resource, which can be a dictionary or a list.
+        :param fhir_field_key: The key (string) or index (integer as string) to retrieve the value for.
+        :return: The value associated with the key or index, or an empty string if not found.
+        """
         try:
-            result = parent[child]
+            result = fhir_resource[fhir_field_key]
         except Exception:
             try:
-                child = int(child)
-                result = parent[child]
+                child = int(fhir_field_key)
+                result = fhir_resource[child]
             except Exception:
                 result = ""
         return result
 
-    # locate a value for a key
-    def _scan_for_value(self, fhir_fields):
-        field_list = fhir_fields.split("|")
+    def _resolve_fhir_path(self, fhir_field_path: str) -> str:
+        """
+        Resolves a FHIR value from a pipe-delimited FHIR path string.
+        This function navigates through FHIR resources stored in `self.fhir_resources`
+        using the provided path, which may include nested dictionaries or lists.
+        Fields prefixed with "#" indicate list-based lookups.
+        """
+        field_list = fhir_field_path.split("|")
+
         # get root field before we iterate
-        rootfield = self.fhir_resource[field_list[0]]
+        resource_per_field = self.fhir_resources[field_list[0]]
         del field_list[0]
         try:
             for field in field_list:
                 if field.startswith("#"):
-                    rootfield = self._locate_list_id(rootfield, field)  # check here for default index??
+                    resource_per_field = self._locate_fhir_item_in_list(
+                        resource_per_field, field
+                    )  # check here for default index??
                 else:
-                    rootfield = self._get_node(rootfield, field)
+                    resource_per_field = self._extract_fhir_node_value(resource_per_field, field)
         except Exception:
-            rootfield = ""
-        return rootfield
+            resource_per_field = ""
+        return resource_per_field
 
-    # get the value list for a key
-    def get_key_value(self, field_name):
+    def get_fhir_value_list(self, field_path: str) -> list[str]:
+        """
+        Retrieves one or more values from a FHIR resource and returns them as a list.
+        :param field_path: The FHIR field path to retrieve the values for.
+        :return: A list of values found at the specified FHIR field path.
+        """
         value = []
         try:
-            response_value = self._scan_for_value(field_name)
+            response_value = self._resolve_fhir_path(field_path)
         except Exception:
             response_value = ""
 
         value.append(response_value)
         return value
 
-    # get the value list for a key
-    def get_key_single_value(self, field_name):
+    def get_fhir_value(self, field_path: str) -> str:
+        """
+        Retrieves a single value from a FHIR resource.
+        :param field_path: The FHIR field path to retrieve the value for.
+        :return: The value as a string, or an empty string if not found.
+        """
         value = ""
         try:
-            response_value = self._scan_for_value(field_name)
+            response_value = self._resolve_fhir_path(field_path)
         except Exception:
             response_value = ""
 

lambdas/shared/src/common/validator/validator.py

@@ -98,10 +98,10 @@ def _validate_expression(
         if "parentExpression" in expression:
             parent_expression = expression["parentExpression"]
             if self._check_error_record_for_fail(parent_expression):
-                error_record = {
-                    "code": ExceptionLevels.PARENT_FAILED,
-                    "message": MESSAGES[ExceptionLevels.PARENT_FAILED] + ", Parent ID: " + parent_expression,
-                }
+                error_record = ErrorReport(
+                    code=ExceptionLevels.PARENT_FAILED,
+                    message=MESSAGES[ExceptionLevels.PARENT_FAILED] + ", Parent ID: " + parent_expression,
+                )
                 self._add_error_record(
                     error_record, expression_error_group, expression_name, expression_id, error_level
                 )
@@ -217,7 +217,7 @@ def run_validation(
     # Report Generation
     # Build the error Report
     def build_error_report(self, event_id):
-        occurrence_date_time = self.data_parser.get_key_single_value("occurrenceDateTime")
+        occurrence_date_time = self.data_parser.get_fhir_value("occurrenceDateTime")
         dq_reporter = DQReporter()
         dq_report = dq_reporter.generate_error_report(event_id, occurrence_date_time, self.error_records)
 

lambdas/shared/tests/test_common/validator/test_parser.py

@@ -14,15 +14,15 @@ def test_parse_fhir_key_exists(self):
 
         fhir_parser = FHIRParser()
         fhir_parser.parse_fhir_file(fhirFilePath)
-        my_value = fhir_parser.get_key_value("vaccineCode|coding|0|code")
+        my_value = fhir_parser.get_fhir_value_list("vaccineCode|coding|0|code")
         self.assertEqual(my_value, ["42223111000001107"])
 
     def test_parse_fhir_key_not_exists(self):
         fhirFilePath = self.fhir_data_folder / "vaccination.json"
 
         fhir_parser = FHIRParser()
         fhir_parser.parse_fhir_file(fhirFilePath)
-        my_value = fhir_parser.get_key_value("vaccineCode|coding|1")
+        my_value = fhir_parser.get_fhir_value_list("vaccineCode|coding|1")
         self.assertEqual(my_value, [""])
-        my_value = fhir_parser.get_key_value("vaccineCode|coding|1|codes")
+        my_value = fhir_parser.get_fhir_value_list("vaccineCode|coding|1|codes")
         self.assertEqual(my_value, [""])