Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
secretKey = fields.Str(allow_none=True)
readOnly = fields.Bool(allow_none=True)
@staticmethod
def schema_config():
return PersistenceEntityConfig
@validates_schema
def validate_persistence(self, data):
existing_claim = data.get("existingClaim")
host_path = data.get("hostPath")
store = data.get("store")
validate_persistence(existing_claim, host_path, store)
class PersistenceEntityConfig(BaseConfig):
SCHEMA = PersistenceEntitySchema
REDUCED_ATTRIBUTES = [
"existingClaim",
"mountPath",
"hostPath",
"store",
"bucket",
"secret",
"secretKey",
"readOnly",
]
def __init__(
self, # noqa
existingClaim=None,
mountPath=None,
class ReplicaSchema(BaseSchema):
replicas = fields.Int(allow_none=True)
environment = fields.Nested(EnvironmentSchema, allow_none=True)
termination = fields.Nested(TerminationSchema, allow_none=True)
init = fields.Nested(InitSchema, allow_none=True)
mounts = fields.Nested(MountsSchema, allow_none=True)
container = fields.Nested(ReplicaContainerSchema, allow_none=True)
@staticmethod
def schema_config():
return ReplicaConfig
class ReplicaConfig(BaseConfig, V1Replica):
SCHEMA = ReplicaSchema
IDENTIFIER = "replication"
REDUCED_ATTRIBUTES = [
"replicas",
"environment",
"termination",
"mounts",
"container",
]
@property
def is_categorical(self):
return False
class MatrixQNormalSchema(BaseSchema):
kind = fields.Str(allow_none=True, validate=validate.Equal("qnormal"))
value = QNormal(allow_none=True)
@staticmethod
def schema_config():
return MatrixQNormalConfig
class MatrixQNormalConfig(BaseConfig):
SCHEMA = MatrixQNormalSchema
IDENTIFIER = "qnormal"
def __init__(self, value, kind=IDENTIFIER):
self.kind = kind
self.value = value
@property
def is_distribution(self):
return True
@property
def is_uniform(self):
return False
@property
@property
def is_uniform(self):
return False
class MatrixLinSpaceSchema(BaseSchema):
kind = fields.Str(allow_none=True, validate=validate.Equal("linspace"))
value = LinSpace(allow_none=True)
@staticmethod
def schema_config():
return MatrixLinSpaceConfig
class MatrixLinSpaceConfig(BaseConfig):
SCHEMA = MatrixLinSpaceSchema
IDENTIFIER = "linspace"
def __init__(self, value, kind=IDENTIFIER):
self.kind = kind
self.value = value
@property
def is_distribution(self):
return False
@property
def is_continuous(self):
return False
@property
def get_config(cls):
if not cls.is_initialized():
return None
config_filepath = cls.get_config_filepath()
with open(config_filepath, "r") as config_file:
config_str = config_file.read()
if issubclass(cls.CONFIG, BaseConfig):
return cls.CONFIG.from_dict(ujson.loads(config_str))
return cls.CONFIG(**ujson.loads(config_str))
return IOConfig
@validates_schema
def validate_io(self, values):
validate_io(
name=values.get("name"),
iotype=values.get("iotype"),
value=values.get("value"),
is_list=values.get("is_list"),
is_optional=values.get("is_optional"),
is_flag=values.get("is_flag"),
options=values.get("options"),
)
class IOConfig(BaseConfig, V1IO):
SCHEMA = IOSchema
IDENTIFIER = "io"
REDUCED_ATTRIBUTES = [
"description",
"type",
"value",
"is_optional",
"is_flag",
"is_list",
"options",
]
def validate_value(self, value, parse=True):
if self.iotype is None:
return value
from polyaxon.schemas.base import BaseConfig, BaseSchema
class ContainerEnvSchema(BaseSchema):
image = fields.Str(required=True)
image_pull_policy = fields.Str(allow_none=True)
sleep_interval = fields.Int(allow_none=True)
outputs_sync_interval = fields.Int(allow_none=True)
logs_sync_interval = fields.Int(allow_none=True)
@staticmethod
def schema_config():
return ContainerEnvConfig
class ContainerEnvConfig(BaseConfig, V1ContainerEnv):
SCHEMA = ContainerEnvSchema
IDENTIFIER = "container"
REDUCED_ATTRIBUTES = [
"image_pull_policy",
"sleep_interval",
"outputs_sync_interval",
"logs_sync_interval",
]
@staticmethod
def schema_config():
return ComponentRefConfig
@validates_schema
def validate_component_ref(self, values):
validate_component_ref(
name=values.get("name"),
url=values.get("url"),
path=values.get("path"),
hub=values.get("hub"),
)
class ComponentRefConfig(BaseConfig, V1ComponentRef):
SCHEMA = ComponentRefSchema
IDENTIFIER = "component_ref"
REDUCED_ATTRIBUTES = ["name", "url", "path", "hub"]
def validate(self):
validate_component_ref(
name=self.name, url=self.url, path=self.path, hub=self.hub
)
from polyaxon.schemas.fields.ref_or_obj import RefOrObject
class CronScheduleSchema(BaseSchema):
kind = fields.Str(allow_none=True, validate=validate.Equal("cron"))
start_at = RefOrObject(fields.LocalDateTime(allow_none=True))
end_at = RefOrObject(fields.LocalDateTime(allow_none=True))
cron = RefOrObject(fields.String(required=True))
depends_on_past = RefOrObject(fields.Bool(allow_none=True))
@staticmethod
def schema_config():
return CronScheduleConfig
class CronScheduleConfig(BaseConfig, V1CronSchedule):
SCHEMA = CronScheduleSchema
IDENTIFIER = "cron"
IDENTIFIER_KIND = True
from polyaxon_sdk import V1ExactTimeSchedule
from polyaxon.schemas.base import BaseConfig, BaseSchema
from polyaxon.schemas.fields.ref_or_obj import RefOrObject
class ExactTimeScheduleSchema(BaseSchema):
kind = fields.Str(allow_none=True, validate=validate.Equal("exact_time"))
start_at = RefOrObject(fields.LocalDateTime(required=True), required=True)
@staticmethod
def schema_config():
return ExactTimeScheduleConfig
class ExactTimeScheduleConfig(BaseConfig, V1ExactTimeSchedule):
SCHEMA = ExactTimeScheduleSchema
IDENTIFIER = "exact_time"
IDENTIFIER_KIND = True