Hi again, is it possible to set

max_retries

on a task at flow run time?

Hey @Robert Hales, would you pass it through a parameter?

It would be the result of another task ideally

Oh that I don’t think can’t be done.

Can it be done via parameter?

I don’t think so. The only potential way I can see this being done is storing the Flow as a script, and then using the KV Store.

Okay thanks!

You can retrieve the task run count from the Prefect context

prefect.context.get("task_run_count", 1)

If you pass the retry number as a parameter the retry conditional can managed internally to the flow/task, since raising the Prefect Retry signal will trigger a retry even if max_retries has been exceeded. https://docs.prefect.io/api/latest/core/task.html#task-2

Wow that’s clever @Sam Cook!

@Robert Hales this is a bit hacky but it could work thanks to the KV Store

"""
First set the key:
from prefect.backend import set_key_value
key_value_uuid = set_key_value(key="variable_nr_retries", value=5)
"""
from datetime import timedelta
import prefect
from prefect import Flow, task
from prefect.backend import get_key_value

KEY_NAME = "variable_nr_retries"


@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")
    <http://logger.info|logger.info>("%s. TaskRun", run_count)
    raise Exception("Failing to test retries...")


with Flow("retry-tester") as flow:
    nr_retries = int(get_key_value(key="variable_nr_retries"))
    retry_test(task_args=dict(max_retries=nr_retries))

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

@Anna Geller (old account) thanks for this will check it out!