Add ability to set up notifications from the Python API

[Samreay]

First check

• [X] I added a descriptive title to this issue.
• [X] I used the GitHub search to find a similar request and didn't find it.
• [X] I searched the Prefect documentation for this feature.

Prefect Version

2.x

Describe the current behavior

Notifications are great. So great, in fact, that we'd love to set them up when we spin up Prefect. We already have a script which runs after Prefect is launched that automates the creation of our AWS, ECS, and S3 blocks. I'd love to also be able to use it to set up things like Slack notifications

Describe the proposed behavior

Right now, you can only use notification blocks (documented here: https://docs.prefect.io/api-ref/prefect/blocks/notifications/#prefect.blocks.notifications.SlackWebhook) to load an existing block and send notifications.

This is separate to a global notification like the one you can configure in the UI (I believe).

If this is already possible and I've missed it, perhaps adding to the main Notification page an example would be useful.

Example Use

Unsure how the functionality should exist, but it would be great if we can create things like SlackWebhook Blocks, save them, and configure them to run on all tasks/flows/a tag subset (aka duplicating the UI functionality with kwargs).

Additional context

No response

[github-actions[bot]]

This issue is stale because it has been open 30 days with no activity. To keep this issue open remove stale label or comment.

[ryantbrennan1]

@Samreay If I'm understanding correctly I think this is already possible but it's not documented. I have a similar automated block setup script as you and this works:

from prefect.blocks.notifications import SlackWebhook

block = SackWebhook(
    url={webhook_url},
    notify_type='failure',
)

block.save('slack-alert')

[WillRaphaelson]

Hey @Samreay - thanks for this. I'm accepting this into our backlog but its hard to tell when we'll get to it. In the interim, I'll just highlight that one way to accomplish this in code is to use the orchestration client directly. Something like:

from prefect.client.orchestration import get_client
import anyio

async def create_notification():
    async with get_client() as client:
        await client.create_flow_run_notification_policy(
            ... args ...
        )

anyio.run(create_notification())

details on that create_flow_run_notification_policy method can be found here

[Samreay]

This is very useful, thanks!