Skip to content

Messaging Backends

django_slack_tools.slack_messages.backends

BaseBackend

Bases: ABC

Abstract base class for messaging backends.

Source code in django_slack_tools/slack_messages/backends/base.py 
 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
class BaseBackend(ABC):
    """Abstract base class for messaging backends."""

    def prepare_message(
        self,
        *,
        policy: SlackMessagingPolicy | None = None,
        channel: str,
        header: MessageHeader,
        body: MessageBody,
    ) -> SlackMessage:
        """Prepare message.

        Args:
            policy: Related policy instance.
            channel: Channel to send message.
            header: Slack message control header.
            body: Slack message body.

        Returns:
            Prepared message.
        """
        _header: dict = policy.header_defaults if policy else {}
        _header.update(dataclasses.asdict(header))

        _body = dataclasses.asdict(body)

        return SlackMessage(policy=policy, channel=channel, header=_header, body=_body)

    def prepare_messages_from_policy(
        self,
        policy: SlackMessagingPolicy,
        *,
        header: MessageHeader,
        context: dict[str, Any],
    ) -> list[SlackMessage]:
        """Prepare messages from policy.

        Args:
            policy: Policy to create messages from.
            header: Common message header.
            context: Message context.

        Returns:
            Prepared messages.
        """
        overridden_reserved = RESERVED_CONTEXT_KWARGS & set(context.keys())
        if overridden_reserved:
            logger.warning(
                "Template keyword argument(s) %s reserved for passing mentions, but already exists."
                " User provided value will override it.",
                ", ".join(f"`{s}`" for s in overridden_reserved),
            )

        messages: list[SlackMessage] = []
        for recipient in policy.recipients.all():
            logger.debug("Sending message to recipient %s", recipient)

            # Initialize template instance
            template = self._get_template_instance_from_policy(policy)

            # Prepare rendering arguments
            render_context = self._get_default_context(policy=policy, recipient=recipient)
            render_context.update(context)
            logger.debug("Context prepared as: %r", render_context)

            # Render template and parse as body
            rendered = template.render(context=render_context)
            body = MessageBody.from_any(rendered)

            # Create message instance
            message = self.prepare_message(policy=policy, channel=recipient.channel, header=header, body=body)
            messages.append(message)

        return SlackMessage.objects.bulk_create(messages)

    def _get_template_instance_from_policy(self, policy: SlackMessagingPolicy) -> BaseTemplate:
        """Get template instance."""
        if policy.template_type == SlackMessagingPolicy.TemplateType.DICT:
            return DictTemplate(policy.template)

        if policy.template_type == SlackMessagingPolicy.TemplateType.DJANGO:
            return DjangoTemplate(file=policy.template)

        if policy.template_type == SlackMessagingPolicy.TemplateType.DJANGO_INLINE:
            return DjangoTemplate(inline=policy.template)

        msg = f"Unsupported template type: {policy.template_type!r}"
        raise ValueError(msg)

    def _get_default_context(self, *, policy: SlackMessagingPolicy, recipient: SlackMessageRecipient) -> dict[str, Any]:
        """Get default context for rendering.

        Following default context keys are created:

        - `policy`: Policy code.
        - `mentions`: List of mentions.
        - `mentions_as_str`: Comma-separated joined string of mentions.
        """
        mentions: list[SlackMention] = list(recipient.mentions.all())
        mentions_as_str = ", ".join(mention.mention for mention in mentions)

        return {
            "policy": policy.code,
            "mentions": mentions,
            "mentions_as_str": mentions_as_str,
        }

    def send_messages(self, *messages: SlackMessage, raise_exception: bool, get_permalink: bool) -> int:
        """Shortcut to send multiple messages.

        Args:
            messages: Messages to send.
            raise_exception: Whether to propagate exceptions.
            get_permalink: Try to get the message permalink via additional Slack API call.

        Returns:
            Count of messages sent successfully.
        """
        num_sent = 0
        for message in messages:
            sent = self.send_message(message=message, raise_exception=raise_exception, get_permalink=get_permalink)
            num_sent += 1 if sent.ok else 0

        return num_sent

    def send_message(
        self,
        message: SlackMessage,
        *,
        raise_exception: bool,
        get_permalink: bool,
    ) -> SlackMessage:
        """Send message.

        Args:
            message: Prepared message.
            raise_exception: Whether to propagate exceptions.
            get_permalink: Try to get the message permalink via additional Slack API call.

        Returns:
            Message sent to Slack.
        """
        try:
            response: SlackResponse
            try:
                response = self._send_message(message)
            except SlackApiError as err:
                if raise_exception:
                    raise

                logger.warning(
                    "Error occurred while sending message but suppressed because `raise_exception` set.",
                    exc_info=err,
                )
                response = err.response

            message.ok = ok = cast(bool, response.get("ok"))
            if ok:
                # Get message TS if OK
                message.ts = cast(str, response.get("ts"))

                # Store thread TS if possible
                data: dict[str, Any] = response.get("message", {})
                message.parent_ts = data.get("thread_ts", "")

                # Get message permalink
                if get_permalink:
                    message.permalink = self._get_permalink(message=message, raise_exception=raise_exception)

            message.request = self._record_request(response)
            message.response = self._record_response(response)
        except:  # noqa: E722
            if raise_exception:
                raise

            logger.exception("Error occurred while sending message but suppressed because `raise_exception` set.")
            message.exception = traceback.format_exc()
        finally:
            message.save()

        message.refresh_from_db()
        return message

    @abstractmethod
    def _send_message(self, message: SlackMessage) -> SlackResponse:
        """Internal implementation of actual 'send message' behavior."""

    @abstractmethod
    def _get_permalink(self, *, message: SlackMessage, raise_exception: bool) -> str:
        """Get a permalink for given message identifier."""

    @abstractmethod
    def _record_request(self, response: SlackResponse) -> Any:
        """Extract request data to be recorded. Should return JSON-serializable object."""

    @abstractmethod
    def _record_response(self, response: SlackResponse) -> Any:
        """Extract response data to be recorded. Should return JSON-serializable object."""

prepare_message(*, policy=None, channel, header, body)

Prepare message.

Parameters:

Name Type Description Default
policy SlackMessagingPolicy | None

Related policy instance.

 None
channel str

Channel to send message.

 required
header MessageHeader

Slack message control header.

 required
body MessageBody

Slack message body.

 required

Returns:

Type Description
SlackMessage

Prepared message.
Source code in django_slack_tools/slack_messages/backends/base.py 
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
def prepare_message(
    self,
    *,
    policy: SlackMessagingPolicy | None = None,
    channel: str,
    header: MessageHeader,
    body: MessageBody,
) -> SlackMessage:
    """Prepare message.

    Args:
        policy: Related policy instance.
        channel: Channel to send message.
        header: Slack message control header.
        body: Slack message body.

    Returns:
        Prepared message.
    """
    _header: dict = policy.header_defaults if policy else {}
    _header.update(dataclasses.asdict(header))

    _body = dataclasses.asdict(body)

    return SlackMessage(policy=policy, channel=channel, header=_header, body=_body)

prepare_messages_from_policy(policy, *, header, context)

Prepare messages from policy.

Parameters:

Name Type Description Default
policy SlackMessagingPolicy

Policy to create messages from.

 required
header MessageHeader

Common message header.

 required
context dict[str, Any]

Message context.

 required

Returns:

Type Description
list[SlackMessage]

Prepared messages.
Source code in django_slack_tools/slack_messages/backends/base.py 
 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
def prepare_messages_from_policy(
    self,
    policy: SlackMessagingPolicy,
    *,
    header: MessageHeader,
    context: dict[str, Any],
) -> list[SlackMessage]:
    """Prepare messages from policy.

    Args:
        policy: Policy to create messages from.
        header: Common message header.
        context: Message context.

    Returns:
        Prepared messages.
    """
    overridden_reserved = RESERVED_CONTEXT_KWARGS & set(context.keys())
    if overridden_reserved:
        logger.warning(
            "Template keyword argument(s) %s reserved for passing mentions, but already exists."
            " User provided value will override it.",
            ", ".join(f"`{s}`" for s in overridden_reserved),
        )

    messages: list[SlackMessage] = []
    for recipient in policy.recipients.all():
        logger.debug("Sending message to recipient %s", recipient)

        # Initialize template instance
        template = self._get_template_instance_from_policy(policy)

        # Prepare rendering arguments
        render_context = self._get_default_context(policy=policy, recipient=recipient)
        render_context.update(context)
        logger.debug("Context prepared as: %r", render_context)

        # Render template and parse as body
        rendered = template.render(context=render_context)
        body = MessageBody.from_any(rendered)

        # Create message instance
        message = self.prepare_message(policy=policy, channel=recipient.channel, header=header, body=body)
        messages.append(message)

    return SlackMessage.objects.bulk_create(messages)

send_message(message, *, raise_exception, get_permalink)

Send message.

Parameters:

Name Type Description Default
message SlackMessage

Prepared message.

 required
raise_exception bool

Whether to propagate exceptions.

 required
get_permalink bool

Try to get the message permalink via additional Slack API call.

 required

Returns:

Type Description
SlackMessage

Message sent to Slack.
Source code in django_slack_tools/slack_messages/backends/base.py 
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
def send_message(
    self,
    message: SlackMessage,
    *,
    raise_exception: bool,
    get_permalink: bool,
) -> SlackMessage:
    """Send message.

    Args:
        message: Prepared message.
        raise_exception: Whether to propagate exceptions.
        get_permalink: Try to get the message permalink via additional Slack API call.

    Returns:
        Message sent to Slack.
    """
    try:
        response: SlackResponse
        try:
            response = self._send_message(message)
        except SlackApiError as err:
            if raise_exception:
                raise

            logger.warning(
                "Error occurred while sending message but suppressed because `raise_exception` set.",
                exc_info=err,
            )
            response = err.response

        message.ok = ok = cast(bool, response.get("ok"))
        if ok:
            # Get message TS if OK
            message.ts = cast(str, response.get("ts"))

            # Store thread TS if possible
            data: dict[str, Any] = response.get("message", {})
            message.parent_ts = data.get("thread_ts", "")

            # Get message permalink
            if get_permalink:
                message.permalink = self._get_permalink(message=message, raise_exception=raise_exception)

        message.request = self._record_request(response)
        message.response = self._record_response(response)
    except:  # noqa: E722
        if raise_exception:
            raise

        logger.exception("Error occurred while sending message but suppressed because `raise_exception` set.")
        message.exception = traceback.format_exc()
    finally:
        message.save()

    message.refresh_from_db()
    return message

send_messages(*messages, raise_exception, get_permalink)

Shortcut to send multiple messages.

Parameters:

Name Type Description Default
messages SlackMessage

Messages to send.

 ()
raise_exception bool

Whether to propagate exceptions.

 required
get_permalink bool

Try to get the message permalink via additional Slack API call.

 required

Returns:

Type Description
int

Count of messages sent successfully.
Source code in django_slack_tools/slack_messages/backends/base.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def send_messages(self, *messages: SlackMessage, raise_exception: bool, get_permalink: bool) -> int:
    """Shortcut to send multiple messages.

    Args:
        messages: Messages to send.
        raise_exception: Whether to propagate exceptions.
        get_permalink: Try to get the message permalink via additional Slack API call.

    Returns:
        Count of messages sent successfully.
    """
    num_sent = 0
    for message in messages:
        sent = self.send_message(message=message, raise_exception=raise_exception, get_permalink=get_permalink)
        num_sent += 1 if sent.ok else 0

    return num_sent

DummyBackend

Bases: BaseBackend

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
class DummyBackend(BaseBackend):
    """An dummy backend that does nothing with message."""

    def prepare_message(self, *args: Any, **kwargs: Any) -> SlackMessage:  # noqa: D102, ARG002
        return SlackMessage(header={}, body={})

    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": True},
            headers={},
            status_code=200,
        )

    def _get_permalink(self, *, message: SlackMessage, raise_exception: bool) -> str:  # noqa: ARG002
        return ""

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

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

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)

SlackBackend

Bases: BaseBackend

Backend actually sending the messages.

Source code in django_slack_tools/slack_messages/backends/slack.py 
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
class SlackBackend(BaseBackend):
    """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(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 _get_permalink(self, *, message: SlackMessage, raise_exception: bool = False) -> str:
        """Get a permalink for given message identifier."""
        if not message.ts:
            msg = "Message timestamp is not set, can't retrieve permalink."
            raise ValueError(msg)

        try:
            _permalink_resp = self._slack_app.client.chat_getPermalink(
                channel=message.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 _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 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
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

SlackRedirectBackend

Bases: SlackBackend

Inherited Slack backend with redirection to specific channels.

Source code in django_slack_tools/slack_messages/backends/slack.py 
 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
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:
        """Prepare message to send, with modified for redirection.

        Args:
            args: Positional arguments to pass to super method.
            channel: Original channel to send message.
            body: Message content.
            kwargs: Keyword arguments to pass to super method.

        Returns:
            Prepared message instance.
        """
        # 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 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
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)

prepare_message(*args, channel, body, **kwargs)

Prepare message to send, with modified for redirection.

Parameters:

Name Type Description Default
args Any

Positional arguments to pass to super method.

 ()
channel str

Original channel to send message.

 required
body MessageBody

Message content.

 required
kwargs Any

Keyword arguments to pass to super method.

 {}

Returns:

Type Description
SlackMessage

Prepared message instance.
Source code in django_slack_tools/slack_messages/backends/slack.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def prepare_message(self, *args: Any, channel: str, body: MessageBody, **kwargs: Any) -> SlackMessage:
    """Prepare message to send, with modified for redirection.

    Args:
        args: Positional arguments to pass to super method.
        channel: Original channel to send message.
        body: Message content.
        kwargs: Keyword arguments to pass to super method.

    Returns:
        Prepared message instance.
    """
    # 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)