Skip to content

prefect_slack.credentials

Credential classes use to store Slack credentials.

SlackCredentials

Bases: Block

Block holding Slack credentials for use in tasks and flows.

Parameters:

Name Type Description Default
token

Bot user OAuth token for the Slack app used to perform actions.

 required

Examples:

Load stored Slack credentials:

from prefect_slack import SlackCredentials
slack_credentials_block = SlackCredentials.load("BLOCK_NAME")

Get a Slack client:

from prefect_slack import SlackCredentials
slack_credentials_block = SlackCredentials.load("BLOCK_NAME")
client = slack_credentials_block.get_client()
Source code in prefect_slack/credentials.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
class SlackCredentials(Block):
    """
    Block holding Slack credentials for use in tasks and flows.

    Args:
        token: Bot user OAuth token for the Slack app used to perform actions.

    Examples:
        Load stored Slack credentials:
        ```python
        from prefect_slack import SlackCredentials
        slack_credentials_block = SlackCredentials.load("BLOCK_NAME")
        ```

        Get a Slack client:
        ```python
        from prefect_slack import SlackCredentials
        slack_credentials_block = SlackCredentials.load("BLOCK_NAME")
        client = slack_credentials_block.get_client()
        ```
    """  # noqa E501

    _block_type_name = "Slack Credentials"
    _logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/c1965ecbf8704ee1ea20d77786de9a41ce1087d1-500x500.png"  # noqa
    _documentation_url = "https://prefecthq.github.io/prefect-slack/credentials/#prefect_slack.credentials.SlackCredentials"  # noqa

    token: SecretStr = Field(
        default=...,
        description="Bot user OAuth token for the Slack app used to perform actions.",
    )

    def get_client(self) -> AsyncWebClient:
        """
        Returns an authenticated `AsyncWebClient` to interact with the Slack API.
        """
        return AsyncWebClient(token=self.token.get_secret_value())

get_client()

Returns an authenticated AsyncWebClient to interact with the Slack API.

Source code in prefect_slack/credentials.py 
45
46
47
48
49
def get_client(self) -> AsyncWebClient:
    """
    Returns an authenticated `AsyncWebClient` to interact with the Slack API.
    """
    return AsyncWebClient(token=self.token.get_secret_value())

SlackWebhook

Bases: NotificationBlock

Block holding a Slack webhook for use in tasks and flows.

Parameters:

Name Type Description Default
url

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

 required

Examples:

Load stored Slack webhook:

from prefect_slack import SlackWebhook
slack_webhook_block = SlackWebhook.load("BLOCK_NAME")

Get a Slack webhook client:

from prefect_slack import SlackWebhook
slack_webhook_block = SlackWebhook.load("BLOCK_NAME")
client = slack_webhook_block.get_client()

Send a notification in Slack:

from prefect_slack import SlackWebhook
slack_webhook_block = SlackWebhook.load("BLOCK_NAME")
slack_webhook_block.notify("Hello, world!")
Source code in prefect_slack/credentials.py 
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
class SlackWebhook(NotificationBlock):
    """
    Block holding a Slack webhook for use in tasks and flows.

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

    Examples:
        Load stored Slack webhook:
        ```python
        from prefect_slack import SlackWebhook
        slack_webhook_block = SlackWebhook.load("BLOCK_NAME")
        ```

        Get a Slack webhook client:
        ```python
        from prefect_slack import SlackWebhook
        slack_webhook_block = SlackWebhook.load("BLOCK_NAME")
        client = slack_webhook_block.get_client()
        ```

        Send a notification in Slack:
        ```python
        from prefect_slack import SlackWebhook
        slack_webhook_block = SlackWebhook.load("BLOCK_NAME")
        slack_webhook_block.notify("Hello, world!")
        ```
    """

    _block_type_name = "Slack Incoming Webhook"
    _logo_url = "https://images.ctfassets.net/gm98wzqotmnx/7dkzINU9r6j44giEFuHuUC/85d4cd321ad60c1b1e898bc3fbd28580/5cb480cd5f1b6d3fbadece79.png?h=250"  # noqa
    _documentation_url = "https://prefecthq.github.io/prefect-slack/credentials/#prefect_slack.credentials.SlackWebhook"  # noqa

    url: SecretStr = Field(
        default=...,
        title="Webhook URL",
        description="Slack webhook URL which can be used to send messages.",
        examples=["https://hooks.slack.com/XXX"],
    )

    def get_client(self) -> AsyncWebhookClient:
        """
        Returns an authenticated `AsyncWebhookClient` to interact with the configured
        Slack webhook.
        """
        return AsyncWebhookClient(url=self.url.get_secret_value())

    @sync_compatible
    async def notify(self, body: str, subject: Optional[str] = None):
        """
        Sends a message to the Slack channel.
        """
        client = self.get_client()

        response = await client.send(text=body)

        # prefect>=2.17.2 added a means for notification blocks to raise errors on
        # failures. This is not available in older versions, so we need to check if the
        # private base class attribute exists before using it.
        if getattr(self, "_raise_on_failure", False):  # pragma: no cover
            try:
                from prefect.blocks.abstract import NotificationError
            except ImportError:
                NotificationError = Exception

            if response.status_code >= 400:
                raise NotificationError(f"Failed to send message: {response.body}")

get_client()

Returns an authenticated AsyncWebhookClient to interact with the configured Slack webhook.

Source code in prefect_slack/credentials.py 
93
94
95
96
97
98
def get_client(self) -> AsyncWebhookClient:
    """
    Returns an authenticated `AsyncWebhookClient` to interact with the configured
    Slack webhook.
    """
    return AsyncWebhookClient(url=self.url.get_secret_value())

notify(body, subject=None) async

Sends a message to the Slack channel.

Source code in prefect_slack/credentials.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
@sync_compatible
async def notify(self, body: str, subject: Optional[str] = None):
    """
    Sends a message to the Slack channel.
    """
    client = self.get_client()

    response = await client.send(text=body)

    # prefect>=2.17.2 added a means for notification blocks to raise errors on
    # failures. This is not available in older versions, so we need to check if the
    # private base class attribute exists before using it.
    if getattr(self, "_raise_on_failure", False):  # pragma: no cover
        try:
            from prefect.blocks.abstract import NotificationError
        except ImportError:
            NotificationError = Exception

        if response.status_code >= 400:
            raise NotificationError(f"Failed to send message: {response.body}")