diff --git a/brian_dashboard_manager/grafana/dashboard.py b/brian_dashboard_manager/grafana/dashboard.py index 5b59a0bfa7e2cc301e84716c02bacb61bcc38b5f..394895f044fa9575d57db38ab9ff662aef650823 100644 --- a/brian_dashboard_manager/grafana/dashboard.py +++ b/brian_dashboard_manager/grafana/dashboard.py @@ -1,21 +1,101 @@ 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