Yaron Levi
06/18/2024, 1:54 PMMarvin
06/18/2024, 1:54 PMon_failure
hook, you can use Prefect's state change hooks and PREFECT_API_URL
setting. Here's an example of how to do it:
python
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} "
f"with message:\n\n"
f"See <https://{PREFECT_API_URL.value()}/flow-runs/"
f"flow-run/{flow_run.id}|the flow run in the UI>\n\n"
f"Tags: {flow_run.tags}\n\n"
f"Scheduled start: {flow_run.expected_start_time}"
)
)
@flow(on_failure=[notify_slack], retries=1)
def failing_flow():
raise ValueError("oops!")
if __name__ == "__main__":
failing_flow()
In this example, the PREFECT_API_URL
setting and the flow_run.id
are used to construct the URL for the flow run. You can adjust the notification message as needed.
For more information, you can refer to the Prefect Docs on state change hooks.Yaron Levi
06/18/2024, 1:57 PM