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())