Task authoring and execution

Flyte tasks in flytekit are the fundamental building blocks of workflows. They are declared using decorators that transform Python functions into structured entities capable of local execution, remote serialization, and containerized execution on a Flyte cluster.

Task Declaration and Metadata

The primary way to define a task is using the @task decorator. This decorator captures the function's signature, docstrings, and execution policies to construct a PythonFunctionTask.

from flytekit import task
from datetime import timedelta

@task(
    retries=3,
    timeout=timedelta(minutes=5),
    cache=True,
    cache_version="1.0",
    interruptible=True
)
def square(x: int) -> int:
    return x * x

Execution Policies

Task behavior is governed by the TaskMetadata class in base_task.py. It tracks several key attributes:

• Caching: Controlled by cache and cache_version. If cache=True, a cache_version must be provided. You can also use cache_serialize=True to ensure that identical task instances execute serially.
• Retries: The retries integer determines how many times Flyte will attempt to re-run the task on failure.
• Timeouts: The timeout parameter (either an int seconds or datetime.timedelta) limits the maximum duration of a single execution.
• Interruptibility: Setting interruptible=True allows Flyte to schedule the task on lower-cost, pre-emptible nodes.

The Task Abstraction Stack

Flytekit uses a layered class hierarchy to manage the transition from Python code to Flyte IDL:

1. Task (base_task.py): The root base class. It handles the core Flyte identity (name, type, interface) and registration in FlyteEntities.
2. PythonTask (base_task.py): Adds a native Python Interface. It is responsible for translating between Flyte's literal types and Python native types using the TypeEngine.
3. PythonAutoContainerTask (python_auto_container.py): Handles containerization details, including image selection, resource requests/limits (CPU, memory), and environment variables.
4. PythonFunctionTask (python_function_task.py): The standard implementation for function-backed tasks. It detects the interface from type hints and executes the user function.

Local vs. Remote Execution

When you call a task locally, flytekit invokes Task.local_execute. This method:

1. Translates native Python inputs into Flyte Literal objects.
2. Checks the LocalTaskCache if caching is enabled.
3. Invokes dispatch_execute, which calls the actual user function.
4. Wraps the results back into Promise objects for use in workflows.

Specialized Task Types

Dynamic Tasks

Dynamic tasks allow you to generate new workflow structures at runtime based on input data. They are declared with the @dynamic decorator, which sets the execution_mode to ExecutionBehavior.DYNAMIC.

from flytekit import dynamic

@task
def t1(a: int) -> str:
    return str(a)

@dynamic
def my_dynamic_subwf(a: int) -> typing.List[str]:
    s = []
    for i in range(a):
        s.append(t1(a=i))
    return s

Internally, PythonFunctionTask.dynamic_execute compiles the function body into a DynamicJobSpec during remote execution, which Flyte Propeller then uses to expand the workflow graph.

Eager Workflows

Eager workflows (declared with @eager) allow for truly dynamic Pythonic execution where every task call is immediately executed on the Flyte backend. This uses EagerAsyncPythonFunctionTask and requires an async def.

from flytekit import task, eager
import asyncio

@eager
async def eager_workflow(x: int) -> int:
    out = await add_one(x=x)
    return await double(x=out)

# Local execution
result = asyncio.run(eager_workflow(x=1))

During backend execution, the EagerAsyncPythonFunctionTask uses a Controller to manage a worker queue that communicates with Flyte Admin to trigger sub-executions.

Serialization and Task Resolvers

To run a task in a container, Flyte needs to know how to re-instantiate the Python task object. This is handled by the TaskResolverMixin. The default command generated by PythonAutoContainerTask.get_container looks like this:

pyflyte-execute --inputs {{.input}} --output-prefix {{.outputPrefix}} \
  --resolver flytekit.core.python_auto_container.default_task_resolver \
  -- task-module my_module task-name my_task_function

The default_task_resolver uses the module path and function name to import and load the task at runtime. If you define tasks dynamically or in nested scopes, you may need a custom resolver, as PythonFunctionTask rejects nested functions by default to ensure they are importable on the cluster.

Implementation Details and Constraints

• Output Handling: Flyte tasks that return nothing still return a VoidPromise. If a task returns multiple values, flytekit expects a tuple or NamedTuple and maps them to the declared output interface.
• Decks: Tasks can generate HTML "Decks" for data visualization. This is enabled via enable_deck=True in the decorator. PythonFunctionTask automatically includes source code and dependency renderers in the generated decks.
• Ignore Inputs: You can use ignore_input_vars in PythonFunctionTask to prevent specific arguments from being included in the Flyte interface, which is useful for injecting client-side configuration that shouldn't be tracked by the platform.
• Type Safety: PythonTask asserts that the number of returned values matches the interface. If a task declared to return one value returns a tuple, _output_to_literal_map will raise a TypeError.