dashboard.py

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__)


# 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)
        return r
    except HTTPError as e:
        logger.error(
            f'Error when provisioning dashboard {dashboard["title"]}: {e.response.text}')
    return None