Skip to content

slack

Push Notifications via Slack

SlackNotifications #

Bases: BaseNotifications

Push Notifications via Slack

Source code in camply/notifications/slack.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
class SlackNotifications(BaseNotifications):
    """
    Push Notifications via Slack
    """

    def __init__(self):
        super().__init__()
        self.session.headers.update({"Content-Type": "application/json"})
        if any([SlackConfig.SLACK_WEBHOOK is None, SlackConfig.SLACK_WEBHOOK == ""]):
            warning_message = (
                "Slack is not configured properly. To send Slack messages "
                "make sure to run `camply configure` or set the "
                "proper environment variable: `SLACK_WEBHOOK`."
            )
            logger.error(warning_message)
            raise EnvironmentError(warning_message)

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

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

        Returns
        -------
        requests.Response
        """
        message_blocks = kwargs.pop("blocks", [])
        message_json = {
            "text": message,
        }
        if message_blocks:
            message_json = {
                "blocks": message_blocks,
            }
        logger.debug(message_json)
        response = self.session.post(
            url=SlackConfig.SLACK_WEBHOOK,
            json=message_json,
        )
        try:
            response.raise_for_status()
        except requests.HTTPError as he:
            logger.warning(
                "Notifications weren't able to be sent to Slack. "
                "Your configuration might be incorrect."
            )
            raise ConnectionError(response.text) from he
        return response

    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(
                    {
                        "type": "mrkdwn",
                        "text": f"*{key}*",
                    }
                )
                if key in ["Permitted Equipment", "Booking Link"]:
                    data_type = "mrkdwn"
                else:
                    data_type = "plain_text"
                fields.append(
                    {
                        "type": data_type,
                        "text": str(value),
                    }
                )

            blocks = []
            blocks.append(
                {
                    "type": "header",
                    "text": {
                        "type": "plain_text",
                        "text": message_title,
                    },
                }
            )
            # Slack only allows 10 fields (k+v) per section
            for chunk in range(0, len(fields) + 1, 10):
                chunk_max = chunk + 10
                blocks.append(
                    {
                        "type": "section",
                        "fields": fields[chunk:chunk_max],
                    }
                )
            self.send_message(
                message=message_title,
                blocks=blocks,
            )

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/slack.py 
 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
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(
                {
                    "type": "mrkdwn",
                    "text": f"*{key}*",
                }
            )
            if key in ["Permitted Equipment", "Booking Link"]:
                data_type = "mrkdwn"
            else:
                data_type = "plain_text"
            fields.append(
                {
                    "type": data_type,
                    "text": str(value),
                }
            )

        blocks = []
        blocks.append(
            {
                "type": "header",
                "text": {
                    "type": "plain_text",
                    "text": message_title,
                },
            }
        )
        # Slack only allows 10 fields (k+v) per section
        for chunk in range(0, len(fields) + 1, 10):
            chunk_max = chunk + 10
            blocks.append(
                {
                    "type": "section",
                    "fields": fields[chunk:chunk_max],
                }
            )
        self.send_message(
            message=message_title,
            blocks=blocks,
        )

send_message(message, **kwargs) #

Send a message via Slack - if environment variables are configured.

Parameters:

Name Type Description Default
message str
required

Returns:

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

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

    Returns
    -------
    requests.Response
    """
    message_blocks = kwargs.pop("blocks", [])
    message_json = {
        "text": message,
    }
    if message_blocks:
        message_json = {
            "blocks": message_blocks,
        }
    logger.debug(message_json)
    response = self.session.post(
        url=SlackConfig.SLACK_WEBHOOK,
        json=message_json,
    )
    try:
        response.raise_for_status()
    except requests.HTTPError as he:
        logger.warning(
            "Notifications weren't able to be sent to Slack. "
            "Your configuration might be incorrect."
        )
        raise ConnectionError(response.text) from he
    return response