715: usdm schema validation

README.md

@@ -534,3 +534,17 @@ Then run normally: `core.exe validate -rest -of -config -commands
 ---
 
 **Note:** Setting `DATASET_SIZE_THRESHOLD=0` tells the engine to use Dask processing for all datasets regardless of size, size threshold defaults to 1/4 of available RAM so datasets larger than this will use Dask. See env.example to see what the CLI .env file should look like
+
+## Updating USDM JSON Schema
+
+Currently, the engine supports USDM JSON Schema validation against versions 3.0 and 4.0. The schema definition files are located at:
+
+- `resources/cache/usdm-3-0-schema.pkl`
+- `resources/cache/usdm-4-0-schema.pkl`
+
+These schema definitions were derived from the OpenAPI specs located in the `https://github.com/cdisc-org/DDF-RA` repo, so in order to update the existing schemas or create a new one, run:
+
+1. `git --no-pager --git-dir DDF-RA.git show --format=format:"%B" {required tag (example: v3.0.0)}:Deliverables/API/USDM_API.json > USDM_API_{required version}.json`
+2. Use `scripts/openapi-to-json.py` script to convert the OpenAPI spec to JSON schema definition
+3. Use `scripts/json_pkl_converter.py` script to convert the JSON file to `.pkl`
+4. Place the `.pkl` file to `resources/cache`

cdisc_rules_engine/dataset_builders/dataset_builder_factory.py

@@ -1,6 +1,9 @@
 # flake8: noqa
 from typing import Type
 
+from cdisc_rules_engine.dataset_builders.json_schema_check_dataset_builder import (
+    JsonSchemaCheckDatasetBuilder,
+)
 from cdisc_rules_engine.dataset_builders.jsonata_dataset_builder import (
     JSONataDatasetBuilder,
 )
@@ -77,6 +80,7 @@ class DatasetBuilderFactory(FactoryInterface):
         RuleTypes.VALUE_CHECK_WITH_DATASET_METADATA.value: ValueCheckDatasetMetadataDatasetBuilder,
         RuleTypes.VALUE_CHECK_WITH_VARIABLE_METADATA.value: ValueCheckVariableMetadataDatasetBuilder,
         RuleTypes.JSONATA.value: JSONataDatasetBuilder,
+        RuleTypes.JSON_SCHEMA_CHECK.value: JsonSchemaCheckDatasetBuilder,
     }
 
     @classmethod

cdisc_rules_engine/dataset_builders/json_schema_check_dataset_builder.py

@@ -0,0 +1,172 @@
+import copy
+import json
+from copy import deepcopy
+import re
+
+from jsonschema import validators, exceptions
+from cdisc_rules_engine.dataset_builders.base_dataset_builder import BaseDatasetBuilder
+from cdisc_rules_engine.models.dataset import DatasetInterface
+from cdisc_rules_engine.utilities.utils import tag_source
+
+
+class JsonSchemaCheckDatasetBuilder(BaseDatasetBuilder):
+    dataset_template = {
+        "json_path": [],
+        "error_attribute": [],
+        "error_value": [],
+        "validator": [],
+        "validator_value": [],
+        "message": [],
+        "dataset": [],
+        "id": [],
+        "_path": [],
+    }
+
+    def build(self, **kwargs) -> DatasetInterface:
+        return self.get_dataset()
+
+    def _get_cached_dataset(self) -> dict[str, list[str]]:
+        cache_key: str = (
+            f"json_schema_validation_result_{self.data_service.dataset_path}"
+        )
+        if cached := self.cache.get(cache_key):
+            return cached
+
+        schema = self.library_metadata.standard_schema_definition
+        cls = validators.validator_for(schema)
+        cls.check_schema(schema)
+        validator = cls(schema)
+
+        errtree = exceptions.ErrorTree(validator.iter_errors(self.data_service.json))
+
+        errlist = copy.deepcopy(self.dataset_template)
+        self.list_errors(errtree, errlist)
+
+        self.cache.add(cache_key, errlist)
+
+        return errlist
+
+    def get_dataset(self, **kwargs) -> DatasetInterface:
+        dataset = self._get_cached_dataset()
+        records = [
+            {key: dataset[key][i] for key in dataset}
+            for i in range(len(next(iter(dataset.values()))))
+        ]
+        filtered = [
+            row for row in records if row["dataset"] == self.dataset_metadata.name
+        ]
+        return tag_source(
+            (
+                self.dataset_implementation.from_records(filtered, **kwargs)
+                if filtered
+                else self.dataset_implementation.from_dict(
+                    self.dataset_template, **kwargs
+                )
+            ),
+            self.dataset_metadata,
+        )
+
+    def list_errors(self, tree: exceptions.ErrorTree, errlist: dict[str, list]):
+        if tree.errors:
+            for ve in tree.errors.values():
+                self.process_error(error=ve, errlist=errlist)
+
+        if len(tree._contents) > 0:
+            for k, v in tree._contents.items():
+                self.list_errors(
+                    tree=v,
+                    errlist=errlist,
+                )
+
+    def get_instance_by_path(self, instance: dict, path_list: list) -> dict:
+        _inst = deepcopy(instance)
+        for p in path_list:
+            _inst = _inst[p]
+        return _inst
+
+    def get_parent_path(self, path_list: list):
+        return list(path_list)[0 : (-1 - int(isinstance(path_list[-1], int)))]
+
+    def parse_error(
+        self,
+        error: exceptions.ValidationError,
+        errlist: dict[str, list],
+        errpath: list,
+    ):
+        errctx = self.get_instance_by_path(self.data_service.json, errpath)
+        errattr = (
+            self.get_attributes_from_message(error.message)
+            if error.validator in ["required", "additionalProperties"]
+            else (
+                "{}[{}]".format(error.absolute_path[-2], error.absolute_path[-1])
+                if isinstance(error.absolute_path[-1], int)
+                else error.absolute_path[-1]
+            )
+        )
+        errlist["json_path"].append(error.json_path)
+        errlist["error_attribute"].append(errattr)
+        errlist["error_value"].append(json.dumps(error.instance))
+        errlist["validator"].append(error.validator)
+        errlist["validator_value"].append(str(error.validator_value))
+        errlist["message"].append(
+            error.message.replace(str(error.instance), f"[Value of {errattr}]")
+            if len(str(error.instance)) > len(errattr) + 11
+            and str(error.instance) in error.message
+            else error.message
+        )
+        errlist["dataset"].append(errctx.get("instanceType", "") if errctx else "")
+        errlist["id"].append(errctx.get("id", "") if errctx else "")
+        errlist["_path"].append("/" + "/".join(map(str, errpath)))
+
+    def list_context_errors(
+        self,
+        error: exceptions.ValidationError,
+        errlist: dict[str, list],
+        skip_subschemas: list = [],
+    ):
+        if error.context:
+            for vec in error.context:
+                if (
+                    skip_subschemas == []
+                    or list(vec.schema_path)[0] not in skip_subschemas
+                ):
+                    self.process_error(error=vec, errlist=errlist)
+
+    def process_error(
+        self, error: exceptions.ValidationError, errlist: dict[str, list]
+    ):
+        if error.validator == "anyOf":
+            skip_ssi = []
+            refs = [
+                ss["$ref"].split("/")[-1]
+                for ss in error.schema["anyOf"]
+                if "$ref" in ss
+            ]
+            for vec in error.context:
+                if (
+                    list(vec.relative_path) == ["instanceType"]
+                    and vec.validator == "const"
+                    and vec.instance in refs
+                ) or (
+                    list(vec.relative_path) == []
+                    and vec.validator == "type"
+                    and vec.validator_value == "null"
+                ):
+                    skip_ssi.append(list(vec.schema_path)[0])
+            self.list_context_errors(
+                error=error, errlist=errlist, skip_subschemas=skip_ssi
+            )
+        else:
+            self.parse_error(
+                error=error,
+                errlist=errlist,
+                errpath=(
+                    error.absolute_path
+                    if error.validator in ["required", "additionalProperties"]
+                    else self.get_parent_path(error.absolute_path)
+                ),
+            )
+            self.list_context_errors(error=error, errlist=errlist)
+
+    def get_attributes_from_message(self, message: str) -> list[str]:
+        return re.findall(r"'([^, ]+)'", message)

cdisc_rules_engine/enums/rule_types.py

@@ -32,3 +32,4 @@ class RuleTypes(BaseEnum):
     )
     VALUE_CHECK_WITH_DATASET_METADATA = "Value Check with Dataset Metadata"
     VALUE_CHECK_WITH_VARIABLE_METADATA = "Value Check with Variable Metadata"
+    JSON_SCHEMA_CHECK = "JSON Schema Check"

cdisc_rules_engine/models/library_metadata_container.py

@@ -7,6 +7,7 @@ class LibraryMetadataContainer:
     def __init__(
         self,
         standard_metadata={},
+        standard_schema_definition={},
         model_metadata={},
         ct_package_metadata={},
         variable_codelist_map={},
@@ -15,6 +16,7 @@ def __init__(
         cache_path: str = "",
     ):
         self._standard_metadata = standard_metadata
+        self._standard_schema_definition = standard_schema_definition
         self._model_metadata = model_metadata
         self._ct_package_metadata = ct_package_metadata
         self._variable_codelist_map = variable_codelist_map
@@ -30,6 +32,14 @@ def standard_metadata(self):
     def standard_metadata(self, value):
         self._standard_metadata = value
 
+    @property
+    def standard_schema_definition(self):
+        return self._standard_schema_definition
+
+    @standard_schema_definition.setter
+    def standard_schema_definition(self, value):
+        self._standard_schema_definition = value
+
     @property
     def variable_codelist_map(self):
         return self._variable_codelist_map

resources/cache/usdm-3-0-schema.pkl

 library_metadata: LibraryMetadataContainer = get_library_metadata_from_cache(args) 
 library_metadata: LibraryMetadataContainer = get_library_metadata_from_cache(args) 

resources/schema/Rule_Type.md

@@ -556,3 +556,17 @@ Attach define xml metadata at variable level
 - `library_variable_data_type`
 - `library_variable_ccode`
 - `variable_has_empty_values`
+
+## JSON Schema Check
+
+#### Columns:
+
+- `json_path`
+- `error_attribute`
+- `error_value`
+- `validator`
+- `validator_value`
+- `message`
+- `dataset`
+- `id`
+- `_path`

scripts/json_pkl_converter.py

@@ -0,0 +1,47 @@
+import argparse
+import os
+import json
+import pickle
+
+
+def parse_arguments():
+    parser = argparse.ArgumentParser(description="Convert between JSON and PKL files.")
+    parser.add_argument(
+        "-i", "--input_file", required=True, help="Input file (.json or .pkl)"
+    )
+    args = parser.parse_args()
+    return args
+
+
+def json_to_pkl(json_path, pkl_path):
+    with open(json_path, "r", encoding="utf-8") as f:
+        data = json.load(f)
+    with open(pkl_path, "wb") as f:
+        pickle.dump(data, f)
+    print(f"Converted {json_path} to {pkl_path}")
+
+
+def pkl_to_json(pkl_path, json_path):
+    with open(pkl_path, "rb") as f:
+        data = pickle.load(f)
+    with open(json_path, "w", encoding="utf-8") as f:
+        json.dump(data, f, ensure_ascii=False, indent=2)
+    print(f"Converted {pkl_path} to {json_path}")
+
+
+def main():
+    args = parse_arguments()
+    input_file = args.input_file
+    base, ext = os.path.splitext(input_file)
+    if ext.lower() == ".json":
+        out_file = base + ".pkl"
+        json_to_pkl(input_file, out_file)
+    elif ext.lower() == ".pkl":
+        out_file = base + ".json"
+        pkl_to_json(input_file, out_file)
+    else:
+        print("Unsupported file extension. Please provide a .json or .pkl file.")
+
+
+if __name__ == "__main__":
+    main()

scripts/openapi-to-json.py

@@ -0,0 +1,59 @@
+import os
+import argparse
+import json
+
+
+def parse_arguments():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("-i", "--input_file", help="USDM OpenAPI JSON file")
+    args = parser.parse_args()
+    return args
+
+
+args = parse_arguments()
+
+filename = os.path.split(args.input_file)[-1]
+
+outfname = "".join(filename.split(".")[0]) + "_schemas"
+
+with open(args.input_file) as f:
+    openapi = json.load(f)
+
+jschema = {"$defs": {}}
+
+
+def replace_deep(data, a, b):
+    if isinstance(data, str):
+        return data.replace(a, b)
+    elif isinstance(data, dict):
+        return {k: replace_deep(v, a, b) for k, v in data.items()}
+    elif isinstance(data, list):
+        return [replace_deep(v, a, b) for v in data]
+    else:
+        # nothing to do?
+        return data
+
+
+for sn, sd in openapi["components"]["schemas"].items():
+    if sn == "Wrapper-Input":
+        for k, v in sd.items():
+            jschema[k] = replace_deep(
+                replace_deep(v, "components/schemas", "$defs"), "-Input", ""
+            )
+    elif not sn.endswith("-Output"):
+        jschema["$defs"][sn.replace("-Input", "")] = replace_deep(
+            replace_deep(sd, "components/schemas", "$defs"), "-Input", ""
+        )
+
+for v in jschema["$defs"].values():
+    v.update({"additionalProperties": False})
+    for pn, pd in v.get("properties", {}).items():
+        if pn in v.get("required", []) and pd.get("type", "") == "array":
+            pd.update({"minItems": 1})
+
+with open(
+    os.path.join("".join(os.path.split(args.input_file)[0:-1]), outfname + ".json"),
+    "w",
+    encoding="utf-8",
+) as f:
+    json.dump(jschema, f, ensure_ascii=False, indent=4)