Core Task and Workflow Abstractions

Flytekit provides a structured way to define units of execution and their orchestration through the Task and WorkflowBase abstractions. These classes bridge the gap between Python-native code and the Flyte engine's internal representation (IDL).

Tasks: The Atomic Units of Execution

In flytekit, a task is the smallest unit of work. While most users interact with tasks via the @task decorator, the underlying behavior is governed by the Task and PythonTask classes in flytekit.core.base_task.

Defining a Task

When you use the @task decorator, flytekit creates an instance of PythonFunctionTask (a subclass of PythonTask). This class captures your function's signature, metadata, and execution logic.

from flytekit import task, Resources

@task(
    retries=3,
    cache=True,
    cache_version="1.0",
    requests=Resources(cpu="1", mem="1Gi"),
    limits=Resources(cpu="2", mem="2Gi")
)
def square(n: int) -> int:
    return n * n

Internally, the Task class (defined in flytekit/core/base_task.py) manages:

• Interface: A TypedInterface that maps Python types to Flyte IDL types.
• Metadata: A TaskMetadata object containing execution settings like retries and caching.
• Execution: The dispatch_execute method, which handles the translation of Flyte literals to Python-native values before calling your function.

Task Metadata and Configuration

The TaskMetadata class allows you to control how Flyte executes your task. If you enable caching (cache=True), you must also provide a cache_version.

# flytekit/core/base_task.py

class TaskMetadata(object):
    cache: bool = False
    cache_version: str = ""
    retries: int = 0
    timeout: Optional[Union[datetime.timedelta, int]] = None
    interruptible: Optional[bool] = None
    # ...

When a task is executed locally, flytekit's local_execute method checks the LocalTaskCache if caching is enabled. If a hit is found, it returns the cached LiteralMap instead of running the task again.

Task Resolvers

For a task to run on a remote Flyte cluster, the container needs to know how to find and load the specific Python task object. This is handled by the TaskResolverMixin. The default resolver, default_task_resolver, uses the module path and function name to rehydrate the task at runtime.

If you need custom loading logic (e.g., loading tasks from a database or a dynamic source), you can implement a custom resolver by inheriting from TaskResolverMixin and overriding load_task and loader_args.

Workflows: Orchestrating Tasks

Workflows organize tasks into a Directed Acyclic Graph (DAG). The WorkflowBase class (found in flytekit/core/workflow.py) is the foundation for all workflows.

Workflow Structure

A workflow is essentially a collection of Node objects, where each node encapsulates a task, a sub-workflow, or a launch plan.

from flytekit import workflow

@workflow
def my_workflow(n: int) -> int:
    result = square(n=n)
    return result

When the @workflow decorator is applied, flytekit:

1. Compiles the function: It executes the function body using Promise objects instead of real values.
2. Links Nodes: Every time a task is called within the workflow, PythonTask.compile is triggered, which calls create_and_link_node to add a new node to the workflow's internal graph.
3. Binds Outputs: The return values of the workflow function are bound to the workflow's output interface.

Execution Flow

The WorkflowBase.__call__ method handles both local execution and remote compilation:

• Local Execution: If no compilation context is active, flytekit executes the tasks locally in the order defined by the DAG, passing real Python values between them.
• Compilation: When preparing to register the workflow with Flyte Admin, flytekit traverses the graph to generate a WorkflowTemplate (a protobuf model) that describes the DAG structure.

Advanced Orchestration: Dynamic Tasks

Sometimes the structure of your DAG depends on runtime data (e.g., processing a variable number of files). In these cases, you use a Dynamic Task.

A dynamic task is a PythonFunctionTask with execution_mode set to DYNAMIC. When executed, it doesn't just return data; it returns a DynamicJobSpec containing a new sub-graph of tasks that Flyte Propeller will then execute.

@task
def dynamic_task(n: int) -> list[int]:
    nodes = []
    for i in range(n):
        nodes.append(square(n=i))
    return nodes

Internally, PythonFunctionTask.compile_into_workflow is called during the execution of a dynamic task to generate this runtime-defined DAG.