API Documentation

This is the API documentation of all the classes and functions relevant for Plugin development. The rest of the code deals with the internal workings of Slack Machine and is very much an implementation detail and subject to change. Therefore it is not documented.

Plugin classes

The following classes form the basis for Plugin development.

machine.plugins.base.MachineBasePlugin

Base class for all Slack Machine plugins

The purpose of this class is two-fold:

1. It acts as a marker-class so Slack Machine can recognize plugins as such
2. It provides a lot of common functionality and convenience methods for plugins to interact with channels and users

Attributes:

Name	Type	Description
settings	CaseInsensitiveDict	Slack Machine settings object that contains all settings that were defined through local_settings.py Plugin developers can use any settings that are defined by the user, and ask users to add new settings specifically for their plugin.

Source code in machine/plugins/base.py

class MachineBasePlugin:
    """Base class for all Slack Machine plugins

    The purpose of this class is two-fold:

    1. It acts as a marker-class so Slack Machine can recognize plugins as such
    2. It provides a lot of common functionality and convenience methods for plugins to
       interact with channels and users

    :var settings: Slack Machine settings object that contains all settings that
        were defined through ``local_settings.py`` Plugin developers can use any
        settings that are defined by the user, and ask users to add new settings
        specifically for their plugin.
    """

    _client: SlackClient
    storage: PluginStorage
    settings: CaseInsensitiveDict
    _fq_name: str

    def __init__(self, client: SlackClient, settings: CaseInsensitiveDict, storage: PluginStorage):
        self._client = client
        self.storage = storage
        self.settings = settings
        self._fq_name = f"{self.__module__}.{self.__class__.__name__}"

    async def init(self) -> None:
        """Initialize plugin

        This method can be implemented by concrete plugin classes. It will be called **once**
        for each plugin, when that plugin is first loaded. You can refer to settings via
        ``self.settings``, and access storage through ``self.storage``, but the Slack client has
        not been initialized yet, so you cannot send or process messages during initialization.

        :return: None
        """
        return None

    @property
    def users(self) -> dict[str, User]:
        """Dictionary of all users in the Slack workspace

        :return: a dictionary of all users in the Slack workspace, where the key is the user id and
            the value is a [`User`][machine.models.user.User] object
        """
        return self._client.users

    @property
    def users_by_email(self) -> dict[str, User]:
        """Dictionary of all users in the Slack workspace by email

        **Note**: not every user might have an email address in their profile, so this
        dictionary might not contain all users in the Slack workspace

        :return: a dictionary of all users in the Slack workspace, where the key is the email and
            the value is a [`User`][machine.models.user.User] object
        """
        return self._client.users

    @property
    def channels(self) -> dict[str, Channel]:
        """List of all channels in the Slack workspace

        This is a list of all channels in the Slack workspace that the bot is aware of. This
        includes all public channels, all private channels the bot is a member of and all DM
        channels the bot is a member of.

        :return: a list of all channels in the Slack workspace, where each channel is a
            :py:class:`~machine.models.channel.Channel` object
        """
        return self._client.channels

    @property
    def web_client(self) -> AsyncWebClient:
        """Slack SDK web client to access the [Slack Web API][slack-web-api]

        This property references an instance of [`AsyncWebClient`][async-web-client]

        [slack-web-api]: https://api.slack.com/web
        [async-web-client]: https://slack.dev/python-slack-sdk/api-docs/slack_sdk/web/async_client.html#slack_sdk.web.async_client.AsyncWebClient
        """  # noqa: E501
        return self._client.web_client

    def find_channel_by_name(self, channel_name: str) -> Channel | None:
        """Find a channel by its name, irrespective of a preceding pound symbol. This does not include DMs.

        :param channel_name: The name of the channel to retrieve.
        :return: The channel if found, None otherwise.
        """
        if channel_name.startswith("#"):
            channel_name = channel_name[1:]
        for c in self.channels.values():
            if c.name_normalized and channel_name.lower() == c.name_normalized.lower():
                return c
        return None

    def get_user_by_id(self, user_id: str) -> User | None:
        """Get a user by their ID.

        :param user_id: The ID of the user to retrieve.
        :return: The user if found, None otherwise.
        """
        return self.users.get(user_id)

    def get_user_by_email(self, email: str) -> User | None:
        """Get a user by their email address.

        :param email: The email address of the user to retrieve.
        :return: The user if found, None otherwise.
        """
        return self._client.get_user_by_email(email)

    @property
    def bot_info(self) -> dict[str, Any]:
        """Information about the bot user in Slack

        This will return a dictionary with information about the bot user in Slack that represents
        Slack Machine

        :return: Bot user
        """
        return self._client.bot_info

    def at(self, user: User) -> str:
        """Create a mention of the provided user

        Create a mention of the provided user in the form of ``<@[user_id]>``. This method is
        convenient when you want to include mentions in your message. This method does not send
        a message, but should be used together with methods like
        :py:meth:`~machine.plugins.base.MachineBasePlugin.say`

        :param user: user your want to mention
        :return: user mention
        """
        return user.fmt_mention()

    async def say(
        self,
        channel: Channel | str,
        text: str | None = None,
        attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
        blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
        thread_ts: str | None = None,
        ephemeral_user: User | str | None = None,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Send a message to a channel

        Send a message to a channel using the WebAPI. Allows for rich formatting using
        `blocks`_ and/or `attachments`_. You can provide blocks and attachments as Python dicts or
        you can use the `convenient classes`_ that the underlying slack client provides.
        Can also reply in-thread and send ephemeral messages, visible to only one user.
        Ephemeral messages and threaded messages are mutually exclusive, and ``ephemeral_user``
        takes precedence over ``thread_ts``
        Any extra kwargs you provide, will be passed on directly to the `chat.postMessage`_ or
        `chat.postEphemeral`_ request.

        .. _attachments: https://api.slack.com/docs/message-attachments
        .. _blocks: https://api.slack.com/reference/block-kit/blocks
        .. _convenient classes:
            https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

        :param channel: :py:class:`~machine.models.channel.Channel` object or id of channel to send
            message to. Can be public or private (group) channel, or DM channel.
        :param text: message text
        :param attachments: optional attachments (see `attachments`_)
        :param blocks: optional blocks (see `blocks`_)
        :param thread_ts: optional timestamp of thread, to send a message in that thread
        :param ephemeral_user: optional user name or id if the message needs to visible
            to a specific user only
        :return: Dictionary deserialized from `chat.postMessage`_ request, or `chat.postEphemeral`_
            if `ephemeral_user` is True.

        .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage
        .. _chat.postEphemeral: https://api.slack.com/methods/chat.postEphemeral
        """
        return await self._client.send(
            channel,
            text=text,
            attachments=attachments,
            blocks=blocks,
            thread_ts=thread_ts,
            ephemeral_user=ephemeral_user,
            **kwargs,
        )

    async def say_scheduled(
        self,
        when: datetime,
        channel: Channel | str,
        text: str,
        attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
        blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
        thread_ts: str | None = None,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Schedule a message to a channel

        This is the scheduled version of :py:meth:`~machine.plugins.base.MachineBasePlugin.say`.
        It behaves the same, but will send the message at the scheduled time.

        :param when: when you want the message to be sent, as :py:class:`datetime.datetime` instance
        :param channel: :py:class:`~machine.models.channel.Channel` object or id of channel to send
            message to. Can be public or private (group) channel, or DM channel.
        :param text: message text
        :param attachments: optional attachments (see `attachments`_)
        :param blocks: optional blocks (see `blocks`_)
        :param thread_ts: optional timestamp of thread, to send a message in that thread
        :return: None

        .. _attachments: https://api.slack.com/docs/message-attachments
        .. _blocks: https://api.slack.com/reference/block-kit/blocks
        """
        return await self._client.send_scheduled(
            when,
            channel,
            text=text,
            attachments=attachments,
            blocks=blocks,
            thread_ts=thread_ts,
            **kwargs,
        )

    async def update(
        self,
        channel: Channel | str,
        ts: str,
        text: str | None = None,
        attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
        blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
        ephemeral_user: User | str | None = None,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Update an existing message

        Update an existing message using the WebAPI. Allows for rich formatting using
        [blocks] and/or [attachments]. You can provide blocks and attachments as Python dicts or
        you can use the [convenient classes] that the underlying slack client provides.
        Can also update in-thread and ephemeral messages, visible to only one user.
        Any extra kwargs you provide, will be passed on directly to the [`chat.update`][chat_update] request.

        [attachments]: https://api.slack.com/docs/message-attachments
        [blocks]: https://api.slack.com/reference/block-kit/blocks
        [convenient classes]: https://github.com/slackapi/python-slack-sdk/tree/main/slack/web/classes
        [chat_update]: https://api.slack.com/methods/chat.update

        :param channel: [`Channel`][machine.models.channel.Channel] object or id of channel to send
            message to. Can be public or private (group) channel, or DM channel.
        :param ts: timestamp of the message to be updated.
        :param text: message text
        :param attachments: optional attachments (see [attachments])
        :param blocks: optional blocks (see [blocks])
        :param ephemeral_user: optional user name or id if the message needs to visible
            to a specific user only
        :return: Dictionary deserialized from [`chat.update`](https://api.slack.com/methods/chat.update) request


        """
        return await self._client.update(
            channel,
            ts=ts,
            text=text,
            attachments=attachments,
            blocks=blocks,
            ephemeral_user=ephemeral_user,
            **kwargs,
        )

    async def delete(
        self,
        channel: Channel | str,
        ts: str,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Delete an existing message

        Delete an existing message using the WebAPI.
        Any extra kwargs you provide, will be passed on directly to the [`chat.delete`][chat_delete] request.

        [chat_delete]: https://api.slack.com/methods/chat.delete

        :param channel: [`Channel`][machine.models.channel.Channel] object or id of channel to send
            message to. Can be public or private (group) channel, or DM channel.
        :param ts: timestamp of the message to be deleted.
        :return: Dictionary deserialized from [`chat.delete`](https://api.slack.com/methods/chat.delete) request


        """
        return await self._client.delete(
            channel,
            ts=ts,
            **kwargs,
        )

    async def react(self, channel: Channel | str, ts: str, emoji: str) -> AsyncSlackResponse:
        """React to a message in a channel

        Add a reaction to a message in a channel. What message to react to, is determined by the
        combination of the channel and the timestamp of the message.

        :param channel: :py:class:`~machine.models.channel.Channel` object or id of channel to send
            message to. Can be public or private (group) channel, or DM channel.
        :param ts: timestamp of the message to react to
        :param emoji: what emoji to react with (should be a string, like 'angel', 'thumbsup', etc.)
        :return: Dictionary deserialized from `reactions.add`_ request.

        .. _reactions.add: https://api.slack.com/methods/reactions.add
        """
        return await self._client.react(channel, ts, emoji)

    async def send_dm(
        self,
        user: User | str,
        text: str | None = None,
        attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
        blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Send a Direct Message

        Send a Direct Message to a user by opening a DM channel and sending a message to it. Allows
        for rich formatting using `blocks`_ and/or `attachments`_. You can provide blocks and
        attachments as Python dicts or you can use the `convenient classes`_ that the underlying
        slack client provides.
        Any extra kwargs you provide, will be passed on directly to the `chat.postMessage`_ request.

        .. _attachments: https://api.slack.com/docs/message-attachments
        .. _blocks: https://api.slack.com/reference/block-kit/blocks
        .. _convenient classes:
            https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

        :param user: :py:class:`~machine.models.user.User` object or id of user to send DM to.
        :param text: message text
        :param attachments: optional attachments (see `attachments`_)
        :param blocks: optional blocks (see `blocks`_)
        :return: Dictionary deserialized from `chat.postMessage`_ response.

        .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage
        """
        return await self._client.send_dm(user, text, attachments=attachments, blocks=blocks, **kwargs)

    async def send_dm_scheduled(
        self,
        when: datetime,
        user: User | str,
        text: str,
        attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
        blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Schedule a Direct Message

        This is the scheduled version of
        :py:meth:`~machine.plugins.base.MachineBasePlugin.send_dm`. It behaves the same, but
        will send the DM at the scheduled time.

        :param when: when you want the message to be sent, as :py:class:`datetime.datetime` instance
        :param user: :py:class:`~machine.models.user.User` object or id of user to send DM to.
        :param text: message text
        :param attachments: optional attachments (see `attachments`_)
        :param blocks: optional blocks (see `blocks`_)
        :return: None

        .. _attachments: https://api.slack.com/docs/message-attachments
        .. _blocks: https://api.slack.com/reference/block-kit/blocks
        """
        return await self._client.send_dm_scheduled(
            when, user, text=text, attachments=attachments, blocks=blocks, **kwargs
        )

    async def open_im(self, users: User | str | list[User | str]) -> str:
        """Open a DM channel with one or more users

        Open a DM channel with one or more users. If the DM channel already exists, the existing channel id
        will be returned. If the DM channel does not exist, a new channel will be created and the
        id of the new channel will be returned.

        :param users: :py:class:`~machine.models.user.User` object or id of user to open DM with.
        :return: id of the DM channel
        """
        return await self._client.open_im(users)

    def emit(self, event: str, **kwargs: Any) -> None:
        """Emit an event

        Emit an event that plugins can listen for. You can include arbitrary data as keyword
        arguments.

        :param event: name of the event
        :param **kwargs: any data you want to emit with the event
        :return: None
        """
        ee.emit(event, self, **kwargs)

    async def pin_message(self, channel: Channel | str, ts: str) -> AsyncSlackResponse:
        """Pin message

        Pin a message in a channel

        :param channel: channel to pin the message in
        :param ts: timestamp of the message to pin
        :return: response from the Slack Web API
        """
        return await self._client.pin_message(channel, ts)

    async def unpin_message(self, channel: Channel | str, ts: str) -> AsyncSlackResponse:
        """Unpin message

        Unpin a message that was previously pinned in a channel

        :param channel: channel where the message is pinned that needs to be unpinned
        :param ts: timestamp of the message to unpin
        :return: response from the Slack Web API
        """
        return await self._client.unpin_message(channel, ts)

    async def set_topic(self, channel: Channel | str, topic: str, **kwargs: Any) -> AsyncSlackResponse:
        """Set channel topic

        Set or update topic for the channel

        :param channel: channel where topic needs to be set or updated
        :param topic: topic for the channel (slack does not support formatting for topics)
        :return: response from the Slack Web API
        """
        return await self._client.set_topic(channel, topic, **kwargs)

    async def open_modal(self, trigger_id: str, view: dict | View, **kwargs: Any) -> AsyncSlackResponse:
        """Open a modal dialog

        Open a modal dialog in response to a user action. The modal dialog can be used to collect
        information from the user, or to display information to the user.

        :param trigger_id: trigger id is provided by Slack when a user action is performed, such as a slash command
            or a button click
        :param view: view definition for the modal dialog
        :return: response from the Slack Web API
        """
        return await self._client.web_client.views_open(trigger_id=trigger_id, view=view, **kwargs)

    async def push_modal(self, trigger_id: str, view: dict | View, **kwargs: Any) -> AsyncSlackResponse:
        """Push a new view onto the stack of a modal that was already opened

        Push a new view onto the stack of a modal that was already opened by a open_modal call. At most 3 views can be
        active in a modal at the same time. For more information on the lifecycle of modals, refer to the
        [relevant Slack documentation](https://api.slack.com/surfaces/modals)

        :param trigger_id: trigger id is provided by Slack when a user action is performed, such as a slash command
            or a button click
        :param view: view definition for the modal dialog
        :return: response from the Slack Web API
        """
        return await self._client.push_modal(trigger_id=trigger_id, view=view, **kwargs)

    async def update_modal(
        self,
        view: dict | View,
        view_id: str | None = None,
        external_id: str | None = None,
        hash: str | None = None,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Update a modal dialog

        Update a modal dialog that was previously opened. You can update the view by providing the view_id or the
        external_id of the modal. external_id has precedence over view_id, but at least one needs to be provided.
        You can also provide a hash of the view that you want to update to prevent race conditions.

        :param view: view definition for the modal dialog
        :param view_id: id of the view to update
        :param external_id: external id of the view to update
        :param hash: hash of the view to update
        :return: response from the Slack Web API
        """
        return await self._client.update_modal(view=view, view_id=view_id, external_id=external_id, hash=hash, **kwargs)

    async def publish_home_tab(
        self, user: User | str, view: dict | View, hash: str | None = None, **kwargs: Any
    ) -> AsyncSlackResponse:
        """Publish a view to the home tab of a user

        Publish a view to the home tab of a user. The view will be visible to the user when they open the home tab of
        your Slack app. This method can be used both to publish a new view for the home tab or update an existing view.
        You can provide a hash of the view that you want to update to prevent race conditions.

        Note: be careful with the use of this method, as you might be overwriting the user's home tab that was set by
        another Slack Machine plugin enabled in your bot.

        :param user: user for whom to publish or update the home tab
        :param view: view definition for the home tab
        :param hash: hash of the view to update
        :return: response from the Slack Web API
        """
        return await self._client.publish_home_tab(user=user, view=view, hash=hash, **kwargs)

users: dict[str, User] property

Dictionary of all users in the Slack workspace

Returns:

Type	Description
dict[str, User]	a dictionary of all users in the Slack workspace, where the key is the user id and the value is a User object

users_by_email: dict[str, User] property

Dictionary of all users in the Slack workspace by email

Note: not every user might have an email address in their profile, so this dictionary might not contain all users in the Slack workspace

Returns:

Type	Description
dict[str, User]	a dictionary of all users in the Slack workspace, where the key is the email and the value is a User object

channels: dict[str, Channel] property

List of all channels in the Slack workspace

This is a list of all channels in the Slack workspace that the bot is aware of. This includes all public channels, all private channels the bot is a member of and all DM channels the bot is a member of.

Returns:

Type	Description
dict[str, Channel]	a list of all channels in the Slack workspace, where each channel is a :py:class:~machine.models.channel.Channel object

web_client: AsyncWebClient property

Slack SDK web client to access the Slack Web API

This property references an instance of AsyncWebClient

bot_info: dict[str, Any] property

Information about the bot user in Slack

This will return a dictionary with information about the bot user in Slack that represents Slack Machine

Returns:

Type	Description
dict[str, Any]	Bot user

init() -> None async

Initialize plugin

This method can be implemented by concrete plugin classes. It will be called once for each plugin, when that plugin is first loaded. You can refer to settings via self.settings, and access storage through self.storage, but the Slack client has not been initialized yet, so you cannot send or process messages during initialization.

Returns:

Type	Description
None	None

Source code in machine/plugins/base.py

async def init(self) -> None:
    """Initialize plugin

    This method can be implemented by concrete plugin classes. It will be called **once**
    for each plugin, when that plugin is first loaded. You can refer to settings via
    ``self.settings``, and access storage through ``self.storage``, but the Slack client has
    not been initialized yet, so you cannot send or process messages during initialization.

    :return: None
    """
    return None

find_channel_by_name(channel_name: str) -> Channel | None

Find a channel by its name, irrespective of a preceding pound symbol. This does not include DMs.

Parameters:

Name	Type	Description	Default
channel_name	str	The name of the channel to retrieve.	required

Returns:

Type	Description
Channel | None	The channel if found, None otherwise.

Source code in machine/plugins/base.py

def find_channel_by_name(self, channel_name: str) -> Channel | None:
    """Find a channel by its name, irrespective of a preceding pound symbol. This does not include DMs.

    :param channel_name: The name of the channel to retrieve.
    :return: The channel if found, None otherwise.
    """
    if channel_name.startswith("#"):
        channel_name = channel_name[1:]
    for c in self.channels.values():
        if c.name_normalized and channel_name.lower() == c.name_normalized.lower():
            return c
    return None

get_user_by_id(user_id: str) -> User | None

Get a user by their ID.

Parameters:

Name	Type	Description	Default
user_id	str	The ID of the user to retrieve.	required

Returns:

Type	Description
User | None	The user if found, None otherwise.

Source code in machine/plugins/base.py

def get_user_by_id(self, user_id: str) -> User | None:
    """Get a user by their ID.

    :param user_id: The ID of the user to retrieve.
    :return: The user if found, None otherwise.
    """
    return self.users.get(user_id)

get_user_by_email(email: str) -> User | None

Get a user by their email address.

Parameters:

Name	Type	Description	Default
email	str	The email address of the user to retrieve.	required

Returns:

Type	Description
User | None	The user if found, None otherwise.

Source code in machine/plugins/base.py

def get_user_by_email(self, email: str) -> User | None:
    """Get a user by their email address.

    :param email: The email address of the user to retrieve.
    :return: The user if found, None otherwise.
    """
    return self._client.get_user_by_email(email)

at(user: User) -> str

Create a mention of the provided user

Create a mention of the provided user in the form of <@[user_id]>. This method is convenient when you want to include mentions in your message. This method does not send a message, but should be used together with methods like :py:meth:~machine.plugins.base.MachineBasePlugin.say

Parameters:

Name	Type	Description	Default
user	User	user your want to mention	required

Returns:

Type	Description
str	user mention

Source code in machine/plugins/base.py

def at(self, user: User) -> str:
    """Create a mention of the provided user

    Create a mention of the provided user in the form of ``<@[user_id]>``. This method is
    convenient when you want to include mentions in your message. This method does not send
    a message, but should be used together with methods like
    :py:meth:`~machine.plugins.base.MachineBasePlugin.say`

    :param user: user your want to mention
    :return: user mention
    """
    return user.fmt_mention()

say(channel: Channel | str, text: str | None = None, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, thread_ts: str | None = None, ephemeral_user: User | str | None = None, **kwargs: Any) -> AsyncSlackResponse async

Send a message to a channel

Send a message to a channel using the WebAPI. Allows for rich formatting using blocks and/or attachments. You can provide blocks and attachments as Python dicts or you can use the convenient classes that the underlying slack client provides. Can also reply in-thread and send ephemeral messages, visible to only one user. Ephemeral messages and threaded messages are mutually exclusive, and ephemeral_user takes precedence over thread_ts Any extra kwargs you provide, will be passed on directly to the chat.postMessage or chat.postEphemeral_ request.

.. _attachments: https://api.slack.com/docs/message-attachments .. _blocks: https://api.slack.com/reference/block-kit/blocks .. _convenient classes: https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

Parameters:

Name	Type	Description	Default
channel	Channel | str	:py:class:~machine.models.channel.Channel object or id of channel to send message to. Can be public or private (group) channel, or DM channel.	required
text	str | None	message text	None
attachments	Sequence[Attachment] | Sequence[dict[str, Any]] | None	optional attachments (see attachments_)	None
blocks	Sequence[Block] | Sequence[dict[str, Any]] | None	optional blocks (see blocks_)	None
thread_ts	str | None	optional timestamp of thread, to send a message in that thread	None
ephemeral_user	User | str | None	optional user name or id if the message needs to visible to a specific user only	None

Returns:

Type	Description
AsyncSlackResponse	Dictionary deserialized from chat.postMessage request, or chat.postEphemeral if ephemeral_user is True. .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage .. _chat.postEphemeral: https://api.slack.com/methods/chat.postEphemeral

Source code in machine/plugins/base.py

async def say(
    self,
    channel: Channel | str,
    text: str | None = None,
    attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
    blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
    thread_ts: str | None = None,
    ephemeral_user: User | str | None = None,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Send a message to a channel

    Send a message to a channel using the WebAPI. Allows for rich formatting using
    `blocks`_ and/or `attachments`_. You can provide blocks and attachments as Python dicts or
    you can use the `convenient classes`_ that the underlying slack client provides.
    Can also reply in-thread and send ephemeral messages, visible to only one user.
    Ephemeral messages and threaded messages are mutually exclusive, and ``ephemeral_user``
    takes precedence over ``thread_ts``
    Any extra kwargs you provide, will be passed on directly to the `chat.postMessage`_ or
    `chat.postEphemeral`_ request.

    .. _attachments: https://api.slack.com/docs/message-attachments
    .. _blocks: https://api.slack.com/reference/block-kit/blocks
    .. _convenient classes:
        https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

    :param channel: :py:class:`~machine.models.channel.Channel` object or id of channel to send
        message to. Can be public or private (group) channel, or DM channel.
    :param text: message text
    :param attachments: optional attachments (see `attachments`_)
    :param blocks: optional blocks (see `blocks`_)
    :param thread_ts: optional timestamp of thread, to send a message in that thread
    :param ephemeral_user: optional user name or id if the message needs to visible
        to a specific user only
    :return: Dictionary deserialized from `chat.postMessage`_ request, or `chat.postEphemeral`_
        if `ephemeral_user` is True.

    .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage
    .. _chat.postEphemeral: https://api.slack.com/methods/chat.postEphemeral
    """
    return await self._client.send(
        channel,
        text=text,
        attachments=attachments,
        blocks=blocks,
        thread_ts=thread_ts,
        ephemeral_user=ephemeral_user,
        **kwargs,
    )

say_scheduled(when: datetime, channel: Channel | str, text: str, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, thread_ts: str | None = None, **kwargs: Any) -> AsyncSlackResponse async

Schedule a message to a channel

This is the scheduled version of :py:meth:~machine.plugins.base.MachineBasePlugin.say. It behaves the same, but will send the message at the scheduled time.

Parameters:

Name	Type	Description	Default
when	datetime	when you want the message to be sent, as :py:class:datetime.datetime instance	required
channel	Channel | str	:py:class:~machine.models.channel.Channel object or id of channel to send message to. Can be public or private (group) channel, or DM channel.	required
text	str	message text	required
attachments	Sequence[Attachment] | Sequence[dict[str, Any]] | None	optional attachments (see attachments_)	None
blocks	Sequence[Block] | Sequence[dict[str, Any]] | None	optional blocks (see blocks_)	None
thread_ts	str | None	optional timestamp of thread, to send a message in that thread	None

Returns:

Type	Description
AsyncSlackResponse	None .. _attachments: https://api.slack.com/docs/message-attachments .. _blocks: https://api.slack.com/reference/block-kit/blocks

Source code in machine/plugins/base.py

async def say_scheduled(
    self,
    when: datetime,
    channel: Channel | str,
    text: str,
    attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
    blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
    thread_ts: str | None = None,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Schedule a message to a channel

    This is the scheduled version of :py:meth:`~machine.plugins.base.MachineBasePlugin.say`.
    It behaves the same, but will send the message at the scheduled time.

    :param when: when you want the message to be sent, as :py:class:`datetime.datetime` instance
    :param channel: :py:class:`~machine.models.channel.Channel` object or id of channel to send
        message to. Can be public or private (group) channel, or DM channel.
    :param text: message text
    :param attachments: optional attachments (see `attachments`_)
    :param blocks: optional blocks (see `blocks`_)
    :param thread_ts: optional timestamp of thread, to send a message in that thread
    :return: None

    .. _attachments: https://api.slack.com/docs/message-attachments
    .. _blocks: https://api.slack.com/reference/block-kit/blocks
    """
    return await self._client.send_scheduled(
        when,
        channel,
        text=text,
        attachments=attachments,
        blocks=blocks,
        thread_ts=thread_ts,
        **kwargs,
    )

update(channel: Channel | str, ts: str, text: str | None = None, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, ephemeral_user: User | str | None = None, **kwargs: Any) -> AsyncSlackResponse async

Update an existing message

Update an existing message using the WebAPI. Allows for rich formatting using blocks and/or attachments. You can provide blocks and attachments as Python dicts or you can use the convenient classes that the underlying slack client provides. Can also update in-thread and ephemeral messages, visible to only one user. Any extra kwargs you provide, will be passed on directly to the chat.update request.

Parameters:

Name	Type	Description	Default
channel	Channel | str	Channel object or id of channel to send message to. Can be public or private (group) channel, or DM channel.	required
ts	str	timestamp of the message to be updated.	required
text	str | None	message text	None
attachments	Sequence[Attachment] | Sequence[dict[str, Any]] | None	optional attachments (see [attachments])	None
blocks	Sequence[Block] | Sequence[dict[str, Any]] | None	optional blocks (see [blocks])	None
ephemeral_user	User | str | None	optional user name or id if the message needs to visible to a specific user only	None

Returns:

Type	Description
AsyncSlackResponse	Dictionary deserialized from chat.update request

Source code in machine/plugins/base.py

async def update(
    self,
    channel: Channel | str,
    ts: str,
    text: str | None = None,
    attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
    blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
    ephemeral_user: User | str | None = None,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Update an existing message

    Update an existing message using the WebAPI. Allows for rich formatting using
    [blocks] and/or [attachments]. You can provide blocks and attachments as Python dicts or
    you can use the [convenient classes] that the underlying slack client provides.
    Can also update in-thread and ephemeral messages, visible to only one user.
    Any extra kwargs you provide, will be passed on directly to the [`chat.update`][chat_update] request.

    [attachments]: https://api.slack.com/docs/message-attachments
    [blocks]: https://api.slack.com/reference/block-kit/blocks
    [convenient classes]: https://github.com/slackapi/python-slack-sdk/tree/main/slack/web/classes
    [chat_update]: https://api.slack.com/methods/chat.update

    :param channel: [`Channel`][machine.models.channel.Channel] object or id of channel to send
        message to. Can be public or private (group) channel, or DM channel.
    :param ts: timestamp of the message to be updated.
    :param text: message text
    :param attachments: optional attachments (see [attachments])
    :param blocks: optional blocks (see [blocks])
    :param ephemeral_user: optional user name or id if the message needs to visible
        to a specific user only
    :return: Dictionary deserialized from [`chat.update`](https://api.slack.com/methods/chat.update) request


    """
    return await self._client.update(
        channel,
        ts=ts,
        text=text,
        attachments=attachments,
        blocks=blocks,
        ephemeral_user=ephemeral_user,
        **kwargs,
    )

delete(channel: Channel | str, ts: str, **kwargs: Any) -> AsyncSlackResponse async

Delete an existing message

Delete an existing message using the WebAPI. Any extra kwargs you provide, will be passed on directly to the chat.delete request.

Parameters:

Name	Type	Description	Default
channel	Channel | str	Channel object or id of channel to send message to. Can be public or private (group) channel, or DM channel.	required
ts	str	timestamp of the message to be deleted.	required

Returns:

Type	Description
AsyncSlackResponse	Dictionary deserialized from chat.delete request

Source code in machine/plugins/base.py

async def delete(
    self,
    channel: Channel | str,
    ts: str,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Delete an existing message

    Delete an existing message using the WebAPI.
    Any extra kwargs you provide, will be passed on directly to the [`chat.delete`][chat_delete] request.

    [chat_delete]: https://api.slack.com/methods/chat.delete

    :param channel: [`Channel`][machine.models.channel.Channel] object or id of channel to send
        message to. Can be public or private (group) channel, or DM channel.
    :param ts: timestamp of the message to be deleted.
    :return: Dictionary deserialized from [`chat.delete`](https://api.slack.com/methods/chat.delete) request


    """
    return await self._client.delete(
        channel,
        ts=ts,
        **kwargs,
    )

react(channel: Channel | str, ts: str, emoji: str) -> AsyncSlackResponse async

React to a message in a channel

Add a reaction to a message in a channel. What message to react to, is determined by the combination of the channel and the timestamp of the message.

Parameters:

Name	Type	Description	Default
channel	Channel | str	:py:class:~machine.models.channel.Channel object or id of channel to send message to. Can be public or private (group) channel, or DM channel.	required
ts	str	timestamp of the message to react to	required
emoji	str	what emoji to react with (should be a string, like 'angel', 'thumbsup', etc.)	required

Returns:

Type	Description
AsyncSlackResponse	Dictionary deserialized from reactions.add_ request. .. _reactions.add: https://api.slack.com/methods/reactions.add

Source code in machine/plugins/base.py

async def react(self, channel: Channel | str, ts: str, emoji: str) -> AsyncSlackResponse:
    """React to a message in a channel

    Add a reaction to a message in a channel. What message to react to, is determined by the
    combination of the channel and the timestamp of the message.

    :param channel: :py:class:`~machine.models.channel.Channel` object or id of channel to send
        message to. Can be public or private (group) channel, or DM channel.
    :param ts: timestamp of the message to react to
    :param emoji: what emoji to react with (should be a string, like 'angel', 'thumbsup', etc.)
    :return: Dictionary deserialized from `reactions.add`_ request.

    .. _reactions.add: https://api.slack.com/methods/reactions.add
    """
    return await self._client.react(channel, ts, emoji)

send_dm(user: User | str, text: str | None = None, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, **kwargs: Any) -> AsyncSlackResponse async

Send a Direct Message

Send a Direct Message to a user by opening a DM channel and sending a message to it. Allows for rich formatting using blocks and/or attachments. You can provide blocks and attachments as Python dicts or you can use the convenient classes that the underlying slack client provides. Any extra kwargs you provide, will be passed on directly to the chat.postMessage request.

.. _attachments: https://api.slack.com/docs/message-attachments .. _blocks: https://api.slack.com/reference/block-kit/blocks .. _convenient classes: https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

Parameters:

Name	Type	Description	Default
user	User | str	:py:class:~machine.models.user.User object or id of user to send DM to.	required
text	str | None	message text	None
attachments	Sequence[Attachment] | Sequence[dict[str, Any]] | None	optional attachments (see attachments_)	None
blocks	Sequence[Block] | Sequence[dict[str, Any]] | None	optional blocks (see blocks_)	None

Returns:

Type	Description
AsyncSlackResponse	Dictionary deserialized from chat.postMessage_ response. .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage

Source code in machine/plugins/base.py

async def send_dm(
    self,
    user: User | str,
    text: str | None = None,
    attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
    blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Send a Direct Message

    Send a Direct Message to a user by opening a DM channel and sending a message to it. Allows
    for rich formatting using `blocks`_ and/or `attachments`_. You can provide blocks and
    attachments as Python dicts or you can use the `convenient classes`_ that the underlying
    slack client provides.
    Any extra kwargs you provide, will be passed on directly to the `chat.postMessage`_ request.

    .. _attachments: https://api.slack.com/docs/message-attachments
    .. _blocks: https://api.slack.com/reference/block-kit/blocks
    .. _convenient classes:
        https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

    :param user: :py:class:`~machine.models.user.User` object or id of user to send DM to.
    :param text: message text
    :param attachments: optional attachments (see `attachments`_)
    :param blocks: optional blocks (see `blocks`_)
    :return: Dictionary deserialized from `chat.postMessage`_ response.

    .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage
    """
    return await self._client.send_dm(user, text, attachments=attachments, blocks=blocks, **kwargs)

send_dm_scheduled(when: datetime, user: User | str, text: str, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, **kwargs: Any) -> AsyncSlackResponse async

Schedule a Direct Message

This is the scheduled version of :py:meth:~machine.plugins.base.MachineBasePlugin.send_dm. It behaves the same, but will send the DM at the scheduled time.

Parameters:

Name	Type	Description	Default
when	datetime	when you want the message to be sent, as :py:class:datetime.datetime instance	required
user	User | str	:py:class:~machine.models.user.User object or id of user to send DM to.	required
text	str	message text	required
attachments	Sequence[Attachment] | Sequence[dict[str, Any]] | None	optional attachments (see attachments_)	None
blocks	Sequence[Block] | Sequence[dict[str, Any]] | None	optional blocks (see blocks_)	None

Returns:

Type	Description
AsyncSlackResponse	None .. _attachments: https://api.slack.com/docs/message-attachments .. _blocks: https://api.slack.com/reference/block-kit/blocks

Source code in machine/plugins/base.py

async def send_dm_scheduled(
    self,
    when: datetime,
    user: User | str,
    text: str,
    attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
    blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Schedule a Direct Message

    This is the scheduled version of
    :py:meth:`~machine.plugins.base.MachineBasePlugin.send_dm`. It behaves the same, but
    will send the DM at the scheduled time.

    :param when: when you want the message to be sent, as :py:class:`datetime.datetime` instance
    :param user: :py:class:`~machine.models.user.User` object or id of user to send DM to.
    :param text: message text
    :param attachments: optional attachments (see `attachments`_)
    :param blocks: optional blocks (see `blocks`_)
    :return: None

    .. _attachments: https://api.slack.com/docs/message-attachments
    .. _blocks: https://api.slack.com/reference/block-kit/blocks
    """
    return await self._client.send_dm_scheduled(
        when, user, text=text, attachments=attachments, blocks=blocks, **kwargs
    )

open_im(users: User | str | list[User | str]) -> str async

Open a DM channel with one or more users

Open a DM channel with one or more users. If the DM channel already exists, the existing channel id will be returned. If the DM channel does not exist, a new channel will be created and the id of the new channel will be returned.

Parameters:

Name	Type	Description	Default
users	User | str | list[User | str]	:py:class:~machine.models.user.User object or id of user to open DM with.	required

Returns:

Type	Description
str	id of the DM channel

Source code in machine/plugins/base.py

async def open_im(self, users: User | str | list[User | str]) -> str:
    """Open a DM channel with one or more users

    Open a DM channel with one or more users. If the DM channel already exists, the existing channel id
    will be returned. If the DM channel does not exist, a new channel will be created and the
    id of the new channel will be returned.

    :param users: :py:class:`~machine.models.user.User` object or id of user to open DM with.
    :return: id of the DM channel
    """
    return await self._client.open_im(users)

emit(event: str, **kwargs: Any) -> None

Emit an event

Emit an event that plugins can listen for. You can include arbitrary data as keyword arguments.

Parameters:

Name	Type	Description	Default
event	str	name of the event	required
**kwargs	Any	any data you want to emit with the event	{}

Returns:

Type	Description
None	None

Source code in machine/plugins/base.py

def emit(self, event: str, **kwargs: Any) -> None:
    """Emit an event

    Emit an event that plugins can listen for. You can include arbitrary data as keyword
    arguments.

    :param event: name of the event
    :param **kwargs: any data you want to emit with the event
    :return: None
    """
    ee.emit(event, self, **kwargs)

pin_message(channel: Channel | str, ts: str) -> AsyncSlackResponse async

Pin message

Pin a message in a channel

Parameters:

Name	Type	Description	Default
channel	Channel | str	channel to pin the message in	required
ts	str	timestamp of the message to pin	required

Returns:

Type	Description
AsyncSlackResponse	response from the Slack Web API

Source code in machine/plugins/base.py

async def pin_message(self, channel: Channel | str, ts: str) -> AsyncSlackResponse:
    """Pin message

    Pin a message in a channel

    :param channel: channel to pin the message in
    :param ts: timestamp of the message to pin
    :return: response from the Slack Web API
    """
    return await self._client.pin_message(channel, ts)

unpin_message(channel: Channel | str, ts: str) -> AsyncSlackResponse async

Unpin message

Unpin a message that was previously pinned in a channel

Parameters:

Name	Type	Description	Default
channel	Channel | str	channel where the message is pinned that needs to be unpinned	required
ts	str	timestamp of the message to unpin	required

Returns:

Type	Description
AsyncSlackResponse	response from the Slack Web API

Source code in machine/plugins/base.py

async def unpin_message(self, channel: Channel | str, ts: str) -> AsyncSlackResponse:
    """Unpin message

    Unpin a message that was previously pinned in a channel

    :param channel: channel where the message is pinned that needs to be unpinned
    :param ts: timestamp of the message to unpin
    :return: response from the Slack Web API
    """
    return await self._client.unpin_message(channel, ts)

set_topic(channel: Channel | str, topic: str, **kwargs: Any) -> AsyncSlackResponse async

Set channel topic

Set or update topic for the channel

Parameters:

Name	Type	Description	Default
channel	Channel | str	channel where topic needs to be set or updated	required
topic	str	topic for the channel (slack does not support formatting for topics)	required

Returns:

Type	Description
AsyncSlackResponse	response from the Slack Web API

Source code in machine/plugins/base.py

async def set_topic(self, channel: Channel | str, topic: str, **kwargs: Any) -> AsyncSlackResponse:
    """Set channel topic

    Set or update topic for the channel

    :param channel: channel where topic needs to be set or updated
    :param topic: topic for the channel (slack does not support formatting for topics)
    :return: response from the Slack Web API
    """
    return await self._client.set_topic(channel, topic, **kwargs)

open_modal(trigger_id: str, view: dict | View, **kwargs: Any) -> AsyncSlackResponse async

Open a modal dialog

Open a modal dialog in response to a user action. The modal dialog can be used to collect information from the user, or to display information to the user.

Parameters:

Name	Type	Description	Default
trigger_id	str	trigger id is provided by Slack when a user action is performed, such as a slash command or a button click	required
view	dict | View	view definition for the modal dialog	required

Returns:

Type	Description
AsyncSlackResponse	response from the Slack Web API

Source code in machine/plugins/base.py

async def open_modal(self, trigger_id: str, view: dict | View, **kwargs: Any) -> AsyncSlackResponse:
    """Open a modal dialog

    Open a modal dialog in response to a user action. The modal dialog can be used to collect
    information from the user, or to display information to the user.

    :param trigger_id: trigger id is provided by Slack when a user action is performed, such as a slash command
        or a button click
    :param view: view definition for the modal dialog
    :return: response from the Slack Web API
    """
    return await self._client.web_client.views_open(trigger_id=trigger_id, view=view, **kwargs)

push_modal(trigger_id: str, view: dict | View, **kwargs: Any) -> AsyncSlackResponse async

Push a new view onto the stack of a modal that was already opened

Push a new view onto the stack of a modal that was already opened by a open_modal call. At most 3 views can be active in a modal at the same time. For more information on the lifecycle of modals, refer to the relevant Slack documentation

Parameters:

Name	Type	Description	Default
trigger_id	str	trigger id is provided by Slack when a user action is performed, such as a slash command or a button click	required
view	dict | View	view definition for the modal dialog	required

Returns:

Type	Description
AsyncSlackResponse	response from the Slack Web API

Source code in machine/plugins/base.py

async def push_modal(self, trigger_id: str, view: dict | View, **kwargs: Any) -> AsyncSlackResponse:
    """Push a new view onto the stack of a modal that was already opened

    Push a new view onto the stack of a modal that was already opened by a open_modal call. At most 3 views can be
    active in a modal at the same time. For more information on the lifecycle of modals, refer to the
    [relevant Slack documentation](https://api.slack.com/surfaces/modals)

    :param trigger_id: trigger id is provided by Slack when a user action is performed, such as a slash command
        or a button click
    :param view: view definition for the modal dialog
    :return: response from the Slack Web API
    """
    return await self._client.push_modal(trigger_id=trigger_id, view=view, **kwargs)

update_modal(view: dict | View, view_id: str | None = None, external_id: str | None = None, hash: str | None = None, **kwargs: Any) -> AsyncSlackResponse async

Update a modal dialog

Update a modal dialog that was previously opened. You can update the view by providing the view_id or the external_id of the modal. external_id has precedence over view_id, but at least one needs to be provided. You can also provide a hash of the view that you want to update to prevent race conditions.

Parameters:

Name	Type	Description	Default
view	dict | View	view definition for the modal dialog	required
view_id	str | None	id of the view to update	None
external_id	str | None	external id of the view to update	None
hash	str | None	hash of the view to update	None

Returns:

Type	Description
AsyncSlackResponse	response from the Slack Web API

Source code in machine/plugins/base.py

async def update_modal(
    self,
    view: dict | View,
    view_id: str | None = None,
    external_id: str | None = None,
    hash: str | None = None,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Update a modal dialog

    Update a modal dialog that was previously opened. You can update the view by providing the view_id or the
    external_id of the modal. external_id has precedence over view_id, but at least one needs to be provided.
    You can also provide a hash of the view that you want to update to prevent race conditions.

    :param view: view definition for the modal dialog
    :param view_id: id of the view to update
    :param external_id: external id of the view to update
    :param hash: hash of the view to update
    :return: response from the Slack Web API
    """
    return await self._client.update_modal(view=view, view_id=view_id, external_id=external_id, hash=hash, **kwargs)

publish_home_tab(user: User | str, view: dict | View, hash: str | None = None, **kwargs: Any) -> AsyncSlackResponse async

Publish a view to the home tab of a user

Publish a view to the home tab of a user. The view will be visible to the user when they open the home tab of your Slack app. This method can be used both to publish a new view for the home tab or update an existing view. You can provide a hash of the view that you want to update to prevent race conditions.

Note: be careful with the use of this method, as you might be overwriting the user's home tab that was set by another Slack Machine plugin enabled in your bot.

Parameters:

Name	Type	Description	Default
user	User | str	user for whom to publish or update the home tab	required
view	dict | View	view definition for the home tab	required
hash	str | None	hash of the view to update	None

Returns:

Type	Description
AsyncSlackResponse	response from the Slack Web API

Source code in machine/plugins/base.py

async def publish_home_tab(
    self, user: User | str, view: dict | View, hash: str | None = None, **kwargs: Any
) -> AsyncSlackResponse:
    """Publish a view to the home tab of a user

    Publish a view to the home tab of a user. The view will be visible to the user when they open the home tab of
    your Slack app. This method can be used both to publish a new view for the home tab or update an existing view.
    You can provide a hash of the view that you want to update to prevent race conditions.

    Note: be careful with the use of this method, as you might be overwriting the user's home tab that was set by
    another Slack Machine plugin enabled in your bot.

    :param user: user for whom to publish or update the home tab
    :param view: view definition for the home tab
    :param hash: hash of the view to update
    :return: response from the Slack Web API
    """
    return await self._client.publish_home_tab(user=user, view=view, hash=hash, **kwargs)

---

machine.plugins.message.Message

A message that was received by the bot

This class represents a message that was received by the bot and passed to one or more plugins. It contains the message (text) itself, and metadata about the message, such as the sender of the message, the channel the message was sent to.

The Message class also contains convenience methods for replying to the message in the right channel, replying to the sender, etc.

Source code in machine/plugins/message.py

class Message:
    """A message that was received by the bot

    This class represents a message that was received by the bot and passed to one or more
    plugins. It contains the message (text) itself, and metadata about the message, such as the
    sender of the message, the channel the message was sent to.

    The ``Message`` class also contains convenience methods for replying to the message in the
    right channel, replying to the sender, etc.
    """

    # TODO: create proper class for msg_event
    def __init__(self, client: SlackClient, msg_event: dict[str, Any]):
        self._client = client
        self._msg_event = msg_event

    @property
    def sender(self) -> User:
        """The sender of the message

        :return: the User the message was sent by
        """
        return self._client.users[self._msg_event["user"]]

    @property
    def channel(self) -> Channel:
        """The channel the message was sent to

        :return: the Channel the message was sent to
        """
        return self._client.channels[self._msg_event["channel"]]

    @property
    def is_dm(self) -> bool:
        channel_id = self._msg_event["channel"]
        return not (channel_id.startswith("C") or channel_id.startswith("G"))

    @property
    def text(self) -> str:
        """The body of the actual message

        :return: the body (text) of the actual message
        """
        return self._msg_event["text"]

    @property
    def at_sender(self) -> str:
        """The sender of the message formatted as mention

        :return: a string representation of the sender of the message, formatted as `mention`_,
            to be used in messages

        .. _mention: https://api.slack.com/docs/message-formatting#linking_to_channels_and_users
        """
        return self.sender.fmt_mention()

    async def say(
        self,
        text: str | None = None,
        attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
        blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
        thread_ts: str | None = None,
        ephemeral: bool = False,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Send a new message to the channel the original message was received in

        Send a new message to the channel the original message was received in, using the WebAPI.
        Allows for rich formatting using `blocks`_ and/or `attachments`_. You can provide blocks
        and attachments as Python dicts or you can use the `convenient classes`_ that the
        underlying slack client provides.
        Can also reply to a thread and send an ephemeral message only visible to the sender of the
        original message. Ephemeral messages and threaded messages are mutually exclusive, and
        ``ephemeral`` takes precedence over ``thread_ts``
        Any extra kwargs you provide, will be passed on directly to the `chat.postMessage`_ or
        `chat.postEphemeral`_ request.

        .. _attachments: https://api.slack.com/docs/message-attachments
        .. _blocks: https://api.slack.com/reference/block-kit/blocks
        .. _convenient classes:
            https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

        :param text: message text
        :param attachments: optional attachments (see `attachments`_)
        :param blocks: optional blocks (see `blocks`_)
        :param thread_ts: optional timestamp of thread, to send a message in that thread
        :param ephemeral: ``True/False`` wether to send the message as an ephemeral message, only
            visible to the sender of the original message
        :return: Dictionary deserialized from `chat.postMessage`_ request, or `chat.postEphemeral`_
            if `ephemeral` is True.

        .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage
        .. _chat.postEphemeral: https://api.slack.com/methods/chat.postEphemeral
        """
        ephemeral_user = self.sender.id if ephemeral else None

        return await self._client.send(
            self.channel.id,
            text=text,
            attachments=attachments,
            blocks=blocks,
            thread_ts=thread_ts,
            ephemeral_user=ephemeral_user,
            **kwargs,
        )

    async def say_scheduled(
        self,
        when: datetime,
        text: str,
        attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
        blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
        thread_ts: str | None = None,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Schedule a message

        This is the scheduled version of :py:meth:`~machine.plugins.base.Message.say`.
        It behaves the same, but will send the message at the scheduled time.

        :param when: when you want the message to be sent, as :py:class:`datetime.datetime` instance
        :param text: message text
        :param attachments: optional attachments (see `attachments`_)
        :param blocks: optional blocks (see `blocks`_)
        :param thread_ts: optional timestamp of thread, to send a message in that thread
        :return: None

        .. _attachments: https://api.slack.com/docs/message-attachments
        .. _blocks: https://api.slack.com/reference/block-kit/blocks
        """
        return await self._client.send_scheduled(
            when,
            self.channel.id,
            text=text,
            attachments=attachments,
            blocks=blocks,
            thread_ts=thread_ts,
            **kwargs,
        )

    async def reply(
        self,
        text: str | None = None,
        attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
        blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
        in_thread: bool = False,
        ephemeral: bool = False,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Reply to the sender of the original message

        Reply to the sender of the original message with a new message, mentioning that user. Rich
        formatting using `blocks`_ and/or `attachments`_ is possible. You can provide blocks
        and attachments as Python dicts or you can use the `convenient classes`_ that the
        underlying slack client provides.
        Can also reply to a thread and send an ephemeral message only visible to the sender of the
        original message. In the case of in-thread response, the sender of the original message
        will not be mentioned. Ephemeral messages and threaded messages are mutually exclusive,
        and ``ephemeral`` takes precedence over ``in_thread``
        Any extra kwargs you provide, will be passed on directly to the `chat.postMessage`_ or
        `chat.postEphemeral`_ request.

        .. _attachments: https://api.slack.com/docs/message-attachments
        .. _blocks: https://api.slack.com/reference/block-kit/blocks
        .. _convenient classes:
            https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

        :param text: message text
        :param attachments: optional attachments (see `attachments`_)
        :param blocks: optional blocks (see `blocks`_)
        :param in_thread: ``True/False`` wether to reply to the original message in-thread
        :param ephemeral: ``True/False`` wether to send the message as an ephemeral message, only
            visible to the sender of the original message
        :return: Dictionary deserialized from `chat.postMessage`_ request, or `chat.postEphemeral`_
            if `ephemeral` is True.

        .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage
        .. _chat.postEphemeral: https://api.slack.com/methods/chat.postEphemeral
        """
        if in_thread and not ephemeral:
            return await self.say(text, attachments=attachments, blocks=blocks, thread_ts=self.ts, **kwargs)
        else:
            text = self._create_reply(text)
            return await self.say(text, attachments=attachments, blocks=blocks, ephemeral=ephemeral, **kwargs)

    async def reply_scheduled(
        self,
        when: datetime,
        text: str,
        attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
        blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
        in_thread: bool = False,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Schedule a reply and send it

        This is the scheduled version of :py:meth:`~machine.plugins.base.Message.reply`.
        It behaves the same, but will send the reply at the scheduled time.

        :param when: when you want the message to be sent, as :py:class:`datetime.datetime` instance
        :param text: message text
        :param attachments: optional attachments (see `attachments`_)
        :param blocks: optional blocks (see `blocks`_)
        :param in_thread: ``True/False`` wether to reply to the original message in-thread
        :return: None

        .. _attachments: https://api.slack.com/docs/message-attachments
        .. _blocks: https://api.slack.com/reference/block-kit/blocks
        """
        if in_thread:
            return await self.say_scheduled(
                when, text, attachments=attachments, blocks=blocks, thread_ts=self.ts, **kwargs
            )
        else:
            text = cast(str, self._create_reply(text))
            return await self.say_scheduled(when, text, attachments=attachments, blocks=blocks, **kwargs)

    async def reply_dm(
        self,
        text: str | None = None,
        attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
        blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Reply to the sender of the original message with a DM

        Reply in a Direct Message to the sender of the original message by opening a DM channel and
        sending a message to it. Allows for rich formatting using `blocks`_ and/or `attachments`_.
        You can provide blocks and attachments as Python dicts or you can use the
        `convenient classes`_ that the underlying slack client provides.
        Any extra kwargs you provide, will be passed on directly to the `chat.postMessage`_ request.

        .. _attachments: https://api.slack.com/docs/message-attachments
        .. _blocks: https://api.slack.com/reference/block-kit/blocks
        .. _convenient classes:
            https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

        :param text: message text
        :param attachments: optional attachments (see `attachments`_)
        :param blocks: optional blocks (see `blocks`_)
        :return: Dictionary deserialized from `chat.postMessage`_ request.

        .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage
        """
        return await self._client.send_dm(self.sender.id, text, attachments=attachments, blocks=blocks, **kwargs)

    async def reply_dm_scheduled(
        self,
        when: datetime,
        text: str,
        attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
        blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Schedule a DM reply and send it

        This is the scheduled version of :py:meth:`~machine.plugins.base.Message.reply_dm`.
        It behaves the same, but will send the DM at the scheduled time.

        :param when: when you want the message to be sent, as :py:class:`datetime.datetime` instance
        :param text: message text
        :param attachments: optional attachments (see `attachments`_)
        :param blocks: optional blocks (see `blocks`_)
        :return: None

        .. _attachments: https://api.slack.com/docs/message-attachments
        .. _blocks: https://api.slack.com/reference/block-kit/blocks
        """
        return await self._client.send_dm_scheduled(
            when, self.sender.id, text=text, attachments=attachments, blocks=blocks, **kwargs
        )

    async def react(self, emoji: str) -> AsyncSlackResponse:
        """React to the original message

        Add a reaction to the original message

        :param emoji: what emoji to react with (should be a string, like 'angel', 'thumbsup', etc.)
        :return: Dictionary deserialized from `reactions.add`_ request.

        .. _reactions.add: https://api.slack.com/methods/reactions.add
        """
        return await self._client.react(self.channel.id, self._msg_event["ts"], emoji)

    def _create_reply(self, text: str | None) -> str | None:
        if not self.is_dm and text is not None:
            return f"{self.at_sender}: {text}"
        else:
            return text

    @property
    def ts(self) -> str:
        """The timestamp of the message

        :return: the timestamp of the message
        """
        return self._msg_event["ts"]

    @property
    def in_thread(self) -> bool:
        """Is message in a thread

        :return: bool
        """
        return "thread_ts" in self._msg_event

    async def pin_message(self) -> AsyncSlackResponse:
        """Pin message

        Pin the current message in the channel it was posted in
        """
        return await self._client.pin_message(self.channel, self.ts)

    def __str__(self) -> str:
        if self.channel.is_im:
            message = f"Message '{self.text}', sent by user @{self.sender.profile.real_name} in DM"
        else:
            message = (
                f"Message '{self.text}', sent by user @{self.sender.profile.real_name} in channel #{self.channel.name}"
            )
        return message

    def __repr__(self) -> str:
        return f"Message(text={self.text!r}, sender={self.sender.profile.real_name!r}, channel={self.channel.name!r})"

sender: User property

The sender of the message

Returns:

Type	Description
User	the User the message was sent by

channel: Channel property

The channel the message was sent to

Returns:

Type	Description
Channel	the Channel the message was sent to

text: str property

The body of the actual message

Returns:

Type	Description
str	the body (text) of the actual message

at_sender: str property

The sender of the message formatted as mention

Returns:

Type	Description
str	a string representation of the sender of the message, formatted as mention_, to be used in messages .. _mention: https://api.slack.com/docs/message-formatting#linking_to_channels_and_users

ts: str property

The timestamp of the message

Returns:

Type	Description
str	the timestamp of the message

in_thread: bool property

Is message in a thread

Returns:

Type	Description
bool	bool

say(text: str | None = None, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, thread_ts: str | None = None, ephemeral: bool = False, **kwargs: Any) -> AsyncSlackResponse async

Send a new message to the channel the original message was received in

Send a new message to the channel the original message was received in, using the WebAPI. Allows for rich formatting using blocks and/or attachments. You can provide blocks and attachments as Python dicts or you can use the convenient classes that the underlying slack client provides. Can also reply to a thread and send an ephemeral message only visible to the sender of the original message. Ephemeral messages and threaded messages are mutually exclusive, and ephemeral takes precedence over thread_ts Any extra kwargs you provide, will be passed on directly to the chat.postMessage or chat.postEphemeral_ request.

.. _attachments: https://api.slack.com/docs/message-attachments .. _blocks: https://api.slack.com/reference/block-kit/blocks .. _convenient classes: https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

Parameters:

Name	Type	Description	Default
text	str | None	message text	None
attachments	Sequence[Attachment] | Sequence[dict[str, Any]] | None	optional attachments (see attachments_)	None
blocks	Sequence[Block] | Sequence[dict[str, Any]] | None	optional blocks (see blocks_)	None
thread_ts	str | None	optional timestamp of thread, to send a message in that thread	None
ephemeral	bool	True/False wether to send the message as an ephemeral message, only visible to the sender of the original message	False

Returns:

Type	Description
AsyncSlackResponse	Dictionary deserialized from chat.postMessage request, or chat.postEphemeral if ephemeral is True. .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage .. _chat.postEphemeral: https://api.slack.com/methods/chat.postEphemeral

Source code in machine/plugins/message.py

async def say(
    self,
    text: str | None = None,
    attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
    blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
    thread_ts: str | None = None,
    ephemeral: bool = False,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Send a new message to the channel the original message was received in

    Send a new message to the channel the original message was received in, using the WebAPI.
    Allows for rich formatting using `blocks`_ and/or `attachments`_. You can provide blocks
    and attachments as Python dicts or you can use the `convenient classes`_ that the
    underlying slack client provides.
    Can also reply to a thread and send an ephemeral message only visible to the sender of the
    original message. Ephemeral messages and threaded messages are mutually exclusive, and
    ``ephemeral`` takes precedence over ``thread_ts``
    Any extra kwargs you provide, will be passed on directly to the `chat.postMessage`_ or
    `chat.postEphemeral`_ request.

    .. _attachments: https://api.slack.com/docs/message-attachments
    .. _blocks: https://api.slack.com/reference/block-kit/blocks
    .. _convenient classes:
        https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

    :param text: message text
    :param attachments: optional attachments (see `attachments`_)
    :param blocks: optional blocks (see `blocks`_)
    :param thread_ts: optional timestamp of thread, to send a message in that thread
    :param ephemeral: ``True/False`` wether to send the message as an ephemeral message, only
        visible to the sender of the original message
    :return: Dictionary deserialized from `chat.postMessage`_ request, or `chat.postEphemeral`_
        if `ephemeral` is True.

    .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage
    .. _chat.postEphemeral: https://api.slack.com/methods/chat.postEphemeral
    """
    ephemeral_user = self.sender.id if ephemeral else None

    return await self._client.send(
        self.channel.id,
        text=text,
        attachments=attachments,
        blocks=blocks,
        thread_ts=thread_ts,
        ephemeral_user=ephemeral_user,
        **kwargs,
    )

say_scheduled(when: datetime, text: str, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, thread_ts: str | None = None, **kwargs: Any) -> AsyncSlackResponse async

Schedule a message

This is the scheduled version of :py:meth:~machine.plugins.base.Message.say. It behaves the same, but will send the message at the scheduled time.

Parameters:

Name	Type	Description	Default
when	datetime	when you want the message to be sent, as :py:class:datetime.datetime instance	required
text	str	message text	required
attachments	Sequence[Attachment] | Sequence[dict[str, Any]] | None	optional attachments (see attachments_)	None
blocks	Sequence[Block] | Sequence[dict[str, Any]] | None	optional blocks (see blocks_)	None
thread_ts	str | None	optional timestamp of thread, to send a message in that thread	None

Returns:

Type	Description
AsyncSlackResponse	None .. _attachments: https://api.slack.com/docs/message-attachments .. _blocks: https://api.slack.com/reference/block-kit/blocks

Source code in machine/plugins/message.py

async def say_scheduled(
    self,
    when: datetime,
    text: str,
    attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
    blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
    thread_ts: str | None = None,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Schedule a message

    This is the scheduled version of :py:meth:`~machine.plugins.base.Message.say`.
    It behaves the same, but will send the message at the scheduled time.

    :param when: when you want the message to be sent, as :py:class:`datetime.datetime` instance
    :param text: message text
    :param attachments: optional attachments (see `attachments`_)
    :param blocks: optional blocks (see `blocks`_)
    :param thread_ts: optional timestamp of thread, to send a message in that thread
    :return: None

    .. _attachments: https://api.slack.com/docs/message-attachments
    .. _blocks: https://api.slack.com/reference/block-kit/blocks
    """
    return await self._client.send_scheduled(
        when,
        self.channel.id,
        text=text,
        attachments=attachments,
        blocks=blocks,
        thread_ts=thread_ts,
        **kwargs,
    )

reply(text: str | None = None, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, in_thread: bool = False, ephemeral: bool = False, **kwargs: Any) -> AsyncSlackResponse async

Reply to the sender of the original message

Reply to the sender of the original message with a new message, mentioning that user. Rich formatting using blocks and/or attachments is possible. You can provide blocks and attachments as Python dicts or you can use the convenient classes that the underlying slack client provides. Can also reply to a thread and send an ephemeral message only visible to the sender of the original message. In the case of in-thread response, the sender of the original message will not be mentioned. Ephemeral messages and threaded messages are mutually exclusive, and ephemeral takes precedence over in_thread Any extra kwargs you provide, will be passed on directly to the chat.postMessage or chat.postEphemeral_ request.

.. _attachments: https://api.slack.com/docs/message-attachments .. _blocks: https://api.slack.com/reference/block-kit/blocks .. _convenient classes: https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

Parameters:

Name	Type	Description	Default
text	str | None	message text	None
attachments	Sequence[Attachment] | Sequence[dict[str, Any]] | None	optional attachments (see attachments_)	None
blocks	Sequence[Block] | Sequence[dict[str, Any]] | None	optional blocks (see blocks_)	None
in_thread	bool	True/False wether to reply to the original message in-thread	False
ephemeral	bool	True/False wether to send the message as an ephemeral message, only visible to the sender of the original message	False

Returns:

Type	Description
AsyncSlackResponse	Dictionary deserialized from chat.postMessage request, or chat.postEphemeral if ephemeral is True. .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage .. _chat.postEphemeral: https://api.slack.com/methods/chat.postEphemeral

Source code in machine/plugins/message.py

async def reply(
    self,
    text: str | None = None,
    attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
    blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
    in_thread: bool = False,
    ephemeral: bool = False,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Reply to the sender of the original message

    Reply to the sender of the original message with a new message, mentioning that user. Rich
    formatting using `blocks`_ and/or `attachments`_ is possible. You can provide blocks
    and attachments as Python dicts or you can use the `convenient classes`_ that the
    underlying slack client provides.
    Can also reply to a thread and send an ephemeral message only visible to the sender of the
    original message. In the case of in-thread response, the sender of the original message
    will not be mentioned. Ephemeral messages and threaded messages are mutually exclusive,
    and ``ephemeral`` takes precedence over ``in_thread``
    Any extra kwargs you provide, will be passed on directly to the `chat.postMessage`_ or
    `chat.postEphemeral`_ request.

    .. _attachments: https://api.slack.com/docs/message-attachments
    .. _blocks: https://api.slack.com/reference/block-kit/blocks
    .. _convenient classes:
        https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

    :param text: message text
    :param attachments: optional attachments (see `attachments`_)
    :param blocks: optional blocks (see `blocks`_)
    :param in_thread: ``True/False`` wether to reply to the original message in-thread
    :param ephemeral: ``True/False`` wether to send the message as an ephemeral message, only
        visible to the sender of the original message
    :return: Dictionary deserialized from `chat.postMessage`_ request, or `chat.postEphemeral`_
        if `ephemeral` is True.

    .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage
    .. _chat.postEphemeral: https://api.slack.com/methods/chat.postEphemeral
    """
    if in_thread and not ephemeral:
        return await self.say(text, attachments=attachments, blocks=blocks, thread_ts=self.ts, **kwargs)
    else:
        text = self._create_reply(text)
        return await self.say(text, attachments=attachments, blocks=blocks, ephemeral=ephemeral, **kwargs)

reply_scheduled(when: datetime, text: str, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, in_thread: bool = False, **kwargs: Any) -> AsyncSlackResponse async

Schedule a reply and send it

This is the scheduled version of :py:meth:~machine.plugins.base.Message.reply. It behaves the same, but will send the reply at the scheduled time.

Parameters:

Name	Type	Description	Default
when	datetime	when you want the message to be sent, as :py:class:datetime.datetime instance	required
text	str	message text	required
attachments	Sequence[Attachment] | Sequence[dict[str, Any]] | None	optional attachments (see attachments_)	None
blocks	Sequence[Block] | Sequence[dict[str, Any]] | None	optional blocks (see blocks_)	None
in_thread	bool	True/False wether to reply to the original message in-thread	False

Returns:

Type	Description
AsyncSlackResponse	None .. _attachments: https://api.slack.com/docs/message-attachments .. _blocks: https://api.slack.com/reference/block-kit/blocks

Source code in machine/plugins/message.py

async def reply_scheduled(
    self,
    when: datetime,
    text: str,
    attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
    blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
    in_thread: bool = False,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Schedule a reply and send it

    This is the scheduled version of :py:meth:`~machine.plugins.base.Message.reply`.
    It behaves the same, but will send the reply at the scheduled time.

    :param when: when you want the message to be sent, as :py:class:`datetime.datetime` instance
    :param text: message text
    :param attachments: optional attachments (see `attachments`_)
    :param blocks: optional blocks (see `blocks`_)
    :param in_thread: ``True/False`` wether to reply to the original message in-thread
    :return: None

    .. _attachments: https://api.slack.com/docs/message-attachments
    .. _blocks: https://api.slack.com/reference/block-kit/blocks
    """
    if in_thread:
        return await self.say_scheduled(
            when, text, attachments=attachments, blocks=blocks, thread_ts=self.ts, **kwargs
        )
    else:
        text = cast(str, self._create_reply(text))
        return await self.say_scheduled(when, text, attachments=attachments, blocks=blocks, **kwargs)

reply_dm(text: str | None = None, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, **kwargs: Any) -> AsyncSlackResponse async

Reply to the sender of the original message with a DM

Reply in a Direct Message to the sender of the original message by opening a DM channel and sending a message to it. Allows for rich formatting using blocks and/or attachments. You can provide blocks and attachments as Python dicts or you can use the convenient classes that the underlying slack client provides. Any extra kwargs you provide, will be passed on directly to the chat.postMessage request.

.. _attachments: https://api.slack.com/docs/message-attachments .. _blocks: https://api.slack.com/reference/block-kit/blocks .. _convenient classes: https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

Parameters:

Name	Type	Description	Default
text	str | None	message text	None
attachments	Sequence[Attachment] | Sequence[dict[str, Any]] | None	optional attachments (see attachments_)	None
blocks	Sequence[Block] | Sequence[dict[str, Any]] | None	optional blocks (see blocks_)	None

Returns:

Type	Description
AsyncSlackResponse	Dictionary deserialized from chat.postMessage_ request. .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage

Source code in machine/plugins/message.py

async def reply_dm(
    self,
    text: str | None = None,
    attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
    blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Reply to the sender of the original message with a DM

    Reply in a Direct Message to the sender of the original message by opening a DM channel and
    sending a message to it. Allows for rich formatting using `blocks`_ and/or `attachments`_.
    You can provide blocks and attachments as Python dicts or you can use the
    `convenient classes`_ that the underlying slack client provides.
    Any extra kwargs you provide, will be passed on directly to the `chat.postMessage`_ request.

    .. _attachments: https://api.slack.com/docs/message-attachments
    .. _blocks: https://api.slack.com/reference/block-kit/blocks
    .. _convenient classes:
        https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

    :param text: message text
    :param attachments: optional attachments (see `attachments`_)
    :param blocks: optional blocks (see `blocks`_)
    :return: Dictionary deserialized from `chat.postMessage`_ request.

    .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage
    """
    return await self._client.send_dm(self.sender.id, text, attachments=attachments, blocks=blocks, **kwargs)

reply_dm_scheduled(when: datetime, text: str, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, **kwargs: Any) -> AsyncSlackResponse async

Schedule a DM reply and send it

This is the scheduled version of :py:meth:~machine.plugins.base.Message.reply_dm. It behaves the same, but will send the DM at the scheduled time.

Parameters:

Name	Type	Description	Default
when	datetime	when you want the message to be sent, as :py:class:datetime.datetime instance	required
text	str	message text	required
attachments	Sequence[Attachment] | Sequence[dict[str, Any]] | None	optional attachments (see attachments_)	None
blocks	Sequence[Block] | Sequence[dict[str, Any]] | None	optional blocks (see blocks_)	None

Returns:

Type	Description
AsyncSlackResponse	None .. _attachments: https://api.slack.com/docs/message-attachments .. _blocks: https://api.slack.com/reference/block-kit/blocks

Source code in machine/plugins/message.py

async def reply_dm_scheduled(
    self,
    when: datetime,
    text: str,
    attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
    blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Schedule a DM reply and send it

    This is the scheduled version of :py:meth:`~machine.plugins.base.Message.reply_dm`.
    It behaves the same, but will send the DM at the scheduled time.

    :param when: when you want the message to be sent, as :py:class:`datetime.datetime` instance
    :param text: message text
    :param attachments: optional attachments (see `attachments`_)
    :param blocks: optional blocks (see `blocks`_)
    :return: None

    .. _attachments: https://api.slack.com/docs/message-attachments
    .. _blocks: https://api.slack.com/reference/block-kit/blocks
    """
    return await self._client.send_dm_scheduled(
        when, self.sender.id, text=text, attachments=attachments, blocks=blocks, **kwargs
    )

react(emoji: str) -> AsyncSlackResponse async

React to the original message

Add a reaction to the original message

Parameters:

Name	Type	Description	Default
emoji	str	what emoji to react with (should be a string, like 'angel', 'thumbsup', etc.)	required

Returns:

Type	Description
AsyncSlackResponse	Dictionary deserialized from reactions.add_ request. .. _reactions.add: https://api.slack.com/methods/reactions.add

Source code in machine/plugins/message.py

async def react(self, emoji: str) -> AsyncSlackResponse:
    """React to the original message

    Add a reaction to the original message

    :param emoji: what emoji to react with (should be a string, like 'angel', 'thumbsup', etc.)
    :return: Dictionary deserialized from `reactions.add`_ request.

    .. _reactions.add: https://api.slack.com/methods/reactions.add
    """
    return await self._client.react(self.channel.id, self._msg_event["ts"], emoji)

pin_message() -> AsyncSlackResponse async

Pin message

Pin the current message in the channel it was posted in

Source code in machine/plugins/message.py

async def pin_message(self) -> AsyncSlackResponse:
    """Pin message

    Pin the current message in the channel it was posted in
    """
    return await self._client.pin_message(self.channel, self.ts)

machine.plugins.command.Command

A Slack command that was received by the bot

This class represents a Slack command that was received by the bot and passed to a plugin. It contains the text that was included when the command was invoked, and metadata about the command, such as the user that invoked the command, the channel the command was invoked in.

The Command class also contains convenience methods for sending messages in the right channel, opening modals etc.

Source code in machine/plugins/command.py

class Command:
    """A Slack command that was received by the bot

    This class represents a Slack command that was received by the bot and passed to a plugin.
    It contains the text that was included when the command was invoked, and metadata about
    the command, such as the user that invoked the command, the channel the command was invoked
    in.

    The `Command` class also contains convenience methods for sending messages in the right
    channel, opening modals etc.
    """

    # TODO: create proper class for cmd_event
    def __init__(self, client: SlackClient, cmd_payload: dict[str, Any]):
        self._client = client
        self._cmd_payload = cmd_payload
        self._webhook_client = AsyncWebhookClient(self._cmd_payload["response_url"])

    @property
    def sender(self) -> User:
        """The sender of the message

        :return: the User the message was sent by
        """
        return self._client.users[self._cmd_payload["user_id"]]

    @property
    def channel(self) -> Channel:
        """The channel the message was sent to

        :return: the Channel the message was sent to
        """
        return self._client.channels[self._cmd_payload["channel_id"]]

    @property
    def is_dm(self) -> bool:
        channel_id = self._cmd_payload["channel_id"]
        return not (channel_id.startswith("C") or channel_id.startswith("G"))

    @property
    def text(self) -> str:
        """The body of the actual message

        :return: the body (text) of the actual message
        """
        return self._cmd_payload["text"]

    @property
    def command(self) -> str:
        """The command that was invoked

        :return: the command that was invoked
        """
        return self._cmd_payload["command"]

    @property
    def response_url(self) -> str:
        """The response url associated with the command

        This is a unique url for this specific command invocation.
        It can be used for sending messages in response to the command.
        This can only be used 5 times within 30 minutes of receiving the payload.

        :return: the response url associated with the command
        """
        return self._cmd_payload["response_url"]

    @property
    def trigger_id(self) -> str:
        """The trigger id associated with the command

        The trigger id can be used to trigger modals

        :return: the trigger id associated with the command
        """
        return self._cmd_payload["trigger_id"]

    async def say(
        self,
        text: str | None = None,
        attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
        blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
        ephemeral: bool = True,
        **kwargs: Any,
    ) -> WebhookResponse:
        """Send a new message to the channel the command was invoked in

        Send a new message to the channel the command was invoked in, using the response_url as a webhook.
        Allows for rich formatting using [blocks] and/or [attachments] . You can provide blocks
        and attachments as Python dicts or you can use the [convenient classes] that the
        underlying slack client provides.
        This will send an ephemeral message by default, only visible to the user that invoked the command.
        You can set `ephemeral` to `False` to make the message visible to everyone in the channel
        Any extra kwargs you provide, will be passed on directly to `AsyncWebhookClient.send()`

        [attachments]: https://api.slack.com/docs/message-attachments
        [blocks]: https://api.slack.com/reference/block-kit/blocks
        [convenient classes]: https://github.com/slackapi/python-slack-sdk/tree/main/slack/web/classes

        :param text: message text
        :param attachments: optional attachments (see [attachments])
        :param blocks: optional blocks (see [blocks])
        :param ephemeral: `True/False` wether to send the message as an ephemeral message, only
            visible to the sender of the original message
        :return: Dictionary deserialized from `AsyncWebhookClient.send()`
        """
        response_type = "ephemeral" if ephemeral else "in_channel"

        return await self._webhook_client.send(
            text=text, attachments=attachments, blocks=blocks, response_type=response_type, **kwargs
        )

    async def open_modal(
        self,
        view: dict | View,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Open a modal in response to the command

        Open a modal in response to the command, using the trigger_id that was returned when the command was invoked.
        Any extra kwargs you provide, will be passed on directly to `AsyncWebClient.views_open()`

        Note: you have to call this method within 3 seconds of receiving the command payload.

        :param view: the view to open
        :return: Dictionary deserialized from `AsyncWebClient.views_open()`
        """
        return await self._client.open_modal(self.trigger_id, view, **kwargs)

sender: User property

The sender of the message

Returns:

Type	Description
User	the User the message was sent by

channel: Channel property

The channel the message was sent to

Returns:

Type	Description
Channel	the Channel the message was sent to

text: str property

The body of the actual message

Returns:

Type	Description
str	the body (text) of the actual message

command: str property

The command that was invoked

Returns:

Type	Description
str	the command that was invoked

response_url: str property

The response url associated with the command

This is a unique url for this specific command invocation. It can be used for sending messages in response to the command. This can only be used 5 times within 30 minutes of receiving the payload.

Returns:

Type	Description
str	the response url associated with the command

trigger_id: str property

The trigger id associated with the command

The trigger id can be used to trigger modals

Returns:

Type	Description
str	the trigger id associated with the command

say(text: str | None = None, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, ephemeral: bool = True, **kwargs: Any) -> WebhookResponse async

Send a new message to the channel the command was invoked in

Send a new message to the channel the command was invoked in, using the response_url as a webhook. Allows for rich formatting using blocks and/or attachments . You can provide blocks and attachments as Python dicts or you can use the convenient classes that the underlying slack client provides. This will send an ephemeral message by default, only visible to the user that invoked the command. You can set ephemeral to False to make the message visible to everyone in the channel Any extra kwargs you provide, will be passed on directly to AsyncWebhookClient.send()

Parameters:

Name	Type	Description	Default
text	str | None	message text	None
attachments	Sequence[Attachment] | Sequence[dict[str, Any]] | None	optional attachments (see [attachments])	None
blocks	Sequence[Block] | Sequence[dict[str, Any]] | None	optional blocks (see [blocks])	None
ephemeral	bool	True/False wether to send the message as an ephemeral message, only visible to the sender of the original message	True

Returns:

Type	Description
WebhookResponse	Dictionary deserialized from AsyncWebhookClient.send()

Source code in machine/plugins/command.py

async def say(
    self,
    text: str | None = None,
    attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
    blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
    ephemeral: bool = True,
    **kwargs: Any,
) -> WebhookResponse:
    """Send a new message to the channel the command was invoked in

    Send a new message to the channel the command was invoked in, using the response_url as a webhook.
    Allows for rich formatting using [blocks] and/or [attachments] . You can provide blocks
    and attachments as Python dicts or you can use the [convenient classes] that the
    underlying slack client provides.
    This will send an ephemeral message by default, only visible to the user that invoked the command.
    You can set `ephemeral` to `False` to make the message visible to everyone in the channel
    Any extra kwargs you provide, will be passed on directly to `AsyncWebhookClient.send()`

    [attachments]: https://api.slack.com/docs/message-attachments
    [blocks]: https://api.slack.com/reference/block-kit/blocks
    [convenient classes]: https://github.com/slackapi/python-slack-sdk/tree/main/slack/web/classes

    :param text: message text
    :param attachments: optional attachments (see [attachments])
    :param blocks: optional blocks (see [blocks])
    :param ephemeral: `True/False` wether to send the message as an ephemeral message, only
        visible to the sender of the original message
    :return: Dictionary deserialized from `AsyncWebhookClient.send()`
    """
    response_type = "ephemeral" if ephemeral else "in_channel"

    return await self._webhook_client.send(
        text=text, attachments=attachments, blocks=blocks, response_type=response_type, **kwargs
    )

open_modal(view: dict | View, **kwargs: Any) -> AsyncSlackResponse async

Open a modal in response to the command

Open a modal in response to the command, using the trigger_id that was returned when the command was invoked. Any extra kwargs you provide, will be passed on directly to AsyncWebClient.views_open()

Note: you have to call this method within 3 seconds of receiving the command payload.

Parameters:

Name	Type	Description	Default
view	dict | View	the view to open	required

Returns:

Type	Description
AsyncSlackResponse	Dictionary deserialized from AsyncWebClient.views_open()

Source code in machine/plugins/command.py

async def open_modal(
    self,
    view: dict | View,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Open a modal in response to the command

    Open a modal in response to the command, using the trigger_id that was returned when the command was invoked.
    Any extra kwargs you provide, will be passed on directly to `AsyncWebClient.views_open()`

    Note: you have to call this method within 3 seconds of receiving the command payload.

    :param view: the view to open
    :return: Dictionary deserialized from `AsyncWebClient.views_open()`
    """
    return await self._client.open_modal(self.trigger_id, view, **kwargs)

machine.plugins.block_action.BlockAction

A Slack block action that was received by the bot

This class represents a block action that was received by the bot and passed to a plugin. Block actions are actions that are triggered by interactions with blocks in Slack messages and modals. This class contains metadata about the block action, such as the action that happened that triggered this handler, the user that triggered the action, the state of the block when the action was triggered, the payload that was received when the action was triggered.

Source code in machine/plugins/block_action.py

class BlockAction:
    """A Slack block action that was received by the bot

    This class represents a block action that was received by the bot and passed to a plugin.
    Block actions are actions that are triggered by interactions with blocks in Slack messages and modals.
    This class contains metadata about the block action, such as the action that happened that triggered this handler,
    the user that triggered the action, the state of the block when the action was triggered, the payload that was
    received when the action was triggered.
    """

    payload: BlockActionsPayload
    triggered_action: Action

    def __init__(self, client: SlackClient, payload: BlockActionsPayload, triggered_action: Action):
        self._client = client
        self.payload = payload
        """The payload that was received by the bot when the action was triggered that this plugin method listens for"""
        self.triggered_action = triggered_action
        """The action that triggered this plugin method"""
        self._webhook_client = AsyncWebhookClient(self.payload.response_url) if self.payload.response_url else None

    @property
    def user(self) -> User:
        """The user that triggered the action

        :return: the user that triggered the action
        """
        return self._client.users[self.payload.user.id]

    @property
    def channel(self) -> Optional[Channel]:
        """The channel the action was triggered in

        :return: the channel the action was triggered in or None if the action was triggered in a modal
        """
        if self.payload.channel is None:
            return None
        return self._client.channels[self.payload.channel.id]

    @property
    def state(self) -> Optional[State]:
        """The state of the block when the action was triggered

        :return: the state of the block when the action was triggered
        """
        return self.payload.state

    @property
    def response_url(self) -> Optional[str]:
        """The response URL for the action

        :return: the response URL for the action or None if the action was triggered in a modal
        """
        return self.payload.response_url

    @property
    def trigger_id(self) -> str:
        """The trigger id associated with the action

        The trigger id can be used to open a modal

        :return: the trigger id for the action
        """
        return self.payload.trigger_id

    async def say(
        self,
        text: Optional[str] = None,
        attachments: Union[Sequence[Attachment], Sequence[dict[str, Any]], None] = None,
        blocks: Union[Sequence[Block], Sequence[dict[str, Any]], None] = None,
        ephemeral: bool = True,
        replace_original: bool = False,
        delete_original: bool = False,
        **kwargs: Any,
    ) -> Optional[WebhookResponse]:
        """Send a new message to the channel the block action was triggered in

        Send a new message to the channel the block action was triggered in, using the response_url as a webhook.
        If the block action happened in a modal, the response_url will be None and this method will not send a message
        but instead log a warning.
        Allows for rich formatting using [blocks] and/or [attachments] . You can provide blocks
        and attachments as Python dicts or you can use the [convenient classes] that the
        underlying slack client provides.
        This will send an ephemeral message by default, only visible to the user that triggered the action.
        You can set `ephemeral` to `False` to make the message visible to everyone in the channel.
        By default, Slack replaces the original message in which the action was triggered. This method overrides this
        behavior. If you want your message to replace the original, set replace_original to True.
        Any extra kwargs you provide, will be passed on directly to `AsyncWebhookClient.send()`

        [attachments]: https://api.slack.com/docs/message-attachments
        [blocks]: https://api.slack.com/reference/block-kit/blocks
        [convenient classes]: https://github.com/slackapi/python-slack-sdk/tree/main/slack/web/classes

        :param text: message text
        :param attachments: optional attachments (see [attachments])
        :param blocks: optional blocks (see [blocks])
        :param ephemeral: `True/False` wether to send the message as an ephemeral message, only
            visible to the user that initiated the action
        :param replace_original: `True/False` whether the message that contains the block from which the action was
            triggered should be replaced by this message
        :param delete_original: `True/False` whether the message that contains the block from which the action was
            triggered should be deleted. No other parameters should be provided.
        :return: Dictionary deserialized from `AsyncWebhookClient.send()`

        """
        if self._webhook_client is None:
            logger.warning(
                "response_url is None, cannot send message. This likely means the action was triggered in a modal."
            )
            return None

        response_type = "ephemeral" if ephemeral else "in_channel"
        return await self._webhook_client.send(
            text=text,
            attachments=attachments,
            blocks=blocks,
            response_type=response_type,
            replace_original=replace_original,
            delete_original=delete_original,
            **kwargs,
        )

    async def send_dm(
        self,
        text: str | None = None,
        attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
        blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Send a DM to the user that triggered the block action

        Send a Direct Message to the user that triggered the block action by opening a DM channel and
        sending a message to it. Allows for rich formatting using `blocks`_ and/or `attachments`_.
        Allows for rich formatting using [blocks] and/or [attachments] . You can provide blocks
        and attachments as Python dicts or you can use the [convenient classes] that the
        underlying slack client provides.
        Any extra kwargs you provide, will be passed on directly to the `chat.postMessage` request.

        [attachments]: https://api.slack.com/docs/message-attachments
        [blocks]: https://api.slack.com/reference/block-kit/blocks
        [convenient classes]: https://github.com/slackapi/python-slack-sdk/tree/main/slack/web/classes

        :param text: message text
        :param attachments: optional attachments (see [attachments])
        :param blocks: optional blocks (see [blocks])
        :return: Dictionary deserialized from [chat.postMessage] response.

        [chat.postMessage]: https://api.slack.com/methods/chat.postMessage
        """
        return await self._client.send_dm(self.user.id, text, attachments=attachments, blocks=blocks, **kwargs)

    async def open_modal(
        self,
        view: dict | View,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Open a modal in response to the block action

        Open a modal in response to the block action, using the trigger_id that was returned when the block action was
        triggered.
        Any extra kwargs you provide, will be passed on directly to `AsyncWebClient.views_open()`

        Note: you have to call this method within 3 seconds of receiving the block action payload.

        :param view: the view to open
        :return: Dictionary deserialized from `AsyncWebClient.views_open()`
        """
        return await self._client.open_modal(self.trigger_id, view, **kwargs)

payload: BlockActionsPayload = payload instance-attribute

The payload that was received by the bot when the action was triggered that this plugin method listens for

triggered_action: Action = triggered_action instance-attribute

The action that triggered this plugin method

user: User property

The user that triggered the action

Returns:

Type	Description
User	the user that triggered the action

channel: Optional[Channel] property

The channel the action was triggered in

Returns:

Type	Description
Optional[Channel]	the channel the action was triggered in or None if the action was triggered in a modal

state: Optional[State] property

The state of the block when the action was triggered

Returns:

Type	Description
Optional[State]	the state of the block when the action was triggered

response_url: Optional[str] property

The response URL for the action

Returns:

Type	Description
Optional[str]	the response URL for the action or None if the action was triggered in a modal

trigger_id: str property

The trigger id associated with the action

The trigger id can be used to open a modal

Returns:

Type	Description
str	the trigger id for the action

say(text: Optional[str] = None, attachments: Union[Sequence[Attachment], Sequence[dict[str, Any]], None] = None, blocks: Union[Sequence[Block], Sequence[dict[str, Any]], None] = None, ephemeral: bool = True, replace_original: bool = False, delete_original: bool = False, **kwargs: Any) -> Optional[WebhookResponse] async

Send a new message to the channel the block action was triggered in

Send a new message to the channel the block action was triggered in, using the response_url as a webhook. If the block action happened in a modal, the response_url will be None and this method will not send a message but instead log a warning. Allows for rich formatting using blocks and/or attachments . You can provide blocks and attachments as Python dicts or you can use the convenient classes that the underlying slack client provides. This will send an ephemeral message by default, only visible to the user that triggered the action. You can set ephemeral to False to make the message visible to everyone in the channel. By default, Slack replaces the original message in which the action was triggered. This method overrides this behavior. If you want your message to replace the original, set replace_original to True. Any extra kwargs you provide, will be passed on directly to AsyncWebhookClient.send()

Parameters:

Name	Type	Description	Default
text	Optional[str]	message text	None
attachments	Union[Sequence[Attachment], Sequence[dict[str, Any]], None]	optional attachments (see [attachments])	None
blocks	Union[Sequence[Block], Sequence[dict[str, Any]], None]	optional blocks (see [blocks])	None
ephemeral	bool	True/False wether to send the message as an ephemeral message, only visible to the user that initiated the action	True
replace_original	bool	True/False whether the message that contains the block from which the action was triggered should be replaced by this message	False
delete_original	bool	True/False whether the message that contains the block from which the action was triggered should be deleted. No other parameters should be provided.	False

Returns:

Type	Description
Optional[WebhookResponse]	Dictionary deserialized from AsyncWebhookClient.send()

Source code in machine/plugins/block_action.py

async def say(
    self,
    text: Optional[str] = None,
    attachments: Union[Sequence[Attachment], Sequence[dict[str, Any]], None] = None,
    blocks: Union[Sequence[Block], Sequence[dict[str, Any]], None] = None,
    ephemeral: bool = True,
    replace_original: bool = False,
    delete_original: bool = False,
    **kwargs: Any,
) -> Optional[WebhookResponse]:
    """Send a new message to the channel the block action was triggered in

    Send a new message to the channel the block action was triggered in, using the response_url as a webhook.
    If the block action happened in a modal, the response_url will be None and this method will not send a message
    but instead log a warning.
    Allows for rich formatting using [blocks] and/or [attachments] . You can provide blocks
    and attachments as Python dicts or you can use the [convenient classes] that the
    underlying slack client provides.
    This will send an ephemeral message by default, only visible to the user that triggered the action.
    You can set `ephemeral` to `False` to make the message visible to everyone in the channel.
    By default, Slack replaces the original message in which the action was triggered. This method overrides this
    behavior. If you want your message to replace the original, set replace_original to True.
    Any extra kwargs you provide, will be passed on directly to `AsyncWebhookClient.send()`

    [attachments]: https://api.slack.com/docs/message-attachments
    [blocks]: https://api.slack.com/reference/block-kit/blocks
    [convenient classes]: https://github.com/slackapi/python-slack-sdk/tree/main/slack/web/classes

    :param text: message text
    :param attachments: optional attachments (see [attachments])
    :param blocks: optional blocks (see [blocks])
    :param ephemeral: `True/False` wether to send the message as an ephemeral message, only
        visible to the user that initiated the action
    :param replace_original: `True/False` whether the message that contains the block from which the action was
        triggered should be replaced by this message
    :param delete_original: `True/False` whether the message that contains the block from which the action was
        triggered should be deleted. No other parameters should be provided.
    :return: Dictionary deserialized from `AsyncWebhookClient.send()`

    """
    if self._webhook_client is None:
        logger.warning(
            "response_url is None, cannot send message. This likely means the action was triggered in a modal."
        )
        return None

    response_type = "ephemeral" if ephemeral else "in_channel"
    return await self._webhook_client.send(
        text=text,
        attachments=attachments,
        blocks=blocks,
        response_type=response_type,
        replace_original=replace_original,
        delete_original=delete_original,
        **kwargs,
    )

send_dm(text: str | None = None, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, **kwargs: Any) -> AsyncSlackResponse async

Send a DM to the user that triggered the block action

Send a Direct Message to the user that triggered the block action by opening a DM channel and sending a message to it. Allows for rich formatting using blocks and/or attachments. Allows for rich formatting using blocks and/or attachments . You can provide blocks and attachments as Python dicts or you can use the convenient classes that the underlying slack client provides. Any extra kwargs you provide, will be passed on directly to the chat.postMessage request.

Parameters:

Name	Type	Description	Default
text	str | None	message text	None
attachments	Sequence[Attachment] | Sequence[dict[str, Any]] | None	optional attachments (see [attachments])	None
blocks	Sequence[Block] | Sequence[dict[str, Any]] | None	optional blocks (see [blocks])	None

Returns:

Type	Description
AsyncSlackResponse	Dictionary deserialized from [chat.postMessage] response. [chat.postMessage]: https://api.slack.com/methods/chat.postMessage

Source code in machine/plugins/block_action.py

async def send_dm(
    self,
    text: str | None = None,
    attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
    blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Send a DM to the user that triggered the block action

    Send a Direct Message to the user that triggered the block action by opening a DM channel and
    sending a message to it. Allows for rich formatting using `blocks`_ and/or `attachments`_.
    Allows for rich formatting using [blocks] and/or [attachments] . You can provide blocks
    and attachments as Python dicts or you can use the [convenient classes] that the
    underlying slack client provides.
    Any extra kwargs you provide, will be passed on directly to the `chat.postMessage` request.

    [attachments]: https://api.slack.com/docs/message-attachments
    [blocks]: https://api.slack.com/reference/block-kit/blocks
    [convenient classes]: https://github.com/slackapi/python-slack-sdk/tree/main/slack/web/classes

    :param text: message text
    :param attachments: optional attachments (see [attachments])
    :param blocks: optional blocks (see [blocks])
    :return: Dictionary deserialized from [chat.postMessage] response.

    [chat.postMessage]: https://api.slack.com/methods/chat.postMessage
    """
    return await self._client.send_dm(self.user.id, text, attachments=attachments, blocks=blocks, **kwargs)

open_modal(view: dict | View, **kwargs: Any) -> AsyncSlackResponse async

Open a modal in response to the block action

Open a modal in response to the block action, using the trigger_id that was returned when the block action was triggered. Any extra kwargs you provide, will be passed on directly to AsyncWebClient.views_open()

Note: you have to call this method within 3 seconds of receiving the block action payload.

Parameters:

Name	Type	Description	Default
view	dict | View	the view to open	required

Returns:

Type	Description
AsyncSlackResponse	Dictionary deserialized from AsyncWebClient.views_open()

Source code in machine/plugins/block_action.py

async def open_modal(
    self,
    view: dict | View,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Open a modal in response to the block action

    Open a modal in response to the block action, using the trigger_id that was returned when the block action was
    triggered.
    Any extra kwargs you provide, will be passed on directly to `AsyncWebClient.views_open()`

    Note: you have to call this method within 3 seconds of receiving the block action payload.

    :param view: the view to open
    :return: Dictionary deserialized from `AsyncWebClient.views_open()`
    """
    return await self._client.open_modal(self.trigger_id, view, **kwargs)

machine.plugins.modals.ModalSubmission

Source code in machine/plugins/modals.py

class ModalSubmission:
    payload: ViewSubmissionPayload

    def __init__(self, client: SlackClient, payload: ViewSubmissionPayload):
        self._client = client
        self.payload = payload

    @property
    def user(self) -> User:
        """The user that submitted the modal

        :return: the user that submitted the modal
        """
        return self._client.users[self.payload.user.id]

    @property
    def view(self) -> View:
        """The view that was submitted including the state of all the elements in the view

        :return: the view that was submitted
        """
        return self.payload.view

    @property
    def trigger_id(self) -> str:
        """The trigger id associated with the submitted modal

        The trigger id can be user ot open another modal

        :return: the trigger id for the modal
        """
        return self.payload.trigger_id

    async def open_modal(
        self,
        view: dict | SlackSDKView,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Open another modal in response to the modal submission

        Open another modal in response to modal submission, using the trigger_id that was returned when the modal was
        submitted.
        Any extra kwargs you provide, will be passed on directly to `AsyncWebClient.views_open()`

        Note: you have to call this method within 3 seconds of receiving the modal submission payload.

        :param view: the view to open
        :return: Dictionary deserialized from `AsyncWebClient.views_open()`
        """
        return await self._client.open_modal(self.trigger_id, view, **kwargs)

    async def push_modal(
        self,
        view: dict | SlackSDKView,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Push a new modal view in response to the modal submission

        Push a new modal view on top of the view stack in response to modal submission, using the trigger_id that was
        returned when the modal was submitted.
        Any extra kwargs you provide, will be passed on directly to `AsyncWebClient.views_push()`

        Note: you have to call this method within 3 seconds of receiving the modal submission payload.

        :param view: the view to push
        :return: Dictionary deserialized from `AsyncWebClient.views_push()`
        """
        return await self._client.push_modal(self.trigger_id, view, **kwargs)

    async def update_modal(
        self,
        view: dict | SlackSDKView,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Update the modal view in response to the modal submission

        Update the modal view in response to modal submission, using the trigger_id that was returned when the modal was
        submitted.
        Any extra kwargs you provide, will be passed on directly to `AsyncWebClient.views_update()`

        Note: you have to call this method within 3 seconds of receiving the modal submission payload.

        :param view: the view to update
        :return: Dictionary deserialized from `AsyncWebClient.views_update()`
        """
        return await self._client.update_modal(view, self.payload.view.id, self.payload.view.external_id, **kwargs)

    async def send_dm(
        self,
        text: str | None = None,
        attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
        blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Send a DM to the user that submitted the modal

        Send a Direct Message to the user that submitted the modal by opening a DM channel and
        sending a message to it.
        Allows for rich formatting using [blocks] and/or [attachments] . You can provide blocks
        and attachments as Python dicts or you can use the [convenient classes] that the
        underlying slack client provides.
        Any extra kwargs you provide, will be passed on directly to the `chat.postMessage` request.

        [attachments]: https://api.slack.com/docs/message-attachments
        [blocks]: https://api.slack.com/reference/block-kit/blocks
        [convenient classes]: https://github.com/slackapi/python-slack-sdk/tree/main/slack/web/classes

        :param text: message text
        :param attachments: optional attachments (see [attachments])
        :param blocks: optional blocks (see [blocks])
        :return: Dictionary deserialized from [chat.postMessage] response.

        [chat.postMessage]: https://api.slack.com/methods/chat.postMessage
        """
        return await self._client.send_dm(self.user.id, text, attachments=attachments, blocks=blocks, **kwargs)

user: User property

The user that submitted the modal

Returns:

Type	Description
User	the user that submitted the modal

view: View property

The view that was submitted including the state of all the elements in the view

Returns:

Type	Description
View	the view that was submitted

trigger_id: str property

The trigger id associated with the submitted modal

The trigger id can be user ot open another modal

Returns:

Type	Description
str	the trigger id for the modal

open_modal(view: dict | SlackSDKView, **kwargs: Any) -> AsyncSlackResponse async

Open another modal in response to the modal submission

Open another modal in response to modal submission, using the trigger_id that was returned when the modal was submitted. Any extra kwargs you provide, will be passed on directly to AsyncWebClient.views_open()

Note: you have to call this method within 3 seconds of receiving the modal submission payload.

Parameters:

Name	Type	Description	Default
view	dict | View	the view to open	required

Returns:

Type	Description
AsyncSlackResponse	Dictionary deserialized from AsyncWebClient.views_open()

Source code in machine/plugins/modals.py

async def open_modal(
    self,
    view: dict | SlackSDKView,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Open another modal in response to the modal submission

    Open another modal in response to modal submission, using the trigger_id that was returned when the modal was
    submitted.
    Any extra kwargs you provide, will be passed on directly to `AsyncWebClient.views_open()`

    Note: you have to call this method within 3 seconds of receiving the modal submission payload.

    :param view: the view to open
    :return: Dictionary deserialized from `AsyncWebClient.views_open()`
    """
    return await self._client.open_modal(self.trigger_id, view, **kwargs)

push_modal(view: dict | SlackSDKView, **kwargs: Any) -> AsyncSlackResponse async

Push a new modal view in response to the modal submission

Push a new modal view on top of the view stack in response to modal submission, using the trigger_id that was returned when the modal was submitted. Any extra kwargs you provide, will be passed on directly to AsyncWebClient.views_push()

Note: you have to call this method within 3 seconds of receiving the modal submission payload.

Parameters:

Name	Type	Description	Default
view	dict | View	the view to push	required

Returns:

Type	Description
AsyncSlackResponse	Dictionary deserialized from AsyncWebClient.views_push()

Source code in machine/plugins/modals.py

async def push_modal(
    self,
    view: dict | SlackSDKView,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Push a new modal view in response to the modal submission

    Push a new modal view on top of the view stack in response to modal submission, using the trigger_id that was
    returned when the modal was submitted.
    Any extra kwargs you provide, will be passed on directly to `AsyncWebClient.views_push()`

    Note: you have to call this method within 3 seconds of receiving the modal submission payload.

    :param view: the view to push
    :return: Dictionary deserialized from `AsyncWebClient.views_push()`
    """
    return await self._client.push_modal(self.trigger_id, view, **kwargs)

update_modal(view: dict | SlackSDKView, **kwargs: Any) -> AsyncSlackResponse async

Update the modal view in response to the modal submission

Update the modal view in response to modal submission, using the trigger_id that was returned when the modal was submitted. Any extra kwargs you provide, will be passed on directly to AsyncWebClient.views_update()

Note: you have to call this method within 3 seconds of receiving the modal submission payload.

Parameters:

Name	Type	Description	Default
view	dict | View	the view to update	required

Returns:

Type	Description
AsyncSlackResponse	Dictionary deserialized from AsyncWebClient.views_update()

Source code in machine/plugins/modals.py

async def update_modal(
    self,
    view: dict | SlackSDKView,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Update the modal view in response to the modal submission

    Update the modal view in response to modal submission, using the trigger_id that was returned when the modal was
    submitted.
    Any extra kwargs you provide, will be passed on directly to `AsyncWebClient.views_update()`

    Note: you have to call this method within 3 seconds of receiving the modal submission payload.

    :param view: the view to update
    :return: Dictionary deserialized from `AsyncWebClient.views_update()`
    """
    return await self._client.update_modal(view, self.payload.view.id, self.payload.view.external_id, **kwargs)

send_dm(text: str | None = None, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, **kwargs: Any) -> AsyncSlackResponse async

Send a DM to the user that submitted the modal

Send a Direct Message to the user that submitted the modal by opening a DM channel and sending a message to it. Allows for rich formatting using blocks and/or attachments . You can provide blocks and attachments as Python dicts or you can use the convenient classes that the underlying slack client provides. Any extra kwargs you provide, will be passed on directly to the chat.postMessage request.

Parameters:

Name	Type	Description	Default
text	str | None	message text	None
attachments	Sequence[Attachment] | Sequence[dict[str, Any]] | None	optional attachments (see [attachments])	None
blocks	Sequence[Block] | Sequence[dict[str, Any]] | None	optional blocks (see [blocks])	None

Returns:

Type	Description
AsyncSlackResponse	Dictionary deserialized from [chat.postMessage] response. [chat.postMessage]: https://api.slack.com/methods/chat.postMessage

Source code in machine/plugins/modals.py

async def send_dm(
    self,
    text: str | None = None,
    attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
    blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Send a DM to the user that submitted the modal

    Send a Direct Message to the user that submitted the modal by opening a DM channel and
    sending a message to it.
    Allows for rich formatting using [blocks] and/or [attachments] . You can provide blocks
    and attachments as Python dicts or you can use the [convenient classes] that the
    underlying slack client provides.
    Any extra kwargs you provide, will be passed on directly to the `chat.postMessage` request.

    [attachments]: https://api.slack.com/docs/message-attachments
    [blocks]: https://api.slack.com/reference/block-kit/blocks
    [convenient classes]: https://github.com/slackapi/python-slack-sdk/tree/main/slack/web/classes

    :param text: message text
    :param attachments: optional attachments (see [attachments])
    :param blocks: optional blocks (see [blocks])
    :return: Dictionary deserialized from [chat.postMessage] response.

    [chat.postMessage]: https://api.slack.com/methods/chat.postMessage
    """
    return await self._client.send_dm(self.user.id, text, attachments=attachments, blocks=blocks, **kwargs)

machine.plugins.modals.ModalClosure

Source code in machine/plugins/modals.py

class ModalClosure:
    payload: ViewClosedPayload

    def __init__(self, client: SlackClient, payload: ViewClosedPayload):
        self._client = client
        self.payload = payload

    @property
    def user(self) -> User:
        """The user that closed the modal

        :return: the user that closed the modal
        """
        return self._client.users[self.payload.user.id]

    @property
    def view(self) -> View:
        """The view that was closed including the state of all the elements in the view when it was closed

        :return: the view that was closed
        """
        return self.payload.view

    async def send_dm(
        self,
        text: str | None = None,
        attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
        blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Send a DM to the user that closed the modal

        Send a Direct Message to the user that closed the modal by opening a DM channel and
        sending a message to it.
        Allows for rich formatting using [blocks] and/or [attachments] . You can provide blocks
        and attachments as Python dicts or you can use the [convenient classes] that the
        underlying slack client provides.
        Any extra kwargs you provide, will be passed on directly to the `chat.postMessage` request.

        [attachments]: https://api.slack.com/docs/message-attachments
        [blocks]: https://api.slack.com/reference/block-kit/blocks
        [convenient classes]: https://github.com/slackapi/python-slack-sdk/tree/main/slack/web/classes

        :param text: message text
        :param attachments: optional attachments (see [attachments])
        :param blocks: optional blocks (see [blocks])
        :return: Dictionary deserialized from [chat.postMessage] response.

        [chat.postMessage]: https://api.slack.com/methods/chat.postMessage
        """
        return await self._client.send_dm(self.user.id, text, attachments=attachments, blocks=blocks, **kwargs)

user: User property

The user that closed the modal

Returns:

Type	Description
User	the user that closed the modal

view: View property

The view that was closed including the state of all the elements in the view when it was closed

Returns:

Type	Description
View	the view that was closed

send_dm(text: str | None = None, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, **kwargs: Any) -> AsyncSlackResponse async

Send a DM to the user that closed the modal

Send a Direct Message to the user that closed the modal by opening a DM channel and sending a message to it. Allows for rich formatting using blocks and/or attachments . You can provide blocks and attachments as Python dicts or you can use the convenient classes that the underlying slack client provides. Any extra kwargs you provide, will be passed on directly to the chat.postMessage request.

Parameters:

Name	Type	Description	Default
text	str | None	message text	None
attachments	Sequence[Attachment] | Sequence[dict[str, Any]] | None	optional attachments (see [attachments])	None
blocks	Sequence[Block] | Sequence[dict[str, Any]] | None	optional blocks (see [blocks])	None

Returns:

Type	Description
AsyncSlackResponse	Dictionary deserialized from [chat.postMessage] response. [chat.postMessage]: https://api.slack.com/methods/chat.postMessage

Source code in machine/plugins/modals.py

async def send_dm(
    self,
    text: str | None = None,
    attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
    blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Send a DM to the user that closed the modal

    Send a Direct Message to the user that closed the modal by opening a DM channel and
    sending a message to it.
    Allows for rich formatting using [blocks] and/or [attachments] . You can provide blocks
    and attachments as Python dicts or you can use the [convenient classes] that the
    underlying slack client provides.
    Any extra kwargs you provide, will be passed on directly to the `chat.postMessage` request.

    [attachments]: https://api.slack.com/docs/message-attachments
    [blocks]: https://api.slack.com/reference/block-kit/blocks
    [convenient classes]: https://github.com/slackapi/python-slack-sdk/tree/main/slack/web/classes

    :param text: message text
    :param attachments: optional attachments (see [attachments])
    :param blocks: optional blocks (see [blocks])
    :return: Dictionary deserialized from [chat.postMessage] response.

    [chat.postMessage]: https://api.slack.com/methods/chat.postMessage
    """
    return await self._client.send_dm(self.user.id, text, attachments=attachments, blocks=blocks, **kwargs)

Decorators

These are the decorators you can use to have Slack Machine respond to specific things (events, messages, etc.)

machine.plugins.decorators

process(slack_event_type: str) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]

Process Slack events of a specific type

This decorator will enable a Plugin method to process Slack events_ of a specific type. The Plugin method will be called for each event of the specified type that the bot receives. The received event will be passed to the method when called.

.. _Slack events: https://api.slack.com/events

Parameters:

Name	Type	Description	Default
slack_event_type	str	type of event the method needs to process. Can be any event supported by the RTM API	required

Returns:

Type	Description
Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]	wrapped method

Source code in machine/plugins/decorators.py

def process(slack_event_type: str) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]:
    """Process Slack events of a specific type

    This decorator will enable a Plugin method to process `Slack events`_ of a specific type. The
    Plugin method will be called for each event of the specified type that the bot receives.
    The received event will be passed to the method when called.

    .. _Slack events: https://api.slack.com/events

    :param slack_event_type: type of event the method needs to process. Can be any event supported
        by the RTM API
    :return: wrapped method
    """

    def process_decorator(f: Callable[P, R]) -> DecoratedPluginFunc[P, R]:
        fn = cast(DecoratedPluginFunc, f)
        fn.metadata = getattr(f, "metadata", Metadata())
        fn.metadata.plugin_actions.process.append(slack_event_type)
        return fn

    return process_decorator

listen_to(regex: str, flags: re.RegexFlag | int = re.IGNORECASE, handle_message_changed: bool = False) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]

Listen to messages matching a regex pattern

This decorator will enable a Plugin method to listen to messages that match a regex pattern. The Plugin method will be called for each message that matches the specified regex pattern. The received :py:class:~machine.plugins.base.Message will be passed to the method when called. Named groups can be used in the regex pattern, to catch specific parts of the message. These groups will be passed to the method as keyword arguments when called.

Parameters:

Name	Type	Description	Default
regex	str	regex pattern to listen for	required
flags	RegexFlag | int	regex flags to apply when matching	IGNORECASE
handle_message_changed	bool	if changed messages should trigger the decorated function	False

Returns:

Type	Description
Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]	wrapped method

Source code in machine/plugins/decorators.py

def listen_to(
    regex: str, flags: re.RegexFlag | int = re.IGNORECASE, handle_message_changed: bool = False
) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]:
    """Listen to messages matching a regex pattern

    This decorator will enable a Plugin method to listen to messages that match a regex pattern.
    The Plugin method will be called for each message that matches the specified regex pattern.
    The received :py:class:`~machine.plugins.base.Message` will be passed to the method when called.
    Named groups can be used in the regex pattern, to catch specific parts of the message. These
    groups will be passed to the method as keyword arguments when called.

    :param regex: regex pattern to listen for
    :param flags: regex flags to apply when matching
    :param handle_message_changed: if changed messages should trigger the decorated function
    :return: wrapped method
    """

    def listen_to_decorator(f: Callable[P, R]) -> DecoratedPluginFunc[P, R]:
        fn = cast(DecoratedPluginFunc, f)
        fn.metadata = getattr(f, "metadata", Metadata())
        fn.metadata.plugin_actions.listen_to.append(MatcherConfig(re.compile(regex, flags), handle_message_changed))
        return fn

    return listen_to_decorator

respond_to(regex: str, flags: re.RegexFlag | int = re.IGNORECASE, handle_message_changed: bool = False) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]

Listen to messages mentioning the bot and matching a regex pattern

This decorator will enable a Plugin method to listen to messages that are directed to the bot (ie. message starts by mentioning the bot) and match a regex pattern. The Plugin method will be called for each message that mentions the bot and matches the specified regex pattern. The received :py:class:~machine.plugins.base.Message will be passed to the method when called. Named groups can be used in the regex pattern, to catch specific parts of the message. These groups will be passed to the method as keyword arguments when called.

Parameters:

Name	Type	Description	Default
regex	str	regex pattern to listen for	required
flags	RegexFlag | int	regex flags to apply when matching	IGNORECASE
handle_message_changed	bool	if changed messages should trigger the decorated function	False

Returns:

Type	Description
Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]	wrapped method

Source code in machine/plugins/decorators.py

def respond_to(
    regex: str, flags: re.RegexFlag | int = re.IGNORECASE, handle_message_changed: bool = False
) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]:
    """Listen to messages mentioning the bot and matching a regex pattern

    This decorator will enable a Plugin method to listen to messages that are directed to the bot
    (ie. message starts by mentioning the bot) and match a regex pattern.
    The Plugin method will be called for each message that mentions the bot and matches the
    specified regex pattern. The received :py:class:`~machine.plugins.base.Message` will be passed
    to the method when called. Named groups can be used in the regex pattern, to catch specific
    parts of the message. These groups will be passed to the method as keyword arguments when
    called.

    :param regex: regex pattern to listen for
    :param flags: regex flags to apply when matching
    :param handle_message_changed: if changed messages should trigger the decorated function
    :return: wrapped method
    """

    def respond_to_decorator(f: Callable[P, R]) -> DecoratedPluginFunc[P, R]:
        fn = cast(DecoratedPluginFunc, f)
        fn.metadata = getattr(f, "metadata", Metadata())
        fn.metadata.plugin_actions.respond_to.append(MatcherConfig(re.compile(regex, flags), handle_message_changed))
        return fn

    return respond_to_decorator

command(slash_command: str) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]

Respond to a slash command

This decorator will enable a Plugin method to respond to slash commands

Parameters:

Name	Type	Description	Default
slash_command	str	the slash command to respond to	required

Returns:

Type	Description
Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]	wrapped method

Source code in machine/plugins/decorators.py

def command(slash_command: str) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]:
    """Respond to a slash command

    This decorator will enable a Plugin method to respond to slash commands

    :param slash_command: the slash command to respond to
    :return: wrapped method
    """

    def command_decorator(f: Callable[P, R]) -> DecoratedPluginFunc[P, R]:
        fn = cast(DecoratedPluginFunc, f)
        fn.metadata = getattr(f, "metadata", Metadata())
        normalized_slash_command = f"/{slash_command}" if not slash_command.startswith("/") else slash_command
        fn.metadata.plugin_actions.commands.append(
            CommandConfig(command=normalized_slash_command, is_generator=inspect.isasyncgenfunction(f))
        )
        return fn

    return command_decorator

action(action_id: Union[re.Pattern[str], str, None] = None, block_id: Union[re.Pattern[str], str, None] = None) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]

Respond to block actions

This decorator will enable a Plugin method to be triggered when certain block actions are received. The Plugin method will be called when a block action event is received for which the action_id and block_id match the provided values. action_id and block_id can be strings, in which case the incoming action_id and block_id must match exactly, or regex patterns, in which case the incoming action_id and block_id must match the regex pattern.

Both action_id and block_id are optional, but at least one of them must be provided.

Parameters:

Name	Type	Description	Default
action_id	Union[Pattern[str], str, None]	the action_id to respond to, can be a string or regex pattern	None
block_id	Union[Pattern[str], str, None]	the block_id to respond to, can be a string or regex pattern	None

Returns:

Type	Description
Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]	wrapped method

Source code in machine/plugins/decorators.py

def action(
    action_id: Union[re.Pattern[str], str, None] = None, block_id: Union[re.Pattern[str], str, None] = None
) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]:
    """Respond to block actions

    This decorator will enable a Plugin method to be triggered when certain block actions are
    received. The Plugin method will be called when a block action event is received for which
    the action_id and block_id match the provided values. action_id and block_id can be strings,
    in which case the incoming action_id and block_id must match exactly, or regex patterns, in
    which case the incoming action_id and block_id must match the regex pattern.

    Both action_id and block_id are optional, but **at least one of them must be provided**.

    :param action_id: the action_id to respond to, can be a string or regex pattern
    :param block_id: the block_id to respond to, can be a string or regex pattern
    :return: wrapped method
    """

    def action_decorator(f: Callable[P, R]) -> DecoratedPluginFunc[P, R]:
        fn = cast(DecoratedPluginFunc, f)
        fn.metadata = getattr(f, "metadata", Metadata())
        if action_id is None and block_id is None:
            raise ValueError("At least one of action_id or block_id must be provided")
        fn.metadata.plugin_actions.actions.append(ActionConfig(action_id=action_id, block_id=block_id))
        return fn

    return action_decorator

modal(callback_id: Union[re.Pattern[str], str]) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]

Respond to modal submissions

This decorator will enable a Plugin method to be triggered when certain modals are submitted. The Plugin method will be called when a modal submission event is received for which the callback_id matches the provided value. The callback_id can be a string or a regex pattern.

Parameters:

Name	Type	Description	Default
callback_id	Union[Pattern[str], str]	the callback_id to respond to, can be a string or regex pattern	required

Returns:

Type	Description
Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]	wrapped method

Source code in machine/plugins/decorators.py

def modal(callback_id: Union[re.Pattern[str], str]) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]:
    """Respond to modal submissions

    This decorator will enable a Plugin method to be triggered when certain modals are submitted.
    The Plugin method will be called when a modal submission event is received for which the
    callback_id matches the provided value. The callback_id can be a string or a regex pattern.

    :param callback_id: the callback_id to respond to, can be a string or regex pattern
    :return: wrapped method
    """

    def modal_decorator(f: Callable[P, R]) -> DecoratedPluginFunc[P, R]:
        fn = cast(DecoratedPluginFunc, f)
        fn.metadata = getattr(f, "metadata", Metadata())
        is_generator = inspect.isasyncgenfunction(f)
        fn.metadata.plugin_actions.modal_submissions.append(
            ModalConfig(callback_id=callback_id, is_generator=is_generator)
        )
        return fn

    return modal_decorator

modal_closed(callback_id: Union[re.Pattern[str], str]) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]

Respond to modal closures

This decorator will enable a Plugin method to be triggered when certain modals are closed. The Plugin method will be called when a modal closure event is received for which the callback_id matches the provided value. The callback_id can be a string or a regex pattern.

Note: in order to receive modal close events, the modal must have the notify_on_close property set to True.

Parameters:

Name	Type	Description	Default
callback_id	Union[Pattern[str], str]	the callback_id to respond to, can be a string or regex pattern	required

Returns:

Type	Description
Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]	wrapped method

Source code in machine/plugins/decorators.py

def modal_closed(callback_id: Union[re.Pattern[str], str]) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]:
    """Respond to modal closures

    This decorator will enable a Plugin method to be triggered when certain modals are closed.
    The Plugin method will be called when a modal closure event is received for which the
    callback_id matches the provided value. The callback_id can be a string or a regex pattern.

    Note: in order to receive modal close events, the modal must have the `notify_on_close` property set to `True`.

    :param callback_id: the callback_id to respond to, can be a string or regex pattern
    :return: wrapped method
    """

    def modal_closed_decorator(f: Callable[P, R]) -> DecoratedPluginFunc[P, R]:
        fn = cast(DecoratedPluginFunc, f)
        fn.metadata = getattr(f, "metadata", Metadata())
        is_generator = inspect.isasyncgenfunction(f)
        if is_generator:
            raise ValueError("Modal closed handlers cannot be async generators")
        fn.metadata.plugin_actions.modal_closures.append(ModalConfig(callback_id=callback_id))
        return fn

    return modal_closed_decorator

schedule(year: int | str | None = None, month: int | str | None = None, day: int | str | None = None, week: int | str | None = None, day_of_week: int | str | None = None, hour: int | str | None = None, minute: int | str | None = None, second: int | str | None = None, start_date: datetime | str | None = None, end_date: datetime | str | None = None, timezone: tzinfo | str | None = None) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]

Schedule a function to be executed according to a crontab-like schedule

The decorated function will be executed according to the schedule provided. Slack Machine uses APScheduler under the hood for scheduling. For more information on the interpretation of the provided parameters, see :class:CronTrigger<apscheduler:apscheduler.triggers.cron.CronTrigger>

Parameters:

Name	Type	Description	Default
year	int|str	4-digit year	None
month	int|str	month (1-12)	None
day	int|str	day of the (1-31)	None
week	int|str	ISO week (1-53)	None
day_of_week	int|str	number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun)	None
hour	int|str	hour (0-23)	None
minute	int|str	minute (0-59)	None
second	int|str	second (0-59)	None
start_date	datetime|str	earliest possible date/time to trigger on (inclusive)	None
end_date	datetime|str	latest possible date/time to trigger on (inclusive)	None
timezone	datetime.tzinfo|str	time zone to use for the date/time calculations (defaults to scheduler timezone)	None

Source code in machine/plugins/decorators.py

def schedule(
    year: int | str | None = None,
    month: int | str | None = None,
    day: int | str | None = None,
    week: int | str | None = None,
    day_of_week: int | str | None = None,
    hour: int | str | None = None,
    minute: int | str | None = None,
    second: int | str | None = None,
    start_date: datetime | str | None = None,
    end_date: datetime | str | None = None,
    timezone: tzinfo | str | None = None,
) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]:
    """Schedule a function to be executed according to a crontab-like schedule

    The decorated function will be executed according to the schedule provided. Slack Machine uses
    APScheduler under the hood for scheduling. For more information on the interpretation of the
    provided parameters, see :class:`CronTrigger<apscheduler:apscheduler.triggers.cron.CronTrigger>`

    :param int|str year: 4-digit year
    :param int|str month: month (1-12)
    :param int|str day: day of the (1-31)
    :param int|str week: ISO week (1-53)
    :param int|str day_of_week: number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun)
    :param int|str hour: hour (0-23)
    :param int|str minute: minute (0-59)
    :param int|str second: second (0-59)
    :param datetime|str start_date: earliest possible date/time to trigger on (inclusive)
    :param datetime|str end_date: latest possible date/time to trigger on (inclusive)
    :param datetime.tzinfo|str timezone: time zone to use for the date/time calculations (defaults
        to scheduler timezone)
    """
    kwargs = locals()

    def schedule_decorator(f: Callable[P, R]) -> DecoratedPluginFunc[P, R]:
        fn = cast(DecoratedPluginFunc, f)
        fn.metadata = getattr(f, "metadata", Metadata())
        fn.metadata.plugin_actions.schedule = kwargs
        return fn

    return schedule_decorator

on(event: str) -> Callable[[Callable[P, R]], Callable[P, R]]

Listen for an event

The decorated function will be called whenever a plugin (or Slack Machine itself) emits an event with the given name.

Parameters:

Name	Type	Description	Default
event	str	name of the event to listen for. Event names are global	required

Source code in machine/plugins/decorators.py

def on(event: str) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """Listen for an event

    The decorated function will be called whenever a plugin (or Slack Machine itself) emits an
    event with the given name.

    :param event: name of the event to listen for. Event names are global
    """

    def on_decorator(f: Callable[P, R]) -> Callable[P, R]:
        ee.add_listener(event, f)
        return f

    return on_decorator

required_settings(settings: list[str] | str) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]

Specify a required setting for a plugin or plugin method

The settings specified with this decorator will be added to the required settings for the plugin. If one or more settings have not been defined by the user, the plugin will not be loaded and a warning will be written to the console upon startup.

Parameters:

Name	Type	Description	Default
settings	list[str] | str	settings that are required (can be list of strings, or single string)	required

Source code in machine/plugins/decorators.py

def required_settings(settings: list[str] | str) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]:
    """Specify a required setting for a plugin or plugin method

    The settings specified with this decorator will be added to the required settings for the
    plugin. If one or more settings have not been defined by the user, the plugin will not be
    loaded and a warning will be written to the console upon startup.

    :param settings: settings that are required (can be list of strings, or single string)
    """

    def required_settings_decorator(f_or_cls: Callable[P, R]) -> DecoratedPluginFunc[P, R]:
        casted_f_or_cls = cast(DecoratedPluginFunc, f_or_cls)
        casted_f_or_cls.metadata = getattr(f_or_cls, "metadata", Metadata())
        if isinstance(settings, list):
            casted_f_or_cls.metadata.required_settings.extend(settings)
        elif isinstance(settings, str):
            casted_f_or_cls.metadata.required_settings.append(settings)
        return casted_f_or_cls

    return required_settings_decorator

require_any_role(required_roles: list[str]) -> Callable[[Callable[..., Awaitable[None]]], Callable[..., Awaitable[None]]]

Specify required roles for a plugin method

To use the plugin method where this decorator is applied, the user must have at least one of the listed roles.

Parameters:

Name	Type	Description	Default
required_roles	list[str]	list of roles required to use the plugin method	required

Source code in machine/plugins/decorators.py

def require_any_role(
    required_roles: list[str],
) -> Callable[[Callable[..., Awaitable[None]]], Callable[..., Awaitable[None]]]:
    """Specify required roles for a plugin method

    To use the plugin method where this decorator is applied, the user must have
    at least one of the listed roles.

    :param required_roles: list of roles required to use the plugin method
    """

    def middle(func: Callable[..., Awaitable[None]]) -> Callable[..., Awaitable[None]]:
        async def wrapper(self: MachineBasePlugin, msg: Message, **kwargs: Any) -> None:
            if await matching_roles_by_user_id(self, msg.sender.id, required_roles):
                logger.debug(f"User {msg.sender} has one of the required roles {required_roles}")
                return await func(self, msg, **kwargs)
            else:
                logger.debug(f"User {msg.sender} does not have any of the required roles {required_roles}")
                ee.emit(
                    "unauthorized-access",
                    self,
                    message=msg,
                    required_roles=required_roles,
                    combinator=RoleCombinator.ANY,
                )
                await msg.say("I'm sorry, but you don't have access to that command", ephemeral=True)
                return None

        # Copy any existing docs and metadata from container function to
        # generated function
        wrapper.__doc__ = func.__doc__
        casted_wrapper = cast(DecoratedPluginFunc, wrapper)
        casted_wrapper.metadata = getattr(func, "metadata", Metadata())
        return casted_wrapper

    return middle

require_all_roles(required_roles: list[str]) -> Callable[[Callable[..., Awaitable[None]]], Callable[..., Awaitable[None]]]

Specify required roles for a plugin method

To use the plugin method where this decorator is applied, the user must have all of the listed roles.

Parameters:

Name	Type	Description	Default
required_roles	list[str]	list of roles required to use the plugin method	required

Source code in machine/plugins/decorators.py

def require_all_roles(
    required_roles: list[str],
) -> Callable[[Callable[..., Awaitable[None]]], Callable[..., Awaitable[None]]]:
    """Specify required roles for a plugin method

    To use the plugin method where this decorator is applied, the user must have
    all of the listed roles.

    :param required_roles: list of roles required to use the plugin method
    """

    def middle(func: Callable[..., Awaitable[None]]) -> Callable[..., Awaitable[None]]:
        async def wrapper(self: MachineBasePlugin, msg: Message, **kwargs: Any) -> None:
            if await matching_roles_by_user_id(self, msg.sender.id, required_roles) == len(required_roles):
                logger.debug(f"User {msg.sender} has all of the required roles {required_roles}")
                return await func(self, msg, **kwargs)
            else:
                logger.debug(f"User {msg.sender} does not have all of the required roles {required_roles}")
                ee.emit(
                    "unauthorized-access",
                    self,
                    message=msg,
                    required_roles=required_roles,
                    combinator=RoleCombinator.ALL,
                )
                await msg.say("I'm sorry, but you don't have access to that command", ephemeral=True)
                return None

        # Copy any existing docs and metadata from container function to
        # generated function
        wrapper.__doc__ = func.__doc__
        casted_wrapper = cast(DecoratedPluginFunc, wrapper)
        casted_wrapper.metadata = getattr(func, "metadata", Metadata())
        return casted_wrapper

    return middle

Models

These classes represent base objects from the Slack API

machine.models.user.User

Bases: BaseModel

User model that represents a user object from the Slack API

Source code in machine/models/user.py

class User(BaseModel):
    """
    User model that represents a user object from the Slack API
    """

    model_config = ConfigDict(frozen=True)

    id: str
    team_id: Optional[str] = None
    name: str
    deleted: Optional[bool] = None
    profile: Profile
    is_bot: bool
    updated: int
    is_app_user: bool
    color: Optional[str] = None
    real_name: Optional[str] = None
    tz: Optional[str] = None
    tz_label: Optional[str] = None
    tz_offset: Optional[int] = None
    is_admin: Optional[bool] = None
    is_owner: Optional[bool] = None
    is_primary_owner: Optional[bool] = None
    is_restricted: Optional[bool] = None
    is_ultra_restricted: Optional[bool] = None
    is_stranger: Optional[bool] = None
    has_2fa: Optional[bool] = None
    locale: Optional[str] = None

    def fmt_mention(self) -> str:
        return f"<@{self.id}>"

machine.models.channel.Channel

Bases: BaseModel

Channel model that represents a channel object from the Slack API

Source code in machine/models/channel.py

class Channel(BaseModel):
    """
    Channel model that represents a channel object from the Slack API
    """

    model_config = ConfigDict(frozen=True)

    id: str
    name: Optional[str] = None
    created: int
    creator: Optional[str] = None
    is_archived: bool
    is_general: Optional[bool] = None
    name_normalized: Optional[str] = None
    is_shared: Optional[bool] = None
    is_org_shared: bool
    is_member: Optional[bool] = None
    is_private: Optional[bool] = None
    is_mpim: Optional[bool] = None
    is_channel: Optional[bool] = None
    is_group: Optional[bool] = None
    is_im: Optional[bool] = None
    user: Optional[str] = None
    topic: Optional[PurposeTopic] = None
    purpose: Optional[PurposeTopic] = None
    previous_names: Optional[list[str]] = None

    @property
    def identifier(self) -> str:
        if self.name:
            return self.name
        else:
            return self.id

machine.models.interactive.View

Bases: BaseModel

Source code in machine/models/interactive.py

class View(BaseModel):
    id: str
    team_id: str
    type: Literal["modal", "home"]
    blocks: list[Block]
    private_metadata: str
    callback_id: str
    state: State
    hash: str
    title: Text
    clear_on_close: bool
    notify_on_close: bool
    close: Optional[Text]
    submit: Optional[Text]
    previous_view_id: Optional[str]
    root_view_id: str
    app_id: str
    external_id: str
    app_installed_team_id: str
    bot_id: str

Storage

Storage is exposed to plugins through the self.storage field. The following class implements the interface plugins can use to interact with the storage backend.

machine.storage.PluginStorage

Class providing access to persistent storage for plugins

This class is the main access point for plugins to work with persistent storage. It is accessible from plugins using self.storage. Data is serialized before sending it to the storage backend, and deserialized upon retrieval. Serialization is done by dill_, so pretty much any Python object can be stored and retrieved.

.. _Dill: https://pypi.python.org/pypi/dill

Source code in machine/storage/__init__.py

class PluginStorage:
    """Class providing access to persistent storage for plugins

    This class is the main access point for plugins to work with persistent storage. It is
    accessible from plugins using ``self.storage``. Data is serialized before sending it to
    the storage backend, and deserialized upon retrieval. Serialization is done by `dill`_, so
    pretty much any Python object can be stored and retrieved.

    .. _Dill: https://pypi.python.org/pypi/dill
    """

    def __init__(self, fq_plugin_name: str, storage_backend: MachineBaseStorage):
        self._fq_plugin_name = fq_plugin_name
        self._storage = storage_backend

    def _gen_unique_key(self, key: str) -> str:
        return f"{self._fq_plugin_name}:{key}"

    def _namespace_key(self, key: str, shared: bool = False) -> str:
        return key if shared else self._gen_unique_key(key)

    async def set(self, key: str, value: Any, expires: int | timedelta | None = None, shared: bool = False) -> None:
        """Store or update a value by key

        :param key: the key under which to store the data
        :param value: the data to store
        :param expires: optional number of seconds after which the data is expired
        :param shared: ``True/False`` wether this data should be shared by other plugins.  Use with
            care, because it pollutes the global namespace of the storage.
        """
        expires = int(expires.total_seconds()) if isinstance(expires, timedelta) else expires
        namespaced_key = self._namespace_key(key, shared)
        pickled_value = dill.dumps(value)
        await self._storage.set(namespaced_key, pickled_value, expires)

    async def get(self, key: str, shared: bool = False) -> Any | None:
        """Retrieve data by key

        :param key: key for the data to retrieve
        :param shared: ``True/False`` wether to retrieve data from the shared (global) namespace.
        :return: the data, or ``None`` if the key cannot be found/has expired
        """
        namespaced_key = self._namespace_key(key, shared)
        value = await self._storage.get(namespaced_key)
        if value:
            return dill.loads(value)
        else:
            return None

    async def has(self, key: str, shared: bool = False) -> bool:
        """Check if the key exists in storage

        Note: this class implements ``__contains__`` so instead of calling
        ``self.storage.has(...)``, you can also use: ``key in self.storage``. This will check the
        *namespaced* version of the key, so it's the same as:
        ``self.storage.has('key', shared=False)``

        :param key: key to check
        :param shared: ``True/False`` wether to check in the shared (global) namespace
        :return: ``True/False`` wether the key exists. Can only return ``True`` if the key has not
            expired.
        """
        namespaced_key = self._namespace_key(key, shared)
        return await self._storage.has(namespaced_key)

    async def delete(self, key: str, shared: bool = False) -> None:
        """Remove a key and its data from storage

        :param key: key to remove
        :param shared: ``True/False`` wether the key to remove should be in the shared (global)
            namespace
        """
        namespaced_key = self._namespace_key(key, shared)
        await self._storage.delete(namespaced_key)

    async def get_storage_size(self) -> int:
        """Calculate the total size of the storage

        :return: the total size of the storage in bytes (integer)
        """
        return await self._storage.size()

    async def get_storage_size_human(self) -> str:
        """Calculate the total size of the storage in human readable format

        :return: the total size of the storage in a human readable string, rounded to the nearest
            applicable division. eg. B for Bytes, KiB for Kilobytes, MiB for Megabytes etc.
        """
        size = await self.get_storage_size()
        return sizeof_fmt(size)

set(key: str, value: Any, expires: int | timedelta | None = None, shared: bool = False) -> None async

Store or update a value by key

Parameters:

Name	Type	Description	Default
key	str	the key under which to store the data	required
value	Any	the data to store	required
expires	int | timedelta | None	optional number of seconds after which the data is expired	None
shared	bool	True/False wether this data should be shared by other plugins. Use with care, because it pollutes the global namespace of the storage.	False

Source code in machine/storage/__init__.py

async def set(self, key: str, value: Any, expires: int | timedelta | None = None, shared: bool = False) -> None:
    """Store or update a value by key

    :param key: the key under which to store the data
    :param value: the data to store
    :param expires: optional number of seconds after which the data is expired
    :param shared: ``True/False`` wether this data should be shared by other plugins.  Use with
        care, because it pollutes the global namespace of the storage.
    """
    expires = int(expires.total_seconds()) if isinstance(expires, timedelta) else expires
    namespaced_key = self._namespace_key(key, shared)
    pickled_value = dill.dumps(value)
    await self._storage.set(namespaced_key, pickled_value, expires)

get(key: str, shared: bool = False) -> Any | None async

Retrieve data by key

Parameters:

Name	Type	Description	Default
key	str	key for the data to retrieve	required
shared	bool	True/False wether to retrieve data from the shared (global) namespace.	False

Returns:

Type	Description
Any | None	the data, or None if the key cannot be found/has expired

Source code in machine/storage/__init__.py

async def get(self, key: str, shared: bool = False) -> Any | None:
    """Retrieve data by key

    :param key: key for the data to retrieve
    :param shared: ``True/False`` wether to retrieve data from the shared (global) namespace.
    :return: the data, or ``None`` if the key cannot be found/has expired
    """
    namespaced_key = self._namespace_key(key, shared)
    value = await self._storage.get(namespaced_key)
    if value:
        return dill.loads(value)
    else:
        return None

has(key: str, shared: bool = False) -> bool async

Check if the key exists in storage

Note: this class implements __contains__ so instead of calling self.storage.has(...), you can also use: key in self.storage. This will check the namespaced version of the key, so it's the same as: self.storage.has('key', shared=False)

Parameters:

Name	Type	Description	Default
key	str	key to check	required
shared	bool	True/False wether to check in the shared (global) namespace	False

Returns:

Type	Description
bool	True/False wether the key exists. Can only return True if the key has not expired.

Source code in machine/storage/__init__.py

async def has(self, key: str, shared: bool = False) -> bool:
    """Check if the key exists in storage

    Note: this class implements ``__contains__`` so instead of calling
    ``self.storage.has(...)``, you can also use: ``key in self.storage``. This will check the
    *namespaced* version of the key, so it's the same as:
    ``self.storage.has('key', shared=False)``

    :param key: key to check
    :param shared: ``True/False`` wether to check in the shared (global) namespace
    :return: ``True/False`` wether the key exists. Can only return ``True`` if the key has not
        expired.
    """
    namespaced_key = self._namespace_key(key, shared)
    return await self._storage.has(namespaced_key)

delete(key: str, shared: bool = False) -> None async

Remove a key and its data from storage

Parameters:

Name	Type	Description	Default
key	str	key to remove	required
shared	bool	True/False wether the key to remove should be in the shared (global) namespace	False

Source code in machine/storage/__init__.py

async def delete(self, key: str, shared: bool = False) -> None:
    """Remove a key and its data from storage

    :param key: key to remove
    :param shared: ``True/False`` wether the key to remove should be in the shared (global)
        namespace
    """
    namespaced_key = self._namespace_key(key, shared)
    await self._storage.delete(namespaced_key)

get_storage_size() -> int async

Calculate the total size of the storage

Returns:

Type	Description
int	the total size of the storage in bytes (integer)

Source code in machine/storage/__init__.py

async def get_storage_size(self) -> int:
    """Calculate the total size of the storage

    :return: the total size of the storage in bytes (integer)
    """
    return await self._storage.size()

get_storage_size_human() -> str async

Calculate the total size of the storage in human readable format

Returns:

Type	Description
str	the total size of the storage in a human readable string, rounded to the nearest applicable division. eg. B for Bytes, KiB for Kilobytes, MiB for Megabytes etc.

Source code in machine/storage/__init__.py

async def get_storage_size_human(self) -> str:
    """Calculate the total size of the storage in human readable format

    :return: the total size of the storage in a human readable string, rounded to the nearest
        applicable division. eg. B for Bytes, KiB for Kilobytes, MiB for Megabytes etc.
    """
    size = await self.get_storage_size()
    return sizeof_fmt(size)

---

New Storage Backends can be implemented by extending the following class:

machine.storage.backends.base.MachineBaseStorage

Bases: ABC

Base class for storage backends

Extending classes should implement the five methods in this base class. Slack Machine takes care of a lot of details regarding the persistent storage of data. So storage backends do not have to deal with the following, because Slack Machine takes care of these:

• Serialization/Deserialization of data
• Namespacing of keys (so data stored by different plugins doesn't clash)

Source code in machine/storage/backends/base.py

class MachineBaseStorage(ABC):
    """Base class for storage backends

    Extending classes should implement the five methods in this base class. Slack Machine takes
    care of a lot of details regarding the persistent storage of data. So storage backends
    **do not** have to deal with the following, because Slack Machine takes care of these:

    - Serialization/Deserialization of data
    - Namespacing of keys (so data stored by different plugins doesn't clash)
    """

    settings: Mapping[str, Any]

    def __init__(self, settings: Mapping[str, Any]):
        self.settings = settings

    async def init(self) -> None:  # noqa: B027 (no-op by design)
        """Initialize the storage backend"""
        pass

    @abstractmethod
    async def get(self, key: str) -> bytes | None:
        """Retrieve data by key

        :param key: key for which to retrieve data
        :return: the raw data for the provided key, as (byte)string. Should return ``None`` when
            the key is unknown or the data has expired.
        """
        ...

    @abstractmethod
    async def set(self, key: str, value: bytes, expires: int | None = None) -> None:
        """Store data by key

        :param key: the key under which to store the data
        :param value: data as (byte)string
        :param expires: optional expiration time in seconds, after which the data should not be
            returned any more.
        """
        ...

    @abstractmethod
    async def delete(self, key: str) -> None:
        """Delete data by key

        :param key: key for which to delete the data
        """
        ...

    @abstractmethod
    async def has(self, key: str) -> bool:
        """Check if the key exists

        :param key: key to check
        :return: ``True/False`` wether the key exists
        """
        ...

    @abstractmethod
    async def size(self) -> int:
        """Calculate the total size of the storage

        :return: total size of storage in bytes (integer)
        """
        ...

    @abstractmethod
    async def close(self) -> None:
        """Close the storage backend"""
        ...

init() -> None async

Initialize the storage backend

Source code in machine/storage/backends/base.py

async def init(self) -> None:  # noqa: B027 (no-op by design)
    """Initialize the storage backend"""
    pass

get(key: str) -> bytes | None abstractmethod async

Retrieve data by key

Parameters:

Name	Type	Description	Default
key	str	key for which to retrieve data	required

Returns:

Type	Description
bytes | None	the raw data for the provided key, as (byte)string. Should return None when the key is unknown or the data has expired.

Source code in machine/storage/backends/base.py

@abstractmethod
async def get(self, key: str) -> bytes | None:
    """Retrieve data by key

    :param key: key for which to retrieve data
    :return: the raw data for the provided key, as (byte)string. Should return ``None`` when
        the key is unknown or the data has expired.
    """
    ...

set(key: str, value: bytes, expires: int | None = None) -> None abstractmethod async

Store data by key

Parameters:

Name	Type	Description	Default
key	str	the key under which to store the data	required
value	bytes	data as (byte)string	required
expires	int | None	optional expiration time in seconds, after which the data should not be returned any more.	None

Source code in machine/storage/backends/base.py

@abstractmethod
async def set(self, key: str, value: bytes, expires: int | None = None) -> None:
    """Store data by key

    :param key: the key under which to store the data
    :param value: data as (byte)string
    :param expires: optional expiration time in seconds, after which the data should not be
        returned any more.
    """
    ...

delete(key: str) -> None abstractmethod async

Delete data by key

Parameters:

Name	Type	Description	Default
key	str	key for which to delete the data	required

Source code in machine/storage/backends/base.py

@abstractmethod
async def delete(self, key: str) -> None:
    """Delete data by key

    :param key: key for which to delete the data
    """
    ...

has(key: str) -> bool abstractmethod async

Check if the key exists

Parameters:

Name	Type	Description	Default
key	str	key to check	required

Returns:

Type	Description
bool	True/False wether the key exists

Source code in machine/storage/backends/base.py

@abstractmethod
async def has(self, key: str) -> bool:
    """Check if the key exists

    :param key: key to check
    :return: ``True/False`` wether the key exists
    """
    ...

size() -> int abstractmethod async

Calculate the total size of the storage

Returns:

Type	Description
int	total size of storage in bytes (integer)

Source code in machine/storage/backends/base.py

@abstractmethod
async def size(self) -> int:
    """Calculate the total size of the storage

    :return: total size of storage in bytes (integer)
    """
    ...

close() -> None abstractmethod async

Close the storage backend

Source code in machine/storage/backends/base.py

@abstractmethod
async def close(self) -> None:
    """Close the storage backend"""
    ...