Skip to content

Slack Messages

django_slack_tools.slack_messages.messenger

DjangoDatabasePersister

Bases: BaseMiddleware

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

Source code in django_slack_tools/slack_messages/messenger/middlewares.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
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/messenger/middlewares.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
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

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/messenger/middlewares.py 
 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
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  # noqa: PLC0415

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

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/messenger/middlewares.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
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

DjangoPolicyTemplateLoader

Bases: BaseTemplateLoader

Django database-backed template loader.

Source code in django_slack_tools/slack_messages/messenger/template_loaders.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)

DjangoTemplate

Bases: BaseTemplate

Template utilizing Django built-in template engine.

Source code in django_slack_tools/slack_messages/messenger/message_templates.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/messenger/message_templates.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

DjangoTemplateLoader

Bases: BaseTemplateLoader

Django filesystem-backed template loader.

Source code in django_slack_tools/slack_messages/messenger/template_loaders.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/messenger/template_loaders.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

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

Handy APIs for sending Slack messages.

slack_message(to, *, messenger_name=None, header=None, template=None, context=None, message=None) 

slack_message(
    to: str,
    *,
    messenger_name: str | None = None,
    header: MessageHeader | dict[str, Any] | None = None,
    message: str,
) -> MessageResponse | None

slack_message(
    to: str,
    *,
    messenger_name: str | None = None,
    header: MessageHeader | dict[str, Any] | None = None,
    template: str | None = None,
    context: dict[str, Any] | None = None,
) -> MessageResponse | None

Shortcut for sending a Slack message.

Parameters:

Name Type Description Default
to str

Recipient.

 required
messenger_name str | None

Messenger name. If not set, default messenger is used.

 None
header MessageHeader | dict[str, Any] | None

Slack message control header.

 None
template str | None

Message template key. Cannot be used with message.

 None
context dict[str, Any] | None

Context for rendering the template. Only used with template.

 None
message str | None

Simple message text. Cannot be used with template.

 None

Returns:

Type Description
MessageResponse | None

Sent message instance or None.
Source code in django_slack_tools/slack_messages/shortcuts.py 
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
def slack_message(  # noqa: PLR0913
    to: str,
    *,
    messenger_name: str | None = None,
    header: MessageHeader | dict[str, Any] | None = None,
    template: str | None = None,
    context: dict[str, Any] | None = None,
    message: str | None = None,
) -> MessageResponse | None:
    """Shortcut for sending a Slack message.

    Args:
        to: Recipient.
        messenger_name: Messenger name. If not set, default messenger is used.
        header: Slack message control header.
        template: Message template key. Cannot be used with `message`.
        context: Context for rendering the template. Only used with `template`.
        message: Simple message text. Cannot be used with `template`.

    Returns:
        Sent message instance or `None`.
    """
    if (template and message) or (not template and not message):
        msg = "Either `template` or `message` must be set, but not both."
        raise ValueError(msg)

    messenger = get_messenger(messenger_name)
    header = MessageHeader.from_any(header)

    if message:
        request = MessageRequest(
            channel=to,
            header=header,
            body=MessageBody(text=message),
            template_key=None,
            context={},
        )
        return messenger.send_request(request)

    context = context or {}
    return messenger.send(to, header=header, template=template, context=context)

django_slack_tools.slack_messages.tasks

Celery utils.

cleanup_old_messages(*, base_ts=None, threshold_seconds=7 * 24 * 60 * 60)

Delete old messages created before given threshold.

Parameters:

Name Type Description Default
threshold_seconds int | None

Threshold seconds. Defaults to 7 days.

 7 * 24 * 60 * 60
base_ts str | None

Base timestamp to calculate the threshold, in ISO format. If falsy, current timestamp will be used.

 None

Returns:

Type Description
int

Number of deleted messages.
Source code in django_slack_tools/slack_messages/tasks.py 
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
@shared_task
def cleanup_old_messages(
    *,
    base_ts: str | None = None,
    threshold_seconds: int | None = 7 * 24 * 60 * 60,  # 7 days
) -> int:
    """Delete old messages created before given threshold.

    Args:
        threshold_seconds: Threshold seconds. Defaults to 7 days.
        base_ts: Base timestamp to calculate the threshold, in ISO format. If falsy, current timestamp will be used.

    Returns:
        Number of deleted messages.
    """
    dt = datetime.fromisoformat(base_ts) if base_ts else timezone.localtime()

    if threshold_seconds is None:
        logger.warning("Threshold seconds not provided, skipping cleanup.")
        return 0

    cleanup_threshold = dt - timedelta(seconds=threshold_seconds)
    logger.debug("Cleaning up messages older than %s.", cleanup_threshold)

    num_deleted, _ = SlackMessage.objects.filter(created__lt=cleanup_threshold).delete()
    logger.info("Deleted %d old messages.", num_deleted)

    return num_deleted

slack_message(*args, **kwargs)

Celery task wrapper for .shortcuts.slack_message.

Parameters:

Name Type Description Default
args Any

Positional arguments.

 ()
kwargs Any

Keyword arguments.

 {}

Returns:

Type Description
str | None

ID of sent message if any, None otherwise.
Source code in django_slack_tools/slack_messages/tasks.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
@shared_task
def slack_message(*args: Any, **kwargs: Any) -> str | None:
    """Celery task wrapper for `.shortcuts.slack_message`.

    Args:
        args: Positional arguments.
        kwargs: Keyword arguments.

    Returns:
        ID of sent message if any, `None` otherwise.
    """
    response = shortcuts.slack_message(*args, **kwargs)
    return response.ts if response else None