schema rules for vaccination procedure code

Author: Akol125

backend/src/models/utils/pre_validator_utils.py

@@ -1,6 +1,6 @@
 from datetime import date, datetime
 from decimal import Decimal
-from typing import Optional, Union
+from typing import Union
 
 from .generic_utils import is_valid_simple_snomed, nhs_number_mod11_check
 
@@ -23,9 +23,6 @@ def for_string(
         if not isinstance(field_value, str):
             raise TypeError(f"{field_location} must be a string")
 
-        if field_value.isspace():
-            raise ValueError(f"{field_location} must be a non-empty string")
-
         if defined_length:
             if len(field_value) != defined_length:
                 raise ValueError(f"{field_location} must be {defined_length} characters")
@@ -49,15 +46,14 @@ def for_string(
     def for_list(
         field_value: list,
         field_location: str,
-        defined_length: Optional[int] = None,
-        max_length: Optional[int] = None,
+        defined_length: int = None,
         elements_are_strings: bool = False,
-        string_element_max_length: Optional[int] = None,
         elements_are_dicts: bool = False,
     ):
         """
-        Apply pre-validation to a list field to ensure it is a non-empty list which meets the length requirements and
-        requirements, if applicable, for each list element to be a non-empty string or non-empty dictionary
+        Apply pre-validation to a list field to ensure it is a non-empty list which meets the length
+        requirements and requirements, if applicable, for each list element to be a non-empty string
+        or non-empty dictionary
         """
         if not isinstance(field_value, list):
             raise TypeError(f"{field_location} must be an array")
@@ -69,12 +65,12 @@ def for_list(
             if len(field_value) == 0:
                 raise ValueError(f"{field_location} must be a non-empty array")
 
-        if max_length is not None and len(field_value) > max_length:
-            raise ValueError(f"{field_location} must be an array of maximum length {max_length}")
-
         if elements_are_strings:
-            for idx, element in enumerate(field_value):
-                PreValidation.for_string(element, f"{field_location}[{idx}]", max_length=string_element_max_length)
+            for element in field_value:
+                if not isinstance(element, str):
+                    raise TypeError(f"{field_location} must be an array of strings")
+                if len(element) == 0:
+                    raise ValueError(f"{field_location} must be an array of non-empty strings")
 
         if elements_are_dicts:
             for element in field_value:
@@ -185,7 +181,6 @@ def for_positive_integer(field_value: int, field_location: str, max_value: int =
         Apply pre-validation to an integer field to ensure that it is a positive integer,
         which does not exceed the maximum allowed value (if applicable)
         """
-        # This check uses type() instead of isinstance() because bool is a subclass of int.
         if type(field_value) is not int:  # pylint: disable=unidiomatic-typecheck
             raise TypeError(f"{field_location} must be a positive integer")
 
@@ -203,7 +198,6 @@ def for_integer_or_decimal(field_value: Union[int, Decimal], field_location: str
         which does not exceed the maximum allowed number of decimal places (if applicable)
         """
         if not (
-            # This check uses type() instead of isinstance() because bool is a subclass of int.
             type(field_value) is int  # pylint: disable=unidiomatic-typecheck
             or type(field_value) is Decimal  # pylint: disable=unidiomatic-typecheck
         ):

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

@@ -8,7 +8,7 @@
 from common.validator.expression_rule import expression_rule_per_field
 from common.validator.lookup_expressions.key_data import KeyData
 from common.validator.lookup_expressions.lookup_data import LookUpData
-from common.validator.validation_utils import check_if_future_date, nhs_number_mod11_check
+from common.validator.validation_utils import check_if_future_date, is_valid_simple_snomed, nhs_number_mod11_check
 
 
 class ExpressionChecker:
@@ -31,8 +31,6 @@ def validate_expression(
                 return self.validation_for_string_values(expression_rule, field_name, field_value)
             case "LIST":
                 return self.validation_for_list(expression_rule, field_name, field_value)
-            case "NHS_NUMBER":
-                return self.validation_for_nhs_number(expression_rule, field_name, field_value)
             case "DATE":
                 return self.validation_for_date(expression_rule, field_name, field_value)
             case "DATETIME":
@@ -45,6 +43,10 @@ def validate_expression(
                 return self.validation_for_boolean(expression_rule, field_name, field_value)
             case "INTDECIMAL":
                 return self.validation_for_integer_or_decimal(expression_rule, field_name, field_value)
+            case "NHS_NUMBER":
+                return self.validation_for_nhs_number(expression_rule, field_name, field_value)
+            case "SNOMED_CODE":
+                return self.validation_for_snomed_code(expression_rule, field_name, field_value)
             case _:
                 return "Schema expression not found! Check your expression type : " + expression_type
 
@@ -317,3 +319,20 @@ def validation_for_nhs_number(self, expression_rule: str, field_name: str, field
             if self.report_unexpected_exception:
                 message = MESSAGES[ExceptionLevels.UNEXPECTED_EXCEPTION] % (e.__class__.__name__, e)
                 return ErrorReport(ExceptionLevels.UNEXPECTED_EXCEPTION, message, None, field_name)
+
+    def validation_for_snomed_code(self, expression_rule: str, field_location: str, field_value: str):
+        """
+        Apply prevalidation to snomed code to ensure that its a valid one.
+        """
+
+        error_message = f"{field_location} is not a valid snomed code"
+
+        try:
+            is_valid = is_valid_simple_snomed(field_value)
+            if not is_valid:
+                raise ValueError(error_message)
+        except ValueError as e:
+            code = ExceptionLevels.RECORD_CHECK_FAILED
+            message = MESSAGES[ExceptionLevels.RECORD_CHECK_FAILED]
+            details = str(e)
+            return ErrorReport(code, message, None, field_location, details)

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

@@ -16,20 +16,6 @@ def check_if_future_date(parsed_value: date | datetime):
     return False
 
 
-def is_valid_simple_snomed(simple_snomed: str) -> bool:
-    "check the snomed code valid or not."
-    min_snomed_length = 6
-    max_snomed_length = 18
-    return (
-        simple_snomed is not None
-        and simple_snomed.isdigit()
-        and simple_snomed[0] != "0"
-        and min_snomed_length <= len(simple_snomed) <= max_snomed_length
-        and validate(simple_snomed)
-        and (simple_snomed[-3:-1] in ("00", "10"))
-    )
-
-
 def nhs_number_mod11_check(nhs_number: str) -> bool:
     """
     Parameters:-
@@ -56,3 +42,17 @@ def nhs_number_mod11_check(nhs_number: str) -> bool:
         is_mod11 = check_digit == int(nhs_number[-1])
 
     return is_mod11
+
+
+def is_valid_simple_snomed(simple_snomed: str) -> bool:
+    "check the snomed code valid or not."
+    min_snomed_length = 6
+    max_snomed_length = 18
+    return (
+        simple_snomed is not None
+        and simple_snomed.isdigit()
+        and simple_snomed[0] != "0"
+        and min_snomed_length <= len(simple_snomed) <= max_snomed_length
+        and validate(simple_snomed)
+        and (simple_snomed[-3:-1] in ("00", "10"))
+    )

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

@@ -195,6 +195,19 @@ def test_vaccination_procedure_code_string_valid_and_invalid(self):
         # Invalid: non-string value
         self.assertIsInstance(checker.validate_expression("STRING", "", field_path, 123456), ErrorReport)
 
+    # After parent check succeeds - SNOMED_CODE for VACCINATION_PROCEDURE_CODE
+    def test_vaccination_procedure_snomed_code_valid_and_invalid(self):
+        checker = self.make_checker()
+        field_path = "extension|0|valueCodeableConcept|coding|0|code"
+        # Valid SNOMED example (passes Verhoeff, doesn't start with 0, length ok, suffix rule)
+        self.assertIsNone(checker.validate_expression("SNOMED_CODE", "", field_path, "1119349007"))
+        # Invalid: empty
+        self.assertIsInstance(checker.validate_expression("SNOMED_CODE", "", field_path, ""), ErrorReport)
+        # Invalid: non-digit
+        self.assertIsInstance(checker.validate_expression("SNOMED_CODE", "", field_path, "ABC123"), ErrorReport)
+        # Invalid: starts with 0
+        self.assertIsInstance(checker.validate_expression("SNOMED_CODE", "", field_path, "012345"), ErrorReport)
+
     # STRING with VACCINATION_PROCEDURE_TERM
     def test_vaccination_procedure_term_string_valid_and_invalid(self):
         checker = self.make_checker()
@@ -388,16 +401,5 @@ def test_postcode_string_rule_valid_and_invalid(self):
         self.assertIsInstance(checker.validate_expression("STRING", "", field_path, ""), ErrorReport)
 
 
-# class DummyParserEx:
-#     def __init__(self, data=None, raise_on_get=False):
-#         self._data = data or {}
-#         self._raise_on_get = raise_on_get
-
-#     def get_key_value(self, field_name):
-#         if self._raise_on_get:
-#             raise RuntimeError("boom")
-#         return [self._data.get(field_name, "")]
-
-
 if __name__ == "__main__":
     unittest.main()

lambdas/shared/tests/test_common/validator/test_schemas/test_schema.json

@@ -226,6 +226,20 @@
       },
       "errorGroup": "completeness"
     },
+    {
+      "expressionId": "01K9ZR2QZ6D9MTX3T7SQMK37C7",
+      "parentExpressionId": "01K8S0YYGDFWJXN2W3THYG24EZ",
+      "fieldNameFHIR": "extension|0|valueCodeableConcept|coding|0|code",
+      "fieldNameFlat": "VACCINATION_PROCEDURE_CODE",
+      "fieldNumber": 17,
+      "errorLevel": 0,
+      "expression": {
+        "expressionName": "Procedure Code Not Empty Check",
+        "expressionType": "SNOMED_CODE",
+        "expressionRule": ""
+      },
+      "errorGroup": "completeness"
+    },
     {
       "expressionId": "01K5EGR0C85HY6MDNN6TTR1K48",
       "fieldNameFHIR": "extension|0|valueCodeableConcept|coding|0|display",

lambdas/shared/tests/test_common/validator/testing_utils/constants.py

@@ -28,7 +28,7 @@
     "PERFORMING_PROFESSIONAL_SURNAME": "SMITH",
     "RECORDED_DATE": "2025-03-06",
     "PRIMARY_SOURCE": True,
-    "VACCINATION_PROCEDURE_CODE": "PROC123",
+    "VACCINATION_PROCEDURE_CODE": "1324681000000101",
     "VACCINATION_PROCEDURE_TERM": "Procedure Term",
     "DOSE_SEQUENCE": 1,
     "VACCINE_PRODUCT_CODE": "VACC123",