Skip to content

Messenger

django_slack_tools.messenger.backends

BaseBackend

Bases: ABC

Abstract base class for messaging backends.

Source code in django_slack_tools/messenger/backends/base.py
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."""

deliver(request)

Deliver message request.

Source code in django_slack_tools/messenger/backends/base.py
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/messenger/backends/dummy.py
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/messenger/backends/logging_.py
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/messenger/backends/slack.py
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/messenger/backends/slack.py
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/messenger/backends/slack.py
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/messenger/backends/slack.py
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.messenger.message_templates

BaseTemplate

Bases: ABC, Generic[_T]

Base class for templates.

Source code in django_slack_tools/messenger/message_templates/base.py
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/messenger/message_templates/base.py
@abstractmethod
def render(self, context: dict[str, Any]) -> Any:
    """Render the template with the given context."""

PythonTemplate

Bases: BaseTemplate[_PyObj]

Template that renders a dictionary.

Source code in django_slack_tools/messenger/message_templates/python.py
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/messenger/message_templates/python.py
def __init__(self, template: _PyObj) -> None:
    """Initialize the template."""
    self.template = template

django_slack_tools.messenger.middlewares

BaseMiddleware

Base class for middleware components.

Source code in django_slack_tools/messenger/middlewares/base.py
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/messenger/middlewares/base.py
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/messenger/middlewares/base.py
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

django_slack_tools.messenger.template_loaders

BaseTemplateLoader

Bases: ABC

Base class for template loaders.

Source code in django_slack_tools/messenger/template_loaders/base.py
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/messenger/template_loaders/base.py
@abstractmethod
def load(self, key: str) -> BaseTemplate | None:
    """Load a template by key."""

TemplateLoadError

Bases: Exception

Base class for template loader errors.

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

TemplateNotFoundError

Bases: TemplateLoadError

Template not found error.

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

django_slack_tools.messenger.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/messenger/messenger.py
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/messenger/messenger.py
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

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

Simplified shortcut for .send_request().

Source code in django_slack_tools/messenger/messenger.py
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/messenger/messenger.py
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.messenger.request

MessageBody

Bases: BaseModel

Source code in django_slack_tools/messenger/request.py
class MessageBody(BaseModel):  # noqa: D101
    model_config = ConfigDict(extra="forbid")

    attachments: Optional[List[dict]] = None

    # See more about blocks at https://api.slack.com/reference/block-kit/blocks
    blocks: Optional[list[dict]] = None

    text: Optional[str] = None
    icon_emoji: Optional[str] = None
    icon_url: Optional[str] = None
    metadata: Optional[dict] = None
    username: Optional[str] = None

    @model_validator(mode="after")
    def _check_at_least_one_field_is_set(self) -> MessageBody:
        if not any((self.attachments, self.blocks, self.text)):
            msg = "At least one of `attachments`, `blocks` and `text` must set"
            raise ValueError(msg)

        return self

    @classmethod
    def from_any(cls, obj: str | MessageBody | dict[str, Any]) -> MessageBody:
        """Create instance from compatible types."""
        if isinstance(obj, cls):
            return obj

        if isinstance(obj, dict):
            return cls.model_validate(obj)

        if isinstance(obj, str):
            return cls(text=obj)

        msg = f"Unsupported type {type(obj)}"
        raise TypeError(msg)

from_any(obj) classmethod

Create instance from compatible types.

Source code in django_slack_tools/messenger/request.py
@classmethod
def from_any(cls, obj: str | MessageBody | dict[str, Any]) -> MessageBody:
    """Create instance from compatible types."""
    if isinstance(obj, cls):
        return obj

    if isinstance(obj, dict):
        return cls.model_validate(obj)

    if isinstance(obj, str):
        return cls(text=obj)

    msg = f"Unsupported type {type(obj)}"
    raise TypeError(msg)

MessageHeader

Bases: BaseModel

Source code in django_slack_tools/messenger/request.py
class MessageHeader(BaseModel):  # noqa: D101
    model_config = ConfigDict(extra="forbid")

    mrkdwn: Optional[str] = None
    parse: Optional[str] = None
    reply_broadcast: Optional[bool] = None
    thread_ts: Optional[str] = None
    unfurl_links: Optional[bool] = None
    unfurl_media: Optional[bool] = None

    @classmethod
    def from_any(cls, obj: MessageHeader | dict[str, Any] | None = None) -> MessageHeader:
        """Create instance from compatible types."""
        if isinstance(obj, cls):
            return obj

        if isinstance(obj, dict):
            return cls.model_validate(obj)

        if obj is None:
            return cls()

        msg = f"Unsupported type {type(obj)}"
        raise TypeError(msg)

from_any(obj=None) classmethod

Create instance from compatible types.

Source code in django_slack_tools/messenger/request.py
@classmethod
def from_any(cls, obj: MessageHeader | dict[str, Any] | None = None) -> MessageHeader:
    """Create instance from compatible types."""
    if isinstance(obj, cls):
        return obj

    if isinstance(obj, dict):
        return cls.model_validate(obj)

    if obj is None:
        return cls()

    msg = f"Unsupported type {type(obj)}"
    raise TypeError(msg)

MessageRequest

Bases: BaseModel

Message request object.

Source code in django_slack_tools/messenger/request.py
class MessageRequest(BaseModel):
    """Message request object."""

    model_config = ConfigDict(extra="forbid")

    id_: str = Field(default_factory=lambda: str(uuid.uuid4()))
    channel: Any

    # Template key is optional to allow lazy initialization of the template key
    template_key: Optional[str]

    context: Dict[str, Any]
    header: MessageHeader

    # Also, the body is optional because it is rendered from the template
    body: Optional[MessageBody] = None

django_slack_tools.messenger.response

MessageResponse

Bases: BaseModel

Response from a messaging backend.

Source code in django_slack_tools/messenger/response.py
class MessageResponse(BaseModel):
    """Response from a messaging backend."""

    request: Optional[MessageRequest] = None
    ok: bool
    error: Optional[Any] = None
    data: Any
    ts: Optional[str] = None
    parent_ts: Optional[str] = None