refactoring fhirpath to use a python dictionary

Author: Akol125

lambdas/shared/Makefile

@@ -3,7 +3,7 @@ TEST_ENV := @PYTHONPATH=src:tests:src/common
 test:
 	$(TEST_ENV) python -m unittest discover -s tests -p "test_*.py" -v
 
-testv:
+test-validator:
 	$(TEST_ENV) python -m unittest discover -s tests/test_common/validator -p "test_*.py" -v
 
 test-list:

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

@@ -7,8 +7,8 @@
                     ("nhs_number,name,age").
 
 Example:
-    >>> parser = CSVRowParser()
-    >>> parser.parse_row_to_dict("9011011,Tom,32", "nhs_number,name,age")
+    >>> parser = CSVLineParser()
+    >>> parser.parse_csv_line("9011011,Tom,32", "nhs_number,name,age")
     >>> parser.csv_file_data
     {'nhs_number': '9011011', 'name': 'Tom', 'age': '32'}
 """

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

@@ -10,7 +10,7 @@ def parse_fhir_file(self, fhir_file_name: str) -> None:
         with open(fhir_file_name) as json_file:
             self.fhir_resources = json.load(json_file)
 
-    # This is used for JSON FHIR Resource data events
+    # This is used to parse FHIR data events passed in as a dictionary
     def parse_fhir_data(self, fhir_data: dict) -> None:
         self.fhir_resources = fhir_data
 

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

@@ -18,41 +18,44 @@ class Validator:
     def __init__(self, schema_file="", data_type: DataType = None, filepath=""):
         self.filepath = filepath
         self.json_data = {}
+        self.fhir_data = {}
         self.schema_file = schema_file
         self.csv_row = ""
         self.csv_header = ""
         self.data_type = data_type
         self.data_parser = ""
         self.error_records: list[ErrorReport] = []
 
-    def _get_csv_line_parser(self, csv_row, csv_header):
-        csv_parser = CSVLineParser()
-        csv_parser.parse_csv_line(csv_row, csv_header)
-        return csv_parser
-
-    def _get_csv_parser(self, filepath):
+    # Retrieve all the Parsers,
+    def _get_csv_parser(self, filepath: str) -> CSVParser:
         csv_parser = CSVParser()
         csv_parser.parse_csv_file(filepath)
         return csv_parser
 
-    def _get_fhir_parser(self, filepath):
-        fhir_parser = FHIRParser()
-        fhir_parser.parse_fhir_file(filepath)
-        return fhir_parser
+    def _get_csv_line_parser(self, csv_row, csv_header) -> CSVLineParser:
+        csv_line_parser = CSVLineParser()
+        csv_line_parser.parse_csv_line(csv_row, csv_header)
+        return csv_line_parser
 
-    def _get_fhir_json_parser(self, fhir_data):
+    def _get_fhir_parser(self, fhir_data: dict) -> FHIRParser:
         fhir_parser = FHIRParser()
         fhir_parser.parse_fhir_data(fhir_data)
         return fhir_parser
 
-    def _get_schema_parser(self, schemafile):
+    def _get_schema_parser(self, schemafile: str) -> SchemaParser:
         schema_parser = SchemaParser()
         schema_parser.parse_schema(schemafile)
         return schema_parser
 
+    # Collect and add error record to the list
     def _add_error_record(
-        self, error_record: ErrorReport, expression_error_group, expression_name, expression_id, error_level
-    ):
+        self,
+        error_record: ErrorReport,
+        expression_error_group: str,
+        expression_name: str,
+        expression_id: str,
+        error_level: ErrorLevels,
+    ) -> None:
         if error_record is not None:
             error_record.error_group = expression_error_group
             error_record.name = expression_name
@@ -61,15 +64,15 @@ def _add_error_record(
             self.error_records.append(error_record)
 
     # Function to help identify a parent failure in the error list
-    def _check_error_record_for_fail(self, expression_id):
+    def _check_error_record_for_fail(self, expression_identifier: str) -> bool:
         for error_record in self.error_records:
-            if error_record.id == expression_id:
+            if error_record.id == expression_identifier:
                 return True
         return False
 
     #  validate a single expression against the data file
     def _validate_expression(
-        self, expression_validator: ExpressionChecker, expression, inc_header_in_row_count
+        self, expression_validator: ExpressionChecker, expression: dict, inc_header_in_row_count: bool
     ) -> ErrorReport | int:
         row = 1
         if inc_header_in_row_count:
@@ -123,34 +126,40 @@ def _validate_expression(
         return row
 
     def validate_fhir(
-        self, filepath, summarise=False, report_unexpected_exception=True, inc_header_in_row_count=True
+        self,
+        fhir_data: dict,
+        summarise: bool = False,
+        report_unexpected_exception: bool = True,
+        inc_header_in_row_count: bool = True,
     ) -> list[ErrorReport]:
         self.data_type = DataType.FHIR
-        self.filepath = filepath
+        self.fhir_data = fhir_data
         return self.run_validation(summarise, report_unexpected_exception, inc_header_in_row_count)
 
     def validate_csv(
-        self, filepath, summarise=False, report_unexpected_exception=True, inc_header_in_row_count=True
+        self,
+        filepath: str,
+        summarise: bool = False,
+        report_unexpected_exception: bool = True,
+        inc_header_in_row_count: bool = True,
     ) -> list[ErrorReport]:
         self.data_type = DataType.CSV
         self.filepath = filepath
         return self.run_validation(summarise, report_unexpected_exception, inc_header_in_row_count)
 
     def validate_csv_row(
-        self, csv_row, csv_header, summarise=False, report_unexpected_exception=True, inc_header_in_row_count=True
+        self,
+        csv_row: str,
+        csv_header: list[str],
+        summarise: bool = False,
+        report_unexpected_exception: bool = True,
+        inc_header_in_row_count: bool = True,
     ) -> list[ErrorReport]:
         self.data_type = DataType.CSVROW
         self.csv_row = csv_row
         self.csv_header = csv_header
         return self.run_validation(summarise, report_unexpected_exception, inc_header_in_row_count)
 
-    def validate_fhir_json(
-        self, json_data, summarise=False, report_unexpected_exception=True, inc_header_in_row_count=True
-    ) -> list[ErrorReport]:
-        self.data_type = DataType.FHIRJSON
-        self.json_data = json_data
-        return self.run_validation(summarise, report_unexpected_exception, inc_header_in_row_count)
-
     # run the validation against the data
     def run_validation(
         self, summarise=False, report_unexpected_exception=True, inc_header_in_row_count=True
@@ -160,10 +169,7 @@ def run_validation(
 
             match self.data_type:
                 case DataType.FHIR:
-                    self.data_parser = self._get_fhir_parser(self.filepath)
-                    self.is_csv = False
-                case DataType.FHIRJSON:
-                    self.data_parser = self._get_fhir_json_parser(self.json_data)
+                    self.data_parser = self._get_fhir_parser(self.fhir_data)
                     self.is_csv = False
                 case DataType.CSV:
                     self.data_parser = self._get_csv_parser(self.filepath)

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

@@ -14,6 +14,7 @@ def setUp(self):
         validation_folder = Path(__file__).resolve().parent
         self.FHIRFilePath = validation_folder / "sample_data/vaccination2.json"
         self.schemaFilePath = validation_folder / "test_schemas/test_schema.json"
+        self.fhir_resources = None
 
     def test_validation(self):
         start = time.time()
@@ -22,8 +23,12 @@ def test_validation(self):
         with open(self.schemaFilePath) as JSON:
             SchemaFile = json.load(JSON)
 
+        with open(self.FHIRFilePath) as json_file:
+            self.fhir_resources = json.load(json_file)
+
         validator = Validator(SchemaFile)  # FHIR File Path not needed
-        error_list = validator.validate_fhir(self.FHIRFilePath, True, True, True)
+        print(f"FHIR Resources Loaded: {len(self.fhir_resources.get('entry', []))} entries")
+        error_list = validator.validate_fhir(self.fhir_resources, True, True, True)
         error_report = validator.build_error_report("25a8cc4d-1875-4191-ac6d-2d63a0ebc64b")  # include eventID if known
 
         failed_validation = validator.has_validation_failed()