How to get notifications when a flow run fails with an `on_failure` state hook

Help
1

How can I get notifications when a flow run fails using an on_failure state hook?

To get notifications when a flow run fails using an on_failure state hook, follow these steps:

  1. Configure a notification block, like a SlackWebhook block
  2. In your flow definition, add an on_failure state hook to the task or flow you want to monitor.
  3. Pass the notification function you defined earlier as the argument to the on_failure state hook.

Here’s a flow illustrating the process:

from prefect import flow
from prefect.blocks.core import Block
from prefect.settings 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()

Note that the on_failure hook will not run until all retries have completed, when the flow run finally enters a Failed state.

Check out the docs on configuring a state hook here.

1 Like