Implement foundation for accessing the Alerting Provisioning API

Author: bursztym

grafana_client/api.py

@@ -12,6 +12,7 @@
 from .client import GrafanaClient
 from .elements import (
     Admin,
+    AlertingProvisioning,
     Annotations,
     Dashboard,
     DashboardVersions,
@@ -56,6 +57,7 @@ def __init__(
         )
         self.url = None
         self.admin = Admin(self.client)
+        self.alertingprovisioning = AlertingProvisioning(self.client)
         self.dashboard = Dashboard(self.client)
         self.dashboard_versions = DashboardVersions(self.client)
         self.datasource = Datasource(self.client, self)

grafana_client/client.py

@@ -161,6 +161,8 @@ def __request_runner(url, json=None, data=None, headers=None):
             # The "Tempo" data source responds with text/plain.
             if r.headers.get("Content-Type", "").startswith("text/"):
                 return r.text
+            elif not r.text:  # Fix for alert rules
+                return r.status_code
             else:
                 return r.json()
 

grafana_client/elements/__init__.py

@@ -1,4 +1,5 @@
 from .admin import Admin
+from .alertingprovisioning import AlertingProvisioning
 from .annotations import Annotations
 from .base import Base
 from .dashboard import Dashboard

grafana_client/elements/alertingprovisioning.py

@@ -0,0 +1,189 @@
+from .base import Base
+
+
+class AlertingProvisioning(Base):
+    def __init__(self, client):
+        super(AlertingProvisioning, self).__init__(client)
+        self.client = client
+
+    def get_alertrule(self, alertrule_uid):
+        """
+
+        :param alertrule_uid:
+        :return:
+        """
+        get_alertrule_path = "/v1/provisioning/alert-rules/%s" % alertrule_uid
+        r = self.client.GET(get_alertrule_path)
+        return r
+
+    def create_alertrule(self, alertrule):
+        """
+        :param alertrule:
+        :return:
+        """
+
+        create_alertrule_path = "/v1/provisioning/alert-rules"
+        r = self.client.POST(create_alertrule_path, json=alertrule)
+        return r
+
+    def update_alertrule(self, alertrule_uid, alertrule):
+        """
+        @param alertrule_uid:
+        @param alertrule:
+        @return:
+        """
+
+        update_alertrule_path = "/v1/provisioning/alert-rules/%s" % alertrule_uid
+        r = self.client.PUT(update_alertrule_path, json=alertrule)
+        return r
+
+    def update_rule_group_interval(self, folder_uid, group_uid, alertrule_group):
+        """
+        :param folder_uid:
+        :param group_uid:
+        :return:
+        """
+        update_rule_group_interval_path = "/v1/provisioning/folder/%s/rule-groups/%s" % (folder_uid, group_uid)
+        r = self.client.PUT(update_rule_group_interval_path, json=alertrule_group)
+        return r
+
+    def delete_alertrule(self, alertrule_uid):
+        """
+        @param alertrule_uid:
+        @param alertrule:
+        @return:
+        """
+
+        delete_alertrule_path = "/v1/provisioning/alert-rules/%s" % alertrule_uid
+        r = self.client.DELETE(delete_alertrule_path)
+        return r
+
+    def get_contactpoints(self):
+        """
+        Gets all contact points
+        @return:
+        """
+        get_contactpoints_path = "/v1/provisioning/contact-points"
+        return self.client.GET(get_contactpoints_path)
+
+    def create_contactpoint(self, contactpoint):
+        """
+        Creates single contact point
+        @param contactpoint:
+        @return:
+        """
+        create_contactpoint_path = "/v1/provisioning/contact-points"
+        return self.client.POST(create_contactpoint_path, json=contactpoint)
+
+    def update_contactpoint(self, contactpoint_uid, contactpoint):
+        """
+        Updates existing contact point
+        @param contactpoint_uid:
+        @param contactpoint:
+        @return:
+        """
+        update_contactpoint_path = "/v1/provisioning/contact-points/%s" % contactpoint_uid
+        return self.client.PUT(update_contactpoint_path, json=contactpoint)
+
+    def delete_contactpoint(self, contactpoint_uid):
+        """
+        Deletes existing contactpoint
+        @param contactpoint_uid:
+        @return:
+        """
+        delete_contactpoint_path = "/v1/provisioning/contact-points/%s" % contactpoint_uid
+        return self.client.DELETE(delete_contactpoint_path)
+
+    def get_notification_policy_tree(self):
+        """
+        Gets notification policy tree
+        @return:
+        """
+        get_notification_policy_tree_path = "/v1/provisioning/policies"
+        return self.client.GET(get_notification_policy_tree_path)
+
+    def set_notification_policy_tree(self, notification_policy_tree):
+        """
+        Sets notification policy tree
+        @param notification_policy_tree:
+        @return:
+        """
+        set_notification_policy_tree_path = "/v1/provisioning/policies"
+        return self.client.PUT(set_notification_policy_tree_path, json=notification_policy_tree)
+
+    def get_mute_timings(self):
+        """
+        Gets all mute timings
+        @return:
+        """
+        get_mute_timings_path = "/v1/provisioning/mute-timings"
+        return self.client.GET(get_mute_timings_path)
+
+    def get_mute_timing(self, mutetiming_name):
+        """
+        Gets single mute timing
+        @return:
+        """
+        get_mute_timing_path = "/v1/provisioning/mute-timings/%s" % mutetiming_name
+        return self.client.GET(get_mute_timing_path)
+
+    def create_mute_timing(self, mutetiming):
+        """
+        Creates single mute timing
+        @return:
+        """
+        create_mute_timing_path = "/v1/provisioning/mute-timings"
+        return self.client.POST(create_mute_timing_path, json=mutetiming)
+
+    def update_mute_timing(self, mutetiming_name, mutetiming):
+        """
+        Updates existing mute timing
+        @return:
+        """
+        update_mute_timing_path = "/v1/provisioning/mute-timings/%s" % mutetiming_name
+        return self.client.PUT(update_mute_timing_path, json=mutetiming)
+
+    def delete_mute_timing(self, mutetiming_name):
+        """
+        Deletes single mute timing
+        @return:
+        """
+        delete_mute_timing_path = "/v1/provisioning/mute-timings/%s" % mutetiming_name
+        return self.client.GET(delete_mute_timing_path)
+
+    def get_templates(self):
+        """
+        Gets all templates
+        @return:
+        """
+        get_templates_path = "/v1/provisioning/templates"
+        return self.client.GET(get_templates_path)
+
+    def get_template(self, template_name):
+        """
+        Gets single template
+        @param template_name:
+        @return:
+        """
+        get_template_path = "/v1/provisioning/templates/%s" % template_name
+        return self.client.GET(get_template_path)
+
+    def create_or_update_template(self, template_name, template):
+        """
+        Creates or updates (if given template_name exists) template
+        @param template_name:
+        @param template:
+        @return:
+        """
+        create_template_path = "/v1/provisioning/templates/%s" % template_name
+        return self.client.PUT(create_template_path, json=template)
+
+    def delete_template(self, template_name):
+        """
+        Deletes template
+        @param template_name:
+        @param template:
+        @return:
+        """
+        create_template_path = "/v1/provisioning/templates/%s" % template_name
+        return self.client.DELETE(create_template_path)

test/elements/test_alertingprovisioning.py

@@ -0,0 +1,58 @@
+import unittest
+
+import requests_mock
+
+from grafana_client import GrafanaApi
+
+ALERTRULE = {
+    "id": 2,
+    "uid": "bUUGqLiVk",
+    "orgID": 1,
+    "folderUID": "4",
+    "ruleGroup": "Any",
+    "title": "ApiAlerts",
+    "condition": "B",
+    "data": [],
+    "updated": "2022-08-17T13:00:21Z",
+    "noDataState": "NoData",
+    "execErrState": "Alerting",
+    "for": 300000000000,
+    "labels": {"alertGroup": "any", "region": "eu-west-1"},
+}
+
+
+class AlertingProvisioningTestCase(unittest.TestCase):
+    def setUp(self):
+        self.grafana = GrafanaApi(("admin", "admin"), host="localhost", url_path_prefix="", protocol="http")
+
+    @requests_mock.Mocker()
+    def test_get_alertrule(self, m):
+        m.get("http://localhost/api/v1/provisioning/alert-rules/bUUGqLiVk", json=ALERTRULE)
+        response = self.grafana.alertingprovisioning.get_alertrule("bUUGqLiVk")
+        self.assertEqual(response["uid"], "bUUGqLiVk")
+
+    @requests_mock.Mocker()
+    def test_delete_alertrule(self, m):
+        m.delete("http://localhost/api/v1/provisioning/alert-rules/bUUGqLiVk", json={"uid": "bUUGqLiVk"})
+        response = self.grafana.alertingprovisioning.delete_alertrule("bUUGqLiVk")
+        self.assertEqual(response["uid"], "bUUGqLiVk")
+
+    @requests_mock.Mocker()
+    def test_create_alertrule(self, m):
+        m.post(
+            "http://localhost/api/v1/provisioning/alert-rules",
+            json=ALERTRULE,
+        )
+
+        response = self.grafana.alertingprovisioning.create_alertrule(ALERTRULE)
+        self.assertEqual(response["uid"], "bUUGqLiVk")
+
+    @requests_mock.Mocker()
+    def test_update_alertrule(self, m):
+        m.put(
+            "http://localhost/api/v1/provisioning/alert-rules/bUUGqLiVk",
+            json=ALERTRULE,
+        )
+
+        response = self.grafana.alertingprovisioning.update_alertrule(alertrule_uid="bUUGqLiVk", alertrule=ALERTRULE)
+        self.assertEqual(response["uid"], "bUUGqLiVk")