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