[Apprise] Improve integration

... by propagating the mqttwarn data dictionary into the Apprise plugin
template arguments.

Author: amotl

CHANGES.rst

@@ -11,6 +11,8 @@ in progress
 - Add versioning based on Git tags, using ``versioningit``. This will aid in
   telling PR- and nightly releases apart from GA releases when running
   ``mqttwarn --version``.
+- Improve Apprise integration by propagating the mqttwarn data dictionary into
+  the Apprise plugin template arguments.
 
 
 2022-11-21 0.31.0

mqttwarn/services/apprise_multi.py

@@ -5,11 +5,14 @@
 __license__   = 'Eclipse Public License - v 1.0 (http://www.eclipse.org/legal/epl-v10.html)'
 
 # https://github.com/caronc/apprise#developers
-from urllib.parse import urlencode
 from collections import OrderedDict
 
 import apprise
 
+from mqttwarn.services.apprise_util import obtain_apprise_arguments, add_url_params, get_all_template_argument_names
+
+APPRISE_ALL_ARGUMENT_NAMES = get_all_template_argument_names()
+
 
 def plugin(srv, item):
     """Send a message to multiple Apprise plugins."""
@@ -39,6 +42,10 @@ def plugin(srv, item):
             # Collect URL parameters.
             params = OrderedDict()
 
+            # Obtain and apply all possible Ntfy parameters from data dictionary.
+            params.update(obtain_apprise_arguments(item, APPRISE_ALL_ARGUMENT_NAMES))
+
+            # Apply addressee information.
             if "recipients" in address:
                 to = ','.join(address["recipients"])
                 if to:
@@ -49,10 +56,8 @@ def plugin(srv, item):
             if "sender_name" in address:
                 params["name"] = address["sender_name"]
 
-            # Add notification services by server url.
-            uri = baseuri
-            if params:
-                uri += '?' + urlencode(params)
+            # Add parameters to Apprise notification URL.
+            uri = add_url_params(baseuri, params)
             srv.logging.info("Adding notification to: {}".format(uri))
             apobj.add(uri)
 

mqttwarn/services/apprise_single.py

@@ -5,11 +5,14 @@
 __license__   = 'Eclipse Public License - v 1.0 (http://www.eclipse.org/legal/epl-v10.html)'
 
 # https://github.com/caronc/apprise#developers
-from urllib.parse import urlencode
 from collections import OrderedDict
 
 import apprise
 
+from mqttwarn.services.apprise_util import obtain_apprise_arguments, add_url_params, get_all_template_argument_names
+
+APPRISE_ALL_ARGUMENT_NAMES = get_all_template_argument_names()
+
 
 def plugin(srv, item):
     """Send a message to a single Apprise plugin."""
@@ -39,17 +42,20 @@ def plugin(srv, item):
 
         # Collect URL parameters.
         params = OrderedDict()
+
+        # Obtain and apply all possible Ntfy parameters from data dictionary.
+        params.update(obtain_apprise_arguments(item, APPRISE_ALL_ARGUMENT_NAMES))
+
+        # Apply addressee information.
         if sender:
             params["from"] = sender
         if to:
             params["to"] = to
         if sender_name:
             params["name"] = sender_name
 
-        # Add notification services by server url.
-        uri = baseuri
-        if params:
-            uri += '?' + urlencode(params)
+        # Add parameters to Apprise notification URL.
+        uri = add_url_params(baseuri, params)
         apobj.add(uri)
 
         # Submit notification.

mqttwarn/services/apprise_util.py

@@ -0,0 +1,53 @@
+# -*- coding: utf-8 -*-
+# (c) 2021-2023 The mqttwarn developers
+from __future__ import absolute_import
+
+from functools import lru_cache
+from urllib.parse import urlparse, urlencode
+
+from apprise import Apprise, ContentLocation
+
+from mqttwarn.model import ProcessorItem
+
+
+@lru_cache(maxsize=None)
+def get_all_template_argument_names():
+    """
+    Inquire all possible parameter names from all Apprise plugins.
+    """
+    a = Apprise(asset=None, location=ContentLocation.LOCAL)
+    results = a.details()
+    plugin_infos = results['schemas']
+
+    all_arg_names = []
+    for plugin_info in plugin_infos:
+        arg_names = plugin_info["details"]["args"].keys()
+        all_arg_names += arg_names
+
+    return sorted(set(all_arg_names))
+
+
+def obtain_apprise_arguments(item: ProcessorItem, arg_names: list) -> dict:
+    """
+    Obtain eventual Apprise parameters from data dictionary.
+
+    https://github.com/caronc/apprise/wiki/Notify_ntfy#parameter-breakdown
+    """
+    params = dict()
+    for arg_name in arg_names:
+        if isinstance(item.data, dict) and arg_name in item.data:
+            params[arg_name] = item.data[arg_name]
+    return params
+
+
+def add_url_params(url: str, params: dict) -> str:
+    """
+    Serialize query parameter dictionary and add it to URL.
+    """
+    url_parsed = urlparse(url)
+    if params:
+        seperator = "?"
+        if url_parsed.query:
+            seperator = "&"
+        url += seperator + urlencode(params)
+    return url

tests/conftest.py

@@ -5,6 +5,9 @@
 
 import pytest
 
+# Needed to make Apprise not be mocked too much.
+from mqttwarn.services.apprise_util import get_all_template_argument_names  # noqa:F401
+
 # Import custom fixtures.
 from mqttwarn.testing.fixtures import mqttwarn_service as srv  # noqa:F401