Source code for ads.jobs.schema.validator
#!/usr/bin/env python
# -*- coding: utf-8; -*-
# Copyright (c) 2021, 2022 Oracle and/or its affiliates.
# Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/
import fsspec
import json
import yaml
from cerberus import Validator
from os import path
JOB_SCHEMA_PATH = "job_schema.json"
RUNTIME_SCHEMA_PATH = "runtime_schema.json"
INFRASTRUCTURE_SCHEMA_PATH = "infrastructure_schema.json"
[docs]def load_schema(schema_path):
if path.exists(path.join(path.dirname(__file__), schema_path)):
with fsspec.open(
path.join(path.dirname(__file__), schema_path), mode="r", encoding="utf8"
) as f:
schema = json.load(f)
return schema
else:
raise FileNotFoundError(f"{schema_path} does not exist")
[docs]class ValidateRuntime:
"""Class used to validate a Runtime YAML"""
def __init__(self, file):
self.file = file
self.schema = load_schema(RUNTIME_SCHEMA_PATH)
[docs] def validate(self):
"""Validates the Runtime YAML
Raises:
ValueError: if invalid
Returns:
[dict]: Returns normalized dictionary if input matches a known schema, else raises error
"""
v = Validator(self.schema)
valid = v.validate(self.file)
if not valid:
raise ValueError(json.dumps(v.errors, indent=2))
return v.normalized(self.file)
[docs]class ValidateInfrastructure:
"""Class used to validate an Engine YAML"""
def __init__(self, file):
self.file = file
self.schema = load_schema(INFRASTRUCTURE_SCHEMA_PATH)
[docs] def validate(self):
"""Validates the Engine YAML
Raises:
ValueError: if invalid
Returns:
[dict]: Returns normalized dictionary if input matches a known schema, else raises error
"""
v = Validator(self.schema)
valid = v.validate(self.file)
if not valid:
raise ValueError(json.dumps(v.errors, indent=2))
return v.normalized(self.file)
[docs]class ValidateJob:
"""Class used to validate a Job YAML"""
def __init__(self, file):
self.file = file
self.schema = load_schema(JOB_SCHEMA_PATH)
[docs] def validate(self):
"""Validates the Job YAML
Raises:
ValueError: if invalid
Returns:
[dict]: Returns normalized dictionary if input matches a known schema, else raises error
"""
v = Validator(self.schema)
job_valid = v.validate(self.file)
infrastructure_valid, runtime_valid = True, True
if "infrastructure" in self.file["spec"]:
infrastructure_valid = ValidateInfrastructure(
self.file["spec"]["infrastructure"]
).validate()
if "runtime" in self.file["spec"]:
runtime_valid = ValidateRuntime(self.file["spec"]["runtime"]).validate()
if not job_valid or not infrastructure_valid or not runtime_valid:
raise ValueError(json.dumps(v.errors, indent=2))
return v.normalized(self.file)
[docs]class ValidatorFactory:
"""ValidatorFactory is a factory class that calls appropriate
Validator class based on the 'kind'
Usage:
spec = {}
validated_dict = ValidatorFactory(spec).validate():
"""
def __init__(self, file):
if type(file) != dict:
raise ValueError("Input is not a dictionary")
else:
self.file = file
[docs] def validate(self):
"""Calls correct validator based on 'kind'
Raises:
TypeError: raised when 'kind' is not known
Returns:
[boolean]: Returns True if input matches a known schema, else False
"""
if "kind" not in self.file:
raise ValueError("No 'kind' found in input spec")
if self.file["kind"] == "job":
return ValidateJob(self.file).validate()
elif self.file["kind"] == "infrastructure":
return ValidateInfrastructure(self.file).validate()
elif self.file["kind"] == "runtime":
return ValidateRuntime(self.file).validate()
else:
raise TypeError(f"Unknown Kind: {self.file['kind']}")