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. 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)