Skip to content

Slack

Classes to ease interaction with Slack

koheesio.notifications.slack.SlackNotification #

Generic Slack notification class via the Blocks API

NOTE: channel parameter is used only with Slack Web API: https://api.slack.com/messaging/sending If webhook is used, the channel specification is not required

Example: 

s = SlackNotification(
    url="slack-webhook-url",
    channel="channel",
    message="Some *markdown* compatible text",
)
s.execute()

channel class-attribute instance-attribute #

channel: Optional[str] = Field(default=None, description='Slack channel id')

headers class-attribute instance-attribute #

headers: Optional[Dict[str, Any]] = {'Content-type': 'application/json'}

message class-attribute instance-attribute #

message: str = Field(default=..., description='The message that gets posted to Slack')

execute #

execute()

Generate payload and send post request

Source code in src/koheesio/notifications/slack.py 
63
64
65
66
67
68
def execute(self):
    """
    Generate payload and send post request
    """
    self.data = self.get_payload()
    HttpPostStep.execute(self)

get_payload #

get_payload()

Generate payload with Block Kit. More details: https://api.slack.com/block-kit

Source code in src/koheesio/notifications/slack.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def get_payload(self):
    """
    Generate payload with `Block Kit`.
    More details: https://api.slack.com/block-kit
    """
    payload = {
        "attachments": [
            {
                "blocks": [
                    {
                        "type": "section",
                        "text": {
                            "type": "mrkdwn",
                            "text": self.message,
                        },
                    }
                ],
            }
        ]
    }

    if self.channel:
        payload["channel"] = self.channel

    return json.dumps(payload)

koheesio.notifications.slack.SlackNotificationWithSeverity #

Slack notification class via the Blocks API with etra severity information and predefined extra fields

Example: from koheesio.steps.integrations.notifications import NotificationSeverity

s = SlackNotificationWithSeverity(
    url="slack-webhook-url",
    channel="channel",
    message="Some *markdown* compatible text"
    severity=NotificationSeverity.ERROR,
    title="Title",
    environment="dev",
    application="Application"
)
s.execute()

application class-attribute instance-attribute #

application: str = Field(default=..., description='Pipeline or application name')

environment class-attribute instance-attribute #

environment: str = Field(default=..., description='Environment description, e.g. dev / qa /prod')

model_config class-attribute instance-attribute #

model_config = ConfigDict(use_enum_values=False)

severity class-attribute instance-attribute #

severity: NotificationSeverity = Field(default=..., description='Severity of the message')

timestamp class-attribute instance-attribute #

timestamp: datetime = Field(default=utcnow(), alias='execution_timestamp', description='Pipeline or application execution timestamp')

title class-attribute instance-attribute #

title: str = Field(default=..., description='Title of your message')

execute #

execute()

Generate payload and send post request

Source code in src/koheesio/notifications/slack.py 
116
117
118
119
120
121
122
def execute(self):
    """
    Generate payload and send post request
    """
    self.message = self.get_payload_message()
    self.data = self.get_payload()
    HttpPostStep.execute(self)

get_payload_message #

get_payload_message()

Generate payload message based on the predefined set of parameters

Source code in src/koheesio/notifications/slack.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
def get_payload_message(self):
    """
    Generate payload message based on the predefined set of parameters
    """
    return dedent(
        f"""
            {self.severity.alert_icon}   *{self.severity.name}:*  {self.title}
            *Environment:* {self.environment}
            *Application:* {self.application}
            *Message:* {self.message}
            *Timestamp:* {self.timestamp}
        """
    )