Sending custom notification when a task fail

Previously we have been using state handlers to send a customized email when any task fails in 1.0.
but in 2.0 with state handlers gone, we would have to add that logic to ALL of our flow/task definitions:

state1 = my_task1(return_state=True)
if is_failed_state(state1): send_some_custom_email()
state2 = my_task2(return_state=True)
if is_failed_state(state2): send_some_custom_email()

This approach seems cumbersome to write and maintain. Is there a way to add this custom notification logic to ALL tasks we create? (Maybe changing the notification feature in 2.0 to support custom email content?)

1 Like

Yes, it’s possible!

This post shows how you can do it by using a custom block:

and this one uses a custom decorator for task-level alerting:

lots of options, please explore various ways and report back what worked best for you :raised_hands:

Cool! These two both seem promising. I will give them a try. Thanks for the fast response here :+1:

1 Like

This is something I was also looking for, thanks for the links! Can I have my own notification block that would call REST API of an internal service instead of calling Slack?

definitely! feel free to give it a try and share your contribution - if you follow the same approach it should work

Hi @anna_geller , I tried out the second link above and it appears that the project currently does not plan to supprot task-level alerting.

Is there any updates on task-level custom alerting? Or is the best shot here still custom blocks which offers flow-level alerting? :slightly_smiling_face:

Also I was scanning through the code base and saw the notification blocks like Slack. How would I register a new notification block like the one here and let Prefect know its existence when running

you could create the Slack block from the UI and call it directly from your flow or task:

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

def send_alert():
    slack_webhook_block = SlackWebhook.load("slacknotificationblocktest") 
    slack_webhook_block.notify("Hello from Prefect!") 

def notification_flow():

Thanks Anna! I was trying to create a notification block that sends email notification on flow/task failure, so it would be really nice if I can register custom notification blocks to the backend (which I assume is running the as a service).

totally doable, just create the block from python code and it will be available in a backend


1 Like