Skip to content

pushover

Push Notifications via Pushover

PushoverNotifications #

Bases: BaseNotifications, StreamHandler

Push Notifications via Pushover + a Logging Handler

Source code in camply/notifications/pushover.py 
 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
class PushoverNotifications(BaseNotifications, logging.StreamHandler):
    """
    Push Notifications via Pushover + a Logging Handler
    """

    def __init__(self, level: Optional[int] = logging.INFO):
        super().__init__()
        self.session.headers.update(PushoverConfig.API_HEADERS)
        logging.StreamHandler.__init__(self)
        self.setLevel(level=level)
        if any([PushoverConfig.PUSH_USER is None, PushoverConfig.PUSH_USER == ""]):
            warning_message = (
                "Pushover is not configured properly. To send pushover messages "
                "make sure to run `camply configure` or set the "
                "proper environment variables: `PUSHOVER_PUSH_USER`, "
                "`PUSHOVER_PUSH_TOKEN`."
            )
            logger.error(warning_message)
            raise EnvironmentError(warning_message)
        self.pushover_token = PushoverConfig.PUSH_TOKEN
        if self.pushover_token in [None, ""]:
            self.pushover_token = base64.b64decode(
                PushoverConfig.PUSHOVER_DEFAULT_API_TOKEN
            ).decode("utf-8")

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

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

        Returns
        -------
        requests.Response
        """
        response = self.session.post(
            url=PushoverConfig.PUSHOVER_API_ENDPOINT,
            params=dict(
                token=self.pushover_token,
                user=PushoverConfig.PUSH_USER,
                message=message,
                **kwargs,
            ),
        )
        try:
            response.raise_for_status()
        except requests.HTTPError as he:
            logger.warning(
                "Notifications weren't able to be sent to Pushover. "
                "Your configuration might be incorrect."
            )
            raise ConnectionError(response.text) from he
        return response

    def emit(self, record: logging.LogRecord):
        """
        Produce a logging record

        Parameters
        ----------
        record: str
            Message to log
        """
        log_formatted_message = "[{:>10}]: {}".format(
            record.levelname.upper(), record.msg
        )
        title = f"Pushover {record.levelname.title()} Message"
        self.send_message(message=log_formatted_message, title=title)

    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():
                if key == "Booking Link":
                    value = f"<a href='{value}'>{value}</a>"
                fields.append(f"<b>{key}:</b> {value}")
            composed_message = "\n".join(fields)
            self.send_message(message=composed_message, title=message_title, html=1)

emit(record) #

Produce a logging record

Parameters:

Name Type Description Default
record LogRecord

Message to log

 required
Source code in camply/notifications/pushover.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def emit(self, record: logging.LogRecord):
    """
    Produce a logging record

    Parameters
    ----------
    record: str
        Message to log
    """
    log_formatted_message = "[{:>10}]: {}".format(
        record.levelname.upper(), record.msg
    )
    title = f"Pushover {record.levelname.title()} Message"
    self.send_message(message=log_formatted_message, title=title)

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/pushover.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
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():
            if key == "Booking Link":
                value = f"<a href='{value}'>{value}</a>"
            fields.append(f"<b>{key}:</b> {value}")
        composed_message = "\n".join(fields)
        self.send_message(message=composed_message, title=message_title, html=1)

send_message(message, **kwargs) #

Send a message via Pushover - if environment variables are configured

Parameters:

Name Type Description Default
message str
required

Returns:

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

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

    Returns
    -------
    requests.Response
    """
    response = self.session.post(
        url=PushoverConfig.PUSHOVER_API_ENDPOINT,
        params=dict(
            token=self.pushover_token,
            user=PushoverConfig.PUSH_USER,
            message=message,
            **kwargs,
        ),
    )
    try:
        response.raise_for_status()
    except requests.HTTPError as he:
        logger.warning(
            "Notifications weren't able to be sent to Pushover. "
            "Your configuration might be incorrect."
        )
        raise ConnectionError(response.text) from he
    return response