Yu(Irene) Chen
04/10/2026, 8:43 AMMarvin
04/10/2026, 8:43 AMMarvin
04/10/2026, 8:46 AMfrom prefect import task, flow, get_run_logger
from prefect.tasks import exponential_backoff
@task(retries=3, retry_delay_seconds=exponential_backoff(2)) # 2s, 4s, 8s
def do_work_then_retry():
logger = get_run_logger()
# 1) Main work
do_side_effects()
finalize_things()
# 2) Force a retry (this attempt will be marked Failed; next attempt will run)
raise RuntimeError("Triggering retry after finishing required steps")
@flow
def run():
do_work_then_retry()
- Conditional retry at the end
@task(retries=5, retry_delay_seconds=[5, 10, 20, 40, 60])
def maybe_retry():
result = compute()
write_artifacts(result) # logs/artifacts are fine to emit before raising
if need_another_attempt(result):
raise Exception("Retry requested")
return result
- Gate retries with a custom condition (decide which failures should retry)
def should_retry(task, task_run, state) -> bool:
exc = state.result(raise_on_failure=False)
# retry only on transient errors you care about
return isinstance(exc, TimeoutError)
@task(retries=3, retry_condition_fn=should_retry, retry_delay_seconds=5)
def do_stuff():
# ...
raise TimeoutError("transient")
Notes:
- There’s no special “retry” exception to raise in 3.x; any exception will trigger the retry logic if retries > 0 (and your retry_condition_fn returns True, if set).
- If you need to avoid repeating side-effects on retries, make your work idempotent, checkpoint state externally, or split the task so the side-effecting part is separate from the retried part.
If you can share what should trigger the retry and whether you need to persist any outputs before retrying, I can tailor the example (e.g., exact backoff, jitter, or task split).