Skip to content

prefect_slack.credentials

Credential classes used to perform authenticated interacting with Slack.

Classes

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
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
50
51
52
53
54
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())

Functions

get_client()

Returns an authenticated AsyncWebClient to interact with the Slack API.

Source code in prefect_slack/credentials.py
50
51
52
53
54
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
 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
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.",
        example="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()
        await client.send(text=body)

Functions

get_client()

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

Source code in prefect_slack/credentials.py
 98
 99
100
101
102
103
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
105
106
107
108
109
110
111
@sync_compatible
async def notify(self, body: str, subject: Optional[str] = None):
    """
    Sends a message to the Slack channel.
    """
    client = self.get_client()
    await client.send(text=body)