Send an alert when a flow fails with a decorator

If you want to send an alert when a flow fails by adding only one decorator, use prefect-alert.

Installation

Install prefect-alert with pip:

pip install prefect-alert

Create a notification block

Blocks enable you to store configuration and provide an interface for interacting with external systems.

First, start with creating a notification block through UI or a Python script:

from prefect.blocks.notifications import SlackWebhook

slack_block = SlackWebhook(url="https://hooks.slack.com/services/XXX/XXX/XXX")
slack_block.save(name="test")

Send an alert

Next, use the block created and the decorator prefect_alert.alert_on_failure to send alert when a flow fails.

Send an alert when a flow fails

from prefect import flow, task 
from prefect.blocks.notifications import SlackWebhook
from prefect_alert import alert_on_failure

@task
def may_fail():
    raise ValueError()

@alert_on_failure(block_type=SlackWebhook, block_name="test")
@flow
def failed_flow():
    res = may_fail()
    return res

if __name__=="__main__":
    failed_flow()

And you will see something like this on your Slack:

Send an alert when an asynchronous flow fails

from prefect import flow, task 
from prefect.blocks.notifications import SlackWebhook
from prefect_alert import alert_on_failure
import asyncio

@task
async def may_fail():
    raise ValueError()

@alert_on_failure(block_type=SlackWebhook, block_name="test")
@flow
async def failed_flow():
    res = await may_fail()
    return res

if __name__=="__main__":
    asyncio.run(failed_flow())
2 Likes