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 2 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

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
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__}"

    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
        """
        pass

    @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 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

    @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)

init() -> 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.

Returns:

Type Description
None

None

Source code in machine/plugins/base.py
46
47
48
49
50
51
52
53
54
55
56
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
    """
    pass

users() -> dict[str, User] property

Dictionary of all users in the Slack workspace

Returns:

Type Description
dict[str, User]

a dictionary of all users in the Slack workspace, where the key is the user id and the value is a User object

Source code in machine/plugins/base.py
58
59
60
61
62
63
64
65
@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

channels() -> dict[str, Channel] property

List of all channels in the Slack workspace

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

Returns:

Type Description
dict[str, Channel]

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

Source code in machine/plugins/base.py
67
68
69
70
71
72
73
74
75
76
77
78
@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

web_client() -> AsyncWebClient property

Slack SDK web client to access the Slack Web API

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

[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

Source code in machine/plugins/base.py
80
81
82
83
84
85
86
87
88
89
@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

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
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
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

bot_info() -> dict[str, Any] property

Information about the bot user in Slack

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

Returns:

Type Description
dict[str, Any]

Bot user

Source code in machine/plugins/base.py
104
105
106
107
108
109
110
111
112
113
@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

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
115
116
117
118
119
120
121
122
123
124
125
126
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
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
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
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
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
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
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
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
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
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
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
291
292
293
294
295
296
297
298
299
300
301
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
303
304
305
306
307
308
309
310
311
312
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
314
315
316
317
318
319
320
321
322
323
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)

machine.plugins.base.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/base.py
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
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], plugin_class_name: str):
        self._client = client
        self._msg_event = msg_event
        self._fq_plugin_name = plugin_class_name

    @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
User

the User the message was sent by

Source code in machine/plugins/base.py
343
344
345
346
347
348
349
@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"]]

channel() -> Channel property

The channel the message was sent to

Returns:

Type Description
Channel

the Channel the message was sent to

Source code in machine/plugins/base.py
351
352
353
354
355
356
357
@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"]]

text() -> str property

The body of the actual message

Returns:

Type Description
str

the body (text) of the actual message

Source code in machine/plugins/base.py
364
365
366
367
368
369
370
@property
def text(self) -> str:
    """The body of the actual message

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

at_sender() -> str property

The sender of the message formatted as mention

Returns:

Type Description
str

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

Source code in machine/plugins/base.py
372
373
374
375
376
377
378
379
380
381
@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()

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/base.py
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
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/base.py
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
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/base.py
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
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/base.py
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
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/base.py
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
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/base.py
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
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/base.py
602
603
604
605
606
607
608
609
610
611
612
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)

ts() -> str property

The timestamp of the message

Returns:

Type Description
str

the timestamp of the message

Source code in machine/plugins/base.py
620
621
622
623
624
625
626
@property
def ts(self) -> str:
    """The timestamp of the message

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

in_thread() -> bool property

Is message in a thread

Returns:

Type Description
bool

bool

Source code in machine/plugins/base.py
628
629
630
631
632
633
634
@property
def in_thread(self) -> bool:
    """Is message in a thread

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

pin_message() -> AsyncSlackResponse async

Pin message

Pin the current message in the channel it was posted in

Source code in machine/plugins/base.py
636
637
638
639
640
641
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)

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
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
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 re.RegexFlag | int

regex flags to apply when matching

re.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
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
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 re.RegexFlag | int

regex flags to apply when matching

re.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
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
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

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
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
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
172
173
174
175
176
177
178
179
180
181
182
183
184
185
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
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
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
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
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
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
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
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
class User(BaseModel):
    """
    User model that represents a user object from the Slack API
    """

    id: str
    team_id: Optional[str]
    name: str
    deleted: Optional[bool]
    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

    class Config:
        allow_mutation = False

    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
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
class Channel(BaseModel):
    """
    Channel model that represents a channel object from the Slack API
    """

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

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

    class Config:
        allow_mutation = False

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"""
    ...