Skip to content

Slack Dispatcher

nautobot_chatops.dispatchers.slack

Dispatcher implementation for sending content to Slack.

SlackDispatcher

Bases: Dispatcher

Dispatch messages and cards to Slack.

Source code in nautobot_chatops/dispatchers/slack.py
 32
 33
 34
 35
 36
 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
 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
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
class SlackDispatcher(Dispatcher):
    """Dispatch messages and cards to Slack."""

    # pylint: disable=too-many-public-methods
    platform_name = "Slack"
    platform_slug = "slack"

    platform_color = "4A154B"  # Slack Aubergine

    command_prefix = settings.PLUGINS_CONFIG["nautobot_chatops"]["slack_slash_command_prefix"]
    """Prefix prepended to all commands, such as "/" or "!" in some clients."""

    def __init__(self, *args, **kwargs):
        """Init a SlackDispatcher."""
        super().__init__(*args, **kwargs)
        self.slack_client = WebClient(token=settings.PLUGINS_CONFIG["nautobot_chatops"]["slack_api_token"])
        self.slack_menu_limit = int(os.getenv("SLACK_MENU_LIMIT", "100"))

    # pylint: disable=too-many-branches
    @classmethod
    @BACKEND_ACTION_LOOKUP.time()
    def platform_lookup(cls, item_type, item_name) -> Optional[str]:
        """Call out to the chat platform to look up, e.g., a specific user ID by name.

        Args:
          item_type (str): One of "organization", "channel", "user"
          item_name (str): Uniquely identifying name of the given item.

        Returns:
          (str, None)
        """
        instance = cls(context=None)
        cursor = None
        if item_type == "organization":  # pylint: disable=no-else-raise
            # The admin_teams_list API requires admin access and only works under Enterprise
            raise NotImplementedError
        elif item_type == "channel":
            while True:
                try:
                    response = instance.slack_client.conversations_list(cursor=cursor, limit=20, exclude_archived=True)
                except SlackApiError as err:
                    if err.response["error"] == "ratelimited":
                        delay = int(err.response.headers["Retry-After"])
                        time.sleep(delay)
                        continue
                    raise err
                for channel in response["channels"]:
                    if channel["name"] == item_name:
                        return channel["id"]
                cursor = response["response_metadata"]["next_cursor"]
                if not cursor:
                    break

        elif item_type == "user":
            while True:
                try:
                    response = instance.slack_client.users_list(cursor=cursor, limit=20)
                except SlackApiError as err:
                    if err.response["error"] == "ratelimited":
                        delay = int(err.response.headers["Retry-After"])
                        time.sleep(delay)
                        continue
                    raise err
                for member in response["members"]:
                    if member["name"] == item_name:
                        return member["id"]
                cursor = response["response_metadata"]["next_cursor"]
                if not cursor:
                    break

        return None

    @classmethod
    def lookup_user_id_by_email(cls, email) -> Optional[str]:
        """Call out to Slack to look up a specific user ID by email.

        Args:
          email (str): Uniquely identifying email address of the user.

        Returns:
          (str, None)
        """
        instance = cls(context=None)
        try:
            response = instance.slack_client.users_lookupByEmail(email=email)
            return response["user"]["id"]
        except SlackApiError as err:
            if err.response["error"] == "users_not_found":
                return None
            raise err

    # More complex APIs for presenting structured data - these typically build on the more basic functions below

    def command_response_header(self, command, subcommand, args, description="information", image_element=None):
        """Construct a consistently forwarded header including the command that was issued.

        Args:
          command (str): Primary command string
          subcommand (str): Secondary command string
          args (list): of tuples, either (arg_name, human_readable_value, literal_value) or (arg_name, literal_value)
          description (str): Short description of what information is contained in the response
          image_element (dict): As constructed by self.image_element()
        """
        fields = []
        for name, value, *_ in args:
            fields.append(self.markdown_element(self.bold(name)))
            fields.append(self.markdown_element(value))

        command = f"{self.command_prefix}{command}"

        block = {
            "type": "section",
            "text": self.markdown_element(
                f"Hey {self.user_mention()}, here is that {description} you requested\n"
                f"Shortcut: `{command} {subcommand} {' '.join(arg[-1] for arg in args)}`"
            ),
        }

        # Add to block "accessory" key if image_element exists. Otherwise do not
        if image_element:
            block["accessory"] = image_element

        # Slack doesn't like it if we send an empty fields list, we have to omit it entirely
        if fields:
            block["fields"] = fields

        return [block]

    # Send various content to the user or channel

    @BACKEND_ACTION_MARKDOWN.time()
    def send_markdown(self, message, ephemeral=None):
        """Send a Markdown-formatted text message to the user/channel specified by the context."""
        try:
            if ephemeral is None:
                ephemeral = settings.PLUGINS_CONFIG["nautobot_chatops"]["send_all_messages_private"]

            if ephemeral:
                self.slack_client.chat_postEphemeral(
                    channel=self.context.get("channel_id"),
                    user=self.context.get("user_id"),
                    text=message,
                    thread_ts=self.context.get("thread_ts"),
                )
            else:
                self.slack_client.chat_postMessage(
                    channel=self.context.get("channel_id"),
                    user=self.context.get("user_id"),
                    text=message,
                    thread_ts=self.context.get("thread_ts"),
                )
        except SlackClientError as slack_error:
            self.send_exception(slack_error)

    # pylint: disable=arguments-differ
    @BACKEND_ACTION_BLOCKS.time()
    def send_blocks(self, blocks, callback_id=None, modal=False, ephemeral=None, title="Your attention please!"):
        """Send a series of formatting blocks to the user/channel specified by the context.

        Slack distinguishes between simple inline interactive elements and modal dialogs. Modals can contain multiple
        inputs in a single dialog; more importantly for our purposes, certain inputs (such as textentry) can ONLY
        be used in modals and will be rejected if we try to use them inline.

        Args:
          blocks (list): List of block contents as constructed by other dispatcher functions.
          callback_id (str): Callback ID string such as "command subcommand arg1 arg2". Required if `modal` is True.
          modal (bool): Whether to send this as a modal dialog rather than an inline block.
          ephemeral (bool): Whether to send this as an ephemeral message (only visible to the targeted user).
          title (str): Title to include on a modal dialog.
        """
        logger.info("Sending blocks: %s", json.dumps(blocks, indent=2))
        if ephemeral is None:
            ephemeral = settings.PLUGINS_CONFIG["nautobot_chatops"]["send_all_messages_private"]
        try:
            if modal:
                if not callback_id:
                    self.send_error("Tried to create a modal dialog without specifying a callback_id")
                    return
                self.slack_client.views_open(
                    trigger_id=self.context.get("trigger_id"),
                    view={
                        "type": "modal",
                        "title": self.text_element(title),
                        "submit": self.text_element("Submit"),
                        "blocks": blocks,
                        # Embed the current channel information into to the modal as modals don't store this otherwise
                        "private_metadata": json.dumps(
                            {"channel_id": self.context.get("channel_id"), "thread_ts": self.context.get("thread_ts")}
                        ),
                        "callback_id": callback_id,
                    },
                )
            elif ephemeral:
                self.slack_client.chat_postEphemeral(
                    channel=self.context.get("channel_id"),
                    user=self.context.get("user_id"),
                    blocks=blocks,
                    thread_ts=self.context.get("thread_ts"),
                )
            else:
                self.slack_client.chat_postMessage(
                    channel=self.context.get("channel_id"),
                    user=self.context.get("user_id"),
                    blocks=blocks,
                    thread_ts=self.context.get("thread_ts"),
                )
        except SlackClientError as slack_error:
            self.send_exception(slack_error)

    @BACKEND_ACTION_SNIPPET.time()
    def send_snippet(self, text, title=None, ephemeral=None):
        """Send a longer chunk of text as a file snippet."""
        if ephemeral is None:
            ephemeral = settings.PLUGINS_CONFIG["nautobot_chatops"]["send_all_messages_private"]

        if self.context.get("channel_name") == "directmessage":
            channels = [self.context.get("user_id")]
        else:
            channels = [self.context.get("channel_id")]
        channels = ",".join(channels)
        logger.info("Sending snippet to %s: %s", channels, text)
        try:
            # Check for the length of the file if the setup is meant to be a private message
            if ephemeral:
                message_list = self.split_message(text, SLACK_PRIVATE_MESSAGE_LIMIT)
                for msg in message_list:
                    # Send the blocks as a list, this needs to be the case for Slack to send appropriately.
                    self.send_blocks(
                        [self.markdown_block(f"```\n{msg}\n```")],
                        ephemeral=ephemeral,
                    )
            else:
                self.slack_client.files_upload(
                    channels=channels, content=text, title=title, thread_ts=self.context.get("thread_ts")
                )
        except SlackClientError as slack_error:
            self.send_exception(slack_error)

    def static_url(self, path):
        """Return a url to access the static file."""
        static_path = str(static(path))
        if static_path.startswith("http"):
            # Static is being provided by Django Storages
            return static_path
        try:
            return f"{self.context['request_scheme']}://{self.context['request_host']}{static_path}"
        except KeyError:
            if SLACK_SOCKET_STATIC_HOST:
                # Use the settings provided Static Host.
                return f"{SLACK_SOCKET_STATIC_HOST}{static_path}"
            return None

    def send_image(self, image_path):
        """Send an image as a file upload."""
        if self.context.get("channel_name") == "directmessage":
            channels = [self.context.get("user_id")]
        else:
            channels = [self.context.get("channel_id")]
        channels = ",".join(channels)
        logger.info("Sending image %s to %s", image_path, channels)
        self.slack_client.files_upload(channels=channels, file=image_path, thread_ts=self.context.get("thread_ts"))

    def send_warning(self, message):
        """Send a warning message to the user/channel specified by the context."""
        self.send_markdown(f":warning: {self.bold(message)} :warning:", ephemeral=True)

    def send_error(self, message):
        """Send an error message to the user/channel specified by the context."""
        self.send_markdown(f":warning: {self.bold(message)} :warning:", ephemeral=True)

    # pylint: disable=unnecessary-pass
    def send_busy_indicator(self):
        """Send a "typing" indicator to show that work is in progress."""
        # Currently the Slack Events API does not support the "user_typing" event.
        # We're trying not to use the legacy Slack RTM API as it's deprecated.
        # So for now, we do nothing.
        pass

    def send_exception(self, exception):
        """Try to report an exception to the user."""
        self.slack_client.chat_postEphemeral(
            channel=self.context.get("channel_id"),
            user=self.context.get("user_id"),
            text=f"Sorry @{self.context.get('user_name')}, an error occurred :sob:\n```{exception}```",
        )

    # pylint: disable=no-self-use
    def delete_message(self, response_url):
        """Delete a message that was previously sent."""
        WebhookClient(response_url).send_dict({"delete_original": "true"})

    # Prompt the user for various basic inputs

    def prompt_for_text(self, action_id, help_text, label, title="Your attention please!"):
        """Prompt the user to enter freeform text into a field.

        Args:
          action_id (str): Identifier string to attach to the "submit" action.
          help_text (str): Markdown string to display as help text.
          label (str): Label text to display adjacent to the text field.
          title (str): Title to include on the modal dialog.
        """
        textentry = {
            "type": "input",
            "block_id": action_id,
            "label": self.text_element(label),
            "element": {"type": "plain_text_input", "action_id": action_id, "placeholder": self.text_element(label)},
        }
        blocks = [self.markdown_block(help_text), textentry]
        # In Slack, a textentry element can ONLY be sent in a modal dialog
        return self.send_blocks(blocks, callback_id=action_id, ephemeral=True, modal=True, title=title)

    def get_prompt_from_menu_choices(self, choices, offset=0):
        """Returns choices list to accommodate for Slack menu limits.

        Args:
          choices (list): List of (display, value) tuples
          offset (int): If set, starts displaying choices at index location from all choices,
                         as Slack only displays 100 options at a time

        Returns:
          choices (list): List of (display, value) tuples accommodating for Slack menu limits
        """
        choice_length = len(choices)
        if choice_length > self.slack_menu_limit:
            try:
                # Since we are showing "Next..." at the end, this isn't required to show to users anymore
                self.send_warning(
                    f"More than {self.slack_menu_limit} options are available."
                    f"Slack limits us to only displaying {self.slack_menu_limit} options at a time."
                )
            except SlackApiError:
                pass
            new_offset = offset + self.slack_menu_limit - 1
            if offset == 0:
                choices = choices[: self.slack_menu_limit - 1]  # 1 is to leave space for 'next' insert
            else:
                choices = choices[offset:new_offset]
            if choice_length > new_offset:
                # Only insert a 'next' offset if we still have more choices left to see
                choices.append(("Next...", f"menu_offset-{new_offset}"))
        return choices

    def prompt_from_menu(
        self, action_id, help_text, choices, default=(None, None), confirm=False, offset=0
    ):  # pylint: disable=too-many-arguments
        """Prompt the user for a selection from a menu.

        Args:
          action_id (str): Identifier string to attach to the "submit" action.
          help_text (str): Markdown string to display as help text.
          choices (list): List of (display, value) tuples.
          default (tuple): Default (display, value) to pre-select.
          confirm (bool): If True, prompt the user to confirm their selection (if the platform supports this).
          offset (int): If set, starts displaying choices at index location from all choices,
                         as Slack only displays 100 options at a time.
        """
        choices = self.get_prompt_from_menu_choices(choices, offset=offset)
        menu = self.select_element(action_id, choices, default=default, confirm=confirm)
        cancel_button = {
            "type": "button",
            "text": self.text_element("Cancel"),
            "action_id": "action",
            "value": "cancel",
        }

        blocks = [self.markdown_block(help_text), self.actions_block(action_id, [menu, cancel_button])]
        return self.send_blocks(blocks, ephemeral=True)

    # Construct content piecemeal, mostly for use with send_blocks()

    def multi_input_dialog(self, command, sub_command, dialog_title, dialog_list):
        """Provide several input fields on a single dialog.

        Args:
            command (str):  The top level command in use. (ex. net)
            sub_command (str): The command being invoked (ex. add-vlan)
            dialog_title (str): Title of the dialog box
            dialog_list (list):  List of dictionaries containing the dialog parameters. See Example below.

        Example:
            For a selection menu::

                {
                    type: "select",
                    label: "label",
                    choices: [("display 1", "value1"), ("display 2", "value 2")],
                    default: ("display 1", "value1"),
                    confirm: False
                }

            For a text dialog::

                {
                    type: "text",
                    label: "text displayed next to field"
                    default: "default-value",
                    optional: True
                }

            Dictionary Fields

            - type: The type of object to create. Currently supported values: select, text
            - label: A text descriptor that will be placed next to the field
            - choices: A list of tuples which populates the choices in a dropdown selector
            - default: (optional) Default choice of a select menu or initial value to put in a text field.
            - confirm: (optional) If set to True, it will display a "Are you sure?" dialog upon submit.
            - optional: (optional) If set to True, the field will return NoneType is not specified.
        """
        blocks = []
        callback_id = f"{command} {sub_command}"
        for i, dialog in enumerate(dialog_list):
            action_id = f"param_{i}"
            if dialog["type"] == "select":
                menu = self.select_element(
                    action_id,
                    dialog["choices"],
                    default=dialog.get("default", (None, None)),
                    confirm=dialog.get("confirm", False),
                )
                blocks.append(self._input_block(action_id, dialog["label"], menu, dialog.get("optional", False)))
            if dialog["type"] == "text":
                textentry = {
                    "type": "plain_text_input",
                    "action_id": action_id,
                    "placeholder": self.text_element(dialog["label"]),
                    "initial_value": dialog.get("default", "") or "",
                }

                blocks.append(self._input_block(action_id, dialog["label"], textentry, dialog.get("optional", False)))

        return self.send_blocks(blocks, callback_id=callback_id, modal=True, ephemeral=False, title=dialog_title)

    def user_mention(self):
        """Markup for a mention of the username/userid specified in our context."""
        return f"<@{self.context['user_id']}>"

    def bold(self, text):
        """Mark text as bold."""
        return f"*{text}*"

    def hyperlink(self, text, url):
        """Create Hyperlinks."""
        return f"<{url}|{text}>"

    def actions_block(self, block_id, actions):
        """Construct a block consisting of a set of action elements."""
        return {"type": "actions", "block_id": block_id, "elements": actions}

    def _input_block(self, block_id, label, element, optional):
        """Construct a block consisting of Input elements."""
        text_obj = self.text_element(label)
        return {"type": "input", "block_id": block_id, "label": text_obj, "element": element, "optional": optional}

    def markdown_block(self, text):
        """Construct a simple Markdown-formatted text block."""
        return {"type": "section", "text": self.markdown_element(text)}

    def image_element(self, url, alt_text=""):
        """Construct an image element that can be embedded in a block if URL is provided."""
        return {"type": "image", "image_url": url, "alt_text": alt_text} if url else {}

    def markdown_element(self, text):
        """Construct a basic Markdown-formatted text element."""
        return {"type": "mrkdwn", "text": text}

    def select_element(self, action_id, choices, default=(None, None), confirm=False):
        """Construct a basic selection menu with the given choices.

        .. note:: Slack's select menu supports option groups, but others such as Adaptive Cards do not;
                  we have to stick to the lowest common denominator, which means no groups.

        TODO: maybe refactor this API so that Slack can use groups and other dispatchers just ignore the groups?

        Args:
          action_id (str): Identifying string to associate with this element
          choices (list): List of (display, value) tuples
          default (tuple: Default (display, value) to preselect
          confirm (bool): If true (and the platform supports it), prompt the user to confirm their selection
        """
        data = {
            "type": "static_select",
            "action_id": action_id,
            "placeholder": self.text_element("Select an option"),
            "options": [{"text": self.text_element(choice), "value": value} for choice, value in choices],
        }
        default_text, default_value = default
        if default_text and default_value:
            data["initial_option"] = {"text": self.text_element(default_text), "value": default_value}
        if confirm:
            data["confirm"] = {
                "title": self.text_element("Are you sure?"),
                "text": self.markdown_element("Are you *really* sure you want to do this?"),
                "confirm": self.text_element("Yes"),
                "deny": self.text_element("No"),
            }

        return data

    def text_element(self, text):
        """Construct a basic plaintext element."""
        return {"type": "plain_text", "text": text}

command_prefix = settings.PLUGINS_CONFIG['nautobot_chatops']['slack_slash_command_prefix'] instance-attribute class-attribute

Prefix prepended to all commands, such as "/" or "!" in some clients.

__init__(*args, **kwargs)

Init a SlackDispatcher.

Source code in nautobot_chatops/dispatchers/slack.py
def __init__(self, *args, **kwargs):
    """Init a SlackDispatcher."""
    super().__init__(*args, **kwargs)
    self.slack_client = WebClient(token=settings.PLUGINS_CONFIG["nautobot_chatops"]["slack_api_token"])
    self.slack_menu_limit = int(os.getenv("SLACK_MENU_LIMIT", "100"))

actions_block(block_id, actions)

Construct a block consisting of a set of action elements.

Source code in nautobot_chatops/dispatchers/slack.py
def actions_block(self, block_id, actions):
    """Construct a block consisting of a set of action elements."""
    return {"type": "actions", "block_id": block_id, "elements": actions}

bold(text)

Mark text as bold.

Source code in nautobot_chatops/dispatchers/slack.py
def bold(self, text):
    """Mark text as bold."""
    return f"*{text}*"

command_response_header(command, subcommand, args, description='information', image_element=None)

Construct a consistently forwarded header including the command that was issued.

Parameters:

Name Type Description Default
command str

Primary command string

required
subcommand str

Secondary command string

required
args list

of tuples, either (arg_name, human_readable_value, literal_value) or (arg_name, literal_value)

required
description str

Short description of what information is contained in the response

'information'
image_element dict

As constructed by self.image_element()

None
Source code in nautobot_chatops/dispatchers/slack.py
def command_response_header(self, command, subcommand, args, description="information", image_element=None):
    """Construct a consistently forwarded header including the command that was issued.

    Args:
      command (str): Primary command string
      subcommand (str): Secondary command string
      args (list): of tuples, either (arg_name, human_readable_value, literal_value) or (arg_name, literal_value)
      description (str): Short description of what information is contained in the response
      image_element (dict): As constructed by self.image_element()
    """
    fields = []
    for name, value, *_ in args:
        fields.append(self.markdown_element(self.bold(name)))
        fields.append(self.markdown_element(value))

    command = f"{self.command_prefix}{command}"

    block = {
        "type": "section",
        "text": self.markdown_element(
            f"Hey {self.user_mention()}, here is that {description} you requested\n"
            f"Shortcut: `{command} {subcommand} {' '.join(arg[-1] for arg in args)}`"
        ),
    }

    # Add to block "accessory" key if image_element exists. Otherwise do not
    if image_element:
        block["accessory"] = image_element

    # Slack doesn't like it if we send an empty fields list, we have to omit it entirely
    if fields:
        block["fields"] = fields

    return [block]

delete_message(response_url)

Delete a message that was previously sent.

Source code in nautobot_chatops/dispatchers/slack.py
def delete_message(self, response_url):
    """Delete a message that was previously sent."""
    WebhookClient(response_url).send_dict({"delete_original": "true"})

get_prompt_from_menu_choices(choices, offset=0)

Returns choices list to accommodate for Slack menu limits.

Parameters:

Name Type Description Default
choices list

List of (display, value) tuples

required
offset int

If set, starts displaying choices at index location from all choices, as Slack only displays 100 options at a time

0

Returns:

Name Type Description
choices list

List of (display, value) tuples accommodating for Slack menu limits

Source code in nautobot_chatops/dispatchers/slack.py
def get_prompt_from_menu_choices(self, choices, offset=0):
    """Returns choices list to accommodate for Slack menu limits.

    Args:
      choices (list): List of (display, value) tuples
      offset (int): If set, starts displaying choices at index location from all choices,
                     as Slack only displays 100 options at a time

    Returns:
      choices (list): List of (display, value) tuples accommodating for Slack menu limits
    """
    choice_length = len(choices)
    if choice_length > self.slack_menu_limit:
        try:
            # Since we are showing "Next..." at the end, this isn't required to show to users anymore
            self.send_warning(
                f"More than {self.slack_menu_limit} options are available."
                f"Slack limits us to only displaying {self.slack_menu_limit} options at a time."
            )
        except SlackApiError:
            pass
        new_offset = offset + self.slack_menu_limit - 1
        if offset == 0:
            choices = choices[: self.slack_menu_limit - 1]  # 1 is to leave space for 'next' insert
        else:
            choices = choices[offset:new_offset]
        if choice_length > new_offset:
            # Only insert a 'next' offset if we still have more choices left to see
            choices.append(("Next...", f"menu_offset-{new_offset}"))
    return choices

Create Hyperlinks.

Source code in nautobot_chatops/dispatchers/slack.py
def hyperlink(self, text, url):
    """Create Hyperlinks."""
    return f"<{url}|{text}>"

image_element(url, alt_text='')

Construct an image element that can be embedded in a block if URL is provided.

Source code in nautobot_chatops/dispatchers/slack.py
def image_element(self, url, alt_text=""):
    """Construct an image element that can be embedded in a block if URL is provided."""
    return {"type": "image", "image_url": url, "alt_text": alt_text} if url else {}

lookup_user_id_by_email(email) classmethod

Call out to Slack to look up a specific user ID by email.

Parameters:

Name Type Description Default
email str

Uniquely identifying email address of the user.

required

Returns:

Type Description
Optional[str]

(str, None)

Source code in nautobot_chatops/dispatchers/slack.py
@classmethod
def lookup_user_id_by_email(cls, email) -> Optional[str]:
    """Call out to Slack to look up a specific user ID by email.

    Args:
      email (str): Uniquely identifying email address of the user.

    Returns:
      (str, None)
    """
    instance = cls(context=None)
    try:
        response = instance.slack_client.users_lookupByEmail(email=email)
        return response["user"]["id"]
    except SlackApiError as err:
        if err.response["error"] == "users_not_found":
            return None
        raise err

markdown_block(text)

Construct a simple Markdown-formatted text block.

Source code in nautobot_chatops/dispatchers/slack.py
def markdown_block(self, text):
    """Construct a simple Markdown-formatted text block."""
    return {"type": "section", "text": self.markdown_element(text)}

markdown_element(text)

Construct a basic Markdown-formatted text element.

Source code in nautobot_chatops/dispatchers/slack.py
def markdown_element(self, text):
    """Construct a basic Markdown-formatted text element."""
    return {"type": "mrkdwn", "text": text}

multi_input_dialog(command, sub_command, dialog_title, dialog_list)

Provide several input fields on a single dialog.

Parameters:

Name Type Description Default
command str

The top level command in use. (ex. net)

required
sub_command str

The command being invoked (ex. add-vlan)

required
dialog_title str

Title of the dialog box

required
dialog_list list

List of dictionaries containing the dialog parameters. See Example below.

required
Example

For a selection menu::

{
    type: "select",
    label: "label",
    choices: [("display 1", "value1"), ("display 2", "value 2")],
    default: ("display 1", "value1"),
    confirm: False
}

For a text dialog::

{
    type: "text",
    label: "text displayed next to field"
    default: "default-value",
    optional: True
}

Dictionary Fields

  • type: The type of object to create. Currently supported values: select, text
  • label: A text descriptor that will be placed next to the field
  • choices: A list of tuples which populates the choices in a dropdown selector
  • default: (optional) Default choice of a select menu or initial value to put in a text field.
  • confirm: (optional) If set to True, it will display a "Are you sure?" dialog upon submit.
  • optional: (optional) If set to True, the field will return NoneType is not specified.
Source code in nautobot_chatops/dispatchers/slack.py
def multi_input_dialog(self, command, sub_command, dialog_title, dialog_list):
    """Provide several input fields on a single dialog.

    Args:
        command (str):  The top level command in use. (ex. net)
        sub_command (str): The command being invoked (ex. add-vlan)
        dialog_title (str): Title of the dialog box
        dialog_list (list):  List of dictionaries containing the dialog parameters. See Example below.

    Example:
        For a selection menu::

            {
                type: "select",
                label: "label",
                choices: [("display 1", "value1"), ("display 2", "value 2")],
                default: ("display 1", "value1"),
                confirm: False
            }

        For a text dialog::

            {
                type: "text",
                label: "text displayed next to field"
                default: "default-value",
                optional: True
            }

        Dictionary Fields

        - type: The type of object to create. Currently supported values: select, text
        - label: A text descriptor that will be placed next to the field
        - choices: A list of tuples which populates the choices in a dropdown selector
        - default: (optional) Default choice of a select menu or initial value to put in a text field.
        - confirm: (optional) If set to True, it will display a "Are you sure?" dialog upon submit.
        - optional: (optional) If set to True, the field will return NoneType is not specified.
    """
    blocks = []
    callback_id = f"{command} {sub_command}"
    for i, dialog in enumerate(dialog_list):
        action_id = f"param_{i}"
        if dialog["type"] == "select":
            menu = self.select_element(
                action_id,
                dialog["choices"],
                default=dialog.get("default", (None, None)),
                confirm=dialog.get("confirm", False),
            )
            blocks.append(self._input_block(action_id, dialog["label"], menu, dialog.get("optional", False)))
        if dialog["type"] == "text":
            textentry = {
                "type": "plain_text_input",
                "action_id": action_id,
                "placeholder": self.text_element(dialog["label"]),
                "initial_value": dialog.get("default", "") or "",
            }

            blocks.append(self._input_block(action_id, dialog["label"], textentry, dialog.get("optional", False)))

    return self.send_blocks(blocks, callback_id=callback_id, modal=True, ephemeral=False, title=dialog_title)

platform_lookup(item_type, item_name) classmethod

Call out to the chat platform to look up, e.g., a specific user ID by name.

Parameters:

Name Type Description Default
item_type str

One of "organization", "channel", "user"

required
item_name str

Uniquely identifying name of the given item.

required

Returns:

Type Description
Optional[str]

(str, None)

Source code in nautobot_chatops/dispatchers/slack.py
@classmethod
@BACKEND_ACTION_LOOKUP.time()
def platform_lookup(cls, item_type, item_name) -> Optional[str]:
    """Call out to the chat platform to look up, e.g., a specific user ID by name.

    Args:
      item_type (str): One of "organization", "channel", "user"
      item_name (str): Uniquely identifying name of the given item.

    Returns:
      (str, None)
    """
    instance = cls(context=None)
    cursor = None
    if item_type == "organization":  # pylint: disable=no-else-raise
        # The admin_teams_list API requires admin access and only works under Enterprise
        raise NotImplementedError
    elif item_type == "channel":
        while True:
            try:
                response = instance.slack_client.conversations_list(cursor=cursor, limit=20, exclude_archived=True)
            except SlackApiError as err:
                if err.response["error"] == "ratelimited":
                    delay = int(err.response.headers["Retry-After"])
                    time.sleep(delay)
                    continue
                raise err
            for channel in response["channels"]:
                if channel["name"] == item_name:
                    return channel["id"]
            cursor = response["response_metadata"]["next_cursor"]
            if not cursor:
                break

    elif item_type == "user":
        while True:
            try:
                response = instance.slack_client.users_list(cursor=cursor, limit=20)
            except SlackApiError as err:
                if err.response["error"] == "ratelimited":
                    delay = int(err.response.headers["Retry-After"])
                    time.sleep(delay)
                    continue
                raise err
            for member in response["members"]:
                if member["name"] == item_name:
                    return member["id"]
            cursor = response["response_metadata"]["next_cursor"]
            if not cursor:
                break

    return None

prompt_for_text(action_id, help_text, label, title='Your attention please!')

Prompt the user to enter freeform text into a field.

Parameters:

Name Type Description Default
action_id str

Identifier string to attach to the "submit" action.

required
help_text str

Markdown string to display as help text.

required
label str

Label text to display adjacent to the text field.

required
title str

Title to include on the modal dialog.

'Your attention please!'
Source code in nautobot_chatops/dispatchers/slack.py
def prompt_for_text(self, action_id, help_text, label, title="Your attention please!"):
    """Prompt the user to enter freeform text into a field.

    Args:
      action_id (str): Identifier string to attach to the "submit" action.
      help_text (str): Markdown string to display as help text.
      label (str): Label text to display adjacent to the text field.
      title (str): Title to include on the modal dialog.
    """
    textentry = {
        "type": "input",
        "block_id": action_id,
        "label": self.text_element(label),
        "element": {"type": "plain_text_input", "action_id": action_id, "placeholder": self.text_element(label)},
    }
    blocks = [self.markdown_block(help_text), textentry]
    # In Slack, a textentry element can ONLY be sent in a modal dialog
    return self.send_blocks(blocks, callback_id=action_id, ephemeral=True, modal=True, title=title)

prompt_from_menu(action_id, help_text, choices, default=(None, None), confirm=False, offset=0)

Prompt the user for a selection from a menu.

Parameters:

Name Type Description Default
action_id str

Identifier string to attach to the "submit" action.

required
help_text str

Markdown string to display as help text.

required
choices list

List of (display, value) tuples.

required
default tuple

Default (display, value) to pre-select.

(None, None)
confirm bool

If True, prompt the user to confirm their selection (if the platform supports this).

False
offset int

If set, starts displaying choices at index location from all choices, as Slack only displays 100 options at a time.

0
Source code in nautobot_chatops/dispatchers/slack.py
def prompt_from_menu(
    self, action_id, help_text, choices, default=(None, None), confirm=False, offset=0
):  # pylint: disable=too-many-arguments
    """Prompt the user for a selection from a menu.

    Args:
      action_id (str): Identifier string to attach to the "submit" action.
      help_text (str): Markdown string to display as help text.
      choices (list): List of (display, value) tuples.
      default (tuple): Default (display, value) to pre-select.
      confirm (bool): If True, prompt the user to confirm their selection (if the platform supports this).
      offset (int): If set, starts displaying choices at index location from all choices,
                     as Slack only displays 100 options at a time.
    """
    choices = self.get_prompt_from_menu_choices(choices, offset=offset)
    menu = self.select_element(action_id, choices, default=default, confirm=confirm)
    cancel_button = {
        "type": "button",
        "text": self.text_element("Cancel"),
        "action_id": "action",
        "value": "cancel",
    }

    blocks = [self.markdown_block(help_text), self.actions_block(action_id, [menu, cancel_button])]
    return self.send_blocks(blocks, ephemeral=True)

select_element(action_id, choices, default=(None, None), confirm=False)

Construct a basic selection menu with the given choices.

.. note:: Slack's select menu supports option groups, but others such as Adaptive Cards do not; we have to stick to the lowest common denominator, which means no groups.

TODO: maybe refactor this API so that Slack can use groups and other dispatchers just ignore the groups?

Parameters:

Name Type Description Default
action_id str

Identifying string to associate with this element

required
choices list

List of (display, value) tuples

required
default tuple

Default (display, value) to preselect

(None, None)
confirm bool

If true (and the platform supports it), prompt the user to confirm their selection

False
Source code in nautobot_chatops/dispatchers/slack.py
def select_element(self, action_id, choices, default=(None, None), confirm=False):
    """Construct a basic selection menu with the given choices.

    .. note:: Slack's select menu supports option groups, but others such as Adaptive Cards do not;
              we have to stick to the lowest common denominator, which means no groups.

    TODO: maybe refactor this API so that Slack can use groups and other dispatchers just ignore the groups?

    Args:
      action_id (str): Identifying string to associate with this element
      choices (list): List of (display, value) tuples
      default (tuple: Default (display, value) to preselect
      confirm (bool): If true (and the platform supports it), prompt the user to confirm their selection
    """
    data = {
        "type": "static_select",
        "action_id": action_id,
        "placeholder": self.text_element("Select an option"),
        "options": [{"text": self.text_element(choice), "value": value} for choice, value in choices],
    }
    default_text, default_value = default
    if default_text and default_value:
        data["initial_option"] = {"text": self.text_element(default_text), "value": default_value}
    if confirm:
        data["confirm"] = {
            "title": self.text_element("Are you sure?"),
            "text": self.markdown_element("Are you *really* sure you want to do this?"),
            "confirm": self.text_element("Yes"),
            "deny": self.text_element("No"),
        }

    return data

send_blocks(blocks, callback_id=None, modal=False, ephemeral=None, title='Your attention please!')

Send a series of formatting blocks to the user/channel specified by the context.

Slack distinguishes between simple inline interactive elements and modal dialogs. Modals can contain multiple inputs in a single dialog; more importantly for our purposes, certain inputs (such as textentry) can ONLY be used in modals and will be rejected if we try to use them inline.

Parameters:

Name Type Description Default
blocks list

List of block contents as constructed by other dispatcher functions.

required
callback_id str

Callback ID string such as "command subcommand arg1 arg2". Required if modal is True.

None
modal bool

Whether to send this as a modal dialog rather than an inline block.

False
ephemeral bool

Whether to send this as an ephemeral message (only visible to the targeted user).

None
title str

Title to include on a modal dialog.

'Your attention please!'
Source code in nautobot_chatops/dispatchers/slack.py
@BACKEND_ACTION_BLOCKS.time()
def send_blocks(self, blocks, callback_id=None, modal=False, ephemeral=None, title="Your attention please!"):
    """Send a series of formatting blocks to the user/channel specified by the context.

    Slack distinguishes between simple inline interactive elements and modal dialogs. Modals can contain multiple
    inputs in a single dialog; more importantly for our purposes, certain inputs (such as textentry) can ONLY
    be used in modals and will be rejected if we try to use them inline.

    Args:
      blocks (list): List of block contents as constructed by other dispatcher functions.
      callback_id (str): Callback ID string such as "command subcommand arg1 arg2". Required if `modal` is True.
      modal (bool): Whether to send this as a modal dialog rather than an inline block.
      ephemeral (bool): Whether to send this as an ephemeral message (only visible to the targeted user).
      title (str): Title to include on a modal dialog.
    """
    logger.info("Sending blocks: %s", json.dumps(blocks, indent=2))
    if ephemeral is None:
        ephemeral = settings.PLUGINS_CONFIG["nautobot_chatops"]["send_all_messages_private"]
    try:
        if modal:
            if not callback_id:
                self.send_error("Tried to create a modal dialog without specifying a callback_id")
                return
            self.slack_client.views_open(
                trigger_id=self.context.get("trigger_id"),
                view={
                    "type": "modal",
                    "title": self.text_element(title),
                    "submit": self.text_element("Submit"),
                    "blocks": blocks,
                    # Embed the current channel information into to the modal as modals don't store this otherwise
                    "private_metadata": json.dumps(
                        {"channel_id": self.context.get("channel_id"), "thread_ts": self.context.get("thread_ts")}
                    ),
                    "callback_id": callback_id,
                },
            )
        elif ephemeral:
            self.slack_client.chat_postEphemeral(
                channel=self.context.get("channel_id"),
                user=self.context.get("user_id"),
                blocks=blocks,
                thread_ts=self.context.get("thread_ts"),
            )
        else:
            self.slack_client.chat_postMessage(
                channel=self.context.get("channel_id"),
                user=self.context.get("user_id"),
                blocks=blocks,
                thread_ts=self.context.get("thread_ts"),
            )
    except SlackClientError as slack_error:
        self.send_exception(slack_error)

send_busy_indicator()

Send a "typing" indicator to show that work is in progress.

Source code in nautobot_chatops/dispatchers/slack.py
def send_busy_indicator(self):
    """Send a "typing" indicator to show that work is in progress."""
    # Currently the Slack Events API does not support the "user_typing" event.
    # We're trying not to use the legacy Slack RTM API as it's deprecated.
    # So for now, we do nothing.
    pass

send_error(message)

Send an error message to the user/channel specified by the context.

Source code in nautobot_chatops/dispatchers/slack.py
def send_error(self, message):
    """Send an error message to the user/channel specified by the context."""
    self.send_markdown(f":warning: {self.bold(message)} :warning:", ephemeral=True)

send_exception(exception)

Try to report an exception to the user.

Source code in nautobot_chatops/dispatchers/slack.py
def send_exception(self, exception):
    """Try to report an exception to the user."""
    self.slack_client.chat_postEphemeral(
        channel=self.context.get("channel_id"),
        user=self.context.get("user_id"),
        text=f"Sorry @{self.context.get('user_name')}, an error occurred :sob:\n```{exception}```",
    )

send_image(image_path)

Send an image as a file upload.

Source code in nautobot_chatops/dispatchers/slack.py
def send_image(self, image_path):
    """Send an image as a file upload."""
    if self.context.get("channel_name") == "directmessage":
        channels = [self.context.get("user_id")]
    else:
        channels = [self.context.get("channel_id")]
    channels = ",".join(channels)
    logger.info("Sending image %s to %s", image_path, channels)
    self.slack_client.files_upload(channels=channels, file=image_path, thread_ts=self.context.get("thread_ts"))

send_markdown(message, ephemeral=None)

Send a Markdown-formatted text message to the user/channel specified by the context.

Source code in nautobot_chatops/dispatchers/slack.py
@BACKEND_ACTION_MARKDOWN.time()
def send_markdown(self, message, ephemeral=None):
    """Send a Markdown-formatted text message to the user/channel specified by the context."""
    try:
        if ephemeral is None:
            ephemeral = settings.PLUGINS_CONFIG["nautobot_chatops"]["send_all_messages_private"]

        if ephemeral:
            self.slack_client.chat_postEphemeral(
                channel=self.context.get("channel_id"),
                user=self.context.get("user_id"),
                text=message,
                thread_ts=self.context.get("thread_ts"),
            )
        else:
            self.slack_client.chat_postMessage(
                channel=self.context.get("channel_id"),
                user=self.context.get("user_id"),
                text=message,
                thread_ts=self.context.get("thread_ts"),
            )
    except SlackClientError as slack_error:
        self.send_exception(slack_error)

send_snippet(text, title=None, ephemeral=None)

Send a longer chunk of text as a file snippet.

Source code in nautobot_chatops/dispatchers/slack.py
@BACKEND_ACTION_SNIPPET.time()
def send_snippet(self, text, title=None, ephemeral=None):
    """Send a longer chunk of text as a file snippet."""
    if ephemeral is None:
        ephemeral = settings.PLUGINS_CONFIG["nautobot_chatops"]["send_all_messages_private"]

    if self.context.get("channel_name") == "directmessage":
        channels = [self.context.get("user_id")]
    else:
        channels = [self.context.get("channel_id")]
    channels = ",".join(channels)
    logger.info("Sending snippet to %s: %s", channels, text)
    try:
        # Check for the length of the file if the setup is meant to be a private message
        if ephemeral:
            message_list = self.split_message(text, SLACK_PRIVATE_MESSAGE_LIMIT)
            for msg in message_list:
                # Send the blocks as a list, this needs to be the case for Slack to send appropriately.
                self.send_blocks(
                    [self.markdown_block(f"```\n{msg}\n```")],
                    ephemeral=ephemeral,
                )
        else:
            self.slack_client.files_upload(
                channels=channels, content=text, title=title, thread_ts=self.context.get("thread_ts")
            )
    except SlackClientError as slack_error:
        self.send_exception(slack_error)

send_warning(message)

Send a warning message to the user/channel specified by the context.

Source code in nautobot_chatops/dispatchers/slack.py
def send_warning(self, message):
    """Send a warning message to the user/channel specified by the context."""
    self.send_markdown(f":warning: {self.bold(message)} :warning:", ephemeral=True)

static_url(path)

Return a url to access the static file.

Source code in nautobot_chatops/dispatchers/slack.py
def static_url(self, path):
    """Return a url to access the static file."""
    static_path = str(static(path))
    if static_path.startswith("http"):
        # Static is being provided by Django Storages
        return static_path
    try:
        return f"{self.context['request_scheme']}://{self.context['request_host']}{static_path}"
    except KeyError:
        if SLACK_SOCKET_STATIC_HOST:
            # Use the settings provided Static Host.
            return f"{SLACK_SOCKET_STATIC_HOST}{static_path}"
        return None

text_element(text)

Construct a basic plaintext element.

Source code in nautobot_chatops/dispatchers/slack.py
def text_element(self, text):
    """Construct a basic plaintext element."""
    return {"type": "plain_text", "text": text}

user_mention()

Markup for a mention of the username/userid specified in our context.

Source code in nautobot_chatops/dispatchers/slack.py
def user_mention(self):
    """Markup for a mention of the username/userid specified in our context."""
    return f"<@{self.context['user_id']}>"