Hi again, we have a requirement to handle two classes of errors: retryable errors and non-retryable errors. I think I saw some examples how this might be achieved somewhere but cant spot them for Prefect 2. In the non-retryable scenario we'd want the task state to go immediately to Failed (bypassing any retry settings configured on the task decorator)

In Prefect 2.0 you can do: For retriable errors:

@task(retries=2, retry_delay_seconds=60)

For non-retriable errors (no retries):

@task

But if I have a single task that can throw either type of error? Is there a context object that can be manipulated, I was thinking something similar to this?

import requests
from prefect import flow, task

@task(retries=3,retry_delay_seconds=5)
def call_api(url):
    try: 
        response = requests.get(url)
        print(response.status_code)
        return response.json()
    except NonRetryableError:
        TaskRunContext.set_state(TaskRun.FAILED)

@flow
def task_retry_bypass(url):
    fact_json = call_api(url)

Generally speaking, there is no need to use try-except in a Prefect flow. That's why you use Prefect so that you don't need to worry about all the different types of exceptions that might need to be caught. The

retries

mean a maximum number of retries: 1. If the task doesn't need a retry and succeeds after its first execution, it's done. 2. If it succeeds after the first retry (even though you configured 3 of them), the same - the task ends successfully after this first retry. 3. If it still fails after 3 retries, then it will be marked as Failed.

Is overriding the task run state an option, it would allow us to benefit from the task level retry configuration - but also skip that when needed. A common scenario would be distinguishing between transient system failures (e.g. a service unavailable) vs permanent errors arising from unexpected application state (e.g. once an object has been deleted there's no point retrying any attempt to modify it)

Good question. You can return any State in your task to influence that: https://orion-docs.prefect.io/concepts/states/

Thanks Anna, will give that a try

You could even do it on a flow level - having normal retries on a task level, but if the task result returns a specific value, you raise an exception on a flow level and end the flow run immediately - just mentioning this because it's also a valid option:

from prefect import task, flow


@task(retries=3, retry_delay_seconds=5)
def call_api(url):
    response = requests.get(url)
    status_code = response.status_code
    logger = get_run_logger()
    <http://logger.info|logger.info>(status_code)
    return response.json()


@flow
def task_retry_bypass():
    res = call_api().wait().result()
    if res["sth"] == "sth":
        raise ValueError("Non retriable error!")


if __name__ == "__main__":
    task_retry_bypass()

This one worked (no retries)

from prefect import flow, task
import prefect

@task(retries=3,retry_delay_seconds=5)
def call_api(url):
    return prefect.orion.schemas.states.StateType.FAILED

@flow
def task_retry_bypass(url):
    fact_json = call_api(url)

task_retry_bypass("foo")
~

We'd like to make this as simple as possible (to reduce learning curve and help with developer experience), so I think we can extend the above and use an inner decorated function to hide all the details

Sure, you can customize it as you wish, it's pretty much just Python so the sky is the limit! 🙂

I hit a probem with this...if the return is

return prefect.orion.schemas.states.StateType.FAILED

Then the retry is skipped BUT the Flow Run ends up in a Completed state - tracing through this occurs because the StateType.FAILED is not a State (so Prefect defaults to Completed for "python object returns") and if we return

return prefect.orion.schemas.states.Failed()

Then client.py: propose_state appears to do a call to set_task_run_state, but the proposed state is rejected and we go into a retry loop

Thanks for the suggestion, I'll pass it on to the product team

I’m facing a similar scenario.. When not worth retrying a task I’d like to bypass the retry configuration