Skip to main content

Task authoring and execution

Declare a task with a typed Python function​

Start with @task when you have a module-level Python function whose annotations define the Flyte interface:

from flytekit import task

@task
def my_task(x: int, y: dict[str, str]) -> str:
    return f"{x}: {y}"

The decorator is the normal construction path for a PythonFunctionTask. It examines the callable, builds its native Interface from the annotations, and returns a task entity rather than an ordinary function wrapper. The task can still be called with keyword arguments, but the call is routed through Flyte's entity call handler so that the same declaration can either create workflow bindings or execute locally, depending on the active FlyteContext.

For a plugin-backed task, pass its configuration through the decorator. The source documents this form with a Spark configuration and retry policy:

@task(task_config=Spark(), retries=3)
def my_task(x: int, y: dict[str, str]) -> str:
    return f"{x}: {y}"

task.task creates TaskMetadata, selects a task implementation based on task_config, and selects AsyncPythonFunctionTask for coroutine functions. The resulting function task retains the callable, derives its name and module information, and uses PythonAutoContainerTask for container and serialization integration.

The task abstraction hierarchy​

Flyte's abstractions separate the IDL-facing task definition from Python-native invocation:

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

Task stores the task type, name, typed Flyte interface, TaskMetadata, security context, documentation, and task-type version. Constructing a task also registers it in FlyteEntities.entities. Its compile() method is abstract at this level, and its serialization hooks—get_container(), get_k8s_pod(), get_sql(), get_custom(), get_config(), and get_extended_resources()—default to None.

PythonTask adds a Python-native Interface, task configuration, environment variables, and Deck selection. It transforms that native interface into a Flyte typed interface, exposes native input and output types, and implements compile() by calling create_and_link_node(). This is the base for Python-backed task types that do not themselves contain a user function. The source explicitly directs function-based tasks to PythonFunctionTask instead.

PythonFunctionTask is the function-backed implementation. Its constructor requires both a task configuration and a callable; it raises ValueError when the callable is missing. It calls transform_function_to_interface(...), removes any ignore_input_vars, derives the task name with extract_task_module(), and stores the function in task_function. In default execution mode, its execute() method is equivalent to:

return self._task_function(**kwargs)

Use PythonInstanceTask instead when an extension has no user-defined function body and its platform-specific execute() implementation supplies the behavior. An instance task is created as an object and invoked according to its declared interface:

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

The concrete subclass must provide the interface-compatible behavior and execute() implementation.

Configure metadata, containers, and Decks​

Pass execution metadata through the decorator or construct TaskMetadata directly for a task implementation. The metadata fields include retries, timeout, interruptibility, caching, deprecation text, pod-template name, Deck generation, and eager execution. Integer timeouts are interpreted as seconds and converted to datetime.timedelta:

from flytekit.core.base_task import TaskMetadata

metadata = TaskMetadata(retries=3, timeout=60)

The constructor validates cache-related combinations. cache=True requires a non-empty cache_version; cache_serialize=True and cache_ignore_input_vars are valid only when caching is enabled. Thus this is invalid:

TaskMetadata(cache=True)

TaskMetadata.retry_strategy produces Flyte's retry model, while to_taskmetadata_model() adds the Flyte SDK runtime metadata and maps the Python fields to the Flyte task model.

Python tasks also accept container, environment, resource, secret, pod-template, and resolver settings through the auto-container layer. If environment is omitted, PythonTask stores an empty dictionary. The configured/default image is used when container_image is not supplied; a per-task container_image can instead be a string or an ImageSpec.

Enable Deck output with enable_deck, and select fields with deck_fields. disable_deck is deprecated, and the two switches cannot be supplied together:

@task(enable_deck=True)
def rendered(x: int) -> int:
    return x + 1

PythonTask validates every requested field against DeckField. PythonFunctionTask additionally renders source-code and Python-dependency Decks when those fields are selected. By default Decks are disabled for PythonTask; EagerAsyncPythonFunctionTask instead defaults them on.

Understand invocation and local execution​

Calling a task is not always the same as calling its underlying Python function:

value = my_task(x=1, y={"kind": "example"})

Task.__call__() delegates to flyte_entity_call_handler. During workflow compilation, PythonTask.compile() creates and links a node. During local execution, Task.local_execute() follows this path:

native values or Promises
  → translate_inputs_to_literals
  → LiteralMap
  → sandbox_execute
  → dispatch_execute
  → output LiteralMap
  → Promise or VoidPromise

PythonTask.dispatch_execute() performs the Python task lifecycle. It calls pre_execute(), converts the input literal map to native values with TypeEngine.literal_map_to_kwargs(), invokes execute(**native_inputs), calls post_execute(), and converts returned values back to literals with TypeEngine. A task with no declared outputs returns a VoidPromise; otherwise the local path wraps outputs in Promise objects using the declared output names.

When both task metadata and LocalConfig.auto() enable caching, local_execute() checks LocalTaskCache before dispatch. A cache hit returns the stored literal map; a miss executes the task and stores the resulting map. cache_overwrite bypasses an existing entry. Local execution also preserves and annotates the original user exception, whereas the dispatch path wraps exceptions for non-local execution as Flyte user-runtime or non-recoverable system exceptions.

Output conversion follows the declared interface. A single-output NamedTuple receives special handling, while a tuple returned for an individual output is rejected. The local path also asserts that the number of produced literals matches the number of declared outputs.

Generate runtime task graphs with dynamic tasks​

Use @dynamic when the task body creates a workflow graph from runtime values:

from flytekit import dynamic

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

A dynamic function is a PythonFunctionTask with ExecutionBehavior.DYNAMIC. Its execute() calls dynamic_execute() instead of directly calling the function. In a real task execution, dynamic_execute() builds a PythonFunctionWorkflow, serializes the generated nodes, and returns a DynamicJobSpec. During local execution it runs the generated workflow directly and translates the resulting values into a literal map.

If Flyte cannot infer runtime dependencies, provide node_dependency_hints, but only on a dynamic task. PythonFunctionTask raises ValueError when hints are supplied with the default execution behavior because static task and workflow dependencies are discovered automatically.

Async and eager execution​

Coroutine functions are represented by AsyncPythonFunctionTask. Its asynchronous call path uses async_flyte_entity_call_handler, and its default execution awaits the function. Async dynamic execution is explicitly unsupported and raises NotImplementedError.

Use @eager for an async eager workflow. This source example runs locally with asyncio:

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"

@eager constructs EagerAsyncPythonFunctionTask. Its constructor forces ExecutionBehavior.EAGER, marks metadata with is_eager=True, and ignores a caller-provided execution_mode. Locally, the task runs the async function under eager-local execution. Remotely, it creates or reuses a Controller worker queue; task calls become backend executions, and the controller's rendered HTML is added to an Eager Executions Deck.

Eager execution is not dynamic compilation: the eager source explicitly separates the async execution path from dynamic execution. Remote eager execution also requires a user-facing execution context containing an execution ID. The implementation uses _F_EE_ROOT when present to propagate the root tag; otherwise it uses the current execution ID name. For local testing against a remote, EagerAsyncPythonFunctionTask.run(remote, ss, **kwargs) creates the controller and runs the async entry point.

Extend task loading and programmatic interfaces​

Serialized containers need to reconstruct the task object. TaskResolverMixin defines that contract through location, name(), load_task(loader_args), loader_args(settings, task), and get_all_tasks(). The default resolver command described by the source has the shape:

pyflyte-execute ... --resolver ... -- task-module <module> task-name <name>

PythonFunctionTask with the default resolver must refer to a module-level function. Nested, inner, or local functions are rejected, except test functions in modules beginning with test_. If a custom decorator wraps a task, use functools.wraps or functools.update_wrapper; otherwise supply a custom TaskResolverMixin for a different loading strategy.

For explicit programmatic interfaces, kwtypes() preserves keyword insertion order and returns an ordered mapping of names to Python types. Reference entities use it directly:

ref_entity = get_reference_entity(
    ResourceType.WORKFLOW,
    "project",
    "dev",
    "my.other.workflow",
    "abc123",
    inputs=kwtypes(a=str, b=int),
    outputs={},
)

Use PythonTask or PythonInstanceTask for platform-defined executors, and PythonFunctionTask for a callable-backed task. The common Task lifecycle, type-engine conversion, promise handling, resolver contract, and auto-container serialization then provide the integration between the Python declaration and Flyte execution.