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

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}```",
        )

    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'] class-attribute instance-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

hyperlink(text, url)

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']}>"