1"""Helpers for building robust Slack messages"""
  2
  3import json
  4import typing as t
  5from enum import Enum
  6from textwrap import dedent
  7
  8
  9SLACK_MAX_TEXT_LENGTH = 3000
 10SLACK_MAX_ALERT_PREVIEW_BLOCKS = 5
 11SLACK_MAX_ATTACHMENTS_BLOCKS = 50
 12CONTINUATION_SYMBOL = "..."
 13
 14
 15TSlackBlock = t.Dict[str, t.Any]
 16
 17
 18class TSlackBlocks(t.TypedDict):
 19    blocks: t.List[TSlackBlock]
 20
 21
 22class TSlackMessage(TSlackBlocks):
 23    attachments: t.List[TSlackBlocks]
 24    text: str
 25
 26
 27class SlackMessageComposer:
 28    """Builds Slack message with primary and secondary blocks"""
 29
 30    def __init__(self, initial_message: t.Optional[TSlackMessage] = None) -> None:
 31        """Initialize the Slack message builder"""
 32        self.slack_message: TSlackMessage = initial_message or {
 33            "text": "",
 34            "blocks": [],
 35            "attachments": [{"blocks": []}],
 36        }
 37
 38    def add_primary_blocks(self, *blocks: TSlackBlock) -> "SlackMessageComposer":
 39        """Add blocks to the message. Blocks are always displayed"""
 40        self.slack_message["blocks"].extend(blocks)
 41        return self
 42
 43    def add_secondary_blocks(self, *blocks: TSlackBlock) -> "SlackMessageComposer":
 44        """Add attachments to the message
 45
 46        Attachments are hidden behind "show more" button. The first 5 attachments
 47        are always displayed. NOTICE: attachments blocks are deprecated by Slack
 48        """
 49        self.slack_message["attachments"][0]["blocks"].extend(blocks)
 50        if len(self.slack_message["attachments"][0]["blocks"]) >= SLACK_MAX_ATTACHMENTS_BLOCKS:
 51            raise ValueError("Too many attachments")
 52        return self
 53
 54    def add_text(self, text: str) -> "SlackMessageComposer":
 55        """Add text to the message
 56
 57        This text is used in places where content cannot be rendered such as: system push notifications, assistive technology such as screen readers, etc.
 58        """
 59        self.slack_message["text"] = normalize_message(text)
 60        return self
 61
 62    def _introspect(self) -> "SlackMessageComposer":
 63        """Print the message to stdout
 64
 65        This is a debugging method. Useful during composition of the message."""
 66        print(json.dumps(self.slack_message, indent=2))
 67        return self
 68
 69
 70def normalize_message(message: t.Union[str, t.List[str], t.Tuple[str], t.Set[str]]) -> str:
 71    """Normalize message to fit Slack's max text length"""
 72    if isinstance(message, (list, tuple, set)):
 73        message = "\n".join(message)
 74    dedented_message = dedent(message)
 75    if len(dedented_message) < SLACK_MAX_TEXT_LENGTH:
 76        return dedent(dedented_message)
 77    return dedent(
 78        dedented_message[: SLACK_MAX_TEXT_LENGTH - len(CONTINUATION_SYMBOL) - 3]
 79        + CONTINUATION_SYMBOL
 80        + dedented_message[-3:]
 81    )
 82
 83
 84def divider_block() -> dict:
 85    """Create a divider block"""
 86    return {"type": "divider"}
 87
 88
 89def fields_section_block(*messages: str) -> dict:
 90    """Create a section block with multiple markdown fields"""
 91    return {
 92        "type": "section",
 93        "fields": [
 94            {
 95                "type": "mrkdwn",
 96                "text": normalize_message(message),
 97            }
 98            for message in messages
 99        ],
100    }
101
102
103def text_section_block(message: str) -> dict:
104    """Create a section block with text"""
105    return {
106        "type": "section",
107        "text": {
108            "type": "mrkdwn",
109            "text": normalize_message(message),
110        },
111    }
112
113
114def preformatted_rich_text_block(message: str) -> dict:
115    """Create a "rich text" block with pre-formatted text (i.e.: a code block).
116
117    Note that this can also be acheived with text_section_block and using markdown's
118    triple backticks.
119    This function avoids issues with the text containing such backticks and also does
120    not require editing the message in order to insert such backticks.
121    """
122    return {
123        "type": "rich_text",
124        "elements": [
125            {
126                "type": "rich_text_preformatted",
127                "elements": [
128                    {
129                        "type": "text",
130                        "text": normalize_message(message),
131                    },
132                ],
133            },
134        ],
135    }
136
137
138def empty_section_block() -> dict:
139    """Create an empty section block"""
140    return {
141        "type": "section",
142        "text": {
143            "type": "mrkdwn",
144            "text": normalize_message("\t"),
145        },
146    }
147
148
149def context_block(*messages: str) -> dict:
150    """Create a context block with multiple fields"""
151    return {
152        "type": "context",
153        "elements": [
154            {
155                "type": "mrkdwn",
156                "text": normalize_message(message),
157            }
158            for message in messages
159        ],
160    }
161
162
163def header_block(message: str) -> dict:
164    """Create a header block"""
165    return {
166        "type": "header",
167        "text": {
168            "type": "plain_text",
169            "text": message,
170        },
171    }
172
173
174def button_action_block(text: str, url: str) -> dict:
175    """Create a button action block"""
176    return {
177        "type": "actions",
178        "elements": [
179            {
180                "type": "button",
181                "text": {"type": "plain_text", "text": text, "emoji": True},
182                "value": text,
183                "url": url,
184            }
185        ],
186    }
187
188
189def compacted_sections_blocks(*messages: str) -> t.List[dict]:
190    """Create a list of compacted sections blocks"""
191    return [
192        {
193            "type": "section",
194            "fields": [
195                {
196                    "type": "mrkdwn",
197                    "text": normalize_message(message),
198                }
199                for message in messages[i : i + 2]
200            ],
201        }
202        for i in range(0, len(messages), 2)
203    ]
204
205
206class SlackAlertIcon(str, Enum):
207    """Enum for status of the alert"""
208
209    # General statuses
210    X = ":x:"
211    OK = ":large_green_circle:"
212    START = ":arrow_forward:"
213    STOP = ":stop_button:"
214    WARN = ":warning:"
215    ALERT = ":rotating_light:"
216
217    # Execution statuses
218    FAILURE = ":rotating_light:"
219    SUCCESS = ":white_check_mark:"
220    WARNING = ":warning:"
221    SKIPPED = ":fast_forward:"
222    PASSED = ":white_check_mark:"
223
224    # Log levels
225    UNKNOWN = ":question:"
226    INFO = ":information_source:"
227    DEBUG = ":beetle:"
228    CRITICAL = ":fire:"
229    FATAL = ":skull_and_crossbones:"
230    EXCEPTION = ":boom:"
231
232    def __str__(self) -> str:
233        return self.value
234
235
236def stringify_list(list_variation: t.Union[t.List[str], str]) -> str:
237    """Prettify and deduplicate list of strings"""
238    if isinstance(list_variation, str):
239        return list_variation
240    if len(list_variation) == 1:
241        return list_variation[0]
242    return " ".join(list_variation)
243
244
245message = SlackMessageComposer