> ## Documentation Index
> Fetch the complete documentation index at: https://prefect-bd373955-pytest-markdown.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# Recover from failure

> Learn how to automatically retry tasks on failure.

In Prefect, a task *fails* if its Python function raises an exception.

To enable retries, pass `retries` and `retry_delay_seconds` parameters to your
task. If the task fails, Prefect will retry it up to the number of `retries` times, waiting
`retry_delay_seconds` seconds between each attempt. If the task fails on the
final retry, Prefect marks the task as *crashed* if the task raised an exception;
or *failed* if it returned a string.

<Note>
  **Retries do not create new task runs**

  A new task run is not created when a task is retried. A new state is added to the state
  history of the original task run.
</Note>

### Example: making an API request

Consider the real-world problem of making an API request. This example
uses the [`httpx`](https://www.python-httpx.org/) library to make an HTTP
request.

```python hl_lines="4"
import httpx

from prefect import flow, task


@task(retries=2, retry_delay_seconds=5)
def get_data_task(
    url: str = "https://api.brittle-service.com/endpoint"
) -> dict:
    response = httpx.get(url)
    
    # If the response status code is anything but a 2xx, httpx will raise
    # an exception. This task doesn't handle the exception, so Prefect will
    # catch the exception and will consider the task run failed.
    response.raise_for_status()
    
    return response.json()
    

@flow
def get_data_flow():
    get_data_task()
```

In this task, if the HTTP request to the brittle API receives any status code
other than a 2xx, 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 delays for more custom retry behavior.
The following task waits for successively increasing intervals of 1, 10, and 100 seconds,
respectively, before the next attempt starts:

```python
from prefect import task

@task(retries=3, retry_delay_seconds=[1, 10, 100])
def some_task_with_manual_backoff_retries():
   ...
```

The `retry_condition_fn` option accepts a callable that returns a boolean.
If the callable returns `True`, the task is retried. If the callable returns `False`,
the task is not retried. The callable accepts three arguments: the task, the task run,
and the state of the task run. The following task retries on HTTP status codes other than 401 or 404:

```python
import httpx
from prefect import flow, task

def retry_handler(task, task_run, state) -> bool:
    """This is a custom retry handler to handle when we want to retry a task"""
    try:
        # Attempt to get the result of the task
        state.result()
    except httpx.HTTPStatusError as exc:
        # Retry on any HTTP status code that is not 401 or 404
        do_not_retry_on_these_codes = [401, 404]
        return exc.response.status_code not in do_not_retry_on_these_codes
    except httpx.ConnectError:
        # Do not retry
        return False
    except:
        # For any other exception, retry
        return True

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

@flow
def get_data_flow(url):
    my_api_call_task(url=url)

if __name__ == "__main__":
    get_data_flow(url="https://httpbin.org/status/503")
```

Additionally, you can pass a callable that accepts the number of retries as an argument and returns a list.
Prefect includes an [`exponential_backoff`](/3.0rc/api-ref/prefect/tasks/#prefect.tasks.exponential_backoff)
utility. This utility automatically generates a list of retry delays that correspond to an exponential backoff retry
strategy. The following flow waits for 10, 20, then 40 seconds before each retry:

```python
from prefect import task
from prefect.tasks import exponential_backoff

@task(retries=3, retry_delay_seconds=exponential_backoff(backoff_factor=10))
def some_task_with_exponential_backoff_retries():
   ...
```

#### Advanced topic: adding "jitter"

With exponential backoff, you may want to add *jitter* to the delay times. Jitter is
a random amount of time added to retry periods that helps prevent "thundering herd" scenarios. This
is when many tasks all retry at the exact same time, potentially overwhelming systems.

You can use the `retry_jitter_factor` option to add variance to the base delay. For example, a retry
delay of 10 seconds with a `retry_jitter_factor` of 0.5 allows 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:

```python
from prefect import task
from prefect.tasks import exponential_backoff

@task(
    retries=3,
    retry_delay_seconds=exponential_backoff(backoff_factor=10),
    retry_jitter_factor=1,
)
def some_task_with_exponential_backoff_retries():
   ...
```

### Configure retry behavior globally with settings

You can set retries and retry delays with the right global settings.
These settings do not override the `retries` or `retry_delay_seconds` that are set in the flow or task decorator.

```
prefect config set PREFECT_FLOW_DEFAULT_RETRIES=2
prefect config set PREFECT_TASK_DEFAULT_RETRIES=2
prefect config set PREFECT_FLOW_DEFAULT_RETRY_DELAY_SECONDS = [1, 10, 100]
prefect config set PREFECT_TASK_DEFAULT_RETRY_DELAY_SECONDS = [1, 10, 100]
```