dashboard.py

"""
Grafana Dashhboard API endpoints wrapper functions.
"""
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


def delete_dashboard(request: TokenRequest, dashboard, folder_id=None):
    try:
        r = None
        uid = dashboard.get('uid')
        if uid:
            return _delete_dashboard(request, uid)
        elif dashboard.get('title'):
            # if a folder ID is not supplied,
            # dashboard title should be globally unique
            dash = _search_dashboard(request, dashboard, folder_id)
            if dash is None:
                return True
            _delete_dashboard(request, dash.get(
                'dashboard', {}).get('uid', ''))

        logger.info(f'Deleted dashboard: {dashboard.get("title")}')
        return r is not None

    except HTTPError:
        dump = json.dumps(dashboard, indent=2)
        logger.exception(
            f'Error when deleting dashboard:\n{dump}')
        return None


# 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 True
    except HTTPError as e:
        if e.response is not None and e.response.status_code == 404:
            return True
        logger.exception(f'Error when deleting dashboard with UID #{uid}')
        return False


# 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


# lists all dashboards, optionally within a folder
def list_dashboards(request: TokenRequest, folder_id=None):
    params = {
        'query': ''
    }
    if folder_id is not None:
        params['folderIds'] = folder_id

    r = request.get('api/search', params=params)
    return r


# Searches for a dashboard with given title
def find_dashboard(request: TokenRequest, title=None):
    param = {
        **({'query': title} if title else {}),
        'type': 'dash-db',
        'limit': 5000,
        'page': 1
    }
    r = request.get('api/search', params=param)
    if r and len(r) > 0:
        if title:
            return r[0]
        else:
            while True:
                param['page'] += 1
                page = request.get('api/search', params=param)
                if len(page) > 0:
                    r.extend(page)
                else:
                    break
            return r

    return None


# Searches Grafana for a dashboard
# matching the title of the provided dashboard.
def _search_dashboard(request: TokenRequest, dashboard: Dict, folder_id=None):
    try:
        params = {
            'query': dashboard["title"]
        }
        if folder_id is not None:
            params['folderIds'] = folder_id

        r = request.get('api/search', params=params)
        if r and isinstance(r, list):
            if len(r) >= 1:
                for dash in r:
                    if dash['title'] == dashboard['title']:
                        definition = _get_dashboard(request, dash['uid'])
                        return definition
        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, folder_id=None):

    title = dashboard['title']
    existing_dashboard = None
    has_uid = dashboard.get('uid') is not None
    if has_uid:
        existing_dashboard = _get_dashboard(request, uid=dashboard['uid'])

    # The title might not match the one that's provisioned with that UID.
    # Try to find it by searching for the title instead.
    if existing_dashboard is not None:
        grafana_title = existing_dashboard['dashboard']['title']
        different = grafana_title != title
    else:
        different = False

    if existing_dashboard is None or different:
        existing_dashboard = _search_dashboard(request, dashboard, folder_id)

    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
    }
    if folder_id:
        payload['folderId'] = folder_id

    try:
        # action = "Updating" if existing_dashboard else "Creating"
        # logger.info(f'{action} dashboard: {title}')
        r = request.post('api/dashboards/db', json=payload)
        return r
    except HTTPError:
        logger.exception(f'Error when provisioning dashboard {title}')
    return None