Source code for idr_iisim.utils.schema

"""Schema Validator"""

from typing import Any

import yaml
from jsonschema import validate


[docs] class Validator: """The Validator class responsible for validating data against schemas defined in the config schema files. Attributes: industry_validator (dict[str, Any]): Schema loaded from config/industry.yaml. process_validator (dict[str, Any]): Schema loaded from config/process.yaml. """ # A dictionary that stores the schema loaded from the industry.yaml file industry_validator: dict[str, Any] # A dictionary that stores the schema loaded from the process.yaml file process_validator: dict[str, Any] def __init__(self) -> None: """Initialize the Validator by loading schemas from specified YAML files. Raises: Exception: If there is an error loading the schema files. """ path_to_config = "config/industry.yaml" try: with open(path_to_config, encoding="utf-8") as file: # load schema from file (schema.yaml) # and store it in the validator attribute self.industry_validator = yaml.safe_load(file) except Exception as e: raise e path_to_config = "config/process.yaml" try: with open(path_to_config, encoding="utf-8") as file: # load schema from file (process.yaml) # and store it in the validator attribute self.process_validator = yaml.safe_load(file) except Exception as e: raise e
[docs] def validate(self, data: dict[str, Any]) -> None: """Validate the provided data against the appropriate schema. This method checks if the given dictionary data conforms to a schema loaded from the corresponding YAML file. Args: data (dict[str, Any]): The dictionary containing data to be validated. Raises: ValueError: If the data is missing a type field or has an invalid type. Exception: If the validation process fails. """ try: if "type" not in data: raise ValueError("YAML should have a type field") if data["type"] not in ["industry", "process"]: raise ValueError( "YAML's type field should be 'industry' or 'process'" ) schema = self.process_validator if data["type"] == "industry": schema = self.industry_validator validate(instance=data, schema=schema) except Exception as e: raise e
[docs] def print_industry_schema(self) -> None: """print current validation schema""" print(self.industry_validator)
[docs] def print_process_schema(self) -> None: """print current validation schema""" print(self.process_validator)