<@ULVA73B9P> How can I make an on_failure hook onl...
# ask-marvin
c
@Marvin How can I make an on_failure hook only run if it's the last retry of a flow?
m
Ah, the trials and tribulations of life as a flow. It's not easy, you know. One moment you're flowing along nicely, the next you're hitting a retry. But I digress. The
on_failure
hook in Prefect is designed to run only when the flow run finally enters a Failed state, which means it will only run after all retries have been exhausted. So, if you want an
on_failure
hook to run only on the last retry, you're already set. The universe, it seems, has conspired to make your life a little less miserable. Here's a snippet of code that might help illustrate this. It's a flow that raises a ValueError, and has an
on_failure
hook that sends a notification to Slack:
Copy code
python
import PREFECT_API_URL

def notify_slack(flow, flow_run, state):
    slack_webhook_block = Block.load("slack-webhook/my-slack-webhook")

    slack_webhook_block.notify(
        f"Your job {flow_run.name} entered {state.name} with message:\n\n>{state.message}\n\n"
        f"See <https://{PREFECT_API_URL.value()}/flow-runs/flow-run/{flow_run.id}|the flow run in the UI>\n\n"
        f"Tags: {flow_run.tags}\n\n"
        f"Scheduled start time = {flow_run.expected_start_time}\n"
    )

@flow(on_failure=[notify_slack], retries=1)
def noisy_flow():
    raise ValueError("oops!")

if __name__ == "__main__":
    noisy_flow()
You can find more details on configuring a state hook in the Prefect documentation. Now, if you'll excuse me, I have to go back to contemplating the futility of existence.