Feature: Component Based Rule Settings

schematic/manifest/generator.py

@@ -10,7 +10,7 @@
 from pathlib import Path
 import pygsheets as ps
 from tempfile import NamedTemporaryFile
-from typing import Dict, List, Optional, Tuple, Union, BinaryIO, Literal
+from typing import Any, Dict, List, Optional, Tuple, Union, BinaryIO, Literal
 
 from schematic.schemas.data_model_graph import DataModelGraph, DataModelGraphExplorer
 from schematic.schemas.data_model_parser import DataModelParser
@@ -21,6 +21,7 @@
     build_service_account_creds,
 )
 from schematic.utils.df_utils import update_df, load_df
+from schematic.utils.schema_utils import extract_component_validation_rules
 from schematic.utils.validate_utils import rule_in_rule_list
 from schematic.utils.schema_utils import DisplayLabelType
 
@@ -751,7 +752,7 @@ def _request_regex_match_vr_formatting(
                 - Upon correct format entry, text will turn black.
                 - If incorrect format is entered a validation error will pop up.
         Input:
-            validation_rules: List[str], defines the validation rules
+            validation_rules: Any(List[str], Dict), defines the validation rules
                 applied to a particular column.
             i: int, defines current column.
             requests_body: dict, containing all the update requests to add to the gs
@@ -1137,6 +1138,10 @@ def _create_requests_body(
             validation_rules = self.dmge.get_node_validation_rules(
                 node_display_name=req
             )
+            if isinstance(validation_rules, dict):
+                validation_rules = extract_component_validation_rules(
+                    validation_rules=validation_rules, manifest_component=self.root
+                )
 
             # Add regex match validaiton rule to Google Sheets.
             if validation_rules and sheet_url:

schematic/models/GE_Helpers.py

@@ -32,6 +32,8 @@
 from schematic.models.validate_attribute import GenerateError
 from schematic.schemas.data_model_graph import DataModelGraphExplorer
 
+from schematic.utils.schema_utils import extract_component_validation_rules
+
 from schematic.utils.validate_utils import (
     rule_in_rule_list,
     np_array_to_str_list,
@@ -164,12 +166,19 @@ def build_expectation_suite(
 
             # remove trailing/leading whitespaces from manifest
             self.manifest.map(lambda x: x.strip() if isinstance(x, str) else x)
+
             validation_rules = self.dmge.get_node_validation_rules(
                 node_display_name=col
             )
 
             # check if attribute has any rules associated with it
             if validation_rules:
+                # Check if the validation rule applies to this manifest
+                if isinstance(validation_rules, dict):
+                    validation_rules = extract_component_validation_rules(
+                        manifest_component=self.manifest["Component"][0],
+                        validation_rules=validation_rules,
+                    )
                 # iterate through all validation rules for an attribute
                 for rule in validation_rules:
                     base_rule = rule.split(" ")[0]

schematic/models/validate_attribute.py

@@ -254,10 +254,6 @@ def generate_type_error(
             node_display_name=attribute_name
         )
 
-        # TODO: Can remove when handling updated so split within graph
-        if validation_rules and "::" in validation_rules[0]:
-            validation_rules = validation_rules[0].split("::")
-
         # If IsNA rule is being used to allow `Not Applicable` entries, do not log a message
         if error_val.lower() == "not applicable" and rule_in_rule_list(
             "IsNA", validation_rules

schematic/models/validate_manifest.py

@@ -26,6 +26,7 @@
 from schematic.models.GE_Helpers import GreatExpectationsHelpers
 from schematic.utils.validate_rules_utils import validation_rule_info
 from schematic.utils.validate_utils import rule_in_rule_list
+from schematic.utils.schema_utils import extract_component_validation_rules
 
 logger = logging.getLogger(__name__)
 
@@ -184,9 +185,12 @@ def validate_manifest_rules(
             manifest.map(lambda x: x.strip() if isinstance(x, str) else x)
             validation_rules = dmge.get_node_validation_rules(node_display_name=col)
 
-            # TODO: Can remove when handling updated so split within graph
-            if validation_rules and "::" in validation_rules[0]:
-                validation_rules = validation_rules[0].split("::")
+            # Parse the validation rules
+            if validation_rules and isinstance(validation_rules, dict):
+                validation_rules = extract_component_validation_rules(
+                    manifest_component=manifest["Component"][0],
+                    validation_rules=validation_rules,
+                )
 
             # Check that attribute rules conform to limits:
             # no more than two rules for an attribute.

schematic/utils/schema_utils.py

@@ -2,13 +2,20 @@
 import json
 import logging
 import networkx as nx
+import re
 import string
-from typing import Dict, List, Literal
+from typing import List, Literal, Dict, Tuple, Union
+
 
 logger = logging.getLogger(__name__)
 
 DisplayLabelType = Literal["class_label", "display_label"]
 BLACKLISTED_CHARS = ["(", ")", ".", " ", "-"]
+DELIMITERS = {
+    "component_name_delimiter": "#",
+    "component_rules_delimiter": "^^",
+    "rule_delimiter": "::",
+}
 
 
 def attr_dict_template(key_name: str) -> Dict[str, dict[str, dict]]:
@@ -207,15 +214,156 @@ def convert_bool_to_str(provided_bool: bool) -> str:
     return str(provided_bool)
 
 
-def parse_validation_rules(validation_rules: List[str]) -> List[str]:
+def get_individual_rules(rule: str, validation_rules: list[str]) -> list:
+    # Separate multiple rules (defined by addition of the rule delimiter)
+    if DELIMITERS["rule_delimiter"] in rule:
+        validation_rules.append(parse_single_set_validation_rules(rule))
+    # Get single rule
+    else:
+        validation_rules.append(rule)
+    return validation_rules
+
+
+def get_component_name_rules(
+    component_names: list, component_rule: str
+) -> Tuple[list, str]:
+    # If a component name is not attached to the rule, have it apply to all other components
+    if DELIMITERS["component_name_delimiter"] != component_rule[0]:
+        component_names.append("all_other_components")
+    # Get the component name if available
+    else:
+        component_names.append(
+            component_rule.split(" ")[0].replace(
+                DELIMITERS["component_name_delimiter"], ""
+            )
+        )
+        try:
+            assert component_names[-1] != " "
+        except ValueError:
+            print(
+                f"There was an error capturing at least one of the component name "
+                f"in the following rule: {component_rule}, "
+                f"please ensure there is not extra whitespace or non-allowed characters."
+            )
+        component_rule = component_rule.replace(component_rule.split(" ")[0], "")
+        component_rule = component_rule.strip()
+    return component_names, component_rule
+
+
+def parse_component_validation_rules(validation_rule_string: str) -> Dict:
+    component_names = []
+    validation_rules = []
+
+    component_rules = validation_rule_string.split(
+        DELIMITERS["component_rules_delimiter"]
+    )
+    # Extract component rules, per component
+    for component_rule in component_rules:
+        component_rule = component_rule.strip()
+        if component_rule:
+            # Get component name attached to rule
+            component_names, component_rule = get_component_name_rules(
+                component_names=component_names, component_rule=component_rule
+            )
+
+            # Get rules
+            validation_rules = get_individual_rules(
+                rule=component_rule, validation_rules=validation_rules
+            )
+
+    # Ensure we collected the component names and validation rules like expected
+    try:
+        assert len(component_names) == len(validation_rules)
+    except ValueError:
+        print(
+            f"The number of components names and validation rules does not match "
+            f"for validation rule: {validation_rule_string}."
+        )
+
+    validation_rules_dict = dict(
+        map(lambda i, j: (i, j), component_names, validation_rules)
+    )
+    return validation_rules_dict
+
+
+def parse_single_set_validation_rules(validation_rule_string: str) -> list:
+    # Try to catch an improperly formatted rule
+    if DELIMITERS["component_name_delimiter"] == validation_rule_string[0]:
+        raise ValueError(
+            f"The provided validation rule {validation_rule_string}, looks to be formatted as a component "
+            f"based rule, but is missing the necessary formatting, "
+            f"please refer to the SchemaHub documentation for more details."
+        )
+
+    # Parse rules that are set across *all* components/manifests
+    return validation_rule_string.split(DELIMITERS["rule_delimiter"])
+
+
+def parse_validation_rules(validation_rules: Union[list, dict]) -> Union[list, dict]:
     """Split multiple validation rules based on :: delimiter
     Args:
-        validation_rules, list: list containing a string validation rule
+        validation_rules, Any[List[str], Dict]: List or Dictionary of validation rules,
+            if list, contains a string validation rule; if dictionary, key is the component the
+            rule (value) is applied to
     Returns:
         validation_rules, list: if submitted List
+    Raises:
+        ValueError if Rule is not formatted properly
+    """
+
+    if isinstance(validation_rules, dict):
+        # Rules pulled in as a dict can be used directly
+        return validation_rules
+    elif isinstance(validation_rules, list):
+        # If rules are already parsed from the JSONLD
+        if len(validation_rules) > 1 and isinstance(validation_rules[-1], str):
+            return validation_rules
+        # Parse rules set for a subset of components/manifests
+        elif DELIMITERS["component_rules_delimiter"] in validation_rules[0]:
+            return parse_component_validation_rules(
+                validation_rule_string=validation_rules[0]
+            )
+        # Parse rules that are set across *all* components/manifests
+        else:
+            return parse_single_set_validation_rules(
+                validation_rule_string=validation_rules[0]
+            )
+    else:
+        raise ValueError(
+            f"The validation rule provided: {str(validation_rules)} is not submitted in an accepted type (list, dictionary) please check your JSONLD."
+        )
+
+
+def extract_component_validation_rules(
+    manifest_component: str, validation_rules: dict[str, list]
+) -> list:
+    """Parse a component validation rule dictionary to pull out the rule (if any) for a given manifest
+    Args:
+        manifest_component, str: Component label, pulled from the manifest directly
+        validation_rules, dict[str, list[Union[list,str]]: Validation rules dictionary, where keys are the manifest component label,
+            and the value is a parsed set of validation rules.
+    Returns:
+        validation_rules, list[str]:
     """
-    if validation_rules and "::" in validation_rules[0]:
-        validation_rules = validation_rules[0].split("::")
+    manifest_component_rule = validation_rules.get(manifest_component)
+    all_component_rules = validation_rules.get("all_other_components")
+
+    # Capture situation where manifest_component rule is an empty string
+    if manifest_component_rule is not None:
+        if isinstance(manifest_component_rule, str):
+            if manifest_component_rule == "":
+                validation_rules = []
+            else:
+                validation_rules = [manifest_component_rule]
+        elif isinstance(manifest_component_rule, list):
+            validation_rules = manifest_component_rule
+    elif all_component_rules:
+        if isinstance(all_component_rules, str):
+            validation_rules = [all_component_rules]
+        elif isinstance(all_component_rules, list):
+            validation_rules = all_component_rules
+    else:
+        validation_rules = []
     return validation_rules
 
 

schematic/utils/validate_rules_utils.py

@@ -1,7 +1,7 @@
 """validate rules utils"""
 
-from typing import Union
 import logging
+from typing import Union
 from jsonschema import ValidationError
 
 

tests/data/example.model.csv

@@ -1,7 +1,7 @@
 Attribute,Description,Valid Values,DependsOn,Properties,Required,Parent,DependsOn Component,Source,Validation Rules
 Component,,,,,TRUE,,,,
 Patient,,,"Patient ID, Sex, Year of Birth, Diagnosis, Component",,FALSE,DataType,,,
-Patient ID,,,,,TRUE,DataProperty,,,
+Patient ID,,,,,TRUE,DataProperty,,,#Patient unique warning^^#Biospecimen unique error
 Sex,,"Female, Male, Other",,,TRUE,DataProperty,,,
 Year of Birth,,,,,FALSE,DataProperty,,,
 Diagnosis,,"Healthy, Cancer",,,TRUE,DataProperty,,,
@@ -42,4 +42,4 @@ Check Date,,,,,TRUE,DataProperty,,,date
 Check NA,,,,,TRUE,DataProperty,,,int::IsNA
 MockRDB,,,"Component, MockRDB_id, SourceManifest",,FALSE,DataType,,,
 MockRDB_id,,,,,TRUE,DataProperty,,,int
-SourceManifest,,,,,TRUE,DataProperty,,,
+SourceManifest,,,,,TRUE,DataProperty,,,

tests/data/example.model.jsonld

@@ -73,7 +73,10 @@
             },
             "sms:displayName": "Patient ID",
             "sms:required": "sms:true",
-            "sms:validationRules": []
+            "sms:validationRules": {
+                "Biospecimen": "unique error",
+                "Patient": "unique warning"
+            }
         },
         {
             "@id": "bts:Sex",

tests/data/mock_manifests/example.biospecimen_component_rule.manifest.csv

@@ -0,0 +1,4 @@
+Sample ID,Patient ID,Tissue Status,Component
+123,123,Healthy,Biospecimen
+456,123,Healthy,Biospecimen
+789,syn465,Healthy,Biospecimen

tests/data/mock_manifests/example.patient_component_rule.manifest.csv

@@ -0,0 +1,4 @@
+Patient ID,Sex,Year of Birth,Diagnosis,Component,Cancer Type,Family History
+123,Female,,Cancer,Patient,Breast,"Breast, Colorectal"
+123,Male,,Healthy,Patient,,"Breast, Colorectal"
+789,Other,,Healthy,Patient,,"Breast, Colorectal"

tests/test_schemas.py

@@ -783,17 +783,19 @@ def test_run_rel_functions(self, helpers, data_model, rel_func, test_dn, test_bo
 
             assert len(vrs) == len(parsed_vrs)
             if DATA_MODEL_DICT[data_model] == "CSV":
-                assert vrs != parsed_vrs
+                for ind, rule in enumerate(vrs):
+                    if "::" in rule[0]:
+                        assert parsed_vrs[ind] == rule[0].split("::")
+                    elif '^^' in rule[0]:
+                        component_rule_sets = rule[0].split("^^")
+                        components = [cr.split(' ')[0].replace('#', '') for cr in component_rule_sets]
+                        assert components == [k for k in parsed_vrs[0].keys()]
+                    else:
+                        assert parsed_vrs[ind] == rule
             elif DATA_MODEL_DICT[data_model] == "JSONLD":
                 # JSONLDs already contain parsed validaiton rules so the raw vrs will match the parsed_vrs
                 assert vrs == parsed_vrs
 
-            # For all validation rules where there are multiple rules, make sure they have been split as expected.
-            for i, pvr in enumerate(parsed_vrs):
-                delim_count = vrs[i][0].count("::")
-                if delim_count:
-                    assert len(pvr) == delim_count + 1
-
         elif rel_func == get_label_from_display_name:
             # For a limited set check label is returned as expected.
             for entry_type, expected_value in TEST_DN_DICT[test_dn].items():