Skip to content
Snippets Groups Projects
Commit 915e1009 authored by Bjarke Madsen's avatar Bjarke Madsen
Browse files

Implement dashboard provisioning

parent a23cbf65
No related branches found
No related tags found
No related merge requests found
import logging
import os
import json
from typing import Dict
from requests.models import HTTPError
from brian_dashboard_manager.grafana.utils.request import TokenRequest
logger = logging.getLogger(__name__)
def provision_dashboard(request: TokenRequest, dashboard: Dict):
del dashboard['uid']
del dashboard['id']
del dashboard['version']
# Returns dictionary for each dashboard JSON definition in supplied directory
def get_dashboard_definitions(dir=None): # pragma: no cover
dashboard_dir = dir or os.path.join(
os.path.dirname(__file__), '../../dashboards/')
for (dirpath, _, filenames) in os.walk(dashboard_dir):
for file in filenames:
if file.endswith('.json'):
filename = os.path.join(dirpath, file)
dashboard = json.load(open(filename, 'r'))
yield dashboard
# Deletes a single dashboard for the organization the API token is registered to.
def _delete_dashboard(request: TokenRequest, uid: int):
try:
r = request.delete(f'api/dashboards/uid/{uid}')
if r and 'deleted' in r.get('message', ''):
return r
except HTTPError as e:
logger.error(
f'Error when deleting dashboard with UID #{uid}: {e.response.text}')
return None
# Deletes all dashboards for the organization the API token is registered to.
def delete_dashboards(request: TokenRequest):
r = request.get('api/search')
if r and len(r) > 0:
for dash in r:
_delete_dashboard(request, dash['uid'])
return True
# Searches Grafana for a dashboard matching the title of the provided dashboard.
def _search_dashboard(request: TokenRequest, dashboard: Dict):
try:
r = request.get('api/search', params={
'query': f'{dashboard["title"]}'
})
if r and isinstance(r, list):
if len(r) >= 1:
for dash in r:
if dash['title'] == dashboard['title']:
return _get_dashboard(request, dash['uid'])
return None
except HTTPError:
return None
# Fetches dashboard with given UID for the token's organization.
def _get_dashboard(request: TokenRequest, uid: int):
try:
r = request.get(f'api/dashboards/uid/{uid}')
except HTTPError:
return None
return r
# Creates or updates (if exists) given dashboard for the token's organization.
# supplied dashboards are JSON blobs exported from GUI with a UID.
def create_dashboard(request: TokenRequest, dashboard: Dict):
existing_dashboard = None
if dashboard.get('uid') is not None:
existing_dashboard = _get_dashboard(request, uid=dashboard['uid'])
else:
existing_dashboard = _search_dashboard(request, dashboard)
if existing_dashboard:
dashboard['uid'] = existing_dashboard['dashboard']['uid']
dashboard['id'] = existing_dashboard['dashboard']['id']
dashboard['version'] = existing_dashboard['dashboard']['version']
else:
# We are creating a new dashboard, delete ID if it exists.
del dashboard['id']
payload = {
'dashboard': dashboard,
'overwrite': False
}
try:
logger.info(
f'{"Updating" if existing_dashboard else "Provisioning"} dashboard: {dashboard["title"]}')
r = request.post('api/dashboards/db', json=payload)
except Exception as e:
logger.error(f'Error when provisioning dashboard: ' + e.response.text)
return True
return r
except HTTPError as e:
logger.error(
f'Error when provisioning dashboard {dashboard["title"]}: {e.response.text}')
return None
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment