Skip to content

notifications

Notifications init file

AppriseNotifications #

Bases: BaseNotifications

Push Notifications via Apprise

Source code in camply/notifications/apprise.py 
16
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
class AppriseNotifications(BaseNotifications):
    """
    Push Notifications via Apprise
    """

    def __init__(self):
        super().__init__()
        try:
            import apprise
        except ImportError as ie:
            raise RuntimeError(
                "Looks like `apprise` isn't installed. Install it with `pip install camply[apprise]`"
            ) from ie

        if any(
            [
                AppriseConfig.APPRISE_URL is None,
            ]
        ):
            warning_message = (
                "Apprise is not configured properly. To send Apprise notifications "
                "make sure to run `camply configure` or set the "
                "proper environment variable: `APPRISE_URL`."
            )
            logger.error(warning_message)
            raise EnvironmentError(warning_message)
        self.client = apprise.Apprise()
        self.client.add(AppriseConfig.APPRISE_URL)
        logger.info("Apprise: will notify specified URL")

    def send_message(self, message: str, **kwargs):
        """
        Send a message via Apprise - if environment variables are configured

        Parameters
        ----------
        message: str
        """
        self.client.notify(
            body=message,
            title="Camply Notification",
        )

    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 = [f"🏕{message_title}", ""]
            for key, value in formatted_dict.items():
                fields.append(f"{key}: {value}")
            fields.append("")
            fields.append("camply, the campsite finder ⛺️")
            composed_message = "\n".join(fields)
            self.send_message(message=composed_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/apprise.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
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 = [f"🏕{message_title}", ""]
        for key, value in formatted_dict.items():
            fields.append(f"{key}: {value}")
        fields.append("")
        fields.append("camply, the campsite finder ⛺️")
        composed_message = "\n".join(fields)
        self.send_message(message=composed_message)

send_message(message, **kwargs) #

Send a message via Apprise - if environment variables are configured

Parameters:

Name Type Description Default
message str
required
Source code in camply/notifications/apprise.py 
46
47
48
49
50
51
52
53
54
55
56
57
def send_message(self, message: str, **kwargs):
    """
    Send a message via Apprise - if environment variables are configured

    Parameters
    ----------
    message: str
    """
    self.client.notify(
        body=message,
        title="Camply Notification",
    )

EmailNotifications #

Bases: BaseNotifications

Notifications via Email

Source code in camply/notifications/email_notifications.py 
 16
 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
class EmailNotifications(BaseNotifications):
    """
    Notifications via Email
    """

    email_subject = EmailConfig.EMAIL_SUBJECT_LINE
    email_from = EmailConfig.EMAIL_FROM_ADDRESS
    email_to = EmailConfig.EMAIL_TO_ADDRESS
    email_username = EmailConfig.EMAIL_USERNAME
    _email_password = EmailConfig.EMAIL_PASSWORD
    email_smtp_server = EmailConfig.EMAIL_SMTP_SERVER
    email_smtp_server_port = EmailConfig.EMAIL_SMTP_PORT

    def __init__(self):
        """
        Data Validation

        **kwargs
            Accepts: from, to, subject, username, password, server, port
        """
        super().__init__()
        # PERFORM SOME VALIDATION
        if any(
            [
                EmailConfig.EMAIL_TO_ADDRESS in [None, ""],
                EmailConfig.EMAIL_USERNAME in [None, ""],
                EmailConfig.EMAIL_PASSWORD in [None, ""],
            ]
        ):
            variable_names = "\n\t".join(EmailConfig.ENVIRONMENT_VARIABLE_NAMES)
            optional_variable_names = "\n\t".join(
                EmailConfig.OPTIONAL_ENVIRONMENT_VARIABLE
            )
            error_message = (
                "Email Notification Auth Parameters not set. Run `camply configure` "
                f"or set the following Environment Variables:\n\t{variable_names}"
                "\nOptional Environment Variables:\n\t"
                f"{optional_variable_names}"
            )
            logger.error(error_message)
            raise EnvironmentError(error_message)
        # ATTEMPT AN EMAIL LOGIN AT INIT TO THROW ERRORS EARLY
        _email_server = SMTP_SSL(
            self.email_smtp_server,
            self.email_smtp_server_port,
        )
        _email_server.ehlo()
        _email_server.login(
            user=self.email_username,
            password=self._email_password,
        )
        _email_server.quit()

    def send_message(self, message: str, **kwargs) -> None:
        """
        Send a message via Email

        Parameters
        ----------
        message: str
            Email Body
        **kwargs
            Accepts: from, to, subject, username, password, server, port

        Returns
        -------
        object
        """
        email = EmailMessage()
        email.set_content(message)
        email["Subject"] = kwargs.get("subject", self.email_subject)
        email["From"] = kwargs.get("from", self.email_from)
        email["To"] = kwargs.get("to", self.email_to)
        email_server_user = kwargs.get("username", self.email_username)
        email_server_password = kwargs.get("password", self._email_password)
        email_server_smtp_server = kwargs.get("server", self.email_smtp_server)
        email_server_smtp_server_port = kwargs.get("port", self.email_smtp_server_port)
        email_server = SMTP_SSL(email_server_smtp_server, email_server_smtp_server_port)
        email_server.ehlo()
        email_server.login(user=email_server_user, password=email_server_password)
        logger.info(f"Sending Email to {email['To']}: {email['Subject']}")
        email_server.send_message(email)
        logger.info("Email sent successfully")
        email_server.quit()

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

        Parameters
        ----------
        campsites: List[AvailableCampsite]
        """
        master_email_body_list = []
        for campsite in campsites:
            message_title, formatted_dict = self.format_standard_campsites(
                campsite=campsite,
            )
            fields = [message_title]
            for key, value in formatted_dict.items():
                if key == "Permitted Equipment":
                    value = value.replace("\n  - ", "\n  \t  - ")
                fields.append(f"\t{key}: {value}")
            composed_message = "\n".join(fields) + "\n\n"
            master_email_body_list.append(composed_message)
        master_email_body = "\n".join(master_email_body_list)
        if len(campsites) > 0:
            self.send_message(message=master_email_body)

__init__() #

Data Validation

**kwargs Accepts: from, to, subject, username, password, server, port

Source code in camply/notifications/email_notifications.py 
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
def __init__(self):
    """
    Data Validation

    **kwargs
        Accepts: from, to, subject, username, password, server, port
    """
    super().__init__()
    # PERFORM SOME VALIDATION
    if any(
        [
            EmailConfig.EMAIL_TO_ADDRESS in [None, ""],
            EmailConfig.EMAIL_USERNAME in [None, ""],
            EmailConfig.EMAIL_PASSWORD in [None, ""],
        ]
    ):
        variable_names = "\n\t".join(EmailConfig.ENVIRONMENT_VARIABLE_NAMES)
        optional_variable_names = "\n\t".join(
            EmailConfig.OPTIONAL_ENVIRONMENT_VARIABLE
        )
        error_message = (
            "Email Notification Auth Parameters not set. Run `camply configure` "
            f"or set the following Environment Variables:\n\t{variable_names}"
            "\nOptional Environment Variables:\n\t"
            f"{optional_variable_names}"
        )
        logger.error(error_message)
        raise EnvironmentError(error_message)
    # ATTEMPT AN EMAIL LOGIN AT INIT TO THROW ERRORS EARLY
    _email_server = SMTP_SSL(
        self.email_smtp_server,
        self.email_smtp_server_port,
    )
    _email_server.ehlo()
    _email_server.login(
        user=self.email_username,
        password=self._email_password,
    )
    _email_server.quit()

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/email_notifications.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def send_campsites(self, campsites: List[AvailableCampsite], **kwargs) -> None:
    """
    Send a message with a campsite object

    Parameters
    ----------
    campsites: List[AvailableCampsite]
    """
    master_email_body_list = []
    for campsite in campsites:
        message_title, formatted_dict = self.format_standard_campsites(
            campsite=campsite,
        )
        fields = [message_title]
        for key, value in formatted_dict.items():
            if key == "Permitted Equipment":
                value = value.replace("\n  - ", "\n  \t  - ")
            fields.append(f"\t{key}: {value}")
        composed_message = "\n".join(fields) + "\n\n"
        master_email_body_list.append(composed_message)
    master_email_body = "\n".join(master_email_body_list)
    if len(campsites) > 0:
        self.send_message(message=master_email_body)

send_message(message, **kwargs) #

Send a message via Email

Parameters:

Name Type Description Default
message str

Email Body

 required
**kwargs

Accepts: from, to, subject, username, password, server, port

 {}

Returns:

Type Description
object
Source code in camply/notifications/email_notifications.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
def send_message(self, message: str, **kwargs) -> None:
    """
    Send a message via Email

    Parameters
    ----------
    message: str
        Email Body
    **kwargs
        Accepts: from, to, subject, username, password, server, port

    Returns
    -------
    object
    """
    email = EmailMessage()
    email.set_content(message)
    email["Subject"] = kwargs.get("subject", self.email_subject)
    email["From"] = kwargs.get("from", self.email_from)
    email["To"] = kwargs.get("to", self.email_to)
    email_server_user = kwargs.get("username", self.email_username)
    email_server_password = kwargs.get("password", self._email_password)
    email_server_smtp_server = kwargs.get("server", self.email_smtp_server)
    email_server_smtp_server_port = kwargs.get("port", self.email_smtp_server_port)
    email_server = SMTP_SSL(email_server_smtp_server, email_server_smtp_server_port)
    email_server.ehlo()
    email_server.login(user=email_server_user, password=email_server_password)
    logger.info(f"Sending Email to {email['To']}: {email['Subject']}")
    email_server.send_message(email)
    logger.info("Email sent successfully")
    email_server.quit()

MultiNotifierProvider #

Bases: BaseNotifications

Notifications Supported from Multiple Providers

Source code in camply/notifications/multi_provider_notifications.py 
 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
125
126
127
128
129
130
class MultiNotifierProvider(BaseNotifications):
    """
    Notifications Supported from Multiple Providers
    """

    def __init__(self, provider: Union[str, List[str], BaseNotifications, None]):
        """
        Initialize with a Notifier Class Object, a string or list of strings

        Parameters
        ----------
        provider: Union[str, List[str], BaseNotifications, None]
            Provider String, Comma Separated Provider String, or list of provider
            strings
        """
        super().__init__()
        self.providers = [SilentNotifications()]
        if isinstance(provider, str):
            provider = [prov_string.strip() for prov_string in provider.split(",")]
        for notifier_object in provider:
            if isinstance(notifier_object, BaseNotifications):
                notifier = notifier_object
            elif isinstance(notifier_object, str):
                notifier = CAMPSITE_NOTIFICATIONS.get(notifier_object.lower(), None)()
            elif notifier_object is None:
                notifier = None
            else:
                raise NotificationError(
                    "You must provide a proper Notification Identifier"
                )
            if notifier is not None and not isinstance(notifier, SilentNotifications):
                self.providers.append(notifier)

    def send_message(self, message: str, **kwargs):
        """
        Send a message

        Parameters
        ----------
        message: str
            Message Text
        **kwargs
            All kwargs passed to underlying notification method
        """
        for provider in self.providers:
            provider.send_message(message=message, **kwargs)

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

        Parameters
        ----------
        campsites: List[AvailableCampsite]
        """
        for provider in self.providers:
            provider.send_campsites(campsites=campsites, **kwargs)

    def log_providers(self) -> None:
        """
        Log All Providers

        Returns
        -------
        None
        """
        provider_names = [str(provider) for provider in self.providers]
        logger.info(f"Notifications active via: {', '.join(provider_names)}")
        if len(self.providers) == 1:
            logger.info(
                f"Only {self.providers[0]} enabled. "
                "I hope you're watching these logs."
            )

    def last_gasp(self, error: Exception) -> None:
        """
        Make a `last gasp` notification before exiting

        Returns
        -------
        None
        """
        logger.info("Exception encountered, emitting notification last gasp.")
        date_string = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        error_string = str(error)
        error_message = (
            "camply encountered an error and exited 😟 "
            f"[{date_string}] - ({error.__class__.__name__}) {error_string}"
        )
        for provider in self.providers:
            if provider.last_gasp is True:
                provider.send_message(error_message)
        raise RuntimeError(error_message) from error

__init__(provider) #

Initialize with a Notifier Class Object, a string or list of strings

Parameters:

Name Type Description Default
provider Union[str, List[str], BaseNotifications, None]

Provider String, Comma Separated Provider String, or list of provider strings

 required
Source code in camply/notifications/multi_provider_notifications.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
def __init__(self, provider: Union[str, List[str], BaseNotifications, None]):
    """
    Initialize with a Notifier Class Object, a string or list of strings

    Parameters
    ----------
    provider: Union[str, List[str], BaseNotifications, None]
        Provider String, Comma Separated Provider String, or list of provider
        strings
    """
    super().__init__()
    self.providers = [SilentNotifications()]
    if isinstance(provider, str):
        provider = [prov_string.strip() for prov_string in provider.split(",")]
    for notifier_object in provider:
        if isinstance(notifier_object, BaseNotifications):
            notifier = notifier_object
        elif isinstance(notifier_object, str):
            notifier = CAMPSITE_NOTIFICATIONS.get(notifier_object.lower(), None)()
        elif notifier_object is None:
            notifier = None
        else:
            raise NotificationError(
                "You must provide a proper Notification Identifier"
            )
        if notifier is not None and not isinstance(notifier, SilentNotifications):
            self.providers.append(notifier)

last_gasp(error) #

Make a last gasp notification before exiting

Returns:

Type Description
None
Source code in camply/notifications/multi_provider_notifications.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def last_gasp(self, error: Exception) -> None:
    """
    Make a `last gasp` notification before exiting

    Returns
    -------
    None
    """
    logger.info("Exception encountered, emitting notification last gasp.")
    date_string = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    error_string = str(error)
    error_message = (
        "camply encountered an error and exited 😟 "
        f"[{date_string}] - ({error.__class__.__name__}) {error_string}"
    )
    for provider in self.providers:
        if provider.last_gasp is True:
            provider.send_message(error_message)
    raise RuntimeError(error_message) from error

log_providers() #

Log All Providers

Returns:

Type Description
None
Source code in camply/notifications/multi_provider_notifications.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def log_providers(self) -> None:
    """
    Log All Providers

    Returns
    -------
    None
    """
    provider_names = [str(provider) for provider in self.providers]
    logger.info(f"Notifications active via: {', '.join(provider_names)}")
    if len(self.providers) == 1:
        logger.info(
            f"Only {self.providers[0]} enabled. "
            "I hope you're watching these logs."
        )

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/multi_provider_notifications.py 
85
86
87
88
89
90
91
92
93
94
def send_campsites(self, campsites: List[AvailableCampsite], **kwargs):
    """
    Send a message with a campsite object

    Parameters
    ----------
    campsites: List[AvailableCampsite]
    """
    for provider in self.providers:
        provider.send_campsites(campsites=campsites, **kwargs)

send_message(message, **kwargs) #

Send a message

Parameters:

Name Type Description Default
message str

Message Text

 required
**kwargs

All kwargs passed to underlying notification method

 {}
Source code in camply/notifications/multi_provider_notifications.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
def send_message(self, message: str, **kwargs):
    """
    Send a message

    Parameters
    ----------
    message: str
        Message Text
    **kwargs
        All kwargs passed to underlying notification method
    """
    for provider in self.providers:
        provider.send_message(message=message, **kwargs)

PushbulletNotifications #

Bases: BaseNotifications

Push Notifications via PushBullet

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

    def __init__(self):
        super().__init__()
        pushbullet_headers = PushbulletConfig.API_HEADERS.copy()
        pushbullet_headers.update({"Access-Token": PushbulletConfig.API_TOKEN})
        self.session.headers.update(pushbullet_headers)
        if any([PushbulletConfig.API_TOKEN is None, PushbulletConfig.API_TOKEN == ""]):
            warning_message = (
                "Pushbullet is not configured properly. To send Pushbullet messages "
                "make sure to run `camply configure` or set the "
                "proper environment variable: `PUSHBULLET_API_TOKEN`."
            )
            logger.error(warning_message)
            raise EnvironmentError(warning_message)

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

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

        Returns
        -------
        requests.Response
        """
        message_type = kwargs.pop("type", "note")
        message_title = kwargs.pop("title", "Camply Notification")
        message_json = dict(
            type=message_type, title=message_title, body=message, **kwargs
        )
        logger.debug(message_json)
        response = self.session.post(
            url=PushbulletConfig.PUSHBULLET_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 Pushbullet. "
                "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(f"{key}: {value}")
            composed_message = "\n".join(fields)
            self.send_message(
                message=composed_message, title=message_title, type="note"
            )

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/pushbullet.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
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(f"{key}: {value}")
        composed_message = "\n".join(fields)
        self.send_message(
            message=composed_message, title=message_title, type="note"
        )

send_message(message, **kwargs) #

Send a message via PushBullet - if environment variables are configured

Parameters:

Name Type Description Default
message str
required

Returns:

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

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

    Returns
    -------
    requests.Response
    """
    message_type = kwargs.pop("type", "note")
    message_title = kwargs.pop("title", "Camply Notification")
    message_json = dict(
        type=message_type, title=message_title, body=message, **kwargs
    )
    logger.debug(message_json)
    response = self.session.post(
        url=PushbulletConfig.PUSHBULLET_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 Pushbullet. "
            "Your configuration might be incorrect."
        )
        raise ConnectionError(response.text) from he
    return response

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

SilentNotifications #

Bases: BaseNotifications

Silent Notifications

Source code in camply/notifications/silent_notifications.py 
15
16
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
class SilentNotifications(BaseNotifications):
    """
    Silent Notifications
    """

    def send_message(self, message: str, **kwargs) -> None:
        """
        Send a message via Email

        Parameters
        ----------
        message: str
            Email Body
        **kwargs
            kwargs are disregarded

        Returns
        -------
        None
        """
        logger.debug(f"SilentNotification: {message}")

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

        Parameters
        ----------
        campsites: List[AvailableCampsite]
        """
        for campsite in campsites:
            campsite_tuple = (
                (
                    f"{campsite.booking_date.strftime('%Y-%m-%d')} - "
                    f"{campsite.booking_end_date.strftime('%Y-%m-%d')}"
                ),
                campsite.campsite_type,
                campsite.campsite_site_name,
                campsite.recreation_area,
                campsite.facility_name,
                campsite.booking_url,
            )
            message_string = "\n\t• " + "\n\t• ".join(campsite_tuple)
            self.send_message(message_string)
            campsite_formatted = pformat(campsite.dict())
            logger.debug("Campsite Info: " + campsite_formatted)

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/silent_notifications.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def send_campsites(self, campsites: List[AvailableCampsite], **kwargs):
    """
    Send a message with a campsite object

    Parameters
    ----------
    campsites: List[AvailableCampsite]
    """
    for campsite in campsites:
        campsite_tuple = (
            (
                f"{campsite.booking_date.strftime('%Y-%m-%d')} - "
                f"{campsite.booking_end_date.strftime('%Y-%m-%d')}"
            ),
            campsite.campsite_type,
            campsite.campsite_site_name,
            campsite.recreation_area,
            campsite.facility_name,
            campsite.booking_url,
        )
        message_string = "\n\t• " + "\n\t• ".join(campsite_tuple)
        self.send_message(message_string)
        campsite_formatted = pformat(campsite.dict())
        logger.debug("Campsite Info: " + campsite_formatted)

send_message(message, **kwargs) #

Send a message via Email

Parameters:

Name Type Description Default
message str

Email Body

 required
**kwargs

kwargs are disregarded

 {}

Returns:

Type Description
None
Source code in camply/notifications/silent_notifications.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def send_message(self, message: str, **kwargs) -> None:
    """
    Send a message via Email

    Parameters
    ----------
    message: str
        Email Body
    **kwargs
        kwargs are disregarded

    Returns
    -------
    None
    """
    logger.debug(f"SilentNotification: {message}")

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

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

TwilioNotifications #

Bases: BaseNotifications

Push Notifications via Twilio

Source code in camply/notifications/twilio.py 
16
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
class TwilioNotifications(BaseNotifications):
    """
    Push Notifications via Twilio
    """

    def __init__(self):
        super().__init__()
        try:
            from twilio.rest import Client
        except ImportError as ie:
            raise RuntimeError(
                "Looks like `twilio` isn't installed. Install it with `pip install camply[twilio]`"
            ) from ie

        if any(
            [
                TwilioConfig.ACCOUNT_SID is None,
                TwilioConfig.ACCOUNT_SID == "",
                TwilioConfig.AUTH_TOKEN is None,
                TwilioConfig.AUTH_TOKEN == "",
            ]
        ):
            warning_message = (
                "Twilio is not configured properly. To send Twilio messages "
                "make sure to run `camply configure` or set the "
                "proper environment variable: `TWILIO_ACCOUNT_SID`, `TWILIO_AUTH_TOKEN`."
            )
            logger.error(warning_message)
            raise EnvironmentError(warning_message)
        self.client = Client(TwilioConfig.ACCOUNT_SID, TwilioConfig.AUTH_TOKEN)
        self.phone_nums = TwilioConfig.DEST_NUMBERS.split(",")
        logger.info(
            "Twilio: will notify these phone numbers: " + ", ".join(self.phone_nums)
        )

    def send_message(self, message: str, **kwargs):
        """
        Send a message via Twilio - if environment variables are configured

        Parameters
        ----------
        message: str
        """
        for phone_num in self.phone_nums:
            self.client.messages.create(
                to=phone_num, from_=TwilioConfig.SOURCE_NUMBER, body=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 = [f"🏕{message_title}", ""]
            for key, value in formatted_dict.items():
                fields.append(f"{key}: {value}")
            fields.append("")
            fields.append("camply, the campsite finder ⛺️")
            composed_message = "\n".join(fields)
            self.send_message(message=composed_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/twilio.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
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 = [f"🏕{message_title}", ""]
        for key, value in formatted_dict.items():
            fields.append(f"{key}: {value}")
        fields.append("")
        fields.append("camply, the campsite finder ⛺️")
        composed_message = "\n".join(fields)
        self.send_message(message=composed_message)

send_message(message, **kwargs) #

Send a message via Twilio - if environment variables are configured

Parameters:

Name Type Description Default
message str
required
Source code in camply/notifications/twilio.py 
51
52
53
54
55
56
57
58
59
60
61
62
def send_message(self, message: str, **kwargs):
    """
    Send a message via Twilio - if environment variables are configured

    Parameters
    ----------
    message: str
    """
    for phone_num in self.phone_nums:
        self.client.messages.create(
            to=phone_num, from_=TwilioConfig.SOURCE_NUMBER, body=message
        )