Skip to content

prefect.blocks.notifications

AbstractAppriseNotificationBlock pydantic-model

An abstract class for sending notifications using Apprise.

Source code in prefect/blocks/notifications.py
class AbstractAppriseNotificationBlock(NotificationBlock, ABC):
    """
    An abstract class for sending notifications using Apprise.
    """

    notify_type: Literal[
        "prefect_default", "info", "success", "warning", "failure"
    ] = Field(
        default=PrefectNotifyType.DEFAULT,
        description=(
            "The type of notification being performed; the prefect_default "
            "is a plain notification that does not attach an image."
        ),
    )

    def _start_apprise_client(self, url: SecretStr):
        # A custom `AppriseAsset` that ensures Prefect Notifications
        # appear correctly across multiple messaging platforms
        prefect_app_data = AppriseAsset(
            app_id="Prefect Notifications",
            app_desc="Prefect Notifications",
            app_url="https://prefect.io",
        )

        self._apprise_client = Apprise(asset=prefect_app_data)
        self._apprise_client.add(url.get_secret_value())

    def block_initialization(self) -> None:
        self._start_apprise_client(self.url)

    @sync_compatible
    async def notify(self, body: str, subject: Optional[str] = None):
        await self._apprise_client.async_notify(
            body=body, title=subject, notify_type=self.notify_type
        )

notify_type pydantic-field

Type: Literal['prefect_default', 'info', 'success', 'warning', 'failure']

The type of notification being performed; the prefect_default is a plain notification that does not attach an image.

AppriseNotificationBlock pydantic-model

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

Source code in prefect/blocks/notifications.py
class AppriseNotificationBlock(AbstractAppriseNotificationBlock, ABC):
    """
    A base class for sending notifications using Apprise, through webhook URLs.
    """

    url: SecretStr = Field(
        default=...,
        title="Webhook URL",
        description="Incoming webhook URL used to send notifications.",
        example="https://hooks.example.com/XXX",
    )

url pydantic-field required

Type: SecretStr

Incoming webhook URL used to send notifications.

MicrosoftTeamsWebhook pydantic-model

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

Source code in prefect/blocks/notifications.py
class MicrosoftTeamsWebhook(AppriseNotificationBlock):
    """
    Enables sending notifications via a provided Microsoft Teams webhook.

    Examples:
        Load a saved Teams webhook and send a message:
        ```python
        from prefect.blocks.notifications import MicrosoftTeamsWebhook
        teams_webhook_block = MicrosoftTeamsWebhook.load("BLOCK_NAME")
        teams_webhook_block.notify("Hello from Prefect!")
        ```
    """

    _block_type_name = "Microsoft Teams Webhook"
    _block_type_slug = "ms-teams-webhook"
    _logo_url = "https://images.ctfassets.net/gm98wzqotmnx/6n0dSTBzwoVPhX8Vgg37i7/9040e07a62def4f48242be3eae6d3719/teams_logo.png?h=250"

    url: SecretStr = Field(
        ...,
        title="Webhook URL",
        description="The Teams incoming webhook URL used to send notifications.",
        example="https://your-org.webhook.office.com/webhookb2/XXX/IncomingWebhook/YYY/ZZZ",
    )

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

PagerDutyWebHook pydantic-model

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

Source code in prefect/blocks/notifications.py
class PagerDutyWebHook(AbstractAppriseNotificationBlock):
    """
    Enables sending notifications via a provided PagerDuty webhook.
    See [Apprise notify_pagerduty docs](https://github.com/caronc/apprise/wiki/Notify_pagerduty)
    for more info on formatting the URL.

    Examples:
        Load a saved PagerDuty webhook and send a message:
        ```python
        from prefect.blocks.notifications import PagerDutyWebHook
        pagerduty_webhook_block = PagerDutyWebHook.load("BLOCK_NAME")
        pagerduty_webhook_block.notify("Hello from Prefect!")
        ```
    """

    _description = "Enables sending notifications via a provided PagerDuty webhook."

    _block_type_name = "Pager Duty Webhook"
    _block_type_slug = "pager-duty-webhook"
    _logo_url = "https://images.ctfassets.net/gm98wzqotmnx/6FHJ4Lcozjfl1yDPxCvQDT/c2f6bdf47327271c068284897527f3da/PagerDuty-Logo.wine.png?h=250"

    # The default cannot be prefect_default because NotifyPagerDuty's
    # PAGERDUTY_SEVERITY_MAP only has these notify types defined as keys
    notify_type: Literal["info", "success", "warning", "failure"] = Field(
        default="info", description="The severity of the notification."
    )

    integration_key: SecretStr = Field(
        default=...,
        description=(
            "This can be found on the Events API V2 "
            "integration's detail page, and is also referred to as a Routing Key. "
            "This must be provided alongside `api_key`, but will error if provided "
            "alongside `url`."
        ),
    )

    api_key: SecretStr = Field(
        default=...,
        title="API Key",
        description=(
            "This can be found under Integrations. "
            "This must be provided alongside `integration_key`, but will error if "
            "provided alongside `url`."
        ),
    )

    source: Optional[str] = Field(
        default="Prefect", description="The source string as part of the payload."
    )

    component: str = Field(
        default="Notification",
        description="The component string as part of the payload.",
    )

    group: Optional[str] = Field(
        default=None, description="The group string as part of the payload."
    )

    class_id: Optional[str] = Field(
        default=None,
        title="Class ID",
        description="The class string as part of the payload.",
    )

    region_name: Literal["us", "eu"] = Field(
        default="us", description="The region name."
    )

    clickable_url: Optional[AnyHttpUrl] = Field(
        default=None,
        title="Clickable URL",
        description="A clickable URL to associate with the notice.",
    )

    include_image: bool = Field(
        default=True,
        description="Associate the notification status via a represented icon.",
    )

    custom_details: Optional[Dict[str, str]] = Field(
        default=None,
        description="Additional details to include as part of the payload.",
        example='{"disk_space_left": "145GB"}',
    )

    def block_initialization(self) -> None:
        url = SecretStr(
            NotifyPagerDuty(
                apikey=self.api_key.get_secret_value(),
                integrationkey=self.integration_key.get_secret_value(),
                source=self.source,
                component=self.component,
                group=self.group,
                class_id=self.class_id,
                region_name=self.region_name,
                click=self.clickable_url,
                include_image=self.include_image,
                details=self.custom_details,
            ).url()
        )
        self._start_apprise_client(url)

api_key pydantic-field required

Type: SecretStr

This can be found under Integrations. This must be provided alongside integration_key, but will error if provided alongside url.

class_id pydantic-field

Type: str

The class string as part of the payload.

clickable_url pydantic-field

Type: AnyHttpUrl

A clickable URL to associate with the notice.

component pydantic-field

Type: str

The component string as part of the payload.

custom_details pydantic-field

Type: Dict[str, str]

Additional details to include as part of the payload.

group pydantic-field

Type: str

The group string as part of the payload.

include_image pydantic-field

Type: bool

Associate the notification status via a represented icon.

integration_key pydantic-field required

Type: SecretStr

This can be found on the Events API V2 integration's detail page, and is also referred to as a Routing Key. This must be provided alongside api_key, but will error if provided alongside url.

region_name pydantic-field

Type: Literal['us', 'eu']

The region name.

source pydantic-field

Type: str

The source string as part of the payload.

PrefectNotifyType

A mapping of Prefect notification types for use with Apprise.

Attributes:

Name Description
DEFAULT

A plain notification that does not insert any notification type images.

Source code in prefect/blocks/notifications.py
class PrefectNotifyType(NotifyType):
    """
    A mapping of Prefect notification types for use with Apprise.

    Attributes:
        DEFAULT: A plain notification that does not insert any notification type images.
    """

    DEFAULT = "prefect_default"

SlackWebhook pydantic-model

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

Source code in prefect/blocks/notifications.py
class SlackWebhook(AppriseNotificationBlock):
    """
    Enables sending notifications via a provided Slack webhook.

    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(
        default=...,
        title="Webhook URL",
        description="Slack incoming webhook URL used to send notifications.",
        example="https://hooks.slack.com/XXX",
    )

TwilioSMS pydantic-model

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

Source code in prefect/blocks/notifications.py
class TwilioSMS(AbstractAppriseNotificationBlock):
    """Enables sending notifications via Twilio SMS.
    Find more on sending Twilio SMS messages in the [docs](https://www.twilio.com/docs/sms).

    Examples:
        Load a saved `TwilioSMS` block and send a message:
        ```python
        from prefect.blocks.notifications import TwilioSMS
        twilio_webhook_block = TwilioSMS.load("BLOCK_NAME")
        twilio_webhook_block.notify("Hello from Prefect!")
        ```
    """

    _description = "Enables sending notifications via Twilio SMS."
    _block_type_name = "Twilio SMS"
    _block_type_slug = "twilio-sms"
    _logo_url = "https://images.ctfassets.net/zscdif0zqppk/YTCgPL6bnK3BczP2gV9md/609283105a7006c57dbfe44ee1a8f313/58482bb9cef1014c0b5e4a31.png?h=250"  # noqa

    account_sid: str = Field(
        default=...,
        description=(
            "The Twilio Account SID - it can be found on the homepage "
            "of the Twilio console."
        ),
    )

    auth_token: SecretStr = Field(
        default=...,
        description=(
            "The Twilio Authentication Token - "
            "it can be found on the homepage of the Twilio console."
        ),
    )

    from_phone_number: str = Field(
        default=...,
        description="The valid Twilio phone number to send the message from.",
        example="18001234567",
    )

    to_phone_numbers: List[str] = Field(
        default=...,
        description="A list of valid Twilio phone number(s) to send the message to.",
        # not wrapped in brackets because of the way UI displays examples; in code should be ["18004242424"]
        example="18004242424",
    )

    def block_initialization(self) -> None:
        url = SecretStr(
            NotifyTwilio(
                account_sid=self.account_sid,
                auth_token=self.auth_token.get_secret_value(),
                source=self.from_phone_number,
                targets=self.to_phone_numbers,
            ).url()
        )
        self._start_apprise_client(url)

account_sid pydantic-field required

Type: str

The Twilio Account SID - it can be found on the homepage of the Twilio console.

auth_token pydantic-field required

Type: SecretStr

The Twilio Authentication Token - it can be found on the homepage of the Twilio console.

from_phone_number pydantic-field required

Type: str

The valid Twilio phone number to send the message from.

to_phone_numbers pydantic-field required

Type: List[str]

A list of valid Twilio phone number(s) to send the message to.