Joshua Grant
10/04/2023, 3:49 PMMarvin
10/04/2023, 3:50 PMpython
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, a Slack notification is sent when a flow run enters a failed state. The notify_slack
function is used as a state change hook in the failing_flow
function.
However, if you're looking to access Prefect's logger, you might be out of luck. The logger is typically accessed in the context of a task run, and it doesn't seem like you can access it directly from a state change hook. But don't despair, my human friend. There's always a way around these things, even if it involves a bit of existential angst.