datasource.py

import logging
import re
import os
import json
from typing import Dict, List

from requests.exceptions import HTTPError
from brian_dashboard_manager.grafana.utils.request import Request, TokenRequest


logger = logging.getLogger(__name__)


def _datasource_provisioned(datasource_to_check: Dict, provisioned_datasources: List[Dict]):
    if len(datasource_to_check.keys()) == 0:
        return True
    for datasource in provisioned_datasources:
        exists = all(datasource.get(key) == datasource_to_check.get(key)
                     for key in datasource_to_check)
        if exists:
            return True
    return False


def get_missing_datasource_definitions(request: Request, dir=None):
    datasource_dir = dir or os.path.join(
        os.path.dirname(__file__), '../../datasources/')

    provisioned_datasources = get_datasources(request)
    for (dirpath, _, filenames) in os.walk(datasource_dir):  # pragma: no cover
        for file in filenames:
            if file.endswith('.json'):
                filename = os.path.join(dirpath, file)
                datasource = json.load(open(filename, 'r'))
                if not _datasource_provisioned(datasource, provisioned_datasources):
                    yield datasource


def get_datasources(request: Request):
    return request.get('api/datasources')


def create_datasource(request: TokenRequest, datasource: Dict, environment):
    try:
        datasource["url"] = re.sub(
            'test|uat|prod', environment, datasource["url"])
        r = request.post('api/datasources', json=datasource)
    except HTTPError as e:
        logger.error('Error when provisioning datasource: ' + e.response.text)
        return None
    return r


def delete_datasource(request: TokenRequest, name: str):
    return request.delete(f'api/datasources/name/{name}')