<@ULVA73B9P> Can I somehow conditionally skip retr...
# ask-marvin
m
@Marvin Can I somehow conditionally skip retries for a given task? For example, when an error occurs that indicates retrying will not succeed. Maybe by raising a specific error or calling some function?
m
thought for 378.5 seconds
Yes. In Prefect 3.x you can make task retries conditional with the
retry_condition_fn
option on a task. This function is called whenever the task raises an exception; if it returns True, Prefect will retry (subject to remaining attempts). If it returns False, Prefect will stop retrying and keep the failure. There isn’t a built-in “do not retry” exception type. The common pattern is to define your own exception type(s) and make your
retry_condition_fn
return False for those. Example 1: stop retrying for a custom non-retryable error
Copy code
from prefect import task

class NonRetryableError(Exception):
    pass

def should_retry(exc: BaseException) -> bool:
    # Don’t retry for explicitly non-retryable errors
    if isinstance(exc, NonRetryableError):
        return False
    # Retry everything else
    return True

@task(
    retries=5,
    retry_delay_seconds=[5, 15, 30, 60, 120],
    retry_condition_fn=should_retry,
)
def do_work(x: int) -> int:
    if x < 0:
        # This will fail the task immediately with no further retries
        raise NonRetryableError("Bad input; retrying will not help.")
    # Simulate a transient error that should be retried
    raise RuntimeError("Temporary issue; please retry.")
Example 2: stop retrying for 4xx HTTP errors but retry others
Copy code
from prefect import task

def retry_on_http_error(exc: BaseException) -> bool:
    # Example for httpx; adapt for your HTTP client
    try:
        import httpx
        if isinstance(exc, httpx.HTTPStatusError):
            status = exc.response.status_code
            # Do not retry on 4xx, do retry on 5xx
            return status >= 500
    except Exception:
        pass
    # Retry by default for other exceptions
    return True

@task(retries=3, retry_condition_fn=retry_on_http_error)
def call_api(...):
    ...
Key points - Set
retries
to the max attempts you want and use
retry_condition_fn
to dynamically permit/deny each retry based on the raised exception. - Raising a specific error alone will not skip retries; you must pair it with logic in
retry_condition_fn
. - You can define
retry_condition_fn
in the task decorator (as above). If you need to change it at call time, you can wrap your function with
task(...)
and construct variants with different options. If you want, share a snippet of your task and the errors you’d like to treat as non-retryable, and I can help tailor the condition function.