Possible to give a task access to how many times it's been retried?

[itsayellow]

I have a Prefect Task in which I want to change its behavior every time it fails and is retried. Is there any way for the task code to have access to how many times it has been retried?

An example would be trying to access a server and failing. It would be nice on retry to be able for my task to know it should try a mirror server, or to sequentially try each new mirror server every time it is retried. If I had a list of mirror servers, the server I access during this run of my Task could be something like server_list[num_retry].

I've been searching the docs and discussions and can't locate anything that describes this. I only find state handlers, but they seem separate from the task code. The first argument to a state handler is the task object itself, but I am unsure if something can be done like I propose with this.

Any help is appreciated, thanks!

[anna-geller]

@itsayellow you can retrieve the number of retries by using the task_run_count from the context and subtracting 1 (the initial run which was not a retry). You could then use it e.g. in a state handler.

Here is a sample flow you could use to test that:

from datetime import timedelta
import prefect
from prefect import Flow, task


@task(max_retries=2, retry_delay=timedelta(seconds=2))
def retry_test():
    logger = prefect.context.get("logger")
    run_count = prefect.context.get("task_run_count")
    logger.info("%s. TaskRun", run_count)
    raise Exception("Failing to test retries...")


with Flow("retry-tester") as flow:
    retry_test()

if __name__ == "__main__":
    flow.run()

[itsayellow]

Thanks @anna-geller , that seems exactly what I'm looking for.

Is there documentation somewhere about what is available in the prefect.context? I thought there might be something useful in it but couldn't locate the documentation myself.

[itsayellow]

https://docs.prefect.io/api/latest/utilities/context.html

Finally found it now that I know what I'm looking for. 😁

[ymtricks]

Is there similar functionality in Prefect 2?

[AMontgomerie]

I also need this functionality in Prefect 2. I have a use case where I need to send a shutdown signal to a sidecar pod, but I only if the flow isn't going to retry again.

[ses4j]

The Prefect 2 equivalent is prefect.context.get_run_context().task_run.run_count