Skip to main content

Task authoring and execution

A Flyte task is more than a Python callable: it has a typed interface, task metadata, a runtime target, and enough information to be reconstructed in a worker. The public @task decorator is the usual entry point for function-backed tasks, while the lower-level task classes are the extension points for plugins and task types whose execution method is supplied by the platform.

The task abstraction layers​

The core inheritance path for a normal Python function task is:

Task
└── PythonTask
    └── PythonAutoContainerTask
        └── PythonFunctionTask

Task in flytekit/core/base_task.py is the IDL-oriented base. Its constructor stores the task type, name, typed interface, metadata, security context, and documentation, and registers the instance in FlyteEntities.entities. Calling a task delegates to flyte_entity_call_handler; depending on the current FlyteContext, that call either contributes a node during workflow compilation or follows the local execution path.

The base class separates the two execution surfaces that subclasses provide:

  • compile() creates a workflow node. The base implementation is abstract in effect (NotImplementedError), while PythonTask.compile() calls create_and_link_node().
  • dispatch_execute() receives a Flyte LiteralMap and invokes the task executor. It is abstract on Task.
  • pre_execute() and execute() are lifecycle hooks and are abstract on Task.

The methods get_container(), get_k8s_pod(), get_sql(), get_custom(), get_config(), and get_extended_resources() return None on Task. Container, pod, SQL, or plugin-specific task classes override the relevant method to describe the hosted runtime target.

PythonTask adds the Python-facing Interface, plugin task_config, environment variables, and deck configuration. It also implements the standard conversion boundary:

LiteralMap inputs
    -> TypeEngine.literal_map_to_kwargs()
    -> execute(**native_inputs)
    -> post_execute()
    -> TypeEngine.async_to_literal()
    -> LiteralMap outputs

Its dispatch_execute() calls pre_execute() before converting inputs, invokes execute(), calls post_execute(), and converts the returned native values according to the declared output interface. PythonTask.construct_node_metadata() carries timeout, retries, and interruptibility into the workflow node.

Declaring a Python function task​

Use @task for a typed function. The function annotations become its Python interface, and calling the resulting task locally uses native values and returns native values or task-output wrappers as appropriate:

from flytekit import task
import typing

@task
def my_task(a: int) -> typing.NamedTuple("OutputsBC", b=int, c=str):
    return a + 2, "hello world"

assert my_task(a=3) == (5, "hello world")

This is the pattern exercised in tests/flytekit/unit/core/test_type_hints.py. Internally, PythonFunctionTask.__init__() calls transform_function_to_interface() and then transform_interface_to_typed_interface(). It also derives the task name from the function's module and name. With the default resolver, the function must be accessible at module scope; nested or local functions cause PythonFunctionTask to raise a ValueError (test modules are explicitly allowed by the resolver check). If a decorator wraps the function, use functools.wraps or functools.update_wrapper, or provide a custom TaskResolverMixin.

The task() function in flytekit/core/task.py packages decorator options into TaskMetadata, selects a plugin with TaskPlugins.find_pythontask_plugin(type(task_config)), and constructs that plugin with the function and its configuration. If the function is a coroutine and the selected plugin is the ordinary PythonFunctionTask, task() changes the class to AsyncPythonFunctionTask automatically. A registered task plugin can therefore add behavior for a particular task_config type without changing the user-facing decorator.

Common task-level configuration is passed directly to the decorator:

from flytekit import Resources, task

@task(
    retries=2,
    timeout=60,
    interruptible=True,
    resources=Resources(cpu=("1", "2"), mem=(1024, 2048), gpu=1),
)
def process(value: int) -> int:
    return value + 1

The integer timeout is interpreted as seconds and becomes a datetime.timedelta in TaskMetadata. The resources form combines requests and limits: tuple values are request/limit pairs, while a scalar applies to both. Do not combine resources with the separate requests or limits arguments.

Task metadata, retries, and caching​

TaskMetadata is the dataclass used by Task and contains task execution policy rather than function code. Its fields include:

  • cache, cache_serialize, cache_version, and cache_ignore_input_vars for caching;
  • retries and timeout for execution policy;
  • interruptible, pod_template_name, and deprecated;
  • generates_deck and is_eager.

Constructing metadata directly validates the cache and timeout combinations:

from flytekit import TaskMetadata

metadata = TaskMetadata(
    cache=True,
    cache_version="v1",
    cache_serialize=True,
    retries=2,
    timeout=60,
)

TaskMetadata.__post_init__() rejects cache=True without a cache version, cache_serialize=True without caching, and cache_ignore_input_vars without caching. A timeout may be an integer number of seconds or a datetime.timedelta; other values raise ValueError. retry_strategy exposes the retry model, and to_taskmetadata_model() converts the metadata to the Flyte IDL model, including timeout, retries, interruptibility, discovery version, deprecation message, and eager state.

For the decorator API, prefer the Cache object. The current task() implementation accepts either a boolean or Cache; when given a Cache, it obtains the version with cache.get_version(VersionParameters(...)), then fills TaskMetadata with the resulting version, serialization setting, and ignored inputs:

from flytekit import Cache, task

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

The older separate cache_version, cache_serialize, and cache_ignore_input_vars arguments remain for compatibility but are deprecated. Passing those old arguments together with a Cache object raises ValueError. The local execution path consults LocalTaskCache when both task caching and local caching are enabled: it builds a literal input map, checks the task name/version/input key, executes on a miss, and stores the resulting literal map. cache_ignore_input_vars is applied when the cache key is looked up and written.

Normal, dynamic, and eager function execution​

PythonFunctionTask.ExecutionBehavior has three values: DEFAULT, DYNAMIC, and EAGER. Normal users select the special modes with @dynamic and @eager; task() documents execution_mode as mainly internal.

Default tasks​

With the default mode, PythonFunctionTask.execute() simply calls the captured function:

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

During PythonTask.dispatch_execute(), Flyte literals are converted into the annotated Python inputs, PythonFunctionTask.execute() invokes add_one(**kwargs), and the return value is converted back into a literal map. In local execution, Task.local_execute() performs the input translation, invokes sandbox_execute(), and wraps declared outputs in Promise objects (or returns a VoidPromise for a task with no outputs).

Dynamic tasks​

Use @dynamic when the function body constructs the workflow graph at execution time:

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

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

assert ranged_int_to_str(a=5) == [
    "fast-2", "fast-3", "fast-4", "fast-5", "fast-6"
]

PythonFunctionTask.execute() routes DYNAMIC tasks to dynamic_execute(). For local execution, dynamic_execute() creates or reuses a PythonFunctionWorkflow, runs it with local dynamic execution state, and translates its native results into a literal map. During task execution, compile_into_workflow() compiles that generated workflow and returns a DynamicJobSpec containing its task templates, nodes, outputs, and subworkflows. A dynamic function that produces no nodes can instead return a strict output literal map.

node_dependency_hints is an optional list of tasks, launch plans, or workflows for dynamic tasks when dependencies cannot be discovered before runtime. Supplying it to a non-dynamic PythonFunctionTask raises ValueError.

Eager async tasks​

Use @eager with an async def function when task calls inside the function should be represented by backend executions rather than only by an in-process Python call graph:

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

@eager
async def simple_eager_workflow(x: int) -> int:
    out = add_one(x=x)
    return out

The eager() decorator constructs EagerAsyncPythonFunctionTask, forces ExecutionBehavior.EAGER, and sets TaskMetadata.is_eager=True. At local execution it switches to EAGER_LOCAL_EXECUTION and awaits the user function. At backend execution it creates a Controller worker queue, installs signal handlers for SIGINT and SIGTERM, and runs the function in eager execution mode. run_with_backend() renders the controller's execution information into an “Eager Executions” deck and converts an EagerException into a non-recoverable system failure.

get_as_workflow() adapts an eager task into an ImperativeWorkflow and adds an EagerFailureHandlerTask as its failure handler. That handler uses EagerFailureTaskResolver and, at runtime, lists child executions tagged with the parent eager execution and terminates those in UNDEFINED, QUEUED, or RUNNING phases.

AsyncPythonFunctionTask is the ordinary async function variant. Its __call__ uses async_flyte_entity_call_handler, and its synchronous execute wrapper awaits the function. It does not support dynamic execution: selecting DYNAMIC for an async task raises NotImplementedError. Eager and dynamic execution are separate modes.

Tasks without a user function​

Not every task has a function whose signature Flyte can inspect. kwtypes() converts keyword type declarations into an ordered mapping and is used by tasks such as SQLTask to declare inputs and outputs explicitly:

sql = SQLTask(
    "my-query",
    query_template="SELECT * FROM ... WHERE ds = '{{ .Inputs.ds }}'",
    inputs=kwtypes(ds=datetime.datetime),
    outputs=kwtypes(results=FlyteSchema),
    metadata=TaskMetadata(retries=2),
)

For plugin authors, PythonInstanceTask is the corresponding base for a task with no user-defined function body. Its platform-defined execute() method must be implemented by the subclass. A task instance is assigned to a module variable and called through its declared interface:

x = MyInstanceTask(name="x", task_config=...)
x(a=5)

The class extends PythonAutoContainerTask and passes its task name, configuration, task type, and optional resolver to the parent. Shell and dbt tasks are examples of this instance-task pattern.

Resolution and rehydration​

Serialization and execution happen in different processes, so Flyte needs a way to identify a task inside its module. TaskResolverMixin defines that contract with location, name, loader_args(), load_task(), and get_all_tasks().

The default resolver serializes a function task into a command shaped like:

pyflyte-execute --inputs s3://path/inputs.pb --output-prefix s3://outputs/location \
  --raw-output-data-prefix /tmp/data \
  --resolver flytekit.core.python_auto_container.default_task_resolver \
  -- \
  task-module repo_root.workflows.example task-name t1

The resolver's loader_args() supplies the module and task name after --. At runtime, the default loader imports the module and retrieves the named task. This is why a default-resolver function must be module-accessible, and why nested functions fail the PythonFunctionTask validation. Supply a custom TaskResolverMixin through task_resolver= when a task cannot be reconstructed using module and variable lookup.

Execution hooks and output handling​

Override pre_execute() when a plugin must establish runtime state before Flyte converts inputs. The Spark plugin demonstrates this ordering: its pre_execute() builds a SparkSession, applies local Spark configuration when execution is not in TASK_EXECUTION mode, and stores the session before the task's inputs are converted. post_execute() runs after execute() and is a no-op in PythonTask; plugins can use it to clean up or adjust results.

A task can signal that its outputs should not be uploaded by raising IgnoreOutputs. The PyTorch elastic task raises it when a worker receives a termination signal and when no rank-zero result exists:

try:
    out = elastic_launch(...)
except SignalException as e:
    logger.exception(f"Elastic launch agent process terminating: {e}")
    raise IgnoreOutputs()

if 0 in out:
    return out[0].return_value
else:
    raise IgnoreOutputs()

IgnoreOutputs is an exception marker defined in base_task. The runtime entrypoint recognizes it when it is carried by the expected user-runtime exception path; distributed workers use it so only the rank-zero result is returned.

Configuration and testing edge cases​

  • Use enable_deck, not disable_deck. disable_deck is deprecated; setting both raises ValueError. Deck fields are retained only when deck output is enabled. The default deck_fields tuple contains source code, dependencies, timeline, input, and output fields.
  • Keep default-resolver task functions at module scope. For wrapped functions, preserve their identity with functools.wraps; otherwise use a custom resolver.
  • Do not combine node_dependency_hints with a static task.
  • Do not combine async tasks with dynamic execution.
  • Use one Resources object for combined requests and limits, or use separate requests and limits; the decorator rejects conflicting resource forms.
  • Task instances are registered during Task.__init__(), so module-level task objects persist in FlyteEntities.entities for the lifetime of the Python process.
  • For unit tests, flytekit.core.testing.task_mock and patch can replace a task's execution with a mock while preserving the task's use in a workflow. This is useful for testing workflow composition without running the task implementation.