Skip to content

Django Integration

django_slack_tools.slack_messages.models

SlackMention

Bases: TimestampMixin, Model

People or group in channels receive messages.

Source code in django_slack_tools/slack_messages/models/mention.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
class SlackMention(TimestampMixin, models.Model):
    """People or group in channels receive messages."""

    class MentionType(models.TextChoices):
        """Possible mention types."""

        USER = "U", _("User")
        "User mentions. e.g. `@lasuillard`."

        GROUP = "G", _("Group")
        "Team mentions. e.g. `@backend`."

        SPECIAL = "S", _("Special")
        "Special mentions. e.g. `@here`, `@channel`, `@everyone`."

        UNKNOWN = "?", _("Unknown")
        "Unknown mention type."

    type = models.CharField(
        verbose_name=_("Type"),
        help_text=_("Type of mentions."),
        max_length=1,
        choices=MentionType.choices,
        blank=False,
    )
    name = models.CharField(
        verbose_name=_("Name"),
        help_text=_("Human-friendly mention name."),
        max_length=128,
    )
    mention_id = models.CharField(
        verbose_name=_("Mention ID"),
        help_text=_("User or group ID, or raw mention itself."),
        max_length=32,
    )

    objects: SlackMentionManager = SlackMentionManager()

    class Meta:  # noqa: D106
        verbose_name = _("Mention")
        verbose_name_plural = _("Mentions")

    def __str__(self) -> str:
        return _("{name} ({type}, {mention_id})").format(
            name=self.name,
            type=self.get_type_display(),
            mention_id=self.mention_id,
        )

    @property
    def mention(self) -> str:
        """Mention string for use in messages, e.g. `"<@{USER_ID}>"`."""
        if self.type == SlackMention.MentionType.USER:
            return f"<@{self.mention_id}>"

        if self.type == SlackMention.MentionType.GROUP:
            return f"<!subteam^{self.mention_id}>"

        return self.mention_id

mention property

Mention string for use in messages, e.g. "<@{USER_ID}>".

MentionType

Bases: TextChoices

Possible mention types.

Source code in django_slack_tools/slack_messages/models/mention.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
class MentionType(models.TextChoices):
    """Possible mention types."""

    USER = "U", _("User")
    "User mentions. e.g. `@lasuillard`."

    GROUP = "G", _("Group")
    "Team mentions. e.g. `@backend`."

    SPECIAL = "S", _("Special")
    "Special mentions. e.g. `@here`, `@channel`, `@everyone`."

    UNKNOWN = "?", _("Unknown")
    "Unknown mention type."
GROUP = ('G', _('Group')) class-attribute instance-attribute

Team mentions. e.g. @backend.

SPECIAL = ('S', _('Special')) class-attribute instance-attribute

Special mentions. e.g. @here, @channel, @everyone.

UNKNOWN = ('?', _('Unknown')) class-attribute instance-attribute

Unknown mention type.

USER = ('U', _('User')) class-attribute instance-attribute

User mentions. e.g. @lasuillard.

SlackMessage

Bases: TimestampMixin, Model

An Slack message.

Source code in django_slack_tools/slack_messages/models/message.py 
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
class SlackMessage(TimestampMixin, models.Model):
    """An Slack message."""

    id = models.CharField(primary_key=True, unique=True, max_length=255, default=uuid.uuid4, editable=False)
    policy = models.ForeignKey(
        SlackMessagingPolicy,
        verbose_name=_("Messaging Policy"),
        help_text=_("Messaging policy for this message."),
        null=True,  # Message can be built from scratch without using templates
        blank=True,
        on_delete=models.SET_NULL,
    )
    channel = models.CharField(
        verbose_name=_("Channel"),
        help_text=_("ID of channel this message sent to."),
        blank=False,
        max_length=128,  # Maximum length of channel name is 80 characters
    )
    header = models.JSONField(
        verbose_name=_("Header"),
        help_text=_(
            "Slack control arguments."
            " Allowed fields are `mrkdwn`, `parse`, `reply_broadcast`, `thread_ts`, `unfurl_links`, `unfurl_media`.",
        ),
        validators=[header_validator],
    )
    body = models.JSONField(
        verbose_name=_("Body"),
        help_text=_(
            "Message body."
            " Allowed fields are `attachments`, `body`, `text`, `icon_emoji`, `icon_url`, `metadata`, `username`.",
        ),
        validators=[body_validator],
    )
    ok = models.BooleanField(
        verbose_name=_("OK"),
        help_text=_("Whether Slack API respond with OK. If never sent, will be `null`."),
        null=True,
        default=None,
    )
    permalink = models.CharField(
        verbose_name=_("Permalink"),
        help_text=_("Permanent link for this message."),
        max_length=256,
        default="",
        blank=True,
    )

    # As ID, `ts` assigned by Slack, it is known after received response
    # By known, `ts` refers to timestamp (Format of `datetime.timestamp()`, e.g. `"1702737142.945359"`)
    ts = models.CharField(
        verbose_name=_("Message ID"),
        help_text=_("ID of an Slack message."),
        max_length=32,
        null=True,
        blank=True,
        unique=True,
    )
    parent_ts = models.CharField(
        verbose_name=_("Thread ID"),
        help_text=_("ID of current conversation thread."),
        max_length=32,
        default="",
        blank=True,
    )

    # Extraneous call detail for debugging
    request = models.JSONField(
        verbose_name=_("Request"),
        help_text=_("Dump of request content for debugging."),
        null=True,
        blank=True,
    )
    response = models.JSONField(
        verbose_name=_("Response"),
        help_text=_("Dump of response content for debugging."),
        null=True,
        blank=True,
    )
    exception = models.TextField(
        verbose_name=_("Exception"),
        help_text=_("Exception message if any."),
        blank=True,
    )

    objects: SlackMessageManager = SlackMessageManager()

    class Meta:  # noqa: D106
        verbose_name = _("Message")
        verbose_name_plural = _("Messages")
        ordering = ("-created",)

    def __str__(self) -> str:
        if self.ok is True:
            return _("Message ({ts}, OK)").format(id=self.id, ts=self.ts)

        if self.ok is False:
            return _("Message ({id}, not OK)").format(id=self.id)

        return _("Message ({id}, not sent)").format(id=self.id)

SlackMessageRecipient

Bases: TimestampMixin, Model

People or group in channels receive messages.

Source code in django_slack_tools/slack_messages/models/message_recipient.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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
class SlackMessageRecipient(TimestampMixin, models.Model):
    """People or group in channels receive messages."""

    alias = models.CharField(
        verbose_name=_("Alias"),
        help_text=_("Alias for this recipient."),
        max_length=256,
        unique=True,
    )
    channel = models.CharField(
        verbose_name=_("Channel"),
        help_text=_("Slack channel ID where messages will be sent."),
        max_length=128,
        blank=False,
    )
    channel_name = models.CharField(
        verbose_name=_("Channel name"),
        help_text=_("Display name of channel."),
        max_length=256,
        blank=True,
        default="",
    )
    mentions = models.ManyToManyField(
        SlackMention,
        verbose_name=_("Mentions"),
        help_text=_("List of mentions."),
        blank=True,
    )

    objects: SlackMessageRecipientManager = SlackMessageRecipientManager()

    class Meta:  # noqa: D106
        verbose_name = _("Recipient")
        verbose_name_plural = _("Recipients")

    def __str__(self) -> str:
        num_mentions = self.mentions.count()

        return _("{alias} ({channel}, {num_mentions} mentions)").format(
            alias=self.alias,
            channel=self.channel,
            num_mentions=num_mentions,
        )

SlackMessagingPolicy

Bases: TimestampMixin, Model

An Slack messaging policy which determines message content and those who receive messages.

Source code in django_slack_tools/slack_messages/models/messaging_policy.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
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
class SlackMessagingPolicy(TimestampMixin, models.Model):
    """An Slack messaging policy which determines message content and those who receive messages."""

    class TemplateType(models.TextChoices):
        """Possible template types."""

        PYTHON = "P", _("Python")
        "Dictionary-based template."

        DJANGO = "DJ", _("Django")
        "Django XML-based template."

        DJANGO_INLINE = "DI", _("Django Inline")
        "Django inline template."

        UNKNOWN = "?", _("Unknown")
        "Unknown template type."

    code = models.CharField(
        verbose_name=_("Code"),
        help_text=_("Unique message code for lookup, mostly by human."),
        max_length=32,
        unique=True,
    )
    enabled = models.BooleanField(
        verbose_name=_("Enabled"),
        help_text=_("Turn on or off current messaging policy."),
        default=True,
    )
    recipients = models.ManyToManyField(
        SlackMessageRecipient,
        verbose_name=_("Message recipients"),
        help_text=_("Those who will receive messages."),
    )
    header_defaults = models.JSONField(
        verbose_name=_("Default header"),
        help_text=_("Default header values applied to messages on creation."),
        validators=[header_validator],
        blank=True,
        default=dict,
    )
    template_type = models.CharField(
        verbose_name=_("Template type"),
        help_text=_("Type of message template."),
        max_length=2,
        choices=TemplateType.choices,
        default=TemplateType.PYTHON,
    )
    template: models.JSONField[Any] = models.JSONField(
        verbose_name=_("Message template object"),
        help_text=_("Dictionary-based template object."),
        null=True,
        blank=True,
    )

    # Type is too obvious but due to limits...
    objects: SlackMessagingPolicyManager = SlackMessagingPolicyManager()

    class Meta:  # noqa: D106
        verbose_name = _("Messaging Policy")
        verbose_name_plural = _("Messaging Policies")

    def __str__(self) -> str:
        num_recipients = self.recipients.all().count()
        if self.enabled:
            return _("{code} (enabled, {num_recipients} recipients)").format(
                code=self.code,
                num_recipients=num_recipients,
            )

        return _("{code} (disabled, {num_recipients} recipients)").format(code=self.code, num_recipients=num_recipients)

TemplateType

Bases: TextChoices

Possible template types.

Source code in django_slack_tools/slack_messages/models/messaging_policy.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
class TemplateType(models.TextChoices):
    """Possible template types."""

    PYTHON = "P", _("Python")
    "Dictionary-based template."

    DJANGO = "DJ", _("Django")
    "Django XML-based template."

    DJANGO_INLINE = "DI", _("Django Inline")
    "Django inline template."

    UNKNOWN = "?", _("Unknown")
    "Unknown template type."
DJANGO = ('DJ', _('Django')) class-attribute instance-attribute

Django XML-based template.

DJANGO_INLINE = ('DI', _('Django Inline')) class-attribute instance-attribute

Django inline template.

PYTHON = ('P', _('Python')) class-attribute instance-attribute

Dictionary-based template.

UNKNOWN = ('?', _('Unknown')) class-attribute instance-attribute

Unknown template type.

django_slack_tools.slack_messages.middlewares.django

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

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

_preprocess_xml(xml)

Normalize XML text nodes.

Source code in django_slack_tools/slack_messages/message_templates/django.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
def _preprocess_xml(xml: str) -> str:
    """Normalize XML text nodes."""
    root = ET.fromstring(xml)  # noqa: S314 ; TODO(lasuillard): Naive belief that XML is safe
    for node in root.iter():
        node.tag = _rename_tag(node.tag)

        if node.tag in ("text", "elements") and node.text:
            text = dedent(node.text)
            text = _remove_single_newline(text)
            logger.debug("Normalized text node: %r -> %r", node.text, text)
            node.text = text

    return ET.tostring(root, encoding="unicode")

_remove_single_newline(text)

Remove a single newline from repeated newlines. If the are just one newline, replace it with space.

Source code in django_slack_tools/slack_messages/message_templates/django.py 
143
144
145
def _remove_single_newline(text: str) -> str:
    """Remove a single newline from repeated newlines. If the are just one newline, replace it with space."""
    return re.sub(r"([\n]+)", lambda m: "\n" * (m.group(1).count("\n") - 1) or " ", text)

_rename_tag(tag)

Rename tags.

Source code in django_slack_tools/slack_messages/message_templates/django.py 
138
139
140
def _rename_tag(tag: str) -> str:
    """Rename tags."""
    return _TAG_MAPPING.get(tag, tag)

_xml_to_dict(xml)

Parse XML string to Python dictionary.

Following transformations are applied by default:

  • Normalize text nodes: remove single newlines and dedent text, etc.
  • Rename tags for syntactic comfort: block -> blocks, element -> elements

Please check the tests for more detailed examples.

Parameters:

Name Type Description Default
xml str

XML string.

 required

Returns:

Type Description
dict

Parsed dictionary. Be aware, the returned value will be the child of
dict

top-level node (e.g. ...), regardless of its key name.
Source code in django_slack_tools/slack_messages/message_templates/django.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def _xml_to_dict(xml: str) -> dict:
    """Parse XML string to Python dictionary.

    Following transformations are applied by default:

    - Normalize text nodes: remove single newlines and dedent text, etc.
    - Rename tags for syntactic comfort: block -> blocks, element -> elements

    Please check the tests for more detailed examples.

    Args:
        xml: XML string.

    Returns:
        Parsed dictionary. Be aware, the returned value will be the child of
        top-level node (e.g. <root>...</root>), regardless of its key name.
    """
    xml = _preprocess_xml(xml)
    obj = xmltodict.parse(
        xml,
        attr_prefix="",
        cdata_key="text",
        force_list=("blocks", "elements", "options"),
        postprocessor=_xml_postprocessor,
    )
    return dict(next(iter(obj.values())))

django_slack_tools.slack_messages.template_loaders.django

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