in dagify/converter/engine.py [0:0]
def load_config(object):
# Validate Template Path Provided
if object.config_file is None:
raise ValueError("dagify: config file not provided")
if file_exists(object.config_file) is False:
raise FileNotFoundError("dagify: config file does not exist")
with open(object.config_file) as stream:
try:
object.config = yaml.safe_load(stream)
except yaml.YAMLError as exc:
raise exc
if object.config is None:
raise ValueError("dagify: No configuration has been loaded")
if object.config["config"]["mappings"] is None:
raise ValueError(
"dagify: Configuration loaded with error, no Operator/JobType Mappings loaded")
# Modify Configration for Standardization:
templatesToValidate = []
for idx, config in enumerate(object.config["config"]["mappings"]):
# Set Command Uppercase
object.config["config"]["mappings"][idx]["job_type"] = \
object.config["config"]["mappings"][idx]["job_type"].upper()
templatesToValidate.append(object.config["config"]["mappings"][idx]["template_name"])
for root, dirs, files in os.walk(object.templates_path):
for file in files:
if file.endswith(".yaml"):
template_name = file.split(".")[0]
if template_name in templatesToValidate:
print(f"{template_name} ready for validation")
# Loads a Single Template into a Dictionary from .yaml file
file_path = os.path.join(root, file)
template = yamale.make_data(file_path)
schema = yamale.make_schema(object.schema, validators=validators)
if template is not None:
try:
yamale.validate(schema, template)
print(f"Validation succeeded for {file}!")
except yamale.YamaleError as e:
print(f"Validation failed for {file}!\n")
for result in e.results:
for error in result.errors:
print(error)
raise ValueError(f"Template {file_path} incompatible")
return