flyte.orchestrator

Tasks that orchestrate other tasks.

@flyte.orchestrator.task turns a function that only calls other tasks into a task that Flyte runs without a pod of its own, so it takes up none of the task pod resources a regular task does. The tasks it calls run as usual.

import flyte
import flyte.orchestrator
import flyte.remote

add = flyte.remote.Task.get("math.add", auto_version="latest")
multiply = flyte.remote.Task.get("math.multiply", auto_version="latest")


@flyte.orchestrator.task
def pipeline(x: int, y: int) -> int:
    return multiply(add(x, y), 2)


run = flyte.run(pipeline, x=3, y=4)

It may only call tasks. For an orchestrator that also calls traces or durable operations, use env.orchestrator.task on a flyte.TaskEnvironment, which runs in the environment’s image.

Directory

Classes

Methods

Methods

task()

def task(
    name: Optional[str] = None,
    queue: Optional[str] = None,
    timeout_ms: int = 30000,
    max_stack_depth: int = 256,
    cache: CacheRequest = 'disable',
    retries: int = 0,
) -> Union[OrchestratorTaskTemplate, Callable[[Callable], OrchestratorTaskTemplate]]

Turn a function that only calls other tasks into an orchestrator task.

The function’s source is shipped in the task template and executed by a sandbox leaseworker, so the orchestrator itself starts without a pod:

add = flyte.remote.Task.get("math.add", auto_version="latest")

@flyte.orchestrator.task
def pipeline(x: int, y: int) -> int:
    return add(add(x, y), 1)

run = flyte.run(pipeline, x=1, y=2)

Tasks are discovered from the function’s globals; call them without await. They can be:

• remote references (flyte.remote.Task.get). Nothing is built, so the run is submitted straight away.
• tasks defined locally with @env.task. Their image and code bundle are built when the orchestrator is run, as for any flyte.run, and their specs travel in the orchestrator’s template.

Values pass between the orchestrator and its tasks as they would between tasks, but the orchestrator itself can only look into plain data (None, bool, int, float, str, and lists and str-keyed dicts of them) and files and directories. A File or Dir is a named tuple of its path, name, format and hash. A dataframe, Pydantic model or dataclass, such as a Volume, is opaque: the orchestrator can pass it on, return it, or keep it in a list or dict, but not read its fields. Nothing can be opened or read from storage inside the orchestrator; call a task for that.

It can also wait for a value someone supplies from the UI or the API, with flyte.new_condition and the condition’s wait(), again without await. While it waits, nothing runs and no pod is held:

@flyte.orchestrator.task
def deploy(model: str) -> str:
    report = evaluate(model)
    approval = flyte.new_condition("approve", prompt=f"Deploy {model}? {report}", timeout=86_400)
    if approval.wait():
        return publish(model)
    return "rejected"

data_type is one of bool, int, float and str, and timeout is in seconds or a timedelta, imported inside the function since only its source is shipped. Webhooks are not supported here.

A condition that is not signalled in time raises flyte.errors.ConditionTimedoutError from wait(), and one that fails or is aborted raises flyte.errors.ConditionFailedError or flyte.errors.ActionAbortedError. The orchestrator runs without the SDK’s exception classes, so inside it these are the builtins TimeoutError, RuntimeError and RuntimeError: except flyte.errors.ConditionTimedoutError catches only timeouts, but except RuntimeError does not catch them.