extern/apprise
1
0
Fork 0
mirror of https://github.com/caronc/apprise.git synced 2024-12-28 01:28:54 +01:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #66 from caronc/34-tag-notification-support

Browse Source 
Added ability to tag notifications; refs #34
This commit is contained in:
Chris Caron 2019-02-09 01:55:16 -05:00 committed by GitHub
commit 43f6aa33e5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 213 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

78
apprise/Apprise.py
View File

@ -121,7 +121,7 @@ class Apprise(object):
self.add(servers)
@staticmethod
def instantiate(url, asset=None, suppress_exceptions=True):
def instantiate(url, asset=None, tag=None, suppress_exceptions=True):
"""
Returns the instance of a instantiated plugin based on the provided
Server URL. If the url fails to be parsed, then None is returned.
@ -162,6 +162,9 @@ class Apprise(object):
logger.error('Could not parse URL: %s' % url)
return None
# Build a list of tags to associate with the newly added notifications
results['tag'] = set(parse_list(tag))
if suppress_exceptions:
try:
# Attempt to create an instance of our plugin using the parsed
@ -184,10 +187,16 @@ class Apprise(object):
return plugin
def add(self, servers, asset=None):
def add(self, servers, asset=None, tag=None):
"""
Adds one or more server URLs into our list.
You can override the global asset if you wish by including it with the
server(s) that you add.
The tag allows you to associate 1 or more tag values to the server(s)
being added. tagging a service allows you to exclusively access them
when calling the notify() function.
"""
# Initialize our return status
@ -202,12 +211,13 @@ class Apprise(object):
self.servers.append(servers)
return True
# build our server listings
servers = parse_list(servers)
for _server in servers:
# Instantiate ourselves an object, this function throws or
# returns None if it fails
instance = Apprise.instantiate(_server, asset=asset)
instance = Apprise.instantiate(_server, asset=asset, tag=tag)
if not instance:
return_status = False
logging.error(
@ -229,13 +239,18 @@ class Apprise(object):
self.servers[:] = []
def notify(self, title, body, notify_type=NotifyType.INFO,
body_format=None):
body_format=None, tag=None):
"""
Send a notification to all of the plugins previously loaded.
If the body_format specified is NotifyFormat.MARKDOWN, it will
be converted to HTML if the Notification type expects this.
if the tag is specified (either a string or a set/list/tuple
of strings), then only the notifications flagged with that
tagged value are notified. By default all added services
are notified (tag=None)
"""
# Initialize our return result
@ -247,8 +262,63 @@ class Apprise(object):
# Tracks conversions
conversion_map = dict()
# Build our tag setup
# - top level entries are treated as an 'or'
# - second level (or more) entries are treated as 'and'
#
# examples:
# tag="tagA, tagB" = tagA or tagB
# tag=['tagA', 'tagB'] = tagA or tagB
# tag=[('tagA', 'tagC'), 'tagB'] = (tagA and tagC) or tagB
# tag=[('tagB', 'tagC')] = tagB and tagC
# Iterate over our loaded plugins
for server in self.servers:
if tag is not None:
if isinstance(tag, (list, tuple, set)):
# using the tags detected; determine if we'll allow the
# notification to be sent or not
matched = False
# Every entry here will be or'ed with the next
for entry in tag:
if isinstance(entry, (list, tuple, set)):
# treat these entries as though all elements found
# must exist in the notification service
tags = set(parse_list(entry))
if len(tags.intersection(
server.tags)) == len(tags):
# our set contains all of the entries found
# in our notification server object
matched = True
break
elif entry in server:
# our entr(ies) match what was found in our server
# object.
matched = True
break
# else: keep looking
if not matched:
# We did not meet any of our and'ed criteria
continue
elif tag not in server:
# one or more tags were defined and they didn't match the
# entry in the current service; move along...
continue
# else: our content was found inside the server, so we're good
# If our code reaches here, we either did not define a tag (it was
# set to None), or we did define a tag and the logic above
# determined we need to notify the service it's associated with
if server.notify_format not in conversion_map:
if body_format == NotifyFormat.MARKDOWN and \
server.notify_format == NotifyFormat.HTML:

23
apprise/plugins/NotifyBase.py
View File

@ -40,6 +40,7 @@ except ImportError:
from ..utils import parse_url
from ..utils import parse_bool
from ..utils import parse_list
from ..utils import is_hostname
from ..common import NOTIFY_TYPES
from ..common import NotifyFormat
@ -127,6 +128,9 @@ class NotifyBase(object):
# Default Notify Format
notify_format = NotifyFormat.TEXT
# Maintain a set of tags to associate with this specific notification
tags = set()
# Logging
logger = logging.getLogger(__name__)
@ -172,6 +176,12 @@ class NotifyBase(object):
# Provide override
self.notify_format = notify_format
if 'tag' in kwargs:
# We want to associate some tags with our notification service.
# the code below gets the 'tag' argument if defined, otherwise
# it just falls back to whatever was already defined globally
self.tags = set(parse_list(kwargs.get('tag', self.tags)))
def throttle(self, throttle_time=None):
"""
A common throttle control
@ -249,6 +259,19 @@ class NotifyBase(object):
color_type=color_type,
)
def __contains__(self, tags):
"""
Returns true if the tag specified is associated with this notification.
tag can also be a tuple, set, and/or list
"""
if isinstance(tags, (tuple, set, list)):
return bool(set(tags) & self.tags)
# return any match
return tags in self.tags
@property
def app_id(self):
return self.asset.app_id

119
test/test_api.py
View File

@ -38,6 +38,8 @@ from apprise import NotifyImageSize
from apprise import __version__
from apprise.Apprise import __load_matrix
import pytest
import requests
import mock
def test_apprise():
@ -110,7 +112,7 @@ def test_apprise():
class BadNotification(NotifyBase):
def __init__(self, **kwargs):
super(BadNotification, self).__init__()
super(BadNotification, self).__init__(**kwargs)
# We fail whenever we're initialized
raise TypeError()
@ -118,7 +120,7 @@ def test_apprise():
class GoodNotification(NotifyBase):
def __init__(self, **kwargs):
super(GoodNotification, self).__init__(
notify_format=NotifyFormat.HTML)
notify_format=NotifyFormat.HTML, **kwargs)
def notify(self, **kwargs):
# Pretend everything is okay
@ -203,9 +205,20 @@ def test_apprise():
assert(len(a) == 0)
# Instantiate a good object
plugin = a.instantiate('good://localhost')
plugin = a.instantiate('good://localhost', tag="good")
assert(isinstance(plugin, NotifyBase))
# Test simple tagging inside of the object
assert("good" in plugin)
assert("bad" not in plugin)
# the in (__contains__ override) is based on or'ed content; so although
# 'bad' isn't tagged as being in the plugin, 'good' is, so the return
# value of this is True
assert(["bad", "good"] in plugin)
assert(set(["bad", "good"]) in plugin)
assert(("bad", "good") in plugin)
# We an add already substatiated instances into our Apprise object
a.add(plugin)
assert(len(a) == 1)
@ -225,6 +238,106 @@ def test_apprise():
assert(len(a) == 0)
@mock.patch('requests.get')
@mock.patch('requests.post')
def test_apprise_tagging(mock_post, mock_get):
"""
API: Apprise() object tagging functionality
"""
# A request
robj = mock.Mock()
setattr(robj, 'raw', mock.Mock())
# Allow raw.read() calls
robj.raw.read.return_value = ''
robj.text = ''
robj.content = ''
mock_get.return_value = robj
mock_post.return_value = robj
# Simulate a successful notification
mock_get.return_value.status_code = requests.codes.ok
mock_post.return_value.status_code = requests.codes.ok
# Create our object
a = Apprise()
# Add entry and assign it to a tag called 'awesome'
assert(a.add('json://localhost/path1/', tag='awesome') is True)
# Add another notification and assign it to a tag called 'awesome'
# and another tag called 'local'
assert(a.add('json://localhost/path2/', tag=['mmost', 'awesome']) is True)
# notify the awesome tag; this would notify both services behind the
# scenes
assert(a.notify(title="my title", body="my body", tag='awesome') is True)
# notify all of the tags
assert(a.notify(
title="my title", body="my body", tag=['awesome', 'mmost']) is True)
# there is nothing to notify using tags 'missing'. However we intentionally
# don't fail as there is value in identifying a tag that simply have
# nothing to notify from while the object itself contains items
assert(a.notify(
title="my title", body="my body", tag='missing') is True)
# Now to test the ability to and and/or notifications
a = Apprise()
# Add a tag by tuple
assert(a.add('json://localhost/tagA/', tag=("TagA", )) is True)
# Add 2 tags by string
assert(a.add('json://localhost/tagAB/', tag="TagA, TagB") is True)
# Add a tag using a set
assert(a.add('json://localhost/tagB/', tag=set(["TagB"])) is True)
# Add a tag by string (again)
assert(a.add('json://localhost/tagC/', tag="TagC") is True)
# Add 2 tags using a list
assert(a.add('json://localhost/tagCD/', tag=["TagC", "TagD"]) is True)
# Add a tag by string (again)
assert(a.add('json://localhost/tagD/', tag="TagD") is True)
# add a tag set by set (again)
assert(a.add('json://localhost/tagCDE/',
tag=set(["TagC", "TagD", "TagE"])) is True)
# Expression: TagC and TagD
# Matches the following only:
# - json://localhost/tagCD/
# - json://localhost/tagCDE/
assert(a.notify(
title="my title", body="my body", tag=[('TagC', 'TagD')]) is True)
# Expression: (TagY and TagZ) or TagX
# Matches nothing
assert(a.notify(
title="my title", body="my body",
tag=[('TagY', 'TagZ'), 'TagX']) is True)
# Expression: (TagY and TagZ) or TagA
# Matches the following only:
# - json://localhost/tagAB/
assert(a.notify(
title="my title", body="my body",
tag=[('TagY', 'TagZ'), 'TagA']) is True)
# Expression: (TagE and TagD) or TagB
# Matches the following only:
# - json://localhost/tagCDE/
# - json://localhost/tagAB/
# - json://localhost/tagB/
assert(a.notify(
title="my title", body="my body",
tag=[('TagE', 'TagD'), 'TagB']) is True)
# Garbage Entries
assert(a.notify(
title="my title", body="my body",
tag=[(object, ), ]) is True)
def test_apprise_notify_formats(tmpdir):
"""
API: Apprise() TextFormat tests