Write and run tasks
Learn the basics of writing tasks.
A Prefect task is a discrete unit of work in a Prefect workflow.
You can turn any Python function into a task by adding an @task
decorator to it.
Tasks can:
- Take inputs, perform work, and return outputs
- Cache their execution across invocations
- Encapsulate workflow logic into reusable units across flows
- Receive metadata about upstream task dependencies and their state before running
- Use automatic logging to capture runtime details, tags, and final state
- Execute concurrently
- Be defined in the same file as the flow or imported from modules
- Be called from flows or other tasks
Flows and tasks share some common features:
- They can be defined using their respective decorator, which accepts configuration settings (see all task settings and flow settings)
- They can have a name, description, and tags for organization and bookkeeping
- They provide retries, timeouts, and other hooks to handle failure and completion events
Example task
Here’s an example of a simple flow with a single task:
Running that flow in the terminal results in output like this:
This task run is tracked in the UI as well.
Tasks are uniquely identified by a task key, which is a hash composed of the task name, the fully qualified name of the function, and any tags. If the task does not have a name specified, the name is derived from the decorated function object.
How big should a task be?
Prefect encourages “small tasks.” Each one should represent a single logical step of your workflow. This allows Prefect to better contain task failures.
There’s nothing stopping you from putting all of your code in a single task. However, if any line of code fails, the entire task fails and must be retried from the beginning. Avoid this by splitting the code into multiple dependent tasks.
Supported functions
Almost any standard Python function can be turned into a Prefect task by adding the @task
decorator.
Prefect uses client-side task run orchestration by default, which significantly improves performance, especially for workflows with many tasks. Task creation and state updates happen locally, reducing API calls to the Prefect server during execution. This enables efficient handling of large-scale workflows and improves reliability when server connectivity is intermittent.
Tasks are always executed in the main thread by default, unless a specific task runner is used to execute them on different threads, processes, or infrastructure. This facilitates native Python debugging and profiling.
Task updates are logged in batch, leading to eventual consistency for task states in the UI and API queries. While this means there may be a slight delay in seeing the most up-to-date task states, it allows for substantial performance improvements and increased workflow scale.
Synchronous functions
The simplest Prefect task is a synchronous Python function. Here’s an example of a synchronous task that prints a message:
Asynchronous functions
Prefect also supports asynchronous Python functions. The resulting tasks are coroutines that can be awaited or run concurrently, following standard async Python behavior.
Class Methods
Prefect supports synchronous and asynchronous methods as tasks, including instance methods, class methods, and static methods. For class methods and static methods, you must apply the appropriate method decorator above the @task
decorator:
Generators
Prefect supports synchronous and asynchronous generators as tasks. The task is considered to be Running
as long as the generator is yielding values. When the generator is exhausted, the task is considered Completed
. Any values yielded by the generator can be consumed by other tasks, and they will automatically record the generator task as their parent.
Generator functions are consumed when returned from tasks
The result of a completed task must be serializable, but generators cannot be serialized. Therefore, if you return a generator from a task, the generator will be fully consumed and its yielded values will be returned as a list. This can lead to unexpected behavior or blocking if the generator is infinite or very large.
Here is an example of proactive generator consumption:
If you need to return a generator without consuming it, you can yield
it instead of using return
.
Values yielded from generator tasks are not considered final results and do not face the same serialization constraints:
Concurrency
Tasks enable concurrent execution, allowing you to execute multiple tasks asynchronously. This concurrency can greatly enhance the efficiency and performance of your workflows.
Expand the script to calculate the average open issues per user by making more requests:
Now you’re fetching the data you need, but the requests happen sequentially.
Tasks expose a submit
method that changes
the execution from sequential to concurrent.
In this example, you also need to use the
result
method to unpack a list of return values:
The logs show that each task is running concurrently:
Task configuration
Tasks allow for customization through optional arguments that can be provided to the task decorator.
Argument | Description |
---|---|
name | An optional name for the task. If not provided, the name is inferred from the function name. |
description | An optional string description for the task. If not provided, the description is pulled from the docstring for the decorated function. |
tags | An optional set of tags associated with runs of this task. These tags are combined with any tags defined by a prefect.tags context at task runtime. |
timeout_seconds | An optional number of seconds indicating a maximum runtime for the task. If the task exceeds this runtime, it will be marked as failed. |
cache_key_fn | An optional callable that, given the task run context and call parameters, generates a string key. If the key matches a previous completed state, that state result is restored instead of running the task again. |
cache_expiration | An optional amount of time indicating how long cached states for this task are restorable; if not provided, cached states will never expire. |
retries | An optional number of times to retry on task run failure. |
retry_delay_seconds | An optional number of seconds to wait before retrying the task after failure. This is only applicable if retries is nonzero. |
log_prints | An optional boolean indicating whether to log print statements. |
See all possible options in the Python SDK docs.
For example, provide optional name
and description
arguments to a task:
Distinguish runs of this task by providing a task_run_name
.
Python’s standard string formatting syntax applies:
Additionally, this setting accepts a function that returns a string for the task run name:
If you need access to information about the task, use the prefect.runtime
module. For example:
Tags
Tags are optional string labels that enable you to identify and group tasks other than by name or flow. Tags are useful to:
- Filter task runs by tag in the UI and through the Prefect REST API.
- Set concurrency limits on task runs by tag.
You may specify tags as a keyword argument on the task decorator.
Alternatively, specify tags when the task is called rather than in its definition with a tags
context manager, .
Timeouts
Task timeouts prevent unintentional long-running tasks. When the duration of execution for a
task exceeds the duration specified in the timeout, a timeout exception is raised and the task is
marked as failed. In the UI, the task is visibly designated as TimedOut
. From the perspective of the
flow, the timed-out task is treated like any other failed task.
Specify timeout durations with the timeout_seconds
keyword argument:
Retries
Prefect can automatically retry task runs on failure. A task run fails if its Python function raises an exception.
To enable retries, pass retries
and retry_delay_seconds
arguments to your
task.
If the task run fails, Prefect will retry it up to retries
times, waiting
retry_delay_seconds
seconds between each attempt.
If the task fails on the final retry, Prefect marks the task as failed.
A new task run is not created when a task is retried. Instead, a new state is added to the state history of the original task run.
Retries are often useful in cases that depend upon external systems, such as making an API request.
The example below uses the httpx
library to make an HTTP
request.
In this task, if the HTTP request to the brittle API receives any status code other than a 2xx (200, 201, etc.), Prefect will retry the task a maximum of two times, waiting five seconds in between retries.
Custom retry behavior
The retry_delay_seconds
option accepts a list of integers for customized retry behavior.
The following task will wait for successively increasing intervals of 1, 10, and 100 seconds, respectively, before the next attempt starts:
The retry_condition_fn
argument accepts a callable that returns a boolean.
If the callable returns True
, the task will be retried.
If the callable returns False
, the task will not be retried.
The callable accepts three arguments: the task, the task run, and the state of the task run.
The following task will retry on HTTP status codes other than 401 or 404:
Additionally, you can pass a callable that accepts the number of retries as an argument and returns a list.
Prefect includes an exponential_backoff
utility that will automatically generate a list of retry delays that correspond to an exponential backoff retry strategy.
The following flow will wait for 10, 20, then 40 seconds before each retry.
Add “jitter” to avoid thundering herds
You can add jitter to retry delay times. Jitter is a random amount of time added to retry periods that helps prevent “thundering herd” scenarios, which is when many tasks retry at the same time, potentially overwhelming systems.
The retry_jitter_factor
option can be used to add variance to the base delay.
For example, a retry delay of 10 seconds with a retry_jitter_factor
of 0.5 will allow a delay up to 15 seconds.
Large values of retry_jitter_factor
provide more protection against “thundering herds,” while keeping the average retry delay time constant.
For example, the following task adds jitter to its exponential backoff so the retry delays will vary up to a maximum delay time of 20, 40, and 80 seconds respectively.
Configure retry behavior globally
Set default retries and retry delays globally through settings.
These settings will not override the retries
or retry_delay_seconds
that are set in the task decorator.
Was this page helpful?