How to send Slack notifications using the SlackTask

1. Create a slack webhook

First, to create a webhook, go to Slack API: Applications | Slack → Create a New App → From Scratch.

Choose a name for your app and select your desired Slack workspace, then confirm “Create App”:

Select incoming webhooks and switch the toggle on.

Next, create a webhook for the Slack channel of your choice (“Add New Webhook to Workspace”).

Select the channel of your choice and confirm with “Allow”:

Once all that’s done, you can copy your Webhook URL and use it in your flow!

Note that you should treat this webhook in the same way as an API key or password. That’s why in the next section, you’ll see how you can store this webhook as a Prefect Secret.

2. Store the webhook in your Prefect Cloud Secrets

Go to: https://cloud.prefect.io/team/secrets

Click on “Add Secret” and create a Secret with a name SLACK_WEBHOOK_URL:

3. Use the SlackTask in your Flow

If you used the default Secret name called SLACK_WEBHOOK_URL, you can call it within your Flow as follows:

from prefect import Flow
from prefect.tasks.notifications import SlackTask

slack = SlackTask()


with Flow("slack_test") as flow:
    slack(message="Hi from a test flow!")

if __name__ == "__main__":
    flow.run()

When you run it, you should see your test message! :tada:

image

If you used a different Secret name, you need to specify it within your task:

from prefect import Flow
from prefect.tasks.notifications import SlackTask

slack = SlackTask(webhook_secret="YOUR_CUSTOM_SECRET_NAME")


with Flow("slack_test") as flow:
    slack(message="Hi from a test flow!")

if __name__ == "__main__":
    flow.run()

4. Using SlackTask in a state handler to send failure notifications

If you use this task within a state handler, you need to explicitly call the task’s run method SlackTask(message=msg).run().

Why?
Because when you call this task in a Flow block as shown in the section above, Prefect automatically calls the task’s run() method. But when you invoke this task in a state handler (i.e. not within an active Flow context), you need to call the task’s run() method explicitly.

import prefect
from prefect import task, Flow
from prefect.tasks.notifications import SlackTask
from typing import cast


def post_to_slack_on_failure(task, old_state, new_state):
    if new_state.is_failed():
        if isinstance(new_state.result, Exception):
            value = "```{}```".format(repr(new_state.result))
        else:
            value = cast(str, new_state.message)
        msg = (
            f"The task `{prefect.context.task_name}` failed "
            f"in a flow run {prefect.context.flow_run_id} "
            f"with an exception {value}"
        )
        SlackTask(message=msg).run()
    return new_state


@task(state_handlers=[post_to_slack_on_failure])
def divide_numbers(a, b):
    return 1 / (b - a)


with Flow(name="flow_failing_with_bad_parameters") as flow:
    result = divide_numbers(1, 1)


if __name__ == "__main__":
    flow.run()

After running the example above, you should get the following Slack message:

Troubleshooting Secrets

If you see an error telling you that Prefect couldn’t find a local secret, it means that you need to configure your settings to use Prefect Cloud Secrets. This page explains how you can do that:

If you are on Prefect Server, then you need to use local Secrets, since Prefect Server doesn’t have a Secrets backend. To see how you can set Secrets for Prefect Server, check out this topic (make sure to replace the GITHUB_ACCESS_TOKEN with SLACK_WEBHOOK_URL - other than that, you can follow all instructions explained there :point_down: ):