How to automatically rerun your workflow when it fails

from prefect import flow


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

from prefect import flow


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

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()

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")

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")

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)

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"

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