VED-951 Implement DQ validation checks (#1074)

Author: dlzhry2nhs

lambdas/filenameprocessor/poetry.lock

@@ -18,7 +18,6 @@ aws-lambda-typing = "~2.20.0"
 moto = "^4"
 requests = "~2.32.5"
 responses = "~0.25.8"
-pydantic = "~1.10.13"
 pyjwt = "~2.10.1"
 cryptography = "~46.0.0"
 cffi = "~1.17.1"

lambdas/filenameprocessor/pyproject.toml

@@ -1,7 +1,9 @@
 from dataclasses import dataclass
 
+from pydantic import ValidationError
+
 from common.data_quality.completeness import DataQualityCompletenessChecker, MissingFields
-from common.data_quality.validator import DataQualityValidator
+from common.data_quality.models.immunization_batch_row_model import ImmunizationBatchRowModel
 from common.models.fhir_converter.converter import Converter
 
 
@@ -21,27 +23,38 @@ def __init__(
         is_batch_csv: bool,
     ):
         self.completeness_checker = completeness_checker
+        self.data_quality_model = ImmunizationBatchRowModel
         self.is_batch_csv = is_batch_csv
 
     def run_checks(self, immunisation: dict) -> DataQualityOutput:
-        data_quality_validator = DataQualityValidator()
-
         if not self.is_batch_csv:
             immunisation = Converter(fhir_data=immunisation).run_conversion()
 
         return DataQualityOutput(
             missing_fields=self._check_completeness(immunisation),
-            invalid_fields=self._check_validity(immunisation, data_quality_validator),
+            invalid_fields=self._check_validity(immunisation),
             timeliness=self._check_timeliness(immunisation),
         )
 
     def _check_completeness(self, immunisation: dict) -> MissingFields:
         return self.completeness_checker.run_checks(immunisation)
 
-    @staticmethod
-    def _check_validity(immunisation: dict, data_quality_validator: DataQualityValidator) -> list[str]:
-        pass
+    def _check_validity(self, immunisation: dict) -> list[str]:
+        """Checks the flat batch csv immunisation data structure against the fields and validation rules defined by the
+        data quality team. Returns the fields that were invalid."""
+        fields_with_errors = []
+
+        try:
+            self.data_quality_model.parse_obj(immunisation)
+        except ValidationError as exc:
+            for error in exc.errors():
+                path_to_field_name = error.get("loc", [])
+
+                if len(path_to_field_name) > 0:
+                    # Model uses a flat structure, so all fields will have a depth of 0
+                    fields_with_errors.append(path_to_field_name[0])
+
+        return fields_with_errors
 
-    @staticmethod
-    def _check_timeliness(immunisation: dict) -> dict[str, int]:
+    def _check_timeliness(self, immunisation: dict) -> dict[str, int]:
         pass

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

@@ -0,0 +1,31 @@
+import datetime
+import decimal
+
+ALLOWED_DOSE_AMOUNTS = {
+    # The below are all present in the existing data, but we should check which ones are acceptable:
+    decimal.Decimal("0"),
+    decimal.Decimal("0.1"),
+    decimal.Decimal("0.2"),
+    decimal.Decimal("0.3"),
+    decimal.Decimal("0.4"),
+    decimal.Decimal("0.5"),
+    decimal.Decimal("0.7"),
+    decimal.Decimal("1"),
+    decimal.Decimal("2"),
+    decimal.Decimal("10"),
+    decimal.Decimal("11"),
+}
+
+ALLOWED_DOSE_UNIT_CODES = {
+    # The below are all present in the existing data, but we should check which ones are acceptable:
+    "258773002",  # ml
+    "3317411000001100",  # dose
+    "3318611000001103",  # pre-filled disposable injection
+    "3319711000001103",  # unit dose
+    "408102007",  # unit dose
+    "413516001",  # ampoule
+    "415818006",  # vial
+}
+
+MIN_ACCEPTED_PAST_DATE = datetime.date(1900, 1, 1)
+MIN_ACCEPTED_EXPIRY_DATE = datetime.date(2020, 1, 1)

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

@@ -0,0 +1,75 @@
+"""Simple immunization model to support data quality validation. The team only requires simple validation on a subset
+of the immunization fields"""
+
+import datetime
+import decimal
+
+from dateutil.relativedelta import relativedelta
+from pydantic import BaseModel, condate, constr, validator
+
+from common.data_quality.constants import (
+    ALLOWED_DOSE_AMOUNTS,
+    ALLOWED_DOSE_UNIT_CODES,
+    MIN_ACCEPTED_EXPIRY_DATE,
+    MIN_ACCEPTED_PAST_DATE,
+)
+from common.data_quality.timeliness import parse_csv_date, parse_csv_datetime
+
+# Consider upgrading fhir.resources 7 -> 8 and pydantic 1 -> 2 to use the more powerful and readable Annotated types
+# and constraints
+NhsNumber = constr(min_length=10, max_length=10, regex=r"^\d{10}$")
+BatchCsvDate = condate(ge=MIN_ACCEPTED_PAST_DATE)
+PersonPostcode = constr(regex=r"^[A-Z]{1,2}\d[A-Z\d]? ?\d[A-Z]{2}$")
+ExpiryDate = condate(ge=MIN_ACCEPTED_EXPIRY_DATE)
+SnomedCode = constr(min_length=6, max_length=18, regex=r"^\d{6,18}$")
+
+
+class ImmunizationBatchRowModel(BaseModel):
+    """Represents the Immunization as provided in batch CSV. Contains the subset of fields for DQ validation."""
+
+    NHS_NUMBER: NhsNumber
+    PERSON_DOB: BatchCsvDate
+    DATE_AND_TIME: BatchCsvDate
+    PERSON_POSTCODE: PersonPostcode
+    EXPIRY_DATE: ExpiryDate  # TODO - check with DQ team. Should these checks be relative to the occurrence datetime?
+    DOSE_AMOUNT: decimal.Decimal  # TODO - check with DQ team. Actual values vary a lot from proposed enum.
+    SITE_OF_VACCINATION_CODE: SnomedCode  # TODO - check with DQ team. Their reqs are quite rudimentary, could change?
+    ROUTE_OF_VACCINATION_CODE: SnomedCode
+    DOSE_UNIT_CODE: str  # TODO - check with DQ team. Their enum does not match up with what this data actually is.
+    INDICATION_CODE: SnomedCode
+
+    @validator("PERSON_DOB", "EXPIRY_DATE", pre=True)
+    def parse_csv_date(cls, value: str) -> datetime.date:
+        return parse_csv_date(value)
+
+    @validator("PERSON_DOB", "DATE_AND_TIME")
+    def ensure_past_date(cls, value: datetime.date) -> datetime.date:
+        if value >= datetime.date.today():
+            raise ValueError("Date must be in the past")
+
+        return value
+
+    @validator("DATE_AND_TIME", pre=True)
+    def parse_csv_datetime(cls, value: str) -> datetime.datetime:
+        return parse_csv_datetime(value)
+
+    @validator("EXPIRY_DATE")
+    def is_expiry_within_a_year(cls, value: datetime.date) -> datetime.date:
+        if value > datetime.date.today() + relativedelta(years=1):
+            raise ValueError("EXPIRY_DATE must be within a year from today")
+
+        return value
+
+    @validator("DOSE_AMOUNT")
+    def is_valid_dose_amount(cls, value: decimal.Decimal) -> decimal.Decimal:
+        if value not in ALLOWED_DOSE_AMOUNTS:
+            raise ValueError("Invalid DOSE_AMOUNT provided")
+
+        return value
+
+    @validator("DOSE_UNIT_CODE")
+    def is_valid_dose_unit_code(cls, value: str) -> str:
+        if value not in ALLOWED_DOSE_UNIT_CODES:
+            raise ValueError("Invalid DOSE_UNIT_CODE provided")
+
+        return value

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

@@ -0,0 +1,20 @@
+"""Module containing functions to aid with timeliness checks"""
+
+import datetime
+
+
+def parse_csv_date(d: str) -> datetime.date:
+    return datetime.datetime.strptime(d, "%Y%m%d").date()
+
+
+def parse_csv_datetime(d: str) -> datetime.datetime:
+    """Parses the custom NHS imms batch CSV datetime format YYYYmmddThmmss and an optional 2-digit timezone offset"""
+    match len(d):
+        case 17:
+            return datetime.datetime.strptime(d[0:15], "%Y%m%dT%H%M%S").replace(
+                tzinfo=datetime.timezone(datetime.timedelta(hours=int(d[15:17])))
+            )
+        case 15:
+            return datetime.datetime.strptime(d, "%Y%m%dT%H%M%S").replace(tzinfo=datetime.timezone.utc)
+        case _:
+            raise ValueError("Invalid datetime format provided")

lambdas/shared/src/common/data_quality/models/immunization_batch_row_model.py

@@ -134,37 +134,37 @@
 
 VALID_BATCH_IMMUNISATION = {
     "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",
+    "PERSON_FORENAME": "JOHN",
+    "PERSON_SURNAME": "DOE",
+    "PERSON_DOB": "19801231",
+    "PERSON_GENDER_CODE": "1",
+    "PERSON_POSTCODE": "AB12 3CD",
+    "DATE_AND_TIME": "20240511T120000",
+    "SITE_CODE": "RJ1",
     "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",
+    "PERFORMING_PROFESSIONAL_FORENAME": "ALICE",
+    "PERFORMING_PROFESSIONAL_SURNAME": "SMITH",
+    "RECORDED_DATE": "20250306",
+    "PRIMARY_SOURCE": "True",
+    "VACCINATION_PROCEDURE_CODE": "1324681000000101",
+    "VACCINATION_PROCEDURE_TERM": "Procedure Term",
     "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",
+    "VACCINE_PRODUCT_CODE": "VACC123",
+    "VACCINE_PRODUCT_TERM": "Vaccine Term",
+    "VACCINE_MANUFACTURER": "Manufacturer XYZ",
+    "BATCH_NUMBER": "BATCH001",
+    "EXPIRY_DATE": "20241202",
     "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_CODE": "258773002",
     "DOSE_UNIT_TERM": "milliliter",
     "INDICATION_CODE": "443684005",
-    "LOCATION_CODE": "EC1111",
+    "LOCATION_CODE": "X99999",
     "LOCATION_CODE_TYPE_URI": "https://fhir.nhs.uk/Id/ods-organization-code",
 }

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

@@ -1,6 +1,78 @@
+import copy
+import datetime
 import unittest
+from unittest.mock import patch
+
+from common.data_quality.checker import DataQualityChecker
+from common.data_quality.completeness import DataQualityCompletenessChecker
+from test_common.data_quality.sample_values import VALID_BATCH_IMMUNISATION
 
 
 class TestDataQualityChecker(unittest.TestCase):
     def setUp(self):
-        super().setUp()
+        # Fix date.today() for all validation tests
+        date_today_patcher = patch("common.data_quality.models.immunization_batch_row_model.datetime", wraps=datetime)
+        self.mock_date_today = date_today_patcher.start()
+        self.mock_date_today.date.today.return_value = datetime.date(2024, 5, 20)
+
+        completeness_checker = DataQualityCompletenessChecker()
+        self.batch_dq_checker = DataQualityChecker(completeness_checker, is_batch_csv=True)
+        self.fhir_json_dq_checker = DataQualityChecker(completeness_checker, is_batch_csv=False)
+
+    def test_check_validity_returns_empty_list_when_data_is_valid(self):
+        validation_result = self.batch_dq_checker._check_validity(VALID_BATCH_IMMUNISATION)
+
+        self.assertEqual([], validation_result)
+
+    def test_check_validity_returns_list_of_invalid_fields_when_invalid_data_provided(self):
+        test_cases = [
+            ("NHS_NUMBER", "1234"),  # Failing min length
+            ("NHS_NUMBER", "1234543543543543534"),  # Failing max length
+            ("NHS_NUMBER", "900000AB09"),  # Failing digit only check
+            ("PERSON_DOB", "18990101"),  # Prior to min accepted past date
+            ("PERSON_DOB", "20240137"),  # Invalid date
+            ("PERSON_DOB", "20240624"),  # Past dates only
+            ("DATE_AND_TIME", "17000511T120000"),  # Prior to min accepted past date
+            ("DATE_AND_TIME", "20241511T120000"),  # Invalid datetime
+            ("DATE_AND_TIME", "20241511T120"),  # Invalid datetime
+            ("DATE_AND_TIME", "20240520T120001"),  # Past dates only
+            ("PERSON_POSTCODE", "AAA12 3B"),
+            ("EXPIRY_DATE", "18990101"),  # Prior to min accepted past date
+            ("EXPIRY_DATE", "20240137"),  # Invalid date
+            ("EXPIRY_DATE", "20250521"),  # Expiry greater than a year away
+            ("DOSE_AMOUNT", "abd"),  # Not a decimal value
+            ("DOSE_AMOUNT", "5.67"),  # Decimal value but not in the permitted list of values
+            ("SITE_OF_VACCINATION_CODE", "1254"),  # Fails snomed code min length
+            ("SITE_OF_VACCINATION_CODE", "12321432543543543534"),  # Fails snomed code max length
+            ("SITE_OF_VACCINATION_CODE", "18756hg098"),  # Fails regex
+            ("ROUTE_OF_VACCINATION_CODE", "1254"),  # Fails snomed code min length
+            ("ROUTE_OF_VACCINATION_CODE", "12321432543543543534"),  # Fails snomed code max length
+            ("ROUTE_OF_VACCINATION_CODE", "18756hg098"),  # Fails regex
+            ("DOSE_UNIT_CODE", "415818088"),  # Dose unit code not in the enums
+            ("INDICATION_CODE", "1254"),  # Fails snomed code min length
+            ("INDICATION_CODE", "12321432543543543534"),  # Fails snomed code max length
+            ("INDICATION_CODE", "18756hg098"),  # Fails regex
+        ]
+
+        for field, failing_value in test_cases:
+            with self.subTest(field=field, failing_value=failing_value):
+                invalid_batch_imms_payload = copy.deepcopy(VALID_BATCH_IMMUNISATION)
+                invalid_batch_imms_payload[field] = failing_value
+
+                validation_result = self.batch_dq_checker._check_validity(invalid_batch_imms_payload)
+
+                self.assertEqual([field], validation_result)
+
+    def test_check_validity_returns_list_of_multiple_invalid_fields_for_multiple_failures(self):
+        invalid_batch_imms_payload = copy.deepcopy(VALID_BATCH_IMMUNISATION)
+        invalid_batch_imms_payload["NHS_NUMBER"] = "12345678901"
+        invalid_batch_imms_payload["EXPIRY_DATE"] = "20240137"
+        invalid_batch_imms_payload["PERSON_POSTCODE"] = "12 ACX"
+        invalid_batch_imms_payload["DOSE_AMOUNT"] = "6.789"
+        invalid_batch_imms_payload["INDICATION_CODE"] = "123"
+
+        validation_result = self.batch_dq_checker._check_validity(invalid_batch_imms_payload)
+
+        self.assertEqual(
+            ["NHS_NUMBER", "PERSON_POSTCODE", "EXPIRY_DATE", "DOSE_AMOUNT", "INDICATION_CODE"], validation_result
+        )