Skip to content

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 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
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 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`_ request.

        .. _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
        )

    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)

users: dict[str, User] property

Dictionary of all users in the Slack workspace

Returns:

Type Description

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

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

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

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 
46
47
48
49
50
51
52
53
54
55
56
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 
103
104
105
106
107
108
109
110
111
112
113
114
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 
116
117
118
119
120
121
122
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 
124
125
126
127
128
129
130
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 
143
144
145
146
147
148
149
150
151
152
153
154
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 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
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 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
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,
    )

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 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
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_ request. .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage
Source code in machine/plugins/base.py 
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
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`_ request.

    .. _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 
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
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
    )

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 
319
320
321
322
323
324
325
326
327
328
329
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 
331
332
333
334
335
336
337
338
339
340
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 
342
343
344
345
346
347
348
349
350
351
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 
353
354
355
356
357
358
359
360
361
362
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)

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 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
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
        """
        if ephemeral:
            ephemeral_user = self.sender.id
        else:
            ephemeral_user = 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

the User the message was sent by

channel: Channel property

The channel the message was sent to

Returns:

Type Description

the Channel the message was sent to

text: str property

The body of the actual message

Returns:

Type Description

the body (text) of the actual message

at_sender: str property

The sender of the message formatted as mention

Returns:

Type Description

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

the timestamp of the message

in_thread: bool property

Is message in a thread

Returns:

Type Description

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 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
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
    """
    if ephemeral:
        ephemeral_user = self.sender.id
    else:
        ephemeral_user = 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 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
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 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
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 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
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 
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
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 
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
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 
290
291
292
293
294
295
296
297
298
299
300
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 
324
325
326
327
328
329
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 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
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()`

        """
        if ephemeral:
            response_type = "ephemeral"
        else:
            response_type = "in_channel"

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

sender: User property

The sender of the message

Returns:

Type Description

the User the message was sent by

channel: Channel property

The channel the message was sent to

Returns:

Type Description

the Channel the message was sent to

text: str property

The body of the actual message

Returns:

Type Description

the body (text) of the actual message

command: str property

The command that was invoked

Returns:

Type Description

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

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

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 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
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()`

    """
    if ephemeral:
        response_type = "ephemeral"
    else:
        response_type = "in_channel"

    return await self._webhook_client.send(
        text=text, attachments=attachments, blocks=blocks, response_type=response_type, **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 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
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 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
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 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
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 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
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())
        if not slash_command.startswith("/"):
            normalized_slash_command = f"/{slash_command}"
        else:
            normalized_slash_command = slash_command
        fn.metadata.plugin_actions.commands.append(
            CommandConfig(command=normalized_slash_command, is_generator=inspect.isasyncgenfunction(f))
        )
        return fn

    return command_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 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
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 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
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 
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
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 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
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 
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
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 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
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 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
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

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 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
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 
33
34
35
36
37
38
39
40
41
42
43
44
45
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 
47
48
49
50
51
52
53
54
55
56
57
58
59
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 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
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 
77
78
79
80
81
82
83
84
85
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 
87
88
89
90
91
92
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 
 94
 95
 96
 97
 98
 99
100
101
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 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
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:
        """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 
22
23
24
async def init(self) -> None:
    """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 
26
27
28
29
30
31
32
33
34
@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 
36
37
38
39
40
41
42
43
44
45
@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 
47
48
49
50
51
52
53
@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 
55
56
57
58
59
60
61
62
@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 
64
65
66
67
68
69
70
@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 
72
73
74
75
@abstractmethod
async def close(self) -> None:
    """Close the storage backend"""
    ...