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.compiler_pb2.CompiledTask
"""
return _compiler_pb2.CompiledTask(
template=self.template # TODO: .to_flyte_idl()
)
def to_flyte_idl(self):
"""
:rtype: flyteidl.core.compiler_pb2.ConnectionSet.IdList
"""
return _compiler_pb2.ConnectionSet.IdList(ids=self.ids)
def to_flyte_idl(self):
"""
:rtype: flyteidl.core.compiler_pb2.CompiledWorkflow
"""
return _compiler_pb2.CompiledWorkflow(
template=self.template.to_flyte_idl(),
connections=self.connections.to_flyte_idl()
)
def to_flyte_idl(self):
"""
:rtype: flyteidl.core.compiler_pb2.ConnectionSet
"""
return _compiler_pb2.ConnectionSet(
upstream={k: v.to_flyte_idl() for k, v in _six.iteritems(self.upstream)},
downstream={k: v.to_flyte_idl() for k, v in _six.iteritems(self.upstream)}
)
def to_flyte_idl(self):
"""
:rtype: flyteidl.core.compiler_pb2.CompiledTask
"""
return _compiler.CompiledTask(
template=self.template.to_flyte_idl()
)