prefect.blocks.notifications

Classes

AbstractAppriseNotificationBlock

An abstract class for sending notifications using Apprise. Methods:

block_initialization

block_initialization(self) -> None

logger

logger(self) -> LoggerOrAdapter
Returns a logger based on whether the NotificationBlock is called from within a flow or task run context. If a run context is present, the logger property returns a run logger. Else, it returns a default logger labeled with the class’s name. Returns:
  • The run logger or a default logger with the class’s name.

notify

notify(self, body: str, subject: str | None = None) -> None

notify

notify(self, body: str, subject: str | None = None) -> None
Send a notification. Args:
  • body: The body of the notification.
  • subject: The subject of the notification.

raise_on_failure

raise_on_failure(self) -> Generator[None, None, None]
Context manager that, while active, causes the block to raise errors if it encounters a failure sending notifications.

AppriseNotificationBlock

A base class for sending notifications using Apprise, through webhook URLs. Methods:

block_initialization

block_initialization(self) -> None

notify

notify(self, body: str, subject: str | None = None)

notify

notify(self, body: str, subject: str | None = None) -> None

SlackWebhook

Enables sending notifications via a provided Slack webhook. Examples: Load a saved Slack webhook and send a message:
from prefect.blocks.notifications import SlackWebhook

slack_webhook_block = SlackWebhook.load("BLOCK_NAME")
slack_webhook_block.notify("Hello from Prefect!")
Methods:

notify

notify(self, body: str, subject: str | None = None)

MicrosoftTeamsWebhook

Enables sending notifications via a provided Microsoft Teams webhook. Examples: Load a saved Teams webhook and send a message:
from prefect.blocks.notifications import MicrosoftTeamsWebhook
teams_webhook_block = MicrosoftTeamsWebhook.load("BLOCK_NAME")
teams_webhook_block.notify("Hello from Prefect!")
Methods:

block_initialization

block_initialization(self) -> None
see https://github.com/caronc/apprise/pull/1172

notify

notify(self, body: str, subject: str | None = None)

PagerDutyWebHook

Enables sending notifications via a provided PagerDuty webhook. See Apprise notify_pagerduty docs for more info on formatting the URL. Examples: Load a saved PagerDuty webhook and send a message:
from prefect.blocks.notifications import PagerDutyWebHook
pagerduty_webhook_block = PagerDutyWebHook.load("BLOCK_NAME")
pagerduty_webhook_block.notify("Hello from Prefect!")
Methods:

block_initialization

block_initialization(self) -> None

block_initialization

block_initialization(self) -> None

notify

notify(self, body: str, subject: str | None = None)
Apprise will combine subject and body by default, so we need to move the body into the custom_details field. custom_details is part of the webhook url, so we need to update the url and restart the client.

notify

notify(self, body: str, subject: str | None = None) -> None

TwilioSMS

Enables sending notifications via Twilio SMS. Find more on sending Twilio SMS messages in the docs. Examples: Load a saved TwilioSMS block and send a message:
from prefect.blocks.notifications import TwilioSMS
twilio_webhook_block = TwilioSMS.load("BLOCK_NAME")
twilio_webhook_block.notify("Hello from Prefect!")
Methods:

block_initialization

block_initialization(self) -> None

block_initialization

block_initialization(self) -> None

notify

notify(self, body: str, subject: str | None = None) -> None

OpsgenieWebhook

Enables sending notifications via a provided Opsgenie webhook. See Apprise notify_opsgenie docs for more info on formatting the URL. Examples: Load a saved Opsgenie webhook and send a message:
from prefect.blocks.notifications import OpsgenieWebhook
opsgenie_webhook_block = OpsgenieWebhook.load("BLOCK_NAME")
opsgenie_webhook_block.notify("Hello from Prefect!")
Methods:

block_initialization

block_initialization(self) -> None

block_initialization

block_initialization(self) -> None

notify

notify(self, body: str, subject: str | None = None) -> None

MattermostWebhook

Enables sending notifications via a provided Mattermost webhook. See Apprise notify_Mattermost docs # noqa Examples: Load a saved Mattermost webhook and send a message:
from prefect.blocks.notifications import MattermostWebhook

mattermost_webhook_block = MattermostWebhook.load("BLOCK_NAME")

mattermost_webhook_block.notify("Hello from Prefect!")
Methods:

block_initialization

block_initialization(self) -> None

block_initialization

block_initialization(self) -> None

notify

notify(self, body: str, subject: str | None = None) -> None

DiscordWebhook

Enables sending notifications via a provided Discord webhook. See Apprise notify_Discord docs # noqa Examples: Load a saved Discord webhook and send a message:
from prefect.blocks.notifications import DiscordWebhook

discord_webhook_block = DiscordWebhook.load("BLOCK_NAME")

discord_webhook_block.notify("Hello from Prefect!")
Methods:

block_initialization

block_initialization(self) -> None

block_initialization

block_initialization(self) -> None

notify

notify(self, body: str, subject: str | None = None) -> None

CustomWebhookNotificationBlock

Enables sending notifications via any custom webhook. All nested string param contains {{key}} will be substituted with value from context/secrets. Context values include: subject, body and name. Examples: Load a saved custom webhook and send a message:
from prefect.blocks.notifications import CustomWebhookNotificationBlock

custom_webhook_block = CustomWebhookNotificationBlock.load("BLOCK_NAME")

custom_webhook_block.notify("Hello from Prefect!")
Methods:

block_initialization

block_initialization(self) -> None

logger

logger(self) -> LoggerOrAdapter
Returns a logger based on whether the NotificationBlock is called from within a flow or task run context. If a run context is present, the logger property returns a run logger. Else, it returns a default logger labeled with the class’s name. Returns:
  • The run logger or a default logger with the class’s name.

notify

notify(self, body: str, subject: str | None = None) -> None

notify

notify(self, body: str, subject: str | None = None) -> None
Send a notification. Args:
  • body: The body of the notification.
  • subject: The subject of the notification.

raise_on_failure

raise_on_failure(self) -> Generator[None, None, None]
Context manager that, while active, causes the block to raise errors if it encounters a failure sending notifications.

SendgridEmail

Enables sending notifications via any sendgrid account. See Apprise Notify_sendgrid docs Examples: Load a saved Sendgrid and send a email message:
from prefect.blocks.notifications import SendgridEmail

sendgrid_block = SendgridEmail.load("BLOCK_NAME")

sendgrid_block.notify("Hello from Prefect!")
Methods:

block_initialization

block_initialization(self) -> None

block_initialization

block_initialization(self) -> None

notify

notify(self, body: str, subject: str | None = None)

notify

notify(self, body: str, subject: str | None = None) -> None