Task authoring and execution

Declaring a task

When Python code needs to run as a Flyte node, decorate a typed function instead of registering a bare callable. flytekit.task is the normal construction path: annotations become the task's input and output interface, while decorator arguments become task metadata and container configuration.

from flytekit import task

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

For a task that uses a plugin configuration, pass the configuration to the same decorator. The decorator documentation shows this form:

@task(task_config=Spark(), retries=3)
def my_task(x: int, y: typing.Dict[str, str]) -> str:
    ...

Spark() and typing are the names used by the source example; the concrete plugin must be available in the environment where the task is declared. The decorator also accepts container_image, environment, requests, limits, resources, secret_requests, pod_template, pod_template_name, accelerator, shared_memory, task_resolver, execution_mode, node_dependency_hints, docs, enable_deck, deck_fields, and pickle_untyped. Unknown keyword arguments are rejected by task rather than silently passed through.

What the decorator constructs

The implementation in task.py first creates a TaskMetadata value from cache settings, retries, interruptibility, deprecation text, and timeout. It then selects a Python-task plugin with TaskPlugins.find_pythontask_plugin(type(task_config)). For a coroutine function, the built-in PythonFunctionTask plugin is replaced with AsyncPythonFunctionTask; a third-party plugin must itself be compatible with AsyncPythonFunctionTask. Finally, the decorator instantiates the selected plugin and calls update_wrapper so the task object retains the decorated function's wrapper metadata.

The resulting hierarchy is:

Task
└── PythonTask
    └── PythonAutoContainerTask
        └── PythonFunctionTask
            ├── AsyncPythonFunctionTask
            └── EagerAsyncPythonFunctionTask

Task is the IDL-facing abstraction. Its constructor stores the task type, name, typed Flyte interface, metadata, task-type version, security context, and documentation, and appends the task to FlyteEntities.entities. It is callable through Task.__call__, but it does not assume that the task has a Python-native signature.

PythonTask adds that native signature as an Interface. Its constructor transforms the native interface into a Flyte typed interface, keeps the Python interface for conversion, and stores plugin configuration and environment variables. PythonFunctionTask supplies the missing function-specific pieces: it calls transform_function_to_interface using the function annotations and Docstring, removes any ignore_input_vars, and derives the task name from the function's module and name.

Calling tasks while authoring workflows

A task call has two meanings, selected by the active Flyte context. During workflow compilation, call the task with keyword inputs to create a node and obtain promise outputs. The imperative workflow example in workflow.py uses this directly:

@task
def t1(a: str) -> str:
    return a + " world"

@task
def t2():
    print("side effect")

wb = Workflow(name="my_workflow")
wb.add_workflow_input("in1", str)
node = wb.add_entity(t1, a=wb.inputs["in1"])
wb.add_entity(t2)

PythonTask.compile delegates to create_and_link_node, which connects the task interface to the workflow's node inputs and returns promises. The node-creation path expects keyword inputs; create_node rejects positional arguments with a Flyte assertion.

For local execution, Task.local_execute first translates native values, promises, and collections containing them into a LiteralMap. It calls sandbox_execute, which establishes a task sandbox and dispatches through dispatch_execute. The literal outputs are then wrapped as Promise values (or as a VoidPromise when the interface has no outputs). Thus the execution path is:

Python/native inputs
  -> Task.local_execute
  -> translate_inputs_to_literals
  -> PythonTask.dispatch_execute
  -> TypeEngine literal-to-native conversion
  -> execute(**native_inputs)
  -> TypeEngine native-to-literal conversion
  -> Promise / VoidPromise outputs

PythonTask.dispatch_execute invokes pre_execute before input conversion, calls execute, invokes post_execute, and converts the result according to the declared output names. It also writes configured decks. Local exceptions retain the original exception with task context added; remote execution wraps user and system failures in Flyte exceptions.

Metadata, caching, and execution configuration

Use TaskMetadata when constructing a task directly or when configuring mapped-task instances. Its fields include retries, timeout, interruptibility, deprecation text, cache behavior, pod-template name, deck generation, and eager state. The decorator's retries, timeout, interruptible, and related arguments are assembled into this object before the task plugin is instantiated.

Caching has explicit validation rules:

from flytekit import TaskMetadata

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

An integer timeout is converted to datetime.timedelta(seconds=...). A truthy timeout of any other type raises ValueError. cache=True requires a nonempty cache_version; cache_serialize=True and cache_ignore_input_vars are valid only when caching is enabled. retry_strategy() converts the retry count into Flyte's retry model, and to_taskmetadata_model() serializes the metadata together with the installed flytekit SDK version.

The public decorator supports the newer Cache object. If cache=True is used without an explicit version, the decorator creates a Cache with the supplied legacy serialization and ignored-input settings. If a Cache object is supplied, combining it with the deprecated cache_serialize, cache_version, or cache_ignore_input_vars arguments raises ValueError; the decorator obtains the version and ignored inputs from the object instead.

Local caching is consulted only when both TaskMetadata.cache and LocalConfig.auto().cache_enabled are true. Unless cache_overwrite is set, Task.local_execute looks up the task name, cache version, literal inputs, and ignored-input names before dispatching. A miss executes the task and stores its literal output map; a hit returns the cached map.

PythonTask passes container-related settings to PythonAutoContainerTask. That layer resolves the task image using SerializationSettings.image_config, supplies resources, environment, secrets, and pod-template information, and generates the default hosted command. At the Python-task level, omitted environment becomes {}. Decks are disabled by default. Set enable_deck=True to enable them and select fields with deck_fields; setting both enable_deck and the deprecated disable_deck raises ValueError. Invalid deck fields also raise ValueError.

Function-backed execution modes

In the default PythonFunctionTask.ExecutionBehavior.DEFAULT mode, execute(**kwargs) directly invokes the wrapped function:

@task
def double(x: int) -> int:
    return x * 2

In DYNAMIC mode, the function body is treated as a workflow generated at runtime. dynamic_execute creates a cached PythonFunctionWorkflow; for a real task execution, compile_into_workflow serializes the generated workflow and returns a DynamicJobSpec (or a literal map when the generated workflow has no nodes). Local execution instead executes that generated workflow with a local dynamic execution state. Native Python control flow is therefore present in the dynamic-task examples:

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

The source specifically demonstrates range(a) and task dependencies inside a dynamic function. node_dependency_hints may be supplied only with dynamic execution; passing hints to a static task raises ValueError because static dependencies are discovered during compilation.

pickle_untyped=True allows untyped outputs to be pickled as a convenience, but the PythonFunctionTask documentation does not recommend it for production. ignore_input_vars removes named inputs from the Flyte interface, so the declared Python signature and the externally visible task interface can differ.

Async and eager functions

The decorator automatically chooses AsyncPythonFunctionTask for an async def function. Its asynchronous call handler awaits the wrapped function in default mode. Async dynamic execution is explicitly unsupported: AsyncPythonFunctionTask.async_execute raises NotImplementedError for ExecutionBehavior.DYNAMIC.

Eager tasks use EagerAsyncPythonFunctionTask. The @eager decorator forces ExecutionBehavior.EAGER, marks TaskMetadata.is_eager=True, and enables decks by default. The runnable local example from task.py is:

from flytekit import task, eager

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

@task
def double(x: int) -> int:
    return x * 2

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

if __name__ == "__main__":
    import asyncio

    result = asyncio.run(eager_workflow(x=1))
    print(f"Result: {result}")  # "Result: 4"

Locally, EagerAsyncPythonFunctionTask.async_execute sets the eager-local execution mode and awaits the user function. Outside local execution it creates or uses a Controller worker queue and calls run_with_backend; calls to Flyte entities are then tracked as executions rather than ordinary in-process calls. The remote path requires a user execution ID, installs SIGINT and SIGTERM handlers, and uses _F_EE_ROOT when present to propagate the root eager tag. The eager documentation states that client-credentials authentication requires client_secret_group and client_secret_key in a PlatformConfig and a Flyte-compatible configuration supplied through Config.auto. Eager workflows support tasks, workflows, and eager entities, but not Flyte conditionals; use ordinary Python if statements in the eager function.

Tasks without a user function

Use PythonInstanceTask when a task has a platform-defined execution method rather than a user-defined function body. It is a thin PythonAutoContainerTask base whose subclasses provide execute; the instance is declared at module scope so the resolver can capture and rehydrate it. The class documentation illustrates the invocation shape x(a=5) for an instance x created with a task name and task-specific configuration.

For Python-native plugin tasks, extend PythonTask and implement the conversion or backend behavior needed by the plugin. SQLTask is a concrete example of the extension boundary: it builds an Interface from explicit input and output dictionaries and stores a normalized query template, but its execute method raises NotImplementedError("Cannot run a SQL Task natively, please mock."). SQL execution therefore belongs to its backend/plugin rather than the local Python function path.

Resolver-based rehydration

A hosted Python auto-container must identify which task object to load when pyflyte-execute starts. TaskResolverMixin defines that contract: a resolver supplies location, name, loader_args(settings, task), load_task(loader_args), and get_all_tasks(). task_name is an optional override.

PythonAutoContainerTask.get_default_command places the resolver location and loader arguments after --:

pyflyte-execute --inputs {{.input}} --output-prefix {{.outputPrefix}} \
  --raw-output-data-prefix {{.rawOutputDataPrefix}} \
  --checkpoint-path {{.checkpointOutputPrefix}} \
  --prev-checkpoint {{.prevCheckpointPrefix}} \
  --resolver <resolver-location> -- <loader-arguments>

The default resolver's loader arguments are task-module, the extracted module name, task-name, and the extracted task name. DefaultTaskResolver.load_task imports the module with importlib.import_module and returns getattr(task_module, task_name). This is why ordinary function tasks must be accessible at module scope when the default resolver is used. Nested, inner, and local functions are rejected by PythonFunctionTask unless the source is a permitted test function or a wrapper preserves module-level identity with functools.wraps/update_wrapper. For a different storage or rehydration scheme, provide a resolver implementing TaskResolverMixin (the codebase also includes resolver implementations such as ClassStorageTaskResolver).

Map tasks and testing

map_task wraps a PythonFunctionTask or PythonInstanceTask and invokes it once for each value in a collection. The source example applies per-instance metadata, concurrency, partial-success policy, and resource overrides:

@task
def my_mappable_task(a: int) -> typing.Optional[str]:
    return str(a)

@workflow
def my_wf(x: typing.List[int]) -> typing.List[typing.Optional[str]]:
    return map_task(
        my_mappable_task,
        metadata=TaskMetadata(retries=1),
        concurrency=10,
        min_success_ratio=0.75,
    )(a=x).with_overrides(requests=Resources(cpu="10M"))

The array/map implementation rejects dynamic and eager function tasks and requires the wrapped task to have at most one output. concurrency=0 represents unbounded concurrency in map_task; a positive value batches the mapped executions.

For unit tests, task_mock patches a Python task's execute method rather than bypassing the task call machinery:

@task
def t1(i: int) -> int:
    pass

with task_mock(t1) as m:
    m.side_effect = lambda x: x
    t1(10)
    # The mock is valid only within this context

The context manager restores the original execute method after the block. The parsed repository slice contains these examples in source docstrings and documentation, but does not contain the referenced test_*.py or example files; the snippets above are the source-provided task, workflow, eager, map, and testing patterns.

Declaring a task​

What the decorator constructs​

Calling tasks while authoring workflows​

Metadata, caching, and execution configuration​

Function-backed execution modes​

Async and eager functions​

Tasks without a user function​

Resolver-based rehydration​

Map tasks and testing​