Skip to content

notifications

Notifications init file

AppriseNotifications #

Bases: BaseNotifications

Push Notifications via Apprise

Source code in camply/notifications/apprise.py
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
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
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
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
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
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
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
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
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
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
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
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
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
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
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
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
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)

    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
        """
        token = (
            PushoverConfig.PUSH_TOKEN
            if PushoverConfig.PUSH_TOKEN not in [None, ""]
            else base64.b64decode(PushoverConfig.PUSHOVER_DEFAULT_API_TOKEN).decode(
                "utf-8"
            )
        )
        response = self.session.post(
            url=PushoverConfig.PUSHOVER_API_ENDPOINT,
            params=dict(
                token=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
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
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
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
    """
    token = (
        PushoverConfig.PUSH_TOKEN
        if PushoverConfig.PUSH_TOKEN not in [None, ""]
        else base64.b64decode(PushoverConfig.PUSHOVER_DEFAULT_API_TOKEN).decode(
            "utf-8"
        )
    )
    response = self.session.post(
        url=PushoverConfig.PUSHOVER_API_ENDPOINT,
        params=dict(
            token=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
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
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
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
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
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
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
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
@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
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
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
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
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
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
        )