Refine API wrapper for "Plugins" once more

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## unreleased
 
+* Refine API wrapper for "Plugins" once more
+
 
 ## 3.8.2 (2023-09-20)
 

grafana_client/client.py

@@ -1,3 +1,5 @@
+from json import JSONDecodeError
+
 import requests
 import requests.auth
 
@@ -118,7 +120,7 @@ def construct_api_url():
                 self.auth = TokenAuth(self.auth)
 
     def __getattr__(self, item):
-        def __request_runner(url, json=None, data=None, headers=None):
+        def __request_runner(url, json=None, data=None, headers=None, accept_empty_json=False):
             __url = "%s%s" % (self.url, url)
             # Sanity checks.
             if json is not None and not isinstance(json, (dict, list)):
@@ -167,9 +169,15 @@ def __request_runner(url, json=None, data=None, headers=None):
                 return None
 
             # The "Tempo" data source responds with text/plain.
-            if r.headers.get("Content-Type", "").startswith("text/"):
+            content_type = r.headers.get("Content-Type", "")
+            if content_type.startswith("text/"):
                 return r.text
-            else:
+            try:
                 return r.json()
+            except JSONDecodeError:
+                if accept_empty_json:
+                    return r.text
+                else:
+                    raise
 
         return __request_runner

grafana_client/elements/plugin.py

@@ -1,5 +1,7 @@
 import logging
 
+from grafana_client.client import GrafanaClientError
+
 from .base import Base
 
 
@@ -9,60 +11,83 @@ def __init__(self, client):
         self.client = client
         self.logger = logging.getLogger(__name__)
 
-    def get_installed_plugins(self):
+    def list(self):
         """
-        :return:
+        Return list of all installed plugins.
         """
         path = "/plugins?embedded=0"
         r = self.client.GET(path)
         return r
 
-    def install_plugin(self, pluginId, version, errors="raise"):
+    def by_id(self, plugin_id):
+        """
+        Return single plugin item selected by plugin identifier.
+        """
+        plugins = self.list()
+        return get_plugin_by_id(plugin_list=plugins, plugin_id=plugin_id)
+
+    def install(self, plugin_id, version=None, errors="raise"):
         """
-        : return:
+        Install a 3rd-party plugin from the plugin store.
         """
         try:
-            path = "/plugins/%s/install" % pluginId
-            r = self.client.POST(path, json={"version": version})
+            path = "/plugins/%s/install" % plugin_id
+            # Unfortunately, this endpoint may respond with an empty JSON,
+            # which needs compensation, because it does not decode well.
+            r = self.client.POST(path, json={"version": version}, accept_empty_json=True)
             return r
+        except GrafanaClientError as ex:
+            # Ignore `Client Error 409: Plugin already installed`.
+            if "409" not in str(ex):
+                raise
         except Exception as ex:
             if errors == "raise":
                 raise
             elif errors == "ignore":
-                self.logger.info(f"Skipped installing plugin {pluginId}: {ex}")
+                self.logger.info(f"Skipped installing plugin {plugin_id}: {ex}")
             else:
                 raise ValueError(f"error={errors} is invalid")
         return None
 
-    def uninstall_plugin(self, pluginId, errors="raise"):
+    def uninstall(self, plugin_id, errors="raise"):
         """
-        : return:
+        Install a 3rd-party plugin from the Grafana instance.
         """
         try:
-            path = "/plugins/%s/uninstall" % pluginId
+            path = "/plugins/%s/uninstall" % plugin_id
             r = self.client.POST(path)
             return r
         except Exception as ex:
             if errors == "raise":
                 raise
             elif errors == "ignore":
-                self.logger.info(f"Skipped uninstalling plugin {pluginId}: {ex}")
+                self.logger.info(f"Skipped uninstalling plugin {plugin_id}: {ex}")
             else:
                 raise ValueError(f"error={errors} is invalid")
         return None
 
-    def health_check_plugin(self, pluginId):
+    def health(self, plugin_id):
         """
-        :return:
+        Run a health check probe on the designated plugin.
         """
-        path = "/plugins/%s/health" % pluginId
+        path = "/plugins/%s/health" % plugin_id
         r = self.client.GET(path)
         return r
 
-    def get_plugin_metrics(self, pluginId):
+    def metrics(self, plugin_id):
         """
-        : return:
+        Inquire metrics of the designated plugin.
         """
-        path = "/plugins/%s/metrics" % pluginId
+        path = "/plugins/%s/metrics" % plugin_id
         r = self.client.GET(path)
         return r
+
+
+def get_plugin_by_id(plugin_list, plugin_id):
+    """
+    Helper function to filter plugin list by identifier.
+    """
+    try:
+        return next(item for item in plugin_list if item["id"] == plugin_id)
+    except StopIteration:
+        raise KeyError(f"Plugin not found: {plugin_id}")

test/elements/test_plugin.py

@@ -10,7 +10,7 @@ def setUp(self):
         self.grafana = GrafanaApi(("admin", "admin"), host="localhost", url_path_prefix="", protocol="http")
 
     @requests_mock.Mocker()
-    def test_get_installed_plugins(self, m):
+    def test_list(self, m):
         m.get(
             "http://localhost/api/plugins?embedded=0",
             json=[
@@ -186,29 +186,29 @@ def test_get_installed_plugins(self, m):
                 },
             ],
         )
-        plugins = self.grafana.plugin.get_installed_plugins()
+        plugins = self.grafana.plugin.list()
         self.assertTrue(len(plugins), 6)
 
     @requests_mock.Mocker()
-    def test_install_plugin(self, m):
+    def test_install(self, m):
         m.post("http://localhost/api/plugins/alertlist/install", json={"message": "Plugin alertlist installed"})
-        response = self.grafana.plugin.install_plugin(pluginId="alertlist", version="1.3.12")
+        response = self.grafana.plugin.install(plugin_id="alertlist", version="1.3.12")
         self.assertEqual(response["message"], "Plugin alertlist installed")
 
     @requests_mock.Mocker()
-    def test_uninstall_plugin(self, m):
+    def test_uninstall(self, m):
         m.post("http://localhost/api/plugins/alertlist/uninstall", json={"message": "Plugin alertlist uninstalled"})
-        response = self.grafana.plugin.uninstall_plugin(pluginId="alertlist")
+        response = self.grafana.plugin.uninstall(plugin_id="alertlist")
         self.assertEqual(response["message"], "Plugin alertlist uninstalled")
 
     @requests_mock.Mocker()
-    def test_get_plugin_health(self, m):
+    def test_health(self, m):
         m.get("http://localhost/api/plugins/alertlist/health", json={"message": "Plugin alertlist healthy"})
-        response = self.grafana.plugin.health_check_plugin(pluginId="alertlist")
+        response = self.grafana.plugin.health(plugin_id="alertlist")
         self.assertEqual(response["message"], "Plugin alertlist healthy")
 
     @requests_mock.Mocker()
-    def test_get_plugin_metrics(self, m):
+    def test_metrics(self, m):
         m.get("http://localhost/api/plugins/grafana-timestream-datasource/metrics", json={"message": "Not found"})
-        response = self.grafana.plugin.get_plugin_metrics(pluginId="grafana-timestream-datasource")
+        response = self.grafana.plugin.metrics(plugin_id="grafana-timestream-datasource")
         self.assertEqual(response["message"], "Not found")