Sending Notifications in Cloud 1.0 Using Automations -> Cloud 2.0 Slack Webhook Blocks and Notifications

Summary

This article is useful for those who want to be alerted on events and changes that occur when running their Prefect workflows. We will be covering 1) how to set up Automations in Cloud 1.0 to send a Slack notification, and 2) how to implement Slack Webhook blocks and Notifications in Cloud 2.0.

Video Demo

What is Staying the Same

Automations in Cloud 1.0 and Notifications in Cloud 2.0 are both designed to alert users about the state changes of their flow runs. Within both UIs, users can specify which flows to apply alerts too, and which state changes trigger the alerts. For Slack alerts, both Automations and Notifications require webhook URLs to send messages to the appropriate channel.

What is Changing

Cloud 2.0 can send slack notifications via two methods, Slack Webhook Blocks and Notifications. Slack Webhook Blocks are created and stored through the Cloud 2.0 UI, and called within flow’s code. The SlackWebhook block is used to the store the webhook URL used for the notification. In 1.0, the URL would be stored as a Prefect secret.

Cloud 2.0 Notifications are created and saved within the UI, and can be configured to fire off only when a flow with certain tags reaches a specified state. When no flow tags are specified during the creation of the Notification, it is applied to all flow runs which enter the state specified by the user. Tags were not used as a filtering method in 1.0.

How do we do this in Prefect 1.0?

Automations allow us to configure actions that are triggered when a specific event or change takes place in our Prefect ecosystem. To create an automation, we need to access the Cloud 1.0 UI and navigate to the Automations page. From there, we are able to customize and save our Automations.

For the purpose of this demo, we are going to create an Automation which notifies us when a very important flow has finished.

Here is our very important flow:

#My Very Important Prefect 1.0 Flow
import prefect
from prefect import task, Flow

#Here is my 1.0 Task
@task
def say_hello():
    logger = prefect.context.get("logger")
    logger.info("Hello, Cloud!")

#Here is my 1.0 Flow
with Flow("hello-flow") as flow:
    say_hello()

# Register the flow under the "FirstProject" project
flow.register(project_name="FirstProject")

Now that our flow is ready, we need to ensure we complete the following housekeeping steps before creating an Automation.

  1. Ensure we have Prefect 1.0 installed into our environment. (this example uses Prefect ver. 1.2.4)
  2. Log into Cloud.
  3. Set the the Prefect Cloud backend.
  4. Authenticate with Prefect Cloud.
  5. Create a Project (this example has a project called “FirstProject”).
  6. Register our important flow to our project.
  7. Start an agent.

The step of registering the flow is essential, because it enables us to set an Automation for the flow within the UI.

Since this Automation is going to be sending a notification to a Slack channel, we need to create an incoming webhook so that we can post the message to Slack. To do this, we can follow the steps for creating an incoming webhook through Slack’s guide. For this demo, we have created an app called ‘Automation-Message’, and have created and incoming web hook for it. The Webhook URL is the last piece of the puzzle we need in order to create an Automation.

Let’s navigate to the Secrets page in the 1.0 UI to save our webhook URL. Once on this page, we can create a new secret containing our webhook, and give it a name so that we can reference it when creating the Automation.

Now we’re ready to create the Automation. From the Automation page, we can start customizing a new automation. For this demo, we are creating an automation that fires off a slack notification (using our webhook) when any run from our flow finishes.

To test this out, we can navigate to the flow page and hit ‘Quick Run’ in the top right hand corner of the screen.

When the flow completes, we will receive a slack notification in the channel we specified when creating our Slack app and Webhook. Nice!

How do we do this in Prefect 2.0?

Slack Webhook Block

In prefect 2.0, we can send Slack notifications in two ways. The first method we will try is by using a Slack Webhook Block. Blocks allow us to store data securely with the Prefect Orion server, and access it directly from our code later. This block will be called within our very important flow:

from prefect import flow, task

@task(name = "SlackNotif")
def test_notif():
    print(f"Hello from Prefect!")
    
@flow(name="Important-2.0-Flow")
def basic_flow():
    test_notif()


basic_flow()

Let’s be sure that we have completed the following housekeeping steps prior to trying out this demo:

  1. Ensure that we have Prefect 2.0 installed in our environment. (This example uses Prefect ver. 2.0.2)
  2. Sign in or register a Prefect Cloud 2.0 account.
  3. Create workspaces for your account.
  4. Create an API key to authorize a local execution environment.
  5. Configure Orion Settings to use Prefect Cloud.
  6. Configure storage.

We also need to have an incoming webhook URL to store within our Slack Webhook block. Follow the steps for creating an incoming webhook through Slack’s guide. For this part of the tutorial, I have created an App called ‘Block-Notification’, and have created an incoming webhook for it.

To create a notification block, we need to access the Cloud 2.0 UI and navigate to the Blocks page. From there, we can create a Slack Webhook block.

Once the block is created, we can add a few lines of code to our original flow so that it can send a message to Slack.

from prefect import flow, task
from prefect.blocks.notifications import SlackWebhook #import the SlackWebhook module


@task(name = "SlackNotif")
def test_notif():
    slack_webhook_block = SlackWebhook.load("slacknotificationblocktest") #load the Slack Webhook block
    slack_webhook_block.notify(f"Hello from Prefect!") #create our notification

@flow(name="Important-2.0-Flow")
def basic_flow():
    test_notif()

basic_flow()

To test this out, we can execute the flow locally within the terminal. When the flow completes, we will receive a slack notification in the channel we specified when creating our Slack app and Webhook. Cool!

Notifications

The second method we can use in Cloud 2.0 to send a message is by creating a Notification. Notifications are similar to Automations, in that they enable us to send an alert based off a specific state change or event that occurs when a flow is running.

To create a Notification, navigate to the Notifications page and select Create Notification. We will be prompted to specify the flow run states which will trigger the notification. We can also specify specific flow tags that filter which flow runs are covered by the notification, and select whether the notification will be sent as an email or slack message. For this demo, we have selected the ‘Slack Webhook’ option, and pasted the web hook URL in the space provided.

Our example notification is designed to do the following:

“If a run of any flow with any tag enters a ‘Completed’ state, send a notification to slack.”

After creating the Notification, we can run the following flow in our terminal and receive a comprehensive message in the slack channel that we specified when creating our webhook.

from prefect import flow, task

@task(name = "SlackNotif")
def test_notif():
    print(f"Hello from Prefect!")
    
@flow(name="Important-2.0-Flow")
def basic_flow():
    test_notif()


basic_flow()

Sweet! Notice that since we did not specify any tags when creating the block, all flow runs which enter a ‘Completed’ state will fire off this notification.

4 Likes

With Add sync compatible on SlackWebhook by ahuang11 · Pull Request #6166 · PrefectHQ/prefect · GitHub merged in prefect==2.0.2, I think you can directly do without async:

from prefect import flow, task
from prefect.blocks.notifications import SlackWebhook #import the SlackWebhook module


@task(name = "SlackNotif")
def test_notif():
    slack_webhook_block = SlackWebhook.load("slacknotificationblocktest") #load the Slack Webhook block
    slack_webhook_block.notify(f"Hello from Prefect!") #create our notification

@flow(name="Important-2.0-Flow")
def basic_flow():
    test_notif()

basic_flow()
2 Likes

Thank you Andrew! Updated to 2.0.2 and it works like a charm.

2 Likes