-
-def get_all_dashboard_uids():
- response = requests.request("GET", url + "search?query=%", headers=headers)
- dashboards = response.json()
- dashboard_uids = []
- for dashboard in dashboards:
- dashboard_uids.append(dashboard['uid'])
- log.debug("Searching for all dashboard uids: %s", dashboard_uids)
- return dashboard_uids
-
-
-def get_dashboard_status(uid):
- response = requests.request("GET", url + "dashboards/uid/" + uid, headers=headers)
- log.debug("Searching for dashboard result: %s", response.text)
- return response
-
-
-def create_dashboard(uid, name, json_file):
- with open(json_file) as f:
- dashboard_data = f.read()
-
- dashboard_data = dashboard_data.replace('OSM_ID', uid).replace('OSM_NAME', name)
-
- response = requests.request("POST", url + "dashboards/db/", data=dashboard_data, headers=headers)
- log.debug("Creating dashboard result: %s", response.text)
- return response
-
-
-def delete_dashboard(uid):
- response = requests.request("DELETE", url + "dashboards/uid/" + uid, headers=headers)
- log.debug("Delete dashboard result: %s", response.text)
- return response
+class GrafanaBackend:
+ def __init__(self, config: Config):
+ self.conf = config
+ self.url = config.get('grafana', 'url')
+ grafana_user = config.get("grafana", "user")
+ grafana_password = config.get("grafana", "password")
+ self.headers = {
+ 'content-type': "application/json",
+ 'authorization': "Basic %s" % base64.b64encode(
+ (grafana_user + ":" + grafana_password).encode("utf-8")).decode()
+ }
+
+ def get_all_dashboard_uids(self):
+ # Gets only dashboards that were automated by OSM (with tag 'osm_automated')
+ response = requests.request("GET", self.url + "/api/search?tag=osm_automated", headers=self.headers)
+ dashboards = response.json()
+ dashboard_uids = []
+ for dashboard in dashboards:
+ dashboard_uids.append(dashboard['uid'])
+ log.debug("Searching for all dashboard uids: %s", dashboard_uids)
+ return dashboard_uids
+
+ def get_dashboard_status(self, uid):
+ response = requests.request("GET", self.url + "/api/dashboards/uid/" + uid, headers=self.headers)
+ log.debug("Searching for dashboard result: %s", response.text)
+ return response
+
+ def create_dashboard(self, uid, name, json_file):
+ with open(json_file) as f:
+ dashboard_data = f.read()
+
+ dashboard_data = dashboard_data.replace('OSM_ID', uid).replace('OSM_NAME', name)
+
+ response = requests.request(
+ "POST", self.url + "/api/dashboards/db/", data=dashboard_data, headers=self.headers)
+ log.debug("Creating dashboard result: %s", response.text)
+ return response
+
+ def delete_dashboard(self, uid):
+ response = requests.request("DELETE", self.url + "/api/dashboards/uid/" + uid, headers=self.headers)
+ log.debug("Delete dashboard result: %s", response.text)
+ return response