Skip to main content

Task authoring and execution

Tasks are the atomic units of computation in Flyte. Every piece of user code that runs on the cluster is ultimately a task. Understanding how flytekit models tasks — from the base class hierarchy down to the @task decorator you use every day — makes it much easier to reason about caching, retries, serialization, and plugin behavior.

The Task Class Hierarchy​

Flytekit organizes tasks in a layered hierarchy, each layer adding more Python-specific capability:

Task  (flytekit.core.base_task)
  └── PythonTask[T]  (flytekit.core.base_task)
        └── PythonAutoContainerTask[T]  (flytekit.core.python_auto_container)
              ├── PythonFunctionTask[T]  (flytekit.core.python_function_task)
              │     └── AsyncPythonFunctionTask[T]
              │           └── EagerAsyncPythonFunctionTask[T]
              └── PythonInstanceTask[T]  (flytekit.core.python_function_task)

Task in flytekit/core/base_task.py is the closest representation to the FlyteIDL TaskTemplate. It carries a task_type string, a name, a TypedInterface, TaskMetadata, an optional SecurityContext, and docs. Every Task instance registers itself in FlyteEntities.entities on construction — that global list is how the serialization tooling discovers all tasks defined in a module.

PythonTask[T] adds a Python-native Interface (as opposed to the IDL TypedInterface), a generic task_config of type T, environment variables, and deck rendering controls. It also provides the central dispatch_execute() implementation that drives input/output translation via TypeEngine.

PythonFunctionTask is what the @task decorator produces. It infers the interface from Python type annotations via transform_function_to_interface(), stores the wrapped callable as _task_function, and supports three ExecutionBehavior modes.

Declaring Tasks with @task​

The simplest use is just decorating a typed function:

from flytekit import task, workflow
import typing

@task(enable_deck=True)
def t1(a: int) -> typing.NamedTuple("OutputsBC", t1_int_output=int, c=str):
    return a + 2, "world"

@task
def t2(a: str, b: str) -> str:
    return b + a

After decoration, t1 is a PythonFunctionTask instance. Flytekit calls update_wrapper(task_instance, decorated_fn) (in flytekit/core/task.py, line 449) so the task still looks like a regular callable — t1.__wrapped__ points back to the original function, and docstrings are preserved. You can stack custom decorators that use functools.wraps and @task will still work:

from functools import wraps
from flytekit import task

def my_decorator(fn):
    @wraps(fn)
    def wrapper(*args, **kwargs):
        print("before")
        return fn(*args, **kwargs)
    return wrapper

@task
@my_decorator
def my_task(x: int) -> int:
    return x + 1

One hard rule: the function must be accessible at module level. The default_task_resolver (used unless you provide a custom one) needs to reconstruct the task at remote execution time by importing the module and looking up the function by name. If PythonFunctionTask.__init__ detects a nested or inner function with the default resolver, it raises a ValueError:

TaskFunction cannot be a nested/inner or local function.
It should be accessible at a module level for Flyte to execute it.

The exception is functions inside test modules whose names start with test_, or functions wrapped with functools.wraps/update_wrapper.

The Full @task Parameter Surface​

Beyond a bare @task, you can configure compute policy, caching, images, secrets, and more:

Retries and timeouts:

import datetime
from flytekit import task

@task(retries=3, timeout=datetime.timedelta(hours=2))
def my_task(x: int) -> int:
    ...

timeout also accepts an integer number of seconds — TaskMetadata.__post_init__ converts it to a timedelta automatically.

Resources — unified form:

from flytekit import task, Resources

@task(resources=Resources(cpu=("1", "2"), mem=(1024, 2048), gpu=1))
def gpu_task() -> None:
    ...

The resources= parameter takes a Resources object where tuple values (request, limit) set both simultaneously. This is the preferred form. Mixing resources= with the older requests= or limits= parameters raises a ValueError. Passing a single scalar value to a resource field sets both request and limit to that value.

Container image:

@task(container_image="ghcr.io/myorg/myimage:v1.2")
def specialised_task() -> None:
    ...

# Template syntax referencing image config entries
@task(container_image="{{.images.default.fqn}}:tag-{{images.default.tag}}")
def another_task() -> None:
    ...

You can also pass an ImageSpec object to build the image automatically.

Secrets:

from flytekit import task, Secret
from os import getenv

secret_token = Secret(
    group="my-group",
    key="token",
    env_var="MY_SECRET",
    mount_requirement=Secret.MountType.ENV_VAR,
)

@task(secret_requests=[secret_token])
def get_token() -> str:
    return getenv("MY_SECRET", "")

Interruptibility:

@task(interruptible=True, retries=2)
def cheap_task(x: int) -> int:
    ...

When interruptible=None (the default), the task inherits the workflow's interruptibility setting. Explicit True or False overrides it.

Async functions decorated with @task automatically receive an AsyncPythonFunctionTask instance instead of a plain PythonFunctionTask. The @task decorator in flytekit/core/task.py checks inspect.iscoroutinefunction(fn) and switches the plugin class accordingly.

TaskMetadata: The Execution Policy Carrier​

TaskMetadata (a @dataclass in flytekit/core/base_task.py) holds everything that governs how a task execution is treated by the Flyte engine. The @task decorator constructs one internally, but you can also create it explicitly when using non-decorator task types like SQLTask or map_task:

from flytekit.core.base_task import TaskMetadata

metadata = TaskMetadata(
    cache=True,
    cache_version="v3",
    cache_serialize=True,
    retries=2,
    timeout=300,          # automatically converted to timedelta(seconds=300)
    interruptible=True,
    deprecated="Use new_task instead",
)

TaskMetadata.__post_init__ validates the combinations:

  • cache=True requires cache_version to be non-empty.
  • cache_serialize=True requires cache=True.
  • cache_ignore_input_vars requires cache=True.

Violating any of these raises ValueError immediately at definition time, not at runtime.

Caching​

The most common use is cache=True with an explicit cache_version:

@task(cache=True, cache_version="v1")
def is_even(n: int) -> bool:
    return n % 2 == 0

Once this runs locally with FLYTE_LOCAL_CACHE_ENABLED=true, the result is stored in ~/.flyte/local-cache. Subsequent calls with n=1 hit the cache and do not re-execute the function. Verified in tests/flytekit/unit/core/test_local_cache.py:

@task(cache=True, cache_version="v1")
def is_even(n: int) -> bool:
    global n_cached_task_calls
    n_cached_task_calls += 1
    return n % 2 == 0

# After check_evenness(n=1) runs once, n_cached_task_calls stays at 1
# even when called again with the same input.

To exclude certain inputs from the cache key (e.g., a logging verbosity flag that doesn't affect output):

@task(cache=True, cache_version="v1", cache_ignore_input_vars=("verbose",))
def compute(x: int, verbose: bool = False) -> int:
    ...

The Cache object (preferred): Instead of raw string versions, use Cache from flytekit.core.cache for auto-versioning based on function source code and container image:

from flytekit.core.cache import Cache

@task(cache=Cache(serialize=True, ignored_inputs=("verbose",)))
def compute(x: int, verbose: bool = False) -> int:
    ...

When Cache.version is not set, Cache.get_version() calls a chain of CachePolicy objects (obtained from get_plugin().get_default_cache_policies()) to hash the function body, container image, and other VersionParameters. The old cache_version, cache_serialize, and cache_ignore_input_vars keyword arguments still work for backwards compatibility, but the @task docstring marks them as deprecated.

The Execution Lifecycle​

Two Execution Paths​

When you call a task, flytekit takes one of two paths depending on context:

Local execution (calling a task directly in Python, or running a workflow locally):

task(**kwargs)
  → __call__ → flyte_entity_call_handler
    → task.local_execute(ctx, **kwargs)
      → translate_inputs_to_literals(...)   # Python values → LiteralMap
      → [cache lookup if metadata.cache]
      → task.sandbox_execute(ctx, literal_map)
        → task.dispatch_execute(ctx, literal_map)

Remote execution (the task runs inside a container on the Flyte cluster):

pyflyte-execute --inputs ... --output-prefix ... --resolver ... -- <loader_args>
  → entrypoint._dispatch_execute(ctx, task_loader, ...)
    → task.dispatch_execute(ctx, input_literal_map)

Both paths converge on dispatch_execute().

Inside dispatch_execute​

PythonTask.dispatch_execute() in flytekit/core/base_task.py does the following:

  1. Calls pre_execute(user_params) — a hook that can modify the execution context (e.g., initialize a Spark session).
  2. Translates the input LiteralMap to Python-native values via TypeEngine.literal_map_to_kwargs().
  3. Calls self.execute(**native_inputs) — the actual user function (or plugin override).
  4. Calls post_execute(user_params, result) — a hook for cleanup or output transformation.
  5. Translates Python outputs back to a LiteralMap via TypeEngine.async_to_literal().
  6. Writes deck HTML if decks are enabled.

Both pre_execute and post_execute are no-ops in PythonTask by default. Plugins override them — for example, the PyTorch plugin uses post_execute to raise IgnoreOutputs for non-rank-0 workers (see below).

IgnoreOutputs​

IgnoreOutputs is a sentinel exception class in flytekit/core/base_task.py:

class IgnoreOutputs(Exception):
    """
    This exception should be used to indicate that the outputs generated by this can be safely ignored.
    This is useful in case of distributed training or peer-to-peer parallel algorithms.
    """
    pass

When dispatch_execute raises FlyteUserRuntimeException(IgnoreOutputs()), the entrypoint's _dispatch_execute catches it and skips writing outputs.pb to blob storage entirely — confirmed by tests/flytekit/unit/bin/test_python_entrypoint.py: after the exception the mock_write_to_file.call_count is 0. Distributed training plugins like flytekitplugins-kfpytorch raise this in post_execute for worker ranks that don't produce the final result.

Dynamic Tasks with @dynamic​

Sometimes you don't know the shape of the workflow at registration time — for example, you want to fan out across a variable-length list of inputs. That's what @dynamic is for.

@dynamic is defined in flytekit/core/dynamic_workflow_task.py as a simple one-liner:

dynamic = functools.partial(task.task, execution_mode=PythonFunctionTask.ExecutionBehavior.DYNAMIC)

It creates a PythonFunctionTask with execution_mode=ExecutionBehavior.DYNAMIC. The function body runs at execution time (not at registration time like a regular workflow), and its output is a DynamicJobSpec — a description of the sub-workflow to execute:

from flytekit import task, dynamic
import typing

@task
def process(a: int) -> str:
    return "fast-" + str(a + 2)

@dynamic
def fan_out(a: int) -> typing.List[str]:
    results = []
    for i in range(a):          # 'a' is a real Python int here
        results.append(process(a=i))
    return results

Locally, fan_out(a=5) runs the function directly and returns ["fast-2", "fast-3", "fast-4", "fast-5", "fast-6"]. On the cluster, PythonFunctionTask.dynamic_execute() detects ExecutionState.Mode.TASK_EXECUTION and calls compile_into_workflow() instead. That method runs the function body in a compilation context, collecting task calls into a DynamicJobSpec with min_successes, tasks, nodes, and subworkflows. Propeller then executes that sub-workflow.

One caveat from the @dynamic docstring: the compiled sub-workflow is subject to the same scale limits as any workflow. Keep dynamic tasks under roughly fifty nodes; for large-scale parallel runs, use map_task instead.

node_dependency_hints is a parameter specific to @dynamic tasks. It lets you declare tasks, launchplans, or workflows that the dynamic task will reference at runtime, enabling them to be pre-registered:

launchplan0 = LaunchPlan.get_or_create(workflow0)

@dynamic(node_dependency_hints=[launchplan0])
def launch_dynamically():
    return [launchplan0] * 10

Passing node_dependency_hints to a non-dynamic task raises a ValueError.

Eager Workflows with @eager​

Eager workflows take a different approach than @dynamic. Rather than compiling a sub-workflow spec, every task call inside an @eager function triggers a real remote execution on the Flyte cluster. Python becomes the orchestrator.

from flytekit import task, eager

@task
def add_one(x: int) -> int:
    return x + 1

@eager(container_image="ghcr.io/myorg/image:v1")
async def simple_eager_workflow(x: int) -> int:
    out = add_one(x=x)      # kicks off a remote execution
    return out

The @eager decorator (in flytekit/core/task.py) creates an EagerAsyncPythonFunctionTask instance. The function must be async def. Key behaviors of EagerAsyncPythonFunctionTask:

  • Always sets metadata.is_eager = True and enable_deck=True on construction.
  • local_execution_mode() returns ExecutionState.Mode.EAGER_LOCAL_EXECUTION so that subtasks called within it follow the eager local path.
  • On backend execution, execute() constructs a Controller (from flytekit.core.worker_queue) connected to FlyteRemote, and delegates to run_with_backend().
  • run_with_backend() awaits the async function and, regardless of success or failure, renders an HTML deck of all eager sub-executions.

Failure cleanup: when an eager workflow fails mid-execution, child executions may still be running. EagerAsyncPythonFunctionTask.get_as_workflow() wraps the eager task in an ImperativeWorkflow and attaches an EagerFailureHandlerTask as the on_failure handler. This cleanup task queries Flyte Admin for all executions tagged with eager-exec: <parent_execution_name> and terminates them in a loop. Its resolver arguments are ['eager', 'failure', 'handler'], and the container command ends with --resolver flytekit.core.python_function_task.eager_failure_task_resolver.

# From tests/flytekit/unit/core/test_eager_cleanup.py
@eager(container_image=eager_image)
async def simple_eager_workflow(x: int) -> int:
    out = add_one(x=x)
    return out

imperative_wf = simple_eager_workflow.get_as_workflow()
# imperative_wf.failure_node.flyte_entity is an EagerFailureHandlerTask

Extending Tasks: Plugin Authoring​

PythonInstanceTask for Execute-Override Tasks​

When you want a task that has a platform-defined execute body (not a user function), subclass PythonInstanceTask:

from flytekit.core.python_function_task import PythonInstanceTask

class MyPluginTask(PythonInstanceTask[MyConfig]):
    def __init__(self, name: str, task_config: MyConfig, **kwargs):
        super().__init__(name=name, task_config=task_config, task_type="my-plugin-task", **kwargs)

    def execute(self, **kwargs):
        # custom execution logic
        ...

PythonInstanceTask ensures the module loader can find the task by variable name at remote execution time, which matters because there's no user function whose module and name to look up.

TaskResolverMixin — Rehydrating Tasks at Runtime​

When a task container starts on the cluster, it needs to reconstruct the Python task object from the command-line arguments baked into the container spec at serialization time. TaskResolverMixin is the abstract interface that defines this:

class TaskResolverMixin(object):
    @property
    @abstractmethod
    def location(self) -> str:  # importable dotted path to the resolver singleton
        pass

    @abstractmethod
    def loader_args(self, settings: SerializationSettings, t: Task) -> List[str]:
        # returns the args appended after '--' in the container command
        pass

    @abstractmethod
    def load_task(self, loader_args: List[str]) -> Task:
        # reconstructs the task from those args
        pass

The default_task_resolver in flytekit/core/python_auto_container.py generates args like ['task-module', 'my.module', 'task-name', 'my_func'], yielding a container command:

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

The location property must be an importable path to a module-level singleton; load_task is called at remote execution time with the same args that loader_args produced at serialization time.

TaskPlugins — The Plugin Registry​

TaskPlugins in flytekit/core/task.py is a class-level registry that maps a task_config type to a PythonFunctionTask subclass:

from flytekit.core.task import TaskPlugins

# In your plugin package's __init__.py:
TaskPlugins.register_pythontask_plugin(MyConfig, MyFunctionTask)

When @task(task_config=MyConfig(...)) is evaluated, TaskPlugins.find_pythontask_plugin(type(task_config)) returns MyFunctionTask and the decorator instantiates it. If no plugin is registered for the config type, it falls back to the base PythonFunctionTask. Attempting to register a duplicate config type with a different class raises TypeError.

Common Gotchas​

Nested functions are rejected. The default task resolver cannot find a function that isn't accessible at module level. This fails:

def outer():
    @task  # raises ValueError
    def inner(x: int) -> int:
        return x

The exception message suggests using functools.wraps or a custom TaskResolverMixin. Test modules (names starting with test_) are exempt from this restriction.

Cache configuration rules are strict. All three of these raise ValueError at definition time:

  • cache=True without cache_version (unless using the Cache object)
  • cache_serialize=True without cache=True
  • cache_ignore_input_vars=(...) without cache=True

disable_deck is deprecated. Since flytekit 1.10.0, use enable_deck=True instead. Both disable_deck and enable_deck cannot be set simultaneously — that also raises ValueError.

resources= conflicts with requests=/limits=. Use either the unified resources=Resources(cpu=("1","2")) form or the separate requests=/limits= form, not both. Mixing raises a ValueError with the message "resource can not be used together with...".

Async tasks and @dynamic don't mix. AsyncPythonFunctionTask.async_execute() raises NotImplementedError if called with ExecutionBehavior.DYNAMIC. Eager and dynamic are mutually exclusive execution modes.

Local caching requires an environment variable. Even if a task declares cache=True, local caching is only active when FLYTE_LOCAL_CACHE_ENABLED is set. Force a cache miss on an already-cached result with FLYTE_LOCAL_CACHE_OVERWRITE=true. The local cache lives at ~/.flyte/local-cache.

Unknown @task arguments raise immediately. Passing an unrecognized keyword (e.g., template= instead of pod_template=) raises ValueError: Unrecognized argument(s) for task: dict_keys(['template']) — caught in the wrapper function before the task is even constructed.