Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def to_flyte_idl(self):
"""
:rtype: flyteidl.core.tasks_pb2.RuntimeMetadata
"""
return _core_task.RuntimeMetadata(
type=self.type,
version=self.version,
flavor=self.flavor
)
def to_flyte_idl(self):
"""
:rtype: flyteidl.core.tasks_pb2.Container
"""
return _core_task.Container(
image=self.image,
command=self.command,
args=self.args,
resources=self.resources.to_flyte_idl(),
env=[_literals_pb2.KeyValuePair(key=k, value=v) for k, v in _six.iteritems(self.env)],
config=[_literals_pb2.KeyValuePair(key=k, value=v) for k, v in _six.iteritems(self.config)]
)
def to_flyte_idl(self):
"""
:rtype: flyteidl.core.tasks_pb2.TaskTemplate
"""
return _core_task.TaskTemplate(
id=self.id.to_flyte_idl(),
type=self.type,
metadata=self.metadata.to_flyte_idl(),
interface=self.interface.to_flyte_idl(),
custom=_json_format.Parse(_json.dumps(self.custom), _struct.Struct()) if self.custom else None,
container=self.container.to_flyte_idl() if self.container else None
)
def to_flyte_idl(self):
"""
:rtype: flyteidl.admin.task_pb2.TaskMetadata
"""
tm = _core_task.TaskMetadata(
discoverable=self.discoverable,
runtime=self.runtime.to_flyte_idl(),
retries=self.retries.to_flyte_idl(),
interruptible=self.interruptible,
discovery_version=self.discovery_version,
deprecated_error_message=self.deprecated_error_message
)
tm.timeout.FromTimedelta(self.timeout)
return tm