Skip to content

telegram

Push Notifications via Telegram

TelegramNotifications #

Bases: BaseNotifications

Push Notifications via Telegram

Source code in camply/notifications/telegram.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
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
class TelegramNotifications(BaseNotifications):
    """
    Push Notifications via Telegram
    """

    def __init__(self):
        super().__init__()
        self.session.headers.update(TelegramConfig.API_HEADERS)
        if any(
            [
                TelegramConfig.BOT_TOKEN is None,
                TelegramConfig.BOT_TOKEN == "",
                TelegramConfig.CHAT_ID is None,
                TelegramConfig.CHAT_ID == "",
            ]
        ):
            warning_message = (
                "Telegram is not configured properly. To send Telegram messages "
                "make sure to run `camply configure` or set the "
                "proper environment variables: `TELEGRAM_BOT_TOKEN` and `TELEGRAM_CHAT_ID`."
            )
            logger.error(warning_message)
            raise EnvironmentError(warning_message)

    def send_message(self, message: str, escaped=False, **kwargs) -> requests.Response:
        """
        Send a message via Telegram - if environment variables are configured

        Parameters
        ----------
        message: str
        escaped: bool

        Returns
        -------
        requests.Response
        """
        if not escaped:
            message = self.escape_text(message)
        message_json = TelegramConfig.API_CONTENT.copy()
        message_json.update({"text": message})
        logger.debug(message_json)
        response = self.session.post(url=TelegramConfig.API_ENDPOINT, json=message_json)
        try:
            response.raise_for_status()
        except requests.HTTPError as he:
            logger.warning(
                "Notifications weren't able to be sent to Telegram. "
                "Your configuration might be incorrect."
            )
            raise ConnectionError(response.text) from he
        return response

    @staticmethod
    def escape_text(message: str) -> str:
        """
        Escape a message for use in Telegram

        Parameters
        ----------
        message: str

        Returns
        -------
        String
        """
        fields = [
            "_",
            "*",
            "[",
            "]",
            "(",
            ")",
            "~",
            "`",
            ">",
            "#",
            "+",
            "-",
            "=",
            "|",
            "{",
            "}",
            ".",
            "!",
        ]
        for f in fields:
            message = message.replace(f, f"\\{f}")
        return message

    def send_campsites(self, campsites: List[AvailableCampsite], **kwargs):
        """
        Send a message with a campsite object

        Parameters
        ----------
        campsites: AvailableCampsite
        """
        for campsite in campsites:
            message_title, formatted_dict = self.format_standard_campsites(
                campsite=campsite,
            )
            fields = []
            for key, value in formatted_dict.items():
                fields.append(self.escape_text(f"{key}: {value}"))
            message_fields = "\n".join(fields)
            message = f"*{self.escape_text(message_title)}*\n{message_fields}"
            self.send_message(message, escaped=True)

escape_text(message) staticmethod #

Escape a message for use in Telegram

Parameters:

Name Type Description Default
message str
required

Returns:

Type Description
String
Source code in camply/notifications/telegram.py 
 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
@staticmethod
def escape_text(message: str) -> str:
    """
    Escape a message for use in Telegram

    Parameters
    ----------
    message: str

    Returns
    -------
    String
    """
    fields = [
        "_",
        "*",
        "[",
        "]",
        "(",
        ")",
        "~",
        "`",
        ">",
        "#",
        "+",
        "-",
        "=",
        "|",
        "{",
        "}",
        ".",
        "!",
    ]
    for f in fields:
        message = message.replace(f, f"\\{f}")
    return message

send_campsites(campsites, **kwargs) #

Send a message with a campsite object

Parameters:

Name Type Description Default
campsites List[AvailableCampsite]
required
Source code in camply/notifications/telegram.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def send_campsites(self, campsites: List[AvailableCampsite], **kwargs):
    """
    Send a message with a campsite object

    Parameters
    ----------
    campsites: AvailableCampsite
    """
    for campsite in campsites:
        message_title, formatted_dict = self.format_standard_campsites(
            campsite=campsite,
        )
        fields = []
        for key, value in formatted_dict.items():
            fields.append(self.escape_text(f"{key}: {value}"))
        message_fields = "\n".join(fields)
        message = f"*{self.escape_text(message_title)}*\n{message_fields}"
        self.send_message(message, escaped=True)

send_message(message, escaped=False, **kwargs) #

Send a message via Telegram - if environment variables are configured

Parameters:

Name Type Description Default
message str
required
escaped
False

Returns:

Type Description
Response
Source code in camply/notifications/telegram.py 
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
def send_message(self, message: str, escaped=False, **kwargs) -> requests.Response:
    """
    Send a message via Telegram - if environment variables are configured

    Parameters
    ----------
    message: str
    escaped: bool

    Returns
    -------
    requests.Response
    """
    if not escaped:
        message = self.escape_text(message)
    message_json = TelegramConfig.API_CONTENT.copy()
    message_json.update({"text": message})
    logger.debug(message_json)
    response = self.session.post(url=TelegramConfig.API_ENDPOINT, json=message_json)
    try:
        response.raise_for_status()
    except requests.HTTPError as he:
        logger.warning(
            "Notifications weren't able to be sent to Telegram. "
            "Your configuration might be incorrect."
        )
        raise ConnectionError(response.text) from he
    return response