VED-948: Data Quality Skeleton (#1059)

* Move delta converter to shared code

* Refactor and move tests into shared

* Move and rename converter files and tests

* Add basic structure for Data Quality Checker

Author: edhall-nhs

lambdas/delta_backend/src/delta.py

@@ -10,8 +10,8 @@
 from common.aws_dynamodb import get_dynamodb_table
 from common.clients import STREAM_NAME, get_sqs_client, logger
 from common.log_firehose import send_log_to_firehose
-from converter import Converter
-from mappings import ActionFlag, EventName, Operation
+from common.models.fhir_converter.converter import Converter
+from common.models.fhir_converter.mappings import ActionFlag, EventName, Operation
 
 failure_queue_url = os.environ["AWS_SQS_QUEUE_URL"]
 delta_table_name = os.environ["DELTA_TABLE_NAME"]

lambdas/delta_backend/tests/test_convert.py

@@ -1,15 +1,13 @@
-import json
 import os
 import unittest
-from copy import deepcopy
 from datetime import datetime
 from unittest.mock import patch
 
 from boto3 import resource as boto3_resource
 from moto import mock_aws
 
-from mappings import ActionFlag, EventName, Operation
-from utils_for_converter_tests import ErrorValuesForTests, ValuesForTests
+from common.models.fhir_converter.mappings import ActionFlag, EventName, Operation
+from utils import ValuesForTests
 
 MOCK_ENV_VARS = {
     "AWS_SQS_QUEUE_URL": "https://sqs.eu-west-2.amazonaws.com/123456789012/test-queue",
@@ -19,7 +17,7 @@
 }
 request_json_data = ValuesForTests.json_data
 with patch.dict("os.environ", MOCK_ENV_VARS):
-    from delta import Converter, handler
+    from delta import handler
 
 
 @patch.dict("os.environ", MOCK_ENV_VARS, clear=True)
@@ -135,74 +133,6 @@ def assert_dynamodb_record(
         expires_at = unfiltered_items[0]["ExpiresAt"]
         self.assertEqual(expires_at - date_time, expected_seconds)
 
-    def test_fhir_converter_json_direct_data(self):
-        """it should convert fhir json data to flat json"""
-        json_data = json.dumps(ValuesForTests.json_data)
-
-        fhir_converter = Converter(json_data)
-        FlatFile = fhir_converter.run_conversion()
-
-        flatJSON = json.dumps(FlatFile)
-        expected_imms_value = deepcopy(ValuesForTests.expected_imms2)  # UPDATE is currently the default action-flag
-        expected_imms = json.dumps(expected_imms_value)
-        self.assertEqual(flatJSON, expected_imms)
-
-        errorRecords = fhir_converter.get_error_records()
-
-        self.assertEqual(len(errorRecords), 0)
-
-    def test_fhir_converter_json_error_scenario_reporting_on(self):
-        """it should convert fhir json data to flat json - error scenarios"""
-        error_test_cases = [
-            ErrorValuesForTests.missing_json,
-            ErrorValuesForTests.json_dob_error,
-        ]
-
-        for test_case in error_test_cases:
-            json_data = json.dumps(test_case)
-
-            fhir_converter = Converter(json_data)
-            fhir_converter.run_conversion()
-
-            errorRecords = fhir_converter.get_error_records()
-
-            # Check if bad data creates error records
-            self.assertTrue(len(errorRecords) > 0)
-
-    def test_fhir_converter_json_error_scenario_reporting_off(self):
-        """it should convert fhir json data to flat json - error scenarios"""
-        error_test_cases = [
-            ErrorValuesForTests.missing_json,
-            ErrorValuesForTests.json_dob_error,
-        ]
-
-        for test_case in error_test_cases:
-            json_data = json.dumps(test_case)
-
-            fhir_converter = Converter(json_data, report_unexpected_exception=False)
-            fhir_converter.run_conversion()
-
-            errorRecords = fhir_converter.get_error_records()
-
-            # Check if bad data creates error records
-            self.assertTrue(len(errorRecords) == 0)
-
-    def test_fhir_converter_json_incorrect_data_scenario_reporting_on(self):
-        """it should convert fhir json data to flat json - error scenarios"""
-
-        with self.assertRaises(ValueError):
-            fhir_converter = Converter(None)
-            errorRecords = fhir_converter.get_error_records()
-            self.assertTrue(len(errorRecords) > 0)
-
-    def test_fhir_converter_json_incorrect_data_scenario_reporting_off(self):
-        """it should convert fhir json data to flat json - error scenarios"""
-
-        with self.assertRaises(ValueError):
-            fhir_converter = Converter(None, report_unexpected_exception=False)
-            errorRecords = fhir_converter.get_error_records()
-            self.assertTrue(len(errorRecords) == 0)
-
     def test_handler_imms_convert_to_flat_json(self):
         """Test that the Imms field contains the correct flat JSON data for CREATE, UPDATE, and DELETE operations."""
         expected_action_flags = [

lambdas/delta_backend/tests/test_delta.py

@@ -7,13 +7,13 @@
 from botocore.exceptions import ClientError
 
 import delta
+from common.models.fhir_converter.mappings import ActionFlag, EventName, Operation
 from delta import (
     handler,
     process_record,
     send_message,
 )
-from mappings import ActionFlag, EventName, Operation
-from utils_for_converter_tests import RecordConfig, ValuesForTests
+from utils import RecordConfig, ValuesForTests
 
 TEST_QUEUE_URL = "https://sqs.eu-west-2.amazonaws.com/123456789012/test-queue"
 os.environ["AWS_SQS_QUEUE_URL"] = TEST_QUEUE_URL

…ckend/tests/utils_for_converter_tests.py lambdas/delta_backend/tests/utils.pylambdas/delta_backend/tests/utils_for_converter_tests.py renamed to lambdas/delta_backend/tests/utils.py lambdas/delta_backend/tests/utils_for_converter_tests.py renamed to lambdas/delta_backend/tests/utils.py

@@ -3,7 +3,7 @@
 from decimal import Decimal
 from typing import List
 
-from mappings import EventName, Operation
+from common.models.fhir_converter.mappings import EventName, Operation
 
 
 class RecordConfig:
@@ -306,44 +306,6 @@ def get_expected_imms(expected_action_flag):
         "CONVERSION_ERRORS": [],
     }
 
-    expected_imms2 = {
-        "NHS_NUMBER": "9000000009",
-        "PERSON_FORENAME": "Sam",
-        "PERSON_SURNAME": "Trailor",
-        "PERSON_DOB": "19650228",
-        "PERSON_GENDER_CODE": "0",
-        "PERSON_POSTCODE": "EC1A 1BB",
-        "DATE_AND_TIME": "20210207T13281700",
-        "SITE_CODE": "B0C4P",
-        "SITE_CODE_TYPE_URI": "https://fhir.nhs.uk/Id/ods-organization-code",
-        "UNIQUE_ID": "ACME-vacc123456",
-        "UNIQUE_ID_URI": "https://supplierABC/identifiers/vacc",
-        "ACTION_FLAG": "UPDATE",
-        "PERFORMING_PROFESSIONAL_FORENAME": "Florence",
-        "PERFORMING_PROFESSIONAL_SURNAME": "Nightingale",
-        "RECORDED_DATE": "20210207",
-        "PRIMARY_SOURCE": "TRUE",
-        "VACCINATION_PROCEDURE_CODE": "13246814444444",
-        "VACCINATION_PROCEDURE_TERM": "Test Value string 123456 COVID vaccination",
-        "DOSE_SEQUENCE": "1",
-        "VACCINE_PRODUCT_CODE": "39114911000001105",
-        "VACCINE_PRODUCT_TERM": "COVID-19 Vaccine Vaxzevria (ChAdOx1 S [recombinant]) not less than 2.5x100,000,000 infectious units/0.5ml dose suspension for injection multidose vials (AstraZeneca UK Ltd) (product)",
-        "VACCINE_MANUFACTURER": "AstraZeneca Ltd",
-        "BATCH_NUMBER": "4120Z001",
-        "EXPIRY_DATE": "20210702",
-        "SITE_OF_VACCINATION_CODE": "368208006",
-        "SITE_OF_VACCINATION_TERM": "Left upper arm structure (body structure)",
-        "ROUTE_OF_VACCINATION_CODE": "78421000",
-        "ROUTE_OF_VACCINATION_TERM": "Intramuscular route (qualifier value)",
-        "DOSE_AMOUNT": "0.5",
-        "DOSE_UNIT_CODE": "ml",
-        "DOSE_UNIT_TERM": "milliliter",
-        "INDICATION_CODE": "443684005",
-        "LOCATION_CODE": "EC1111",
-        "LOCATION_CODE_TYPE_URI": "https://fhir.nhs.uk/Id/ods-organization-code",
-        "CONVERSION_ERRORS": [],
-    }
-
 
 class ErrorValuesForTests:
     json_dob_error = {

lambdas/shared/src/common/data_quality/__init__.py

@@ -0,0 +1,50 @@
+from dataclasses import dataclass
+
+from common.data_quality.completeness import DataQualityCompletenessChecker
+from common.data_quality.validator import DataQualityValidator
+from common.models.fhir_converter.converter import Converter
+from common.models.fhir_converter.mappings import ActionFlag
+
+
+@dataclass
+class DataQualityOutput:
+    incomplete_fields: dict[str, list[str]]
+    invalid_fields: list[str]
+    timeliness: dict[str, int]
+
+
+class DataQualityChecker:
+    """Runs data quality checks against an Immunisation and creates a Data Quality Output object"""
+
+    def __init__(
+        self,
+        immunisation: dict,
+        action_flag: ActionFlag,
+        completeness_checker: DataQualityCompletenessChecker,
+        data_quality_validator: DataQualityValidator,
+        is_batch_csv: bool,
+    ):
+        self.immunisation = immunisation
+        self.fhir_converter = Converter(fhir_data=immunisation, action_flag=action_flag)
+        self.completeness_checker = completeness_checker
+        self.data_quality_validator = data_quality_validator
+        self.is_batch_csv = is_batch_csv
+
+    def run_checks(self) -> DataQualityOutput:
+        if not self.is_batch_csv:
+            self.immunisation = self.fhir_converter.run_conversion()
+
+        return DataQualityOutput(
+            incomplete_fields=self._check_completeness(),
+            invalid_fields=self._check_validity(),
+            timeliness=self._check_timeliness(),
+        )
+
+    def _check_completeness(self) -> dict[str, list[str]]:
+        pass
+
+    def _check_validity(self) -> list[str]:
+        pass
+
+    def _check_timeliness(self) -> dict[str, int]:
+        pass

lambdas/shared/src/common/data_quality/checker.py

@@ -0,0 +1,3 @@
+class DataQualityCompletenessChecker:
+    def check_completeness(self, immunisation: dict) -> dict[str, list[str]]:
+        pass

lambdas/shared/src/common/data_quality/completeness.py

@@ -0,0 +1,2 @@
+class DataQualityValidator:
+    pass