prefect.blocks.notifications

NotificationBlock pydantic-model

A Block base class for sending notifications.

Source code in prefect/blocks/notifications.py

class NotificationBlock(Block, ABC):
    """
    A `Block` base class for sending notifications.
    """

    _block_schema_capabilities = ["notify"]

    @abstractmethod
    async def notify(self, body: str, subject: Optional[str] = None):
        """
        Send a notification
        """

NotificationBlock.notify async

Send a notification

Source code in prefect/blocks/notifications.py

@abstractmethod
async def notify(self, body: str, subject: Optional[str] = None):
    """
    Send a notification
    """

SlackWebhook pydantic-model

Enables sending notifications via a provided Slack webhook.

Parameters:

Name	Description	Default
url	Slack webhook URL which can be used to send messages (e.g. https://hooks.slack.com/XXX). SecretStr	required

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!")

Source code in prefect/blocks/notifications.py

class SlackWebhook(NotificationBlock):
    """
    Enables sending notifications via a provided Slack webhook.

    Args:
        url (SecretStr): Slack webhook URL which can be used to send messages
            (e.g. `https://hooks.slack.com/XXX`).

    Examples:
        Load a saved Slack webhook and send a message:
        ```python
        from prefect.blocks.notifications import SlackWebhook

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

    _block_type_name = "Slack Webhook"
    _logo_url = "https://images.ctfassets.net/gm98wzqotmnx/7dkzINU9r6j44giEFuHuUC/85d4cd321ad60c1b1e898bc3fbd28580/5cb480cd5f1b6d3fbadece79.png?h=250"

    url: SecretStr = Field(..., title="Webhook URL")

    def block_initialization(self) -> None:
        from slack_sdk.webhook.async_client import AsyncWebhookClient

        self._async_webhook_client = AsyncWebhookClient(url=self.url.get_secret_value())

    @sync_compatible
    async def notify(self, body: str, subject: Optional[str] = None):
        await self._async_webhook_client.send(text=body)