Sending custom notification when a task fail

Archive
, ,
1

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
2

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:

3

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

1 Like
4

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?

5

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

6

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 flow_run_notification.py?

7

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


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

@flow
def notification_flow():
    send_alert()
8

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 flow_run_notification.py as a service).

9

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

i.e. your_block.save(“somename”)

1 Like