is there anyway to set the

max_retries

on a task from the UI (or without having to upload new code)? I was hoping I could template the value and pass it to the @task decorator, but it doesn't look like you can use templates with the

max_retries

argument

Hey @Kathryn Klarich, am pretty sure there isn’t a way unless you use the interactive API which would be convoluted. Some people use a state handler to put conditional logic on the retries if you need it to be dynamic. State handlers can get parameters from the context

Hmm, so the state handler would call the task ?

So the state handler is called between state transitions of the task. If the

old_state

is

Failed

and the new state is

Retrying

, you can have some kind of conditional logic (maybe based on parameters) to not retry or to return

Success

.

ah ok, that's what i was thinking

Check this:

import prefect
from prefect import task, Flow, Parameter
import datetime
from prefect.engine.state import Retrying, Success

def my_state_handler(task, old_state, new_state):
    test = prefect.context.get("parameters")
    real_max_retries = test['real_max_retries']

    run_count = prefect.context.get("task_run_count")

    if old_state.is_failed() and (run_count > real_max_retries):
        print("finished real max retries")
        return Success("test")

    return 

@task(max_retries=10, retry_delay=datetime.timedelta(seconds=2), state_handlers=[my_state_handler])
def data():
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>("I am trying a task")
    raise ValueError()
    return

with Flow("test") as flow:
    real_max_retries = Parameter("real_max_retries", 5)()
    data()

flow.run()

cool, thanks! So if i return a new state (e.g. failed , success) from the statehandler, it will prevent it from retrying?

Yes, it will replace the new state