Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform Zulip Notification
"""
headers = {
'User-Agent': self.app_id,
'Content-Type': 'application/x-www-form-urlencoded; charset=utf-8',
}
# error tracking (used for function return)
has_error = False
# Prepare our notification URL
url = self.notify_url.format(
org=self.organization,
hostname=self.hostname,
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform Twilio Notification
"""
# error tracking (used for function return)
has_error = False
# Prepare our headers
headers = {
'User-Agent': self.app_id,
'Accept': 'application/json',
}
# Prepare our payload
payload = {
'Body': body,
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform Emby Notification
"""
if not self.is_authenticated and not self.login():
# Authenticate if we aren't already
return False
# Acquire our list of sessions
sessions = self.sessions().keys()
if not sessions:
self.logger.warning('There were no Emby sessions to notify.')
# We don't need to fail; there really is no one to notify
return True
url = '%s://%s' % (self.schema, self.host)
if self.port:
def notify(self, body, title=None, notify_type=NotifyType.INFO,
overflow=None, **kwargs):
"""
Performs notification
"""
# Handle situations where the title is None
title = '' if not title else title
# Apply our overflow (if defined)
for chunk in self._apply_overflow(body=body, title=title,
overflow=overflow):
# Send notification
if not self.send(body=chunk['body'], title=chunk['title'],
notify_type=notify_type):
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform XML Notification
"""
# prepare XML Object
headers = {
'User-Agent': self.app_id,
'Content-Type': 'application/xml'
}
# Apply any/all header over-rides defined
headers.update(self.headers)
re_map = {
'{MESSAGE_TYPE}': NotifyXML.escape_html(
notify_type, whitespace=False),
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform Pushjet Notification
"""
params = {
'secret': self.secret_key,
}
# prepare Pushjet Object
payload = {
'message': body,
'title': title,
'link': None,
'level': None,
}
URL masks.
"""
# Application Identifier
app_id = 'Apprise'
# Application Description
app_desc = 'Apprise Notifications'
# Provider URL
app_url = 'https://github.com/caronc/apprise'
# A Simple Mapping of Colors; For every NOTIFY_TYPE identified,
# there should be a mapping to it's color here:
html_notify_map = {
NotifyType.INFO: '#3AA3E3',
NotifyType.SUCCESS: '#3AA337',
NotifyType.FAILURE: '#A32037',
NotifyType.WARNING: '#CACF29',
}
# The default color to return if a mapping isn't found in our table above
default_html_color = '#888888'
# The default image extension to use
default_extension = '.png'
# The default theme
theme = 'default'
# Image URL Mask
image_url_mask = \
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
wrapper to _send since we can alert more then one channel
"""
# Call the _send_ function applicable to whatever mode we're in
# - calls _send_webhook_notification if the mode variable is set
# - calls _send_basic_notification if the mode variable is not set
return getattr(self, '_send_{}_notification'.format(self.mode))(
body=body, title=title, notify_type=notify_type, **kwargs)
def send(self, body, title='', notify_type=NotifyType.INFO, attach=None,
**kwargs):
"""
Perform Slack Notification
"""
# error tracking (used for function return)
has_error = False
# Perform Formatting
title = self._re_formatting_rules.sub( # pragma: no branch
lambda x: self._re_formatting_map[x.group()], title,
)
body = self._re_formatting_rules.sub( # pragma: no branch
lambda x: self._re_formatting_map[x.group()], body,
)
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
"""
Perform DBus Notification
"""
if not self._enabled or MAINLOOP_MAP[self.schema] is None:
self.logger.warning(
"{} notifications could not be loaded.".format(self.schema))
return False
# Acquire our session
try:
session = SessionBus(mainloop=MAINLOOP_MAP[self.schema])
except DBusException:
# Handle exception
self.logger.warning('Failed to send DBus notification.')