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() -> None

Generate payload and send post request

Source code in src/koheesio/notifications/slack.py

def execute(self) -> None:
    """
    Generate payload and send post request
    """
    self.data = self.get_payload()
    HttpPostStep.execute(self)

get_payload

get_payload() -> str

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

Source code in src/koheesio/notifications/slack.py

def get_payload(self) -> str:
    """
    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  # type: ignore[assignment]

    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_factory=utc_now,
    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() -> None

Generate payload and send post request

Source code in src/koheesio/notifications/slack.py

def execute(self) -> None:
    """
    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() -> str

Generate payload message based on the predefined set of parameters

Source code in src/koheesio/notifications/slack.py

def get_payload_message(self) -> str:
    """
    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}
        """
    )