Yes. In Prefect 3.x you can control whether a task retries based on the failure using

retry_condition_fn

on the

@task

decorator. - The function is called after a failure with

(task, task_run, state)

- Return True to proceed with retrying; return False to skip remaining retries and fail immediately - It can be sync or async - To inspect the actual exception, call

state.result()

(it will re-raise the underlying error for you to check) Example: retry only on transient errors, skip on validation errors

from prefect import flow, task

def retry_on_transient(task, task_run, state) -> bool:
    try:
        state.result()  # will raise the underlying exception
    except (TimeoutError, ConnectionError):
        return True   # retry these
    except ValueError:
        return False  # do not retry these
    except Exception:
        return True   # default: retry others

@task(retries=3, retry_delay_seconds=5, retry_condition_fn=retry_on_transient)
def do_work(x: int) -> int:
    # your logic that may raise
    return x * 2

@flow
def my_flow():
    do_work(1)

if __name__ == "__main__":
    my_flow()

HTTP example: retry on 5xx, skip 4xx

import httpx
from prefect import task

def http_retry_condition(task, task_run, state) -> bool:
    try:
        state.result()
    except httpx.HTTPStatusError as exc:
        # retry on server errors only
        return 500 <= exc.response.status_code < 600
    except Exception:
        return True

@task(retries=3, retry_condition_fn=http_retry_condition)
def call_api(url: str):
    r = httpx.get(url)
    r.raise_for_status()
    return r.json()

If you can share which exceptions you want to skip (or keep), I can tailor the condition function for your case.

Absolutely. Here’s a

retry_condition_fn

tailored for

requests

+

response.raise_for_status()

with your rules: - If `X-RateLimit-Remaining-Quota-Minute == 0`: sleep 60s, then return True (retry) - If `X-RateLimit-Remaining-Quota-Month == 0`: return False (do not retry) - For other

requests.HTTPError

cases: return True (retry) - For non-HTTP errors: return True (retry) Drop-in example:

import time
import requests
from prefect import flow, task

def retry_on_rate_limits(task, task_run, state) -> bool:
    try:
        # Will re-raise the underlying exception
        state.result()
    except requests.HTTPError as exc:
        resp = exc.response
        # Be defensive if response is missing
        if resp is not None:
            # Headers are case-insensitive in requests
            minute = resp.headers.get("X-RateLimit-Remaining-Quota-Minute")
            month = resp.headers.get("X-RateLimit-Remaining-Quota-Month")

            def to_int(value):
                try:
                    return int(value) if value is not None else None
                except (TypeError, ValueError):
                    return None

            minute_val = to_int(minute)
            month_val = to_int(month)

            # Month quota exhausted: do not retry
            if month_val == 0:
                return False

            # Minute quota exhausted: wait 60s then retry
            if minute_val == 0:
                time.sleep(60)
                return True

        # For other HTTPError cases, retry
        return True

    except Exception:
        # Non-HTTP errors: retry
        return True

    # Shouldn't reach here for failed states, but default to retry
    return True


@task(
    retries=5,
    # If you sleep in the condition fn, avoid extra built-in delay
    retry_delay_seconds=0,
    retry_condition_fn=retry_on_rate_limits,
)
def call_api(url: str):
    r = requests.get(url, timeout=30)
    r.raise_for_status()
    return r.json()


@flow
def my_flow():
    call_api("<https://api.example.com/data>")

Notes: - This uses

requests.HTTPError

(raised by

response.raise_for_status()

). - If headers are absent or non-integer, it falls back to retrying. - Since we sleep in the condition function, I set

retry_delay_seconds=0

to avoid double delays. If you prefer Prefect-managed delays, remove the

time.sleep(60)

and we can discuss an alternative strategy.