How to automatically rerun your workflow when it fails

Set the number of retries

To rerun your workflow when an exception is raised, set the retries parameter to an integer.

from prefect import flow


@flow(retries=3)
def troublesome_workflow():
    raise Exception("oopsie daisy")

Retry nested flows

When a parent flow retries a direct call to a nested flow, Prefect reuses the child’s completed result if it was persisted. Set persist_result=True on the child flow to enable result persistence. If the child failed or completed without a persisted result, the parent retry creates a new child flow run. The new run has its own retry budget, while the previous run keeps its state and history. Because the child has a new flow-run ID, task caches that include the flow-run ID do not carry over from the previous child run.

Wait between retries

To add a wait between retries, pass a value to the retry_delay_seconds parameter.

from prefect import flow


@flow(retries=3, retry_delay_seconds=2.7182)
def troublesome_workflow():
    raise Exception("oopsie daisy")

Retry parts of a workflow

You can configure retries for individual tasks in a workflow to limit the scope of a retry.

import random

import httpx
from prefect import flow, task


@task
def consistent_task() -> dict:
    print("I won't rerun even if the task after me fails")


@task(retries=4, retry_delay_seconds=5)
def mercurial_task() -> dict:
    # Will fail half of the time
    url = f"https://httpbin.org/status/{random.choice([200, 500])}"

    response = httpx.get(url)
    response.raise_for_status()
    

@flow
def my_workflow():
    consistent_task()
    mercurial_task()

Retry with configurable delay

A task’s retry delays can also be defined as a list of integers for different delays between retries.

from prefect import task


@task(retries=4, retry_delay_seconds=[1, 2, 4, 8])
def melancholic_task():
    raise Exception("We used to see each other so much more often")

You can also use the exponential_backoff utility to generate a list of retry delays that correspond to an exponential backoff retry strategy.

from prefect import task
from prefect.tasks import exponential_backoff


@task(retries=4, retry_delay_seconds=exponential_backoff(backoff_factor=2))
def melancholic_task():
    raise Exception("We used to see each other so much more often")

Retry with a custom condition

Whether or not a task should be retried can be determined dynamically by passing a callable to the retry_condition_fn parameter.

import httpx
from prefect import flow, task


def retry_handler(task, task_run, state) -> bool:
    """
    Retry handler that skips retries if the HTTP status code is 401 or 404.
    """
    try:
        state.result()
    except httpx.HTTPStatusError as exc:
        do_not_retry_on_these_codes = [401, 404]
        return exc.response.status_code not in do_not_retry_on_these_codes
    except httpx.ConnectError:
        return False
    except:
        return True


@task(retries=1, retry_condition_fn=retry_handler)
def api_call_task(url):
    response = httpx.get(url)
    response.raise_for_status()
    return response.json()


@flow
def my_workflow(url):
    api_call_task(url=url)

If a callable passed to retry_condition_fn returns True, the task will be retried. Otherwise, the task will exit with an exception.

Add jitter to retry delays

To add a random amount of time to retry delays, pass a value to the retry_jitter_factor parameter.

import time

from prefect import task
from prefect.tasks import exponential_backoff

last_run_time = time.time()

@task(
    retries=3,
    retry_delay_seconds=exponential_backoff(backoff_factor=5),
    retry_jitter_factor=3,
)
def some_task_with_exponential_backoff_retries():
    global last_run_time
    print(f"Time between retries: {time.time() - last_run_time}")
    if time.time() - last_run_time < 10:
        last_run_time = time.time()
        raise Exception("I could fail some more")
    return "That's enough failure"

Adding jitter to the retry delays avoids multiple tasks introducing load to external systems by failing and retrying at the exact same cadence.

Configure task retry behavior globally

You can set the default retries and retry delays for all tasks via Prefect’s settings. The default values can be overridden on a per-task basis via the retries and retry_delay_seconds parameters.

prefect config set PREFECT_TASKS_DEFAULT_RETRIES=2
prefect config set PREFECT_TASKS_DEFAULT_RETRY_DELAY_SECONDS="1,10,100"

Workflows

Deployments

Configuration

Automations

Workflow Infrastructure

Prefect Cloud

Prefect Self-hosted

AI

Migrate

How to automatically rerun your workflow when it fails

Set the number of retries

Retry nested flows

Wait between retries

Retry parts of a workflow

Retry with configurable delay

Retry with a custom condition

Add jitter to retry delays

Configure task retry behavior globally

​Set the number of retries

​Retry nested flows

​Wait between retries

​Retry parts of a workflow

​Retry with configurable delay

​Retry with a custom condition

​Add jitter to retry delays

​Configure task retry behavior globally

Set the number of retries

Retry nested flows

Wait between retries

Retry parts of a workflow

Retry with configurable delay

Retry with a custom condition

Add jitter to retry delays

Configure task retry behavior globally