Newer
Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
from typing import Tuple
from grafanalib.core import Dashboard
from grafanalib._gen import DashboardEncoder
import json
import requests
import os
import dotenv
def get_dashboard_json(dashboard, overwrite=False, message="Updated by grafanlib", folder=0):
'''
get_dashboard_json generates JSON from grafanalib Dashboard object
:param dashboard - Dashboard() created via grafanalib
'''
# grafanalib generates json which need to pack to "dashboard" root element
return json.dumps(
{
"dashboard": dashboard.to_json_data(),
"overwrite": overwrite,
"message": message,
"folderId": folder,
}, sort_keys=True, indent=2, cls=DashboardEncoder)
def upload_to_grafana(json, server, api_key, verify=True):
'''
upload_to_grafana tries to upload dashboard to grafana and prints response
:param json - dashboard json generated by grafanalib
:param server - grafana server name
:param api_key - grafana api key with read and write privileges
'''
headers = {'Authorization': f"Bearer {api_key}", 'Content-Type': 'application/json'}
r = requests.post(f"https://{server}/api/dashboards/db", data=json, headers=headers, verify=verify)
# TODO: add error handling
print(f"{r.status_code} - {r.content}")
def get_folder_list(server, api_key, verify=True):
'''
upload_to_grafana tries to upload dashboard to grafana and prints response
:param server - grafana server name
:param api_key - grafana api key with read and write privileges
'''
headers = {'Authorization': f"Bearer {api_key}", 'Content-Type': 'application/json'}
r = requests.get(f"https://{server}/api/folders", headers=headers, verify=verify)
# TODO: add error handling
print(f"{r.status_code} - {r.content}")
def load_config_from_env(env_path: str = ".env") -> Tuple[str, str]:
if os.path.exists(env_path):
dotenv.load_dotenv(env_path)
grafana_api_key = os.getenv("GRAFANA_API_KEY")
grafana_server = os.getenv("GRAFANA_SERVER")
return grafana_server, grafana_api_key
def upload_dashboard(dashboard: Dashboard, folder: int) -> None:
grafana_server, grafana_api_key = load_config_from_env()
dashboard_json = get_dashboard_json(dashboard, overwrite=True, folder=folder)
upload_to_grafana(dashboard_json, grafana_server, grafana_api_key)