mattermost_notifier.py

import logging
from typing import Any, Dict, List

import requests

from ..entry import Entry
from .notification_sender import NotificationSender, NotificationSenderError
from .utils import get_env

logger = logging.getLogger(__name__)

MESSAGES_TEMPLATE = (
    "* Une nouvelle version de {name} est disponible [{version}]({url}) !"
)


class MattermostNotifier(NotificationSender):
    """
    Sends a message in a Mattermost channel listing updated softwares
    """

    def __init__(self, mattermost_host: str, channel_id: str, token: str):
        self.mattermost_host = mattermost_host
        self.channel_id = channel_id
        self.token = token

    @classmethod
    def from_env(cls) -> "MattermostNotifier":
        """
        Initializes a MattermostNotifier using environment variables
        """
        return cls(
            get_env("MM_SERVER_HOST"), get_env("MM_CHANNEL_ID"), get_env("MM_TOKEN")
        )

    @staticmethod
    def _entry_to_message(entry: Entry) -> str:
        return MESSAGES_TEMPLATE.format(
            name=entry.name, version=entry.version, url=entry.url
        )

    def _get_api_root(self) -> str:
        return f"https://{self.mattermost_host}/api/v4"

    def notify_updates(self, entries: List[Entry]) -> None:
        req_headers: Dict[str, str] = {"Authorization": f"Bearer {self.token}"}
        req_data: Dict[str, Any] = {
            "channel_id": self.channel_id,
            "message": "\n".join(map(MattermostNotifier._entry_to_message, entries)),
        }
        full_url = self._get_api_root() + "/posts"

        logger.debug("Sending entries to mattermost server %s", self.mattermost_host)

        try:
            response = requests.post(full_url, json=req_data, headers=req_headers)
        except TypeError as err:
            raise NotificationSenderError(
                f"Could not serialize entries to Mattermost server {self.mattermost_host}: {err}"
            ) from err
        except requests.exceptions.ConnectionError as err:
            raise NotificationSenderError(
                f"Could not post message to Mattermost server {self.mattermost_host}: {err}"
            ) from err

        if response.status_code != requests.codes.created:
            raise NotificationSenderError(
                f"Mattermost server {self.mattermost_host} replied with status={response.status_code} (content={response.content})"
            )