Hi! I have a somewhat flaky Prefect task that runs some ETL. I’ve already decorated the task with retry logic (

max_retries

,

retry_delay

, etc), and there’s a task after the ETL task to send a notification to slack (not quite using the Slack integration, but querying the Slack API directly). That said, is it possible to hook into the retry logic such that whenever that ETL task gets retried, I can execute some logic? For example, “whenever this task is going to be put into the

RETRY

state, run this code”? Would this be a state handler like the

terminal_state_handler

on the

Flow

, but for the specific task? Or just a state handler passed to

Flow(state_handlers=[send_slack_message_on_retry])

?

Hey @Danny Vilela, you can do it with state handlers (not terminal). You can do

@task(state_handlers = […]

and then inside that state handler your signature would be `(task, old_state, new_state)`and then you would new

new_state.is_retrying()

as your condition

🤦‍♂️ Totally missed it. Thank you so much for all your hard work @Kevin Kho!!

No problem!

Ah, one last question: I’ve previously asked about using the

prefect.context.task_run_count

context variable, but is that context variable accessible within a state handler? Do we need to pass the task itself somehow? I know

max_retries

is an attribute on

Task

so we have the denominator, but what about getting the number of retries for a given task within a state handler?

yes! just get it with

prefect.context.get(…)

wow that is magical. thank you!