Add a fail with no retry signal/state

[marvin-robot]

Opened from the Prefect Public Slack Community

thomas.weatherston: Hey All, I have a task that makes a request to an API and I only want to retry the task if I receive a certain status code from the response (eg. 500). Status codes like 404 I'd like to fail the task and anything else return the result. I'd still like to have the max retries and retry delay but I can't seem to figure out how to fail the task without it retrying! Any ideas?

kevin701: Hey <@U01KMB55X47>, you’ll need to raise the signal like https://docs.prefect.io/core/concepts/execution.html#state-signals|this . For your case you’ll look at the response code and then raise the appropriate signal

thomas.weatherston: Hey <@U01QEJ9PP53> thanks for the response! I've been attempting to use the FAIL signal to fail the task but it still seems to get retried. A small example:

from prefect import Flow, task
from prefect.engine import signals
from datetime import timedelta


@task(max_retries=5, retry_delay=timedelta(seconds=1))
def test():
    raise signals.FAIL()


with Flow("test-flow") as f:
    test()

f.run()

kevin701: I see. Do you need it in a FAILURE state? Because you can SKIP it but that would be a SUCCESS state

thomas.weatherston: Yeah ideally I'd like it to fail without retrying

jenny: Hi <@U01KMB55X47> - I believe we don't currently have a signal/state that would separate out a fail-no retry and fail-with retry (though I'm happy to open a ticket to suggest it as I think it could be helpful!). For your flow, would a different task failing work? You could fail your first task on the status you want retries on and then for other statuses pass a result to a downstream task to raise the fail signal.

thomas.weatherston: Hey <@ULXMV9SD7>! Ah yes, I reckon I could I could restructure our tasks to work like that, thanks very much :slightly_smiling_face: (a fail-no retry signal would also be awesome, if you can find time to add it in the future!)

jenny: Great. <@ULVA73B9P> open "Add a fail with no retry signal/state"

Original thread can be found here.

[novotl]

+1 I was just looking for a similar thing. Sometimes you just know no matter how much you retry it will always fail.

By the way, you might be able to get away with the ENDRUN, it depends on whether you have cleanup tasks that need to run if your tasks fail. Fail no-retry signal would be awesome though.

from datetime import timedelta

import prefect
from prefect import Flow, task
from prefect.engine import state
from prefect.engine.runner import ENDRUN


@task(max_retries=5, retry_delay=timedelta(seconds=1))
def foo():
    prefect.context.logger.info("running")
    raise ENDRUN(state=state.Failed())


with Flow("test-flow") as f:
    foo()

if __name__ == '__main__':
    f.run()