Hey i was trying to figure out is it possible for me to setu

Question

Hey i was trying to figure out is it possible for me to setup a task that if it fails I can have it retry 3 times i figured this part out But if it still fails I want to kick off another task so the rest of the flow can continue So basically i m trying to swap my task if it fails to another task as a backup so my entire flow can continue on

Anna Geller · Answer

That s an interesting question Here is an example using state handler for that use case here after 3 unsuccessful task runs we re sending a Slack message as part of a 3rd retry you can replace it with any action you would like to take ```from datetime import timedelta import prefect from prefect import Flow task from prefect tasks notifications import SlackTask def take action after threee failed task runs task old state new state allowing 3 retries on the 4th task run we we give up retrying and do something else if new state is retrying and new state run count == 3 take some action SlackTask message= 3 task runs didn t cut it doing sth else now run return new state task max retries=3 retry delay=timedelta seconds=1 state handlers= take action after threee failed task runs def retry test logger = prefect context get logger run count = prefect context get task run count <http logger info|logger info> s TaskRun run count raise Exception Failing in order to test retries with Flow retry tester as flow retry test if name == main flow run ``` The output when running this ``` 2022 02 27 18 27 02+0100 INFO prefect FlowRunner | Beginning Flow run for retry tester 2022 02 27 18 27 02+0100 INFO prefect TaskRunner | Task retry test Starting task run 2022 02 27 18 27 02+0100 INFO prefect retry test | 1 TaskRun 2022 02 27 18 27 02+0100 ERROR prefect TaskRunner | Task retry test Exception encountered during task execution Traceback most recent call last File Users anna conda envs prefectCloudFlows lib python3 8 site packages prefect engine task runner py line 876 in get task run state value = prefect utilities executors run task with timeout File Users anna conda envs prefectCloudFlows lib python3 8 site packages prefect utilities executors py line 468 in run task with timeout return task run args kwargs type ignore File Users anna PycharmProjects prefectCloudFlows 000 Community do sth on third retry py line 24 in retry test raise Exception Failing in order to test retries Exception Failing in order to test retries 2022 02 27 18 27 02+0100 INFO prefect TaskRunner | Task retry test Finished task run for task with final state Retrying 2022 02 27 18 27 02+0100 INFO prefect FlowRunner | Flow run RUNNING terminal tasks are incomplete 2022 02 27 18 27 02+0100 INFO prefect retry tester | Waiting for next available Task run at 2022 02 27T17 27 03 683739+00 00 2022 02 27 18 27 03+0100 INFO prefect FlowRunner | Beginning Flow run for retry tester 2022 02 27 18 27 03+0100 INFO prefect TaskRunner | Task retry test Starting task run 2022 02 27 18 27 03+0100 INFO prefect retry test | 2 TaskRun 2022 02 27 18 27 03+0100 ERROR prefect TaskRunner | Task retry test Exception encountered during task execution Traceback most recent call last File Users anna conda envs prefectCloudFlows lib python3 8 site packages prefect engine task runner py line 876 in get task run state value = prefect utilities executors run task with timeout File Users anna conda envs prefectCloudFlows lib python3 8 site packages prefect utilities executors py line 468 in run task with timeout return task run args kwargs type ignore File Users anna PycharmProjects prefectCloudFlows 000 Community do sth on third retry py line 24 in retry test raise Exception Failing in order to test retries Exception Failing in order to test retries 2022 02 27 18 27 03+0100 INFO prefect TaskRunner | Task retry test Finished task run for task with final state Retrying 2022 02 27 18 27 03+0100 INFO prefect FlowRunner | Flow run RUNNING terminal tasks are incomplete 2022 02 27 18 27 03+0100 INFO prefect retry tester | Waiting for next available Task run at 2022 02 27T17 27 04 732207+00 00 2022 02 27 18 27 04+0100 INFO prefect FlowRunner | Beginning Flow run for retry tester 2022 02 27 18 27 04+0100 INFO prefect TaskRunner | Task retry test Starting task run 2022 02 27 18 27 04+0100 INFO prefect retry test | 3 TaskRun 2022 02 27 18 27 04+0100 ERROR prefect TaskRunner | Task retry test Exception encountered during task execution Traceback most recent call last File Users anna conda envs prefectCloudFlows lib python3 8 site packages prefect engine task runner py line 876 in get task run state value = prefect utilities executors run task with timeout File Users anna conda envs prefectCloudFlows lib python3 8 site packages prefect utilities executors py line 468 in run task with timeout return task run args kwargs type ignore File Users anna PycharmProjects prefectCloudFlows 000 Community do sth on third retry py line 24 in retry test raise Exception Failing in order to test retries Exception Failing in order to test retries 2022 02 27 18 27 05+0100 INFO prefect TaskRunner | Task retry test Finished task run for task with final state Retrying 2022 02 27 18 27 05+0100 INFO prefect FlowRunner | Flow run RUNNING terminal tasks are incomplete 2022 02 27 18 27 05+0100 INFO prefect retry tester | Waiting for next available Task run at 2022 02 27T17 27 05 763641+00 00 2022 02 27 18 27 05+0100 INFO prefect FlowRunner | Beginning Flow run for retry tester 2022 02 27 18 27 05+0100 INFO prefect TaskRunner | Task retry test Starting task run 2022 02 27 18 27 05+0100 INFO prefect retry test | 4 TaskRun 2022 02 27 18 27 05+0100 ERROR prefect TaskRunner | Task retry test Exception encountered during task execution Traceback most recent call last File Users anna conda envs prefectCloudFlows lib python3 8 site packages prefect engine task runner py line 876 in get task run state value = prefect utilities executors run task with timeout File Users anna conda envs prefectCloudFlows lib python3 8 site packages prefect utilities executors py line 468 in run task with timeout return task run args kwargs type ignore File Users anna PycharmProjects prefectCloudFlows 000 Community do sth on third retry py line 24 in retry test raise Exception Failing in order to test retries Exception Failing in order to test retries 2022 02 27 18 27 05+0100 INFO prefect TaskRunner | Task retry test Finished task run for task with final state Failed 2022 02 27 18 27 05+0100 INFO prefect FlowRunner | Flow run FAILED some reference tasks failed ```

Adi Gandra · Answer

But wouldn t the flow fail because the task failed

Adi Gandra · Answer

How can i have the rest of the flow continue on after the state handler takes care of it

Adi Gandra · Answer

can i manipulate the new state that is returned

Anna Geller · Answer

Absolutely you can just return any state you want ```from datetime import timedelta import prefect from prefect import Flow task from prefect engine state import Success from prefect tasks notifications import SlackTask def take action after threee failed task runs task old state new state allowing 3 retries on the 4th task run we we give up retrying and do something else if new state is retrying and new state run count == 3 take some action SlackTask message= 3 task runs didn t cut it doing sth else now run return Success message= We now consider this task run successful return new state task max retries=3 retry delay=timedelta seconds=1 state handlers= take action after threee failed task runs def retry test logger = prefect context get logger run count = prefect context get task run count <http logger info|logger info> s TaskRun run count raise Exception Failing in order to test retries with Flow retry tester as flow retry test if name == main flow run ```

Anna Geller · Answer

``` 2022 02 27 19 35 59+0100 INFO prefect retry test | 3 TaskRun 2022 02 27 19 35 59+0100 ERROR prefect TaskRunner | Task retry test Exception encountered during task execution Traceback most recent call last File Users anna conda envs prefectCloudFlows lib python3 8 site packages prefect engine task runner py line 876 in get task run state value = prefect utilities executors run task with timeout File Users anna conda envs prefectCloudFlows lib python3 8 site packages prefect utilities executors py line 468 in run task with timeout return task run args kwargs type ignore File Users anna PycharmProjects prefectCloudFlows 000 Community do sth on third retry py line 26 in retry test raise Exception Failing in order to test retries Exception Failing in order to test retries 2022 02 27 19 35 59+0100 INFO prefect TaskRunner | Task retry test Finished task run for task with final state Success 2022 02 27 19 35 59+0100 INFO prefect FlowRunner | Flow run SUCCESS all reference tasks succeeded```

Adi Gandra · Answer

Perfect this makes sense One last question if i m in is retrying and new state run count == 3 block does that mean the task is running again

Adi Gandra · Answer

or does returning success mean the task won t run again

Anna Geller · Answer

you re correct returning success means it won t run again

Adi Gandra · Answer

Thanks