Kathryn Klarich
09/14/2021, 5:41 PMmax_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
argumentKevin Kho
Kathryn Klarich
09/14/2021, 5:54 PMKevin Kho
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
.Kathryn Klarich
09/14/2021, 5:57 PMKevin Kho
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()
Kevin Kho
max_retries
is below real_max_retries
but this will only run 5 times even if max_retries
is 10. And then you can configure that value through the Parameter
Kathryn Klarich
09/14/2021, 6:24 PMKevin Kho