Skip to content

multi_provider_notifications

Default Notifier: Silent + Extras

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)