Workflow composition and nodes

Flyte workflows are declarative structures that define a Directed Acyclic Graph (DAG) of execution units called nodes. In flytekit, you compose these workflows by decorating Python functions with the @workflow decorator.

Workflow Compilation and DAG Construction

When you decorate a function with @workflow, flytekit treats the function body as a blueprint rather than a standard Python script. The function is executed at compile-time (serialization-time) to discover the tasks and their dependencies.

During this phase, the inputs passed to tasks are not actual values but Promises—placeholders that represent data that will exist at runtime.

from flytekit import task, workflow

@task
def get_greeting(name: str) -> str:
    return f"Hello, {name}!"

@task
def emphasize(text: str) -> str:
    return f"{text}!!!"

@workflow
def my_workflow(name: str) -> str:
    # 'greeting' is a Promise here, not a string
    greeting = get_greeting(name=name)
    # Passing the Promise to another task creates a data dependency
    return emphasize(text=greeting)

Internally, flytekit uses the FlyteContextManager to track these calls. Every time a task is invoked within a workflow, flytekit creates a Node and adds it to the workflow's internal representation.

Working with Nodes

A Node (defined in flytekit.core.node.Node) is the fundamental unit of a workflow graph. While most nodes are created automatically when you call a task, you can also manage them explicitly.

Explicit Node Creation

If you need to define execution order between tasks that do not share data, use create_node from flytekit.core.node_creation. This is common when one task must finish before another starts (e.g., a setup task and a processing task).

You can enforce order using the >> operator (which calls Node.runs_before):

from flytekit import task, workflow
from flytekit.core.node_creation import create_node

@task
def setup():
    print("Setting up...")

@task
def do_work():
    print("Working...")

@workflow
def manual_dependencies_wf():
    setup_node = create_node(setup)
    work_node = create_node(do_work)
    
    # setup_node must complete before work_node starts
    setup_node >> work_node

Accessing Node Outputs

When using create_node, the return value is a Node object. To pass its outputs to subsequent tasks, access them via the .o0, .o1, etc., attributes or the .outputs dictionary.

@task
def produce_data() -> (int, str):
    return 1, "data"

@workflow
def output_access_wf():
    node = create_node(produce_data)
    # Accessing the first output (index 0)
    consume_task(val=node.o0) 

Customizing Nodes with Overrides

The Node.with_overrides() method allows you to modify the execution settings for a specific task invocation within a workflow. This is useful for adjusting resources or retries without changing the underlying task definition.

Supported overrides include:

• Resources: requests and limits using flytekit.Resources.
• Retries: retries as an integer.
• Timeout: timeout as a datetime.timedelta or integer seconds.
• Caching: cache and cache_version.
• Metadata: node_name (to change the ID in the Flyte UI) and interruptible.

from datetime import timedelta
from flytekit import Resources

@workflow
def overrides_wf(val: int):
    # Override resources and retries for this specific call
    node = create_node(my_task, val=val).with_overrides(
        node_name="high-priority-task",
        requests=Resources(cpu="2", mem="4Gi"),
        retries=3,
        timeout=timedelta(minutes=10),
        interruptible=True
    )

Internal Implementation of Overrides

When with_overrides is called, it modifies the Node instance's internal _metadata and _resources attributes. For example, the _override_node_metadata method handles the logic for converting integer timeouts into datetime.timedelta and validating cache settings:

# From flytekit/core/node.py
def _override_node_metadata(self, name, timeout, retries, ...):
    if timeout is not Node.TIMEOUT_OVERRIDE_SENTINEL:
        if isinstance(timeout, int):
            node_metadata._timeout = datetime.timedelta(seconds=timeout)
        elif isinstance(timeout, datetime.timedelta):
            node_metadata._timeout = timeout
    # ... sets retries, interruptible, etc.

Workflow Metadata and Policies

The @workflow decorator accepts several parameters that control the behavior of the entire DAG:

• failure_policy: Defines what happens when a node fails. WorkflowFailurePolicy.FAIL_IMMEDIATELY (default) stops the workflow, while FAIL_AFTER_EXECUTABLE_NODES_COMPLETE allows independent branches to finish.
• on_failure: A task or workflow to execute if the main workflow fails. This is often used for cleanup or notifications.
• interruptible: Sets the default interruptible state for all nodes in the workflow.

from flytekit.core.workflow import WorkflowFailurePolicy

@workflow(
    failure_policy=WorkflowFailurePolicy.FAIL_AFTER_EXECUTABLE_NODES_COMPLETE,
    interruptible=True
)
def robust_workflow(val: int):
    ...

These settings are encapsulated in WorkflowMetadata and WorkflowMetadataDefaults classes during the wrapping process in flytekit.core.workflow.workflow.