Skip to content

Messaging Backends

django_slack_tools.slack_messages.backends.dummy.DummyBackend

Bases: BackendBase

An dummy backend that does nothing with message.

Source code in django_slack_tools/slack_messages/backends/dummy.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
class DummyBackend(BackendBase):
    """An dummy backend that does nothing with message."""

    def send_message(self, *args: Any, **kwargs: Any) -> SlackMessage:  # noqa: ARG002
        """This backend will not do anything, just like dummy."""
        return SlackMessage()

    def _prepare_message(self, *args: Any, **kwargs: Any) -> SlackMessage:  # noqa: ARG002
        return SlackMessage()

    def _send_message(self, *args: Any, **kwargs: Any) -> SlackResponse:  # noqa: ARG002
        return SlackResponse(
            client=None,
            http_verb="POST",
            api_url="https://www.slack.com/api/chat.postMessage",
            req_args={},
            data={"ok": False},
            headers={},
            status_code=200,
        )

    def _record_request(self, *args: Any, **kwargs: Any) -> Any: ...

    def _record_response(self, *args: Any, **kwargs: Any) -> Any: ...

send_message(*args, **kwargs)

This backend will not do anything, just like dummy.

Source code in django_slack_tools/slack_messages/backends/dummy.py 
20
21
22
def send_message(self, *args: Any, **kwargs: Any) -> SlackMessage:  # noqa: ARG002
    """This backend will not do anything, just like dummy."""
    return SlackMessage()

django_slack_tools.slack_messages.backends.logging.LoggingBackend

Bases: DummyBackend

Backend that log the message rather than sending it.

Source code in django_slack_tools/slack_messages/backends/logging.py 
17
18
19
20
21
22
23
24
25
26
27
28
class LoggingBackend(DummyBackend):
    """Backend that log the message rather than sending it."""

    def _send_message(self, *args: Any, **kwargs: Any) -> SlackResponse:
        logger.debug("Sending an message with following args=%r, kwargs=%r", args, kwargs)
        return super()._send_message(*args, **kwargs)

    def _record_request(self, *args: Any, **kwargs: Any) -> Any:
        logger.debug("Recording request with args=%r, kwargs=%r", args, kwargs)

    def _record_response(self, *args: Any, **kwargs: Any) -> Any:
        logger.debug("Recording response with args=%r, kwargs=%r", args, kwargs)

django_slack_tools.slack_messages.backends.slack.SlackBackend

Bases: BackendBase

Backend actually sending the messages.

Source code in django_slack_tools/slack_messages/backends/slack.py 
 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
class SlackBackend(BackendBase):
    """Backend actually sending the messages."""

    def __init__(
        self,
        *,
        slack_app: App | Callable[[], App] | str,
    ) -> None:
        """Initialize backend.

        Args:
            slack_app: Slack app instance or import string.
        """
        if isinstance(slack_app, str):
            slack_app = import_string(slack_app)

        if callable(slack_app):
            slack_app = slack_app()

        if not isinstance(slack_app, App):
            msg = "Couldn't resolve provided app spec into Slack app instance."
            raise ImproperlyConfigured(msg)

        self._slack_app = slack_app

    def send_message(  # noqa: PLR0913
        self,
        message: SlackMessage | None = None,
        *,
        policy: SlackMessagingPolicy | None = None,
        channel: str | None = None,
        header: MessageHeader | None = None,
        body: MessageBody | None = None,
        raise_exception: bool,
        get_permalink: bool = False,
    ) -> SlackMessage:
        """Send Slack message.

        Args:
            message: Externally prepared message.
                If not given, make one using `channel`, `header` and `body` parameters.
            policy: Messaging policy to create message with.
            channel: Channel to send message.
            header: Message header that controls how message will sent.
            body: Message body describing content of the message.
            raise_exception: Whether to re-raise caught exception while sending messages.
            get_permalink: Try to get the message permalink via extraneous Slack API calls.

        Returns:
            Sent Slack message.
        """
        if not message:
            if not (channel and header and body):
                msg = (
                    "Call signature mismatch for overload."
                    " If `message` not provided, `channel`, `header` and `body` all must given."
                )
                raise TypeError(msg)

            message = self._prepare_message(policy=policy, channel=channel, header=header, body=body)

        try:
            # Send Slack message
            response: SlackResponse
            try:
                response = self._send_message(message=message)
            except SlackApiError as err:
                if raise_exception:
                    raise

                logger.exception(
                    "Error occurred while sending Slack message, but ignored because `raise_exception` not set.",
                )
                response = err.response

            # Update message detail
            ok: bool | None = response.get("ok")
            message.ok = ok
            if ok:
                # `str` if OK, otherwise `None`
                message.ts = cast(str, response.get("ts"))
                message.parent_ts = response.get("message", {}).get("thread_ts", "")  # type: ignore[call-overload]
                if get_permalink:
                    message.permalink = self._get_permalink(
                        channel=message.channel,
                        message_ts=message.ts,
                        raise_exception=raise_exception,
                    )

            message.request = self._record_request(response)
            message.response = self._record_response(response)
        except:
            message.exception = traceback.format_exc()

            # Don't omit raise with flag `raise_exception` here
            raise
        finally:
            message.save()

        return message

    def _get_permalink(self, *, channel: str, message_ts: str, raise_exception: bool = False) -> str:
        """Get a permalink for given message identifier."""
        try:
            _permalink_resp = self._slack_app.client.chat_getPermalink(
                channel=channel,
                message_ts=message_ts,
            )
        except SlackApiError:
            if raise_exception:
                raise

            logger.exception(
                "Error occurred while sending retrieving message's permalink,"
                " but ignored as `raise_exception` not set.",
            )
            return ""

        return _permalink_resp.get("permalink", default="")

    def _send_message(self, *, message: SlackMessage) -> SlackResponse:
        header = message.header or {}
        body = message.body or {}
        return self._slack_app.client.chat_postMessage(channel=message.channel, **header, **body)

    def _record_request(self, response: SlackResponse) -> dict[str, Any]:
        # Remove auth header (token) from request before recording
        response.req_args.get("headers", {}).pop("Authorization", None)

        return response.req_args

    def _record_response(self, response: SlackResponse) -> dict[str, Any]:
        return {
            "http_verb": response.http_verb,
            "api_url": response.api_url,
            "status_code": response.status_code,
            "headers": response.headers,
            "data": response.data,
        }

__init__(*, slack_app)

Initialize backend.

Parameters:

Name Type Description Default
slack_app App | Callable[[], App] | str

Slack app instance or import string.

 required
Source code in django_slack_tools/slack_messages/backends/slack.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def __init__(
    self,
    *,
    slack_app: App | Callable[[], App] | str,
) -> None:
    """Initialize backend.

    Args:
        slack_app: Slack app instance or import string.
    """
    if isinstance(slack_app, str):
        slack_app = import_string(slack_app)

    if callable(slack_app):
        slack_app = slack_app()

    if not isinstance(slack_app, App):
        msg = "Couldn't resolve provided app spec into Slack app instance."
        raise ImproperlyConfigured(msg)

    self._slack_app = slack_app

send_message(message=None, *, policy=None, channel=None, header=None, body=None, raise_exception, get_permalink=False)

Send Slack message.

Parameters:

Name Type Description Default
message SlackMessage | None

Externally prepared message. If not given, make one using channel, header and body parameters.

 None
policy SlackMessagingPolicy | None

Messaging policy to create message with.

 None
channel str | None

Channel to send message.

 None
header MessageHeader | None

Message header that controls how message will sent.

 None
body MessageBody | None

Message body describing content of the message.

 None
raise_exception bool

Whether to re-raise caught exception while sending messages.

 required
get_permalink bool

Try to get the message permalink via extraneous Slack API calls.

 False

Returns:

Type Description
SlackMessage

Sent Slack message.
Source code in django_slack_tools/slack_messages/backends/slack.py 
 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
def send_message(  # noqa: PLR0913
    self,
    message: SlackMessage | None = None,
    *,
    policy: SlackMessagingPolicy | None = None,
    channel: str | None = None,
    header: MessageHeader | None = None,
    body: MessageBody | None = None,
    raise_exception: bool,
    get_permalink: bool = False,
) -> SlackMessage:
    """Send Slack message.

    Args:
        message: Externally prepared message.
            If not given, make one using `channel`, `header` and `body` parameters.
        policy: Messaging policy to create message with.
        channel: Channel to send message.
        header: Message header that controls how message will sent.
        body: Message body describing content of the message.
        raise_exception: Whether to re-raise caught exception while sending messages.
        get_permalink: Try to get the message permalink via extraneous Slack API calls.

    Returns:
        Sent Slack message.
    """
    if not message:
        if not (channel and header and body):
            msg = (
                "Call signature mismatch for overload."
                " If `message` not provided, `channel`, `header` and `body` all must given."
            )
            raise TypeError(msg)

        message = self._prepare_message(policy=policy, channel=channel, header=header, body=body)

    try:
        # Send Slack message
        response: SlackResponse
        try:
            response = self._send_message(message=message)
        except SlackApiError as err:
            if raise_exception:
                raise

            logger.exception(
                "Error occurred while sending Slack message, but ignored because `raise_exception` not set.",
            )
            response = err.response

        # Update message detail
        ok: bool | None = response.get("ok")
        message.ok = ok
        if ok:
            # `str` if OK, otherwise `None`
            message.ts = cast(str, response.get("ts"))
            message.parent_ts = response.get("message", {}).get("thread_ts", "")  # type: ignore[call-overload]
            if get_permalink:
                message.permalink = self._get_permalink(
                    channel=message.channel,
                    message_ts=message.ts,
                    raise_exception=raise_exception,
                )

        message.request = self._record_request(response)
        message.response = self._record_response(response)
    except:
        message.exception = traceback.format_exc()

        # Don't omit raise with flag `raise_exception` here
        raise
    finally:
        message.save()

    return message

django_slack_tools.slack_messages.backends.slack.SlackRedirectBackend

Bases: SlackBackend

Inherited Slack backend with redirection to specific channels.

Source code in django_slack_tools/slack_messages/backends/slack.py 
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
class SlackRedirectBackend(SlackBackend):
    """Inherited Slack backend with redirection to specific channels."""

    def __init__(self, *, slack_app: App | str, redirect_channel: str, inform_redirect: bool = True) -> None:
        """Initialize backend.

        Args:
            slack_app: Slack app instance or import string.
            redirect_channel: Slack channel to redirect.
            inform_redirect: Whether to append an attachment informing that the message has been redirected.
                Defaults to `True`.
        """
        self.redirect_channel = redirect_channel
        self.inform_redirect = inform_redirect

        super().__init__(slack_app=slack_app)

    def _prepare_message(self, *args: Any, channel: str, body: MessageBody, **kwargs: Any) -> SlackMessage:
        # Modify channel to force messages always sent to specific channel
        # Add an attachment that informing message has been redirected
        if self.inform_redirect:
            body.attachments = [
                self._make_inform_attachment(original_channel=channel),
                *(body.attachments or []),
            ]

        return super()._prepare_message(*args, channel=self.redirect_channel, body=body, **kwargs)

    def _make_inform_attachment(self, *, original_channel: str) -> dict[str, Any]:
        msg_redirect_inform = _(
            ":warning:  This message was originally sent to channel *{channel}* but redirected here.",
        )

        return {
            "color": "#eb4034",
            "text": msg_redirect_inform.format(channel=original_channel),
        }

__init__(*, slack_app, redirect_channel, inform_redirect=True)

Initialize backend.

Parameters:

Name Type Description Default
slack_app App | str

Slack app instance or import string.

 required
redirect_channel str

Slack channel to redirect.

 required
inform_redirect bool

Whether to append an attachment informing that the message has been redirected. Defaults to True.

 True
Source code in django_slack_tools/slack_messages/backends/slack.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
def __init__(self, *, slack_app: App | str, redirect_channel: str, inform_redirect: bool = True) -> None:
    """Initialize backend.

    Args:
        slack_app: Slack app instance or import string.
        redirect_channel: Slack channel to redirect.
        inform_redirect: Whether to append an attachment informing that the message has been redirected.
            Defaults to `True`.
    """
    self.redirect_channel = redirect_channel
    self.inform_redirect = inform_redirect

    super().__init__(slack_app=slack_app)