Skip to content

Messenger Framework

django_slack_tools.slack_messages.messenger

Messenger

Messenger class that sends message using templates and middlewares.

Components evaluated in order:

  1. Request processing middlewares
  2. Load template by key and render message in-place
  3. Send message
  4. Response processing middlewares (in reverse order)
Source code in django_slack_tools/slack_messages/messenger.py 
 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
class Messenger:
    """Messenger class that sends message using templates and middlewares.

    Components evaluated in order:

    1. Request processing middlewares
    2. Load template by key and render message in-place
    4. Send message
    5. Response processing middlewares (in reverse order)
    """

    def __init__(
        self,
        *,
        template_loaders: Sequence[BaseTemplateLoader],
        middlewares: Sequence[BaseMiddleware],
        messaging_backend: BaseBackend,
    ) -> None:
        """Initialize the Messenger.

        Args:
            template_loaders: A sequence of template loaders.
                It is tried in order to load the template and the first one that returns a template is used.
            middlewares: A sequence of middlewares.
                Middlewares are applied in the order they are provided for request, and in reverse order for response.
            messaging_backend: The messaging backend to be used.
        """
        # Validate the template loaders
        for tl in template_loaders:
            if not isinstance(tl, BaseTemplateLoader):
                msg = f"Expected inherited from {BaseTemplateLoader!s}, got {type(tl)}"
                raise TypeError(msg)

        self.template_loaders = template_loaders

        # Validate the middlewares
        for mw in middlewares:
            if not isinstance(mw, BaseMiddleware):
                msg = f"Expected inherited from {BaseMiddleware!s}, got {type(mw)}"
                raise TypeError(msg)

        self.middlewares = middlewares

        # Validate the messaging backend
        if not isinstance(messaging_backend, BaseBackend):
            msg = f"Expected inherited from {BaseBackend!s}, got {type(messaging_backend)}"
            raise TypeError(msg)

        self.messaging_backend = messaging_backend

    def send(
        self,
        to: str,
        *,
        template: str | None = None,
        context: dict[str, str],
        header: MessageHeader | dict[str, Any] | None = None,
    ) -> MessageResponse | None:
        """Simplified shortcut for `.send_request()`."""
        header = MessageHeader.model_validate(header or {})
        request = MessageRequest(template_key=template, channel=to, context=context, header=header)
        return self.send_request(request=request)

    def send_request(self, request: MessageRequest) -> MessageResponse | None:
        """Sends a message request and processes the response."""
        logger.info("Sending request: %s", request)
        _request = self._process_request(request)
        if _request is None:
            return None

        self._render_message(_request)
        response = self._deliver_message(_request)
        _response = self._process_response(response)
        if _response is None:
            return None

        logger.info("Response: %s", _response)
        return response

    def _process_request(self, request: MessageRequest) -> MessageRequest | None:
        """Processes the request with middlewares in forward order."""
        for middleware in self.middlewares:
            logger.debug("Processing request (%s) with middleware %s", request, middleware)
            new_request = middleware.process_request(request)
            if new_request is None:
                logger.warning("Middleware %s returned `None`, skipping remaining middlewares", middleware)
                return None

            request = new_request

        logger.debug("Request after processing: %s", request)
        return request

    def _render_message(self, request: MessageRequest) -> None:
        """Updates the request with rendered message, in-place."""
        if request.body is not None:
            logger.debug("Request already has a body, skipping rendering")
            return

        if request.template_key is None:
            msg = "Template key is required to render the message"
            raise ValueError(msg)

        template = self._get_template(request.template_key)
        logger.debug("Rendering request %s with template: %s", request, template)
        rendered = template.render(request.context)
        request.body = MessageBody.model_validate(rendered)

    def _get_template(self, key: str) -> BaseTemplate:
        """Loads the template by key."""
        for loader in self.template_loaders:
            template = loader.load(key)
            if template is not None:
                return template

        msg = f"Template with key '{key}' not found"
        raise TemplateNotFoundError(msg)

    def _deliver_message(self, request: MessageRequest) -> MessageResponse:
        """Invoke the messaging backend to deliver the message."""
        logger.debug("Delivering message request: %s", request)
        response = self.messaging_backend.deliver(request)
        logger.debug("Response after delivery: %s", response)
        return response

    def _process_response(self, response: MessageResponse) -> MessageResponse | None:
        """Processes the response with middlewares in reverse order."""
        for middleware in reversed(self.middlewares):
            logger.debug("Processing response (%s) with middleware: %s", response, middleware)
            new_response = middleware.process_response(response)
            if new_response is None:
                logger.warning("Middleware %s returned `None`, skipping remaining middlewares", middleware)
                return None

            response = new_response

        logger.debug("Response after processing: %s", response)
        return response

__init__(*, template_loaders, middlewares, messaging_backend)

Initialize the Messenger.

Parameters:

Name Type Description Default
template_loaders Sequence[BaseTemplateLoader]

A sequence of template loaders. It is tried in order to load the template and the first one that returns a template is used.

 required
middlewares Sequence[BaseMiddleware]

A sequence of middlewares. Middlewares are applied in the order they are provided for request, and in reverse order for response.

 required
messaging_backend BaseBackend

The messaging backend to be used.

 required
Source code in django_slack_tools/slack_messages/messenger.py 
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
def __init__(
    self,
    *,
    template_loaders: Sequence[BaseTemplateLoader],
    middlewares: Sequence[BaseMiddleware],
    messaging_backend: BaseBackend,
) -> None:
    """Initialize the Messenger.

    Args:
        template_loaders: A sequence of template loaders.
            It is tried in order to load the template and the first one that returns a template is used.
        middlewares: A sequence of middlewares.
            Middlewares are applied in the order they are provided for request, and in reverse order for response.
        messaging_backend: The messaging backend to be used.
    """
    # Validate the template loaders
    for tl in template_loaders:
        if not isinstance(tl, BaseTemplateLoader):
            msg = f"Expected inherited from {BaseTemplateLoader!s}, got {type(tl)}"
            raise TypeError(msg)

    self.template_loaders = template_loaders

    # Validate the middlewares
    for mw in middlewares:
        if not isinstance(mw, BaseMiddleware):
            msg = f"Expected inherited from {BaseMiddleware!s}, got {type(mw)}"
            raise TypeError(msg)

    self.middlewares = middlewares

    # Validate the messaging backend
    if not isinstance(messaging_backend, BaseBackend):
        msg = f"Expected inherited from {BaseBackend!s}, got {type(messaging_backend)}"
        raise TypeError(msg)

    self.messaging_backend = messaging_backend

_deliver_message(request)

Invoke the messaging backend to deliver the message.

Source code in django_slack_tools/slack_messages/messenger.py 
139
140
141
142
143
144
def _deliver_message(self, request: MessageRequest) -> MessageResponse:
    """Invoke the messaging backend to deliver the message."""
    logger.debug("Delivering message request: %s", request)
    response = self.messaging_backend.deliver(request)
    logger.debug("Response after delivery: %s", response)
    return response

_get_template(key)

Loads the template by key.

Source code in django_slack_tools/slack_messages/messenger.py 
129
130
131
132
133
134
135
136
137
def _get_template(self, key: str) -> BaseTemplate:
    """Loads the template by key."""
    for loader in self.template_loaders:
        template = loader.load(key)
        if template is not None:
            return template

    msg = f"Template with key '{key}' not found"
    raise TemplateNotFoundError(msg)

_process_request(request)

Processes the request with middlewares in forward order.

Source code in django_slack_tools/slack_messages/messenger.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
def _process_request(self, request: MessageRequest) -> MessageRequest | None:
    """Processes the request with middlewares in forward order."""
    for middleware in self.middlewares:
        logger.debug("Processing request (%s) with middleware %s", request, middleware)
        new_request = middleware.process_request(request)
        if new_request is None:
            logger.warning("Middleware %s returned `None`, skipping remaining middlewares", middleware)
            return None

        request = new_request

    logger.debug("Request after processing: %s", request)
    return request

_process_response(response)

Processes the response with middlewares in reverse order.

Source code in django_slack_tools/slack_messages/messenger.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
def _process_response(self, response: MessageResponse) -> MessageResponse | None:
    """Processes the response with middlewares in reverse order."""
    for middleware in reversed(self.middlewares):
        logger.debug("Processing response (%s) with middleware: %s", response, middleware)
        new_response = middleware.process_response(response)
        if new_response is None:
            logger.warning("Middleware %s returned `None`, skipping remaining middlewares", middleware)
            return None

        response = new_response

    logger.debug("Response after processing: %s", response)
    return response

_render_message(request)

Updates the request with rendered message, in-place.

Source code in django_slack_tools/slack_messages/messenger.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def _render_message(self, request: MessageRequest) -> None:
    """Updates the request with rendered message, in-place."""
    if request.body is not None:
        logger.debug("Request already has a body, skipping rendering")
        return

    if request.template_key is None:
        msg = "Template key is required to render the message"
        raise ValueError(msg)

    template = self._get_template(request.template_key)
    logger.debug("Rendering request %s with template: %s", request, template)
    rendered = template.render(request.context)
    request.body = MessageBody.model_validate(rendered)

send(to, *, template=None, context, header=None)

Simplified shortcut for .send_request().

Source code in django_slack_tools/slack_messages/messenger.py 
71
72
73
74
75
76
77
78
79
80
81
82
def send(
    self,
    to: str,
    *,
    template: str | None = None,
    context: dict[str, str],
    header: MessageHeader | dict[str, Any] | None = None,
) -> MessageResponse | None:
    """Simplified shortcut for `.send_request()`."""
    header = MessageHeader.model_validate(header or {})
    request = MessageRequest(template_key=template, channel=to, context=context, header=header)
    return self.send_request(request=request)

send_request(request)

Sends a message request and processes the response.

Source code in django_slack_tools/slack_messages/messenger.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def send_request(self, request: MessageRequest) -> MessageResponse | None:
    """Sends a message request and processes the response."""
    logger.info("Sending request: %s", request)
    _request = self._process_request(request)
    if _request is None:
        return None

    self._render_message(_request)
    response = self._deliver_message(_request)
    _response = self._process_response(response)
    if _response is None:
        return None

    logger.info("Response: %s", _response)
    return response

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 
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
class BaseBackend(ABC):
    """Abstract base class for messaging backends."""

    def deliver(self, request: MessageRequest) -> MessageResponse:
        """Deliver message request."""
        if request.body is None:
            msg = "Message body is required."
            raise ValueError(msg)

        try:
            response = self._send_message(
                channel=request.channel,
                header=request.header,
                body=request.body,
            )
            error = None
        except SlackApiError as err:
            response = err.response
            error = traceback.format_exc()

        ok = cast("bool", response.get("ok"))
        data: Any
        if ok:
            ts = cast("Optional[str]", response.get("ts", None))
            data = response.get("message", {})
            parent_ts = data.get("thread_ts", None)
        else:
            ts = None
            data = response.data
            parent_ts = None

        return MessageResponse(
            request=request,
            ok=ok,
            error=error,
            data=data,
            ts=ts,
            parent_ts=parent_ts,
        )

    @abstractmethod
    def _send_message(self, *, channel: str, header: MessageHeader, body: MessageBody) -> SlackResponse:
        """Internal implementation of actual 'send message' behavior."""

_send_message(*, channel, header, body) abstractmethod

Internal implementation of actual 'send message' behavior.

Source code in django_slack_tools/slack_messages/backends/base.py 
63
64
65
@abstractmethod
def _send_message(self, *, channel: str, header: MessageHeader, body: MessageBody) -> SlackResponse:
    """Internal implementation of actual 'send message' behavior."""

deliver(request)

Deliver message request.

Source code in django_slack_tools/slack_messages/backends/base.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
def deliver(self, request: MessageRequest) -> MessageResponse:
    """Deliver message request."""
    if request.body is None:
        msg = "Message body is required."
        raise ValueError(msg)

    try:
        response = self._send_message(
            channel=request.channel,
            header=request.header,
            body=request.body,
        )
        error = None
    except SlackApiError as err:
        response = err.response
        error = traceback.format_exc()

    ok = cast("bool", response.get("ok"))
    data: Any
    if ok:
        ts = cast("Optional[str]", response.get("ts", None))
        data = response.get("message", {})
        parent_ts = data.get("thread_ts", None)
    else:
        ts = None
        data = response.data
        parent_ts = None

    return MessageResponse(
        request=request,
        ok=ok,
        error=error,
        data=data,
        ts=ts,
        parent_ts=parent_ts,
    )

DummyBackend

Bases: BaseBackend

An dummy backend that does nothing with message.

Source code in django_slack_tools/slack_messages/backends/dummy.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
class DummyBackend(BaseBackend):
    """An dummy backend that does nothing with message."""

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

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

SlackBackend

Bases: BaseBackend

Backend actually sending the messages.

Source code in django_slack_tools/slack_messages/backends/slack.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
class SlackBackend(BaseBackend):
    """Backend actually sending the messages."""

    def __init__(self, *, slack_app: 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 not isinstance(slack_app, App):
            msg = f"Expected {App!s} instance, got {type(slack_app)}"
            raise TypeError(msg)

        self._slack_app = slack_app

    def _send_message(self, *, channel: str, header: MessageHeader, body: MessageBody) -> SlackResponse:
        return self._slack_app.client.chat_postMessage(
            channel=channel,
            **header.model_dump(),
            **body.model_dump(),
        )

__init__(*, slack_app)

Initialize backend.

Parameters:

Name Type Description Default
slack_app App | str

Slack app instance or import string.

 required
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
def __init__(self, *, slack_app: 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 not isinstance(slack_app, App):
        msg = f"Expected {App!s} instance, got {type(slack_app)}"
        raise TypeError(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 
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
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 _send_message(self, *, channel: str, header: MessageHeader, body: MessageBody) -> SlackResponse:
        if self.inform_redirect:
            attachments = body.attachments or []
            attachments.append(
                self._make_inform_attachment(original_channel=channel),
            )
            body.attachments = attachments

        return self._slack_app.client.chat_postMessage(
            channel=self.redirect_channel,
            **header.model_dump(),
            **body.model_dump(),
        )

    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 
52
53
54
55
56
57
58
59
60
61
62
63
64
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)

django_slack_tools.slack_messages.middlewares

BaseMiddleware

Base class for middleware components.

Source code in django_slack_tools/slack_messages/middlewares/base.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
class BaseMiddleware:
    """Base class for middleware components."""

    def process_request(self, request: MessageRequest) -> MessageRequest | None:  # pragma: no cover
        """Process the incoming requests.

        Args:
            request: Message request.

        Returns:
            MessageRequest objects or `None`.
        """
        return request

    def process_response(self, response: MessageResponse) -> MessageResponse | None:  # pragma: no cover
        """Processes a sequence of MessageResponse objects and returns the processed sequence.

        Args:
            response: Message response.

        Returns:
            MessageResponse objects or `None`.
        """
        return response

process_request(request)

Process the incoming requests.

Parameters:

Name Type Description Default
request MessageRequest

Message request.

 required

Returns:

Type Description
MessageRequest | None

MessageRequest objects or None.
Source code in django_slack_tools/slack_messages/middlewares/base.py 
14
15
16
17
18
19
20
21
22
23
def process_request(self, request: MessageRequest) -> MessageRequest | None:  # pragma: no cover
    """Process the incoming requests.

    Args:
        request: Message request.

    Returns:
        MessageRequest objects or `None`.
    """
    return request

process_response(response)

Processes a sequence of MessageResponse objects and returns the processed sequence.

Parameters:

Name Type Description Default
response MessageResponse

Message response.

 required

Returns:

Type Description
MessageResponse | None

MessageResponse objects or None.
Source code in django_slack_tools/slack_messages/middlewares/base.py 
25
26
27
28
29
30
31
32
33
34
def process_response(self, response: MessageResponse) -> MessageResponse | None:  # pragma: no cover
    """Processes a sequence of MessageResponse objects and returns the processed sequence.

    Args:
        response: Message response.

    Returns:
        MessageResponse objects or `None`.
    """
    return response

DjangoDatabasePersister

Bases: BaseMiddleware

Persist message history to database. If request is None, will do nothing.

Source code in django_slack_tools/slack_messages/middlewares/django.py 
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
class DjangoDatabasePersister(BaseMiddleware):
    """Persist message history to database. If request is `None`, will do nothing."""

    def __init__(self, *, slack_app: App | None = None, get_permalink: bool = False) -> None:
        """Initialize the middleware.

        Args:
            slack_app: Slack app instance to use for certain tasks, such as getting permalinks.
            get_permalink: If `True`, will try to get the permalink of the message.
        """
        if get_permalink and not isinstance(slack_app, App):
            msg = "`slack_app` must be an instance of `App` if `get_permalink` is set `True`."
            raise ValueError(msg)

        self.slack_app = slack_app
        self.get_permalink = get_permalink

    def process_response(self, response: MessageResponse) -> MessageResponse | None:  # noqa: D102
        request = response.request
        if request is None:
            logger.warning("No request found in response, skipping persister.")
            return response

        logger.debug("Getting permalink for message: %s", response)
        if self.get_permalink:  # noqa: SIM108
            permalink = self._get_permalink(channel=request.channel, ts=response.ts)
        else:
            permalink = ""

        logger.debug("Persisting message history to database: %s", response)
        try:
            history = SlackMessage(
                id=request.id_,
                channel=request.channel,
                header=request.header.model_dump(),
                body=request.body.model_dump() if request.body else {},
                ok=response.ok,
                permalink=permalink,
                ts=response.ts,
                parent_ts=response.parent_ts or "",
                request=request.model_dump(),
                response=response.model_dump(exclude={"request"}),
                exception=response.error or "",
            )
            history.save()
        except Exception:
            logger.exception("Error while saving message history: %s", response)

        return response

    def _get_permalink(self, *, channel: str, ts: str | None) -> str:
        """Get permalink of the message. It returns empty string on error."""
        if not self.slack_app:
            logger.warning("Slack app not provided, cannot get permalink.")
            return ""

        if not ts:
            logger.warning("No message ts provided, cannot get permalink.")
            return ""

        try:
            response = self.slack_app.client.chat_getPermalink(channel=channel, message_ts=ts)
            return response.get("permalink", default="")
        except SlackApiError as err:
            logger.debug("Error while getting permalink: %s", exc_info=err)
            return ""

__init__(*, slack_app=None, get_permalink=False)

Initialize the middleware.

Parameters:

Name Type Description Default
slack_app App | None

Slack app instance to use for certain tasks, such as getting permalinks.

 None
get_permalink bool

If True, will try to get the permalink of the message.

 False
Source code in django_slack_tools/slack_messages/middlewares/django.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
def __init__(self, *, slack_app: App | None = None, get_permalink: bool = False) -> None:
    """Initialize the middleware.

    Args:
        slack_app: Slack app instance to use for certain tasks, such as getting permalinks.
        get_permalink: If `True`, will try to get the permalink of the message.
    """
    if get_permalink and not isinstance(slack_app, App):
        msg = "`slack_app` must be an instance of `App` if `get_permalink` is set `True`."
        raise ValueError(msg)

    self.slack_app = slack_app
    self.get_permalink = get_permalink

Get permalink of the message. It returns empty string on error.

Source code in django_slack_tools/slack_messages/middlewares/django.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def _get_permalink(self, *, channel: str, ts: str | None) -> str:
    """Get permalink of the message. It returns empty string on error."""
    if not self.slack_app:
        logger.warning("Slack app not provided, cannot get permalink.")
        return ""

    if not ts:
        logger.warning("No message ts provided, cannot get permalink.")
        return ""

    try:
        response = self.slack_app.client.chat_getPermalink(channel=channel, message_ts=ts)
        return response.get("permalink", default="")
    except SlackApiError as err:
        logger.debug("Error while getting permalink: %s", exc_info=err)
        return ""

DjangoDatabasePolicyHandler

Bases: BaseMiddleware

Middleware to handle Slack messaging policies stored in the database.

Be cautious when using this middleware because it includes functionality to distribute messages to multiple recipients, which could lead to unwanted infinite loop or recursion if used improperly.

This middleware contains a secondary protection against infinite loops by injecting a context key to the message context. If the key is found in the context, the middleware will stop the message from being sent. So be careful when modifying the context.

Source code in django_slack_tools/slack_messages/middlewares/django.py 
 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
class DjangoDatabasePolicyHandler(BaseMiddleware):
    """Middleware to handle Slack messaging policies stored in the database.

    Be cautious when using this middleware because it includes functionality to distribute messages to multiple recipients,
    which could lead to unwanted infinite loop or recursion if used improperly.

    This middleware contains a secondary protection against infinite loops by injecting a context key to the message context.
    If the key is found in the context, the middleware will stop the message from being sent. So be careful when modifying the context.
    """  # noqa: E501

    _RECURSION_DETECTION_CONTEXT_KEY = "__final__"
    """Recursion detection key injected to message context for fanned-out messages to provide secondary protection against infinite loops."""  # noqa: E501

    def __init__(
        self,
        *,
        messenger: Messenger | str,
        on_policy_not_exists: OnPolicyNotExists = "error",
    ) -> None:
        """Initialize the middleware.

        This middleware will load the policy from the database and send the message to all recipients.

        Args:
            messenger: Messenger instance or name to use for sending messages.
                The messenger instance should be different from the one used in the policy handler,
                because this middleware cannot properly handle fanned-out messages modified by this middleware.
                Also, there are chances of infinite loops if the same messenger is used.
            on_policy_not_exists: Action to take when policy is not found.
        """
        if on_policy_not_exists not in ("create", "default", "error"):
            msg = f'Unknown value for `on_policy_not_exists`: "{on_policy_not_exists}"'
            raise ValueError(msg)

        self._messenger = messenger
        self.on_policy_not_exists = on_policy_not_exists

    # * It's not desirable to put import in the method,
    # * but it's the only way to avoid circular imports for now (what's the fix?)
    @property
    def messenger(self) -> Messenger:
        """Get the messenger instance. If it's a string, will get the messenger from the app settings."""
        if isinstance(self._messenger, str):
            from django_slack_tools.app_settings import get_messenger

            self._messenger = get_messenger(self._messenger)

        return self._messenger

    def process_request(self, request: MessageRequest) -> MessageRequest | None:  # noqa: D102
        # TODO(lasuillard): Hacky way to stop the request, need to find a better way
        #                   Some extra field (request.meta) could be added to share control context
        if request.context.get(self._RECURSION_DETECTION_CONTEXT_KEY, False):
            return request

        code = request.channel
        policy = self._get_policy(code=code)
        if not policy.enabled:
            logger.debug("Policy %s is disabled, skipping further messaging", policy)
            return None

        requests: list[MessageRequest] = []
        for recipient in policy.recipients.all():
            default_context = self._get_default_context(recipient)
            context = {
                **default_context,
                **request.context,
                self._RECURSION_DETECTION_CONTEXT_KEY: True,
            }
            header = MessageHeader.model_validate(
                {
                    **policy.header_defaults,
                    **request.header.model_dump(),
                },
            )
            req = MessageRequest(channel=recipient.channel, template_key=policy.code, context=context, header=header)
            requests.append(req)

        # TODO(lasuillard): How to provide users the access the newly created messages?
        #                   currently, it's possible with persisters but it would require some additional work
        # TODO(lasuillard): Can `sys.setrecursionlimit` be used to prevent spamming if recursion occurs?
        for req in requests:
            self.messenger.send_request(req)

        # Stop current request
        return None

    def _get_policy(self, *, code: str) -> SlackMessagingPolicy:
        """Get the policy for the given code."""
        try:
            policy = SlackMessagingPolicy.objects.get(code=code)
        except SlackMessagingPolicy.DoesNotExist:
            if self.on_policy_not_exists == "create":
                logger.warning("No policy found for template key, creating one: %s", code)
                policy = self._create_policy(code=code)
            elif self.on_policy_not_exists == "default":
                policy = SlackMessagingPolicy.objects.get(code="DEFAULT")
            elif self.on_policy_not_exists == "error":
                raise
            else:
                msg = f'Unknown value for `on_policy_not_exists`: "{self.on_policy_not_exists}"'
                raise ValueError(msg) from None

        return policy

    def _create_policy(self, *, code: str) -> SlackMessagingPolicy:
        """Create a policy with the given code.

        Policy created is disabled by default, thus no message will be sent.
        To modify the default policy creation behavior, simply override this method.
        """
        policy = SlackMessagingPolicy.objects.create(
            code=code,
            enabled=False,
            template_type=SlackMessagingPolicy.TemplateType.UNKNOWN,
        )
        default_recipients = SlackMessageRecipient.objects.filter(alias="DEFAULT")
        policy.recipients.set(default_recipients)
        return policy

    def _get_default_context(self, recipient: SlackMessageRecipient) -> dict:
        """Create default context for the recipient."""
        mentions = [mention.mention for mention in recipient.mentions.all()]
        return {
            "mentions": mentions,
        }

_RECURSION_DETECTION_CONTEXT_KEY = '__final__' class-attribute instance-attribute

Recursion detection key injected to message context for fanned-out messages to provide secondary protection against infinite loops.

messenger property

Get the messenger instance. If it's a string, will get the messenger from the app settings.

__init__(*, messenger, on_policy_not_exists='error')

Initialize the middleware.

This middleware will load the policy from the database and send the message to all recipients.

Parameters:

Name Type Description Default
messenger Messenger | str

Messenger instance or name to use for sending messages. The messenger instance should be different from the one used in the policy handler, because this middleware cannot properly handle fanned-out messages modified by this middleware. Also, there are chances of infinite loops if the same messenger is used.

 required
on_policy_not_exists OnPolicyNotExists

Action to take when policy is not found.

 'error'
Source code in django_slack_tools/slack_messages/middlewares/django.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def __init__(
    self,
    *,
    messenger: Messenger | str,
    on_policy_not_exists: OnPolicyNotExists = "error",
) -> None:
    """Initialize the middleware.

    This middleware will load the policy from the database and send the message to all recipients.

    Args:
        messenger: Messenger instance or name to use for sending messages.
            The messenger instance should be different from the one used in the policy handler,
            because this middleware cannot properly handle fanned-out messages modified by this middleware.
            Also, there are chances of infinite loops if the same messenger is used.
        on_policy_not_exists: Action to take when policy is not found.
    """
    if on_policy_not_exists not in ("create", "default", "error"):
        msg = f'Unknown value for `on_policy_not_exists`: "{on_policy_not_exists}"'
        raise ValueError(msg)

    self._messenger = messenger
    self.on_policy_not_exists = on_policy_not_exists

_create_policy(*, code)

Create a policy with the given code.

Policy created is disabled by default, thus no message will be sent. To modify the default policy creation behavior, simply override this method.

Source code in django_slack_tools/slack_messages/middlewares/django.py 
198
199
200
201
202
203
204
205
206
207
208
209
210
211
def _create_policy(self, *, code: str) -> SlackMessagingPolicy:
    """Create a policy with the given code.

    Policy created is disabled by default, thus no message will be sent.
    To modify the default policy creation behavior, simply override this method.
    """
    policy = SlackMessagingPolicy.objects.create(
        code=code,
        enabled=False,
        template_type=SlackMessagingPolicy.TemplateType.UNKNOWN,
    )
    default_recipients = SlackMessageRecipient.objects.filter(alias="DEFAULT")
    policy.recipients.set(default_recipients)
    return policy

_get_default_context(recipient)

Create default context for the recipient.

Source code in django_slack_tools/slack_messages/middlewares/django.py 
213
214
215
216
217
218
def _get_default_context(self, recipient: SlackMessageRecipient) -> dict:
    """Create default context for the recipient."""
    mentions = [mention.mention for mention in recipient.mentions.all()]
    return {
        "mentions": mentions,
    }

_get_policy(*, code)

Get the policy for the given code.

Source code in django_slack_tools/slack_messages/middlewares/django.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def _get_policy(self, *, code: str) -> SlackMessagingPolicy:
    """Get the policy for the given code."""
    try:
        policy = SlackMessagingPolicy.objects.get(code=code)
    except SlackMessagingPolicy.DoesNotExist:
        if self.on_policy_not_exists == "create":
            logger.warning("No policy found for template key, creating one: %s", code)
            policy = self._create_policy(code=code)
        elif self.on_policy_not_exists == "default":
            policy = SlackMessagingPolicy.objects.get(code="DEFAULT")
        elif self.on_policy_not_exists == "error":
            raise
        else:
            msg = f'Unknown value for `on_policy_not_exists`: "{self.on_policy_not_exists}"'
            raise ValueError(msg) from None

    return policy

django_slack_tools.slack_messages.message_templates

BaseTemplate

Bases: ABC, Generic[_T]

Base class for templates.

Source code in django_slack_tools/slack_messages/message_templates/base.py 
10
11
12
13
14
15
16
17
class BaseTemplate(ABC, Generic[_T]):
    """Base class for templates."""

    template: _T

    @abstractmethod
    def render(self, context: dict[str, Any]) -> Any:
        """Render the template with the given context."""

render(context) abstractmethod

Render the template with the given context.

Source code in django_slack_tools/slack_messages/message_templates/base.py 
15
16
17
@abstractmethod
def render(self, context: dict[str, Any]) -> Any:
    """Render the template with the given context."""

DjangoTemplate

Bases: BaseTemplate

Template utilizing Django built-in template engine.

Source code in django_slack_tools/slack_messages/message_templates/django.py 
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
class DjangoTemplate(BaseTemplate):
    """Template utilizing Django built-in template engine."""

    template: Template

    @overload
    def __init__(self, *, file: str, engine: BaseEngine | None = None) -> None: ...  # pragma: no cover

    @overload
    def __init__(self, *, inline: str, engine: BaseEngine | None = None) -> None: ...  # pragma: no cover

    def __init__(
        self,
        *,
        file: str | None = None,
        inline: str | None = None,
        engine: BaseEngine | None = None,
    ) -> None:
        """Initialize template.

        Args:
            file: Path to file with template.
            inline: XML inline template.
            engine: Template engine to use. Defaults to Django engine.

        Raises:
            TypeError: Some of the arguments are missing or multiple are provided.
        """
        engine = engines["django"] if engine is None else engine

        if len([value for value in (file, inline) if value is not None]) != 1:
            msg = "Exactly one of 'file' or 'inline' must be provided."
            raise TypeError(msg)

        if file:
            template = engine.get_template(file)
        elif inline:
            template = engine.from_string(inline)
        else:  # pragma: no cover
            msg = "Unreachable code"
            raise NotImplementedError(msg)

        self.template = template  # type: ignore[assignment] # False-positive error

    def render(self, context: dict[str, Any]) -> Any:  # noqa: D102
        logger.debug("Rendering template with context: %r", context)
        rendered = self.template.render(context=context)  # type: ignore[arg-type] # False-positive error
        return _xml_to_dict(rendered)

__init__(*, file=None, inline=None, engine=None) 

__init__(
    *, file: str, engine: BaseEngine | None = None
) -> None

__init__(
    *, inline: str, engine: BaseEngine | None = None
) -> None

Initialize template.

Parameters:

Name Type Description Default
file str | None

Path to file with template.

 None
inline str | None

XML inline template.

 None
engine BaseEngine | None

Template engine to use. Defaults to Django engine.

 None

Raises:

Type Description
TypeError

Some of the arguments are missing or multiple are provided.
Source code in django_slack_tools/slack_messages/message_templates/django.py 
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
def __init__(
    self,
    *,
    file: str | None = None,
    inline: str | None = None,
    engine: BaseEngine | None = None,
) -> None:
    """Initialize template.

    Args:
        file: Path to file with template.
        inline: XML inline template.
        engine: Template engine to use. Defaults to Django engine.

    Raises:
        TypeError: Some of the arguments are missing or multiple are provided.
    """
    engine = engines["django"] if engine is None else engine

    if len([value for value in (file, inline) if value is not None]) != 1:
        msg = "Exactly one of 'file' or 'inline' must be provided."
        raise TypeError(msg)

    if file:
        template = engine.get_template(file)
    elif inline:
        template = engine.from_string(inline)
    else:  # pragma: no cover
        msg = "Unreachable code"
        raise NotImplementedError(msg)

    self.template = template  # type: ignore[assignment] # False-positive error

PythonTemplate

Bases: BaseTemplate[_PyObj]

Template that renders a dictionary.

Source code in django_slack_tools/slack_messages/message_templates/python.py 
14
15
16
17
18
19
20
21
22
23
24
25
class PythonTemplate(BaseTemplate[_PyObj]):
    """Template that renders a dictionary."""

    def __init__(self, template: _PyObj) -> None:
        """Initialize the template."""
        self.template = template

    def render(self, context: dict[str, Any]) -> _PyObj:  # noqa: D102
        logger.debug("Rendering template %r with context %r", self.template, context)
        result = _format_obj(self.template, context=context)
        logger.debug("Rendered template %r to %r", self.template, result)
        return result

__init__(template)

Initialize the template.

Source code in django_slack_tools/slack_messages/message_templates/python.py 
17
18
19
def __init__(self, template: _PyObj) -> None:
    """Initialize the template."""
    self.template = template

django_slack_tools.slack_messages.template_loaders

BaseTemplateLoader

Bases: ABC

Base class for template loaders.

Source code in django_slack_tools/slack_messages/template_loaders/base.py 
11
12
13
14
15
16
class BaseTemplateLoader(ABC):
    """Base class for template loaders."""

    @abstractmethod
    def load(self, key: str) -> BaseTemplate | None:
        """Load a template by key."""

load(key) abstractmethod

Load a template by key.

Source code in django_slack_tools/slack_messages/template_loaders/base.py 
14
15
16
@abstractmethod
def load(self, key: str) -> BaseTemplate | None:
    """Load a template by key."""

DjangoPolicyTemplateLoader

Bases: BaseTemplateLoader

Django database-backed template loader.

Source code in django_slack_tools/slack_messages/template_loaders/django.py 
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
class DjangoPolicyTemplateLoader(BaseTemplateLoader):
    """Django database-backed template loader."""

    def load(self, key: str) -> PythonTemplate | DjangoTemplate | None:  # noqa: D102
        return self._get_template_from_policy(policy_or_code=key)

    def _get_template_from_policy(
        self,
        policy_or_code: SlackMessagingPolicy | str,
    ) -> PythonTemplate | DjangoTemplate | None:
        """Get template instance."""
        if isinstance(policy_or_code, str):
            try:
                policy = SlackMessagingPolicy.objects.get(code=policy_or_code)
            except SlackMessagingPolicy.DoesNotExist:
                logger.warning("Policy not found: %s", policy_or_code)
                return None
        else:
            policy = policy_or_code

        if policy.template_type == SlackMessagingPolicy.TemplateType.PYTHON:
            return PythonTemplate(policy.template)

        if policy.template_type == SlackMessagingPolicy.TemplateType.DJANGO:
            try:
                return DjangoTemplate(file=policy.template)
            except TemplateDoesNotExist:
                logger.debug("Template not found: %s", policy.template)
                return None

        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)

_get_template_from_policy(policy_or_code)

Get template instance.

Source code in django_slack_tools/slack_messages/template_loaders/django.py 
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
def _get_template_from_policy(
    self,
    policy_or_code: SlackMessagingPolicy | str,
) -> PythonTemplate | DjangoTemplate | None:
    """Get template instance."""
    if isinstance(policy_or_code, str):
        try:
            policy = SlackMessagingPolicy.objects.get(code=policy_or_code)
        except SlackMessagingPolicy.DoesNotExist:
            logger.warning("Policy not found: %s", policy_or_code)
            return None
    else:
        policy = policy_or_code

    if policy.template_type == SlackMessagingPolicy.TemplateType.PYTHON:
        return PythonTemplate(policy.template)

    if policy.template_type == SlackMessagingPolicy.TemplateType.DJANGO:
        try:
            return DjangoTemplate(file=policy.template)
        except TemplateDoesNotExist:
            logger.debug("Template not found: %s", policy.template)
            return None

    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)

DjangoTemplateLoader

Bases: BaseTemplateLoader

Django filesystem-backed template loader.

Source code in django_slack_tools/slack_messages/template_loaders/django.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
class DjangoTemplateLoader(BaseTemplateLoader):
    """Django filesystem-backed template loader."""

    def __init__(self, *, engine: BaseEngine | None = None) -> None:
        """Initialize template loader.

        Args:
            engine: Template engine to use. Defaults to Django engine.
        """
        self.engine = engines["django"] if engine is None else engine

    def load(self, key: str) -> DjangoTemplate | None:  # noqa: D102
        try:
            return DjangoTemplate(file=key, engine=self.engine)
        except TemplateDoesNotExist:
            logger.debug("Template not found: %s", key)
            return None

__init__(*, engine=None)

Initialize template loader.

Parameters:

Name Type Description Default
engine BaseEngine | None

Template engine to use. Defaults to Django engine.

 None
Source code in django_slack_tools/slack_messages/template_loaders/django.py 
24
25
26
27
28
29
30
def __init__(self, *, engine: BaseEngine | None = None) -> None:
    """Initialize template loader.

    Args:
        engine: Template engine to use. Defaults to Django engine.
    """
    self.engine = engines["django"] if engine is None else engine

TemplateLoadError

Bases: Exception

Base class for template loader errors.

Source code in django_slack_tools/slack_messages/template_loaders/errors.py 
4
5
class TemplateLoadError(Exception):
    """Base class for template loader errors."""

TemplateNotFoundError

Bases: TemplateLoadError

Template not found error.

Source code in django_slack_tools/slack_messages/template_loaders/errors.py 
8
9
class TemplateNotFoundError(TemplateLoadError):
    """Template not found error."""