From 54f02c6ec12ae06ff65d894c1878b681e82b7943 Mon Sep 17 00:00:00 2001 From: Andreas Motl Date: Sat, 27 Apr 2024 11:25:10 +0200 Subject: [PATCH] Chore: Format code and activate linters using Black and Ruff --- .github/workflows/tests.yml | 9 +- grafana_import/cli.py | 581 ++++++++++++++++---------------- grafana_import/constants.py | 6 +- grafana_import/grafana.py | 642 +++++++++++++++++------------------- grafana_import/service.py | 2 +- grafana_import/util.py | 35 +- pyproject.toml | 87 +++-- setup.py | 22 +- tests/conftest.py | 4 +- tests/test_cli.py | 3 +- tests/test_core.py | 2 +- tests/util.py | 26 +- 12 files changed, 690 insertions(+), 729 deletions(-) diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 17117e3..25ab0d6 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -56,9 +56,14 @@ jobs: # Install package in editable mode. pip install --use-pep517 --prefer-binary --editable=.[test,develop] - - name: Run linter and software tests + - name: Check code style + if: matrix.python-version != '3.6' && matrix.python-version != '3.7' run: | - poe check + poe lint + + - name: Run software tests + run: | + poe test - name: Upload coverage to Codecov uses: codecov/codecov-action@v4 diff --git a/grafana_import/cli.py b/grafana_import/cli.py index 911f14e..c777ce4 100644 --- a/grafana_import/cli.py +++ b/grafana_import/cli.py @@ -1,5 +1,3 @@ -#!/usr/bin/python3 -# -*- coding: utf-8 -*- """ Created on Mon March 15th 2021 @@ -9,312 +7,311 @@ V 0.0.0 - 2021/03/15 - JF. PIK - initial version """ -import logging -from pathlib import Path - -#*********************************************************************************************** -# -# -# TODO: -#*********************************************************************************************** - - -from grafana_import.constants import (PKG_NAME, PKG_VERSION, CONFIG_NAME) - -import argparse, json, sys, os, re, traceback +import argparse +import json +import logging +import os +import re +import sys +import traceback from datetime import datetime import grafana_client.client as GrafanaApi + import grafana_import.grafana as Grafana +from grafana_import.constants import CONFIG_NAME, PKG_NAME, PKG_VERSION from grafana_import.service import watchdog_service +from grafana_import.util import grafana_settings, load_yaml_config, read_dashboard_file, setup_logging -from grafana_import.util import load_yaml_config, grafana_settings, read_dashboard_file, setup_logging - -#****************************************************************************************** config = None -#****************************************************************************************** + def save_dashboard(config, args, base_path, dashboard_name, dashboard, action): - output_file = base_path - file_name = dashboard_name - - if 'exports_path' in config['general'] and \ - not re.search(r'^(\.|\/)?/', config['general']['exports_path']): - output_file = os.path.join(output_file, config['general']['exports_path'] ) - - if 'export_suffix' in config['general']: - file_name += datetime.today().strftime(config['general']['export_suffix']) - - if 'meta' in dashboard and 'folderId' in dashboard['meta'] and dashboard['meta']['folderId'] != 0: - file_name = dashboard['meta']['folderTitle'] + '_' + file_name - - file_name = Grafana.remove_accents_and_space( file_name ) - output_file = os.path.join(output_file, file_name + '.json') - try: - output = open(output_file, 'w') - except OSError as e: - print('File {0} error: {1}.'.format(output_file, e.strerror)) - sys.exit(2) - - content = None - if args.pretty: - content = json.dumps( dashboard['dashboard'], sort_keys=True, indent=2 ) - else: - content = json.dumps( dashboard['dashboard'] ) - output.write( content ) - output.close() - print(f"OK: Dashboard '{dashboard_name}' {action} to: {output_file}") - -#****************************************************************************************** -class myArgs: - attrs = [ 'pattern' - , 'base_path', 'config_file' - , 'grafana', 'dashboard_name' - , 'pretty', 'overwrite', 'allow_new', 'verbose' - ] - def __init__(self): + output_file = base_path + file_name = dashboard_name + + if "exports_path" in config["general"] and not re.search(r"^(\.|\/)?/", config["general"]["exports_path"]): + output_file = os.path.join(output_file, config["general"]["exports_path"]) + + if "export_suffix" in config["general"]: + file_name += datetime.today().strftime(config["general"]["export_suffix"]) + + if "meta" in dashboard and "folderId" in dashboard["meta"] and dashboard["meta"]["folderId"] != 0: + file_name = dashboard["meta"]["folderTitle"] + "_" + file_name - for attr in myArgs.attrs: - setattr(self, attr, None) + file_name = Grafana.remove_accents_and_space(file_name) + output_file = os.path.join(output_file, file_name + ".json") + try: + output = open(output_file, "w") + except OSError as e: + print("File {0} error: {1}.".format(output_file, e.strerror)) + sys.exit(2) - def __repr__(self): - obj = {} - for attr in myArgs.attrs: - val = getattr(self, attr) - if not val is None: - obj[attr] = val - return json.dumps(obj) + content = None + if args.pretty: + content = json.dumps(dashboard["dashboard"], sort_keys=True, indent=2) + else: + content = json.dumps(dashboard["dashboard"]) + output.write(content) + output.close() + print(f"OK: Dashboard '{dashboard_name}' {action} to: {output_file}") + + +class myArgs: + attrs = [ + "pattern", + "base_path", + "config_file", + "grafana", + "dashboard_name", + "pretty", + "overwrite", + "allow_new", + "verbose", + ] + + def __init__(self): + + for attr in myArgs.attrs: + setattr(self, attr, None) + + def __repr__(self): + obj = {} + for attr in myArgs.attrs: + val = getattr(self, attr) + if val is not None: + obj[attr] = val + return json.dumps(obj) logger = logging.getLogger(__name__) - -#*********************************************************************************************** + def main(): - setup_logging() - - #****************************************************************** - # get command line arguments - - parser = argparse.ArgumentParser(description='play with grafana dashboards json files.') - - parser.add_argument('-a', '--allow_new' - , action='store_true' - , default=False - , help='if a dashboard with same name exists in an another folder, allow to create a new dashboard with same name it that folder.') - - parser.add_argument('-b', '--base_path' - , help='set base directory to find default files.') - parser.add_argument('-c', '--config_file' - , help='path to config files.') - - parser.add_argument('-d', '--dashboard_name' - , help='name of dashboard to export.') - - parser.add_argument('-u', '--grafana_url' - , help='Grafana URL to connect to.' - , required=False) - - parser.add_argument('-g', '--grafana_label' - , help='label in the config file that represents the grafana to connect to.' - , default='default') - - parser.add_argument('-f', '--grafana_folder' - , help='the folder name where to import into Grafana.') - - parser.add_argument('-i', '--dashboard_file' - , help='path to the dashboard file to import into Grafana.') - - parser.add_argument('-o', '--overwrite' - , action='store_true' - , default=False - , help='if a dashboard with same name exists in same folder, overwrite it with this new one.') - - parser.add_argument('-r', '--reload' - , action='store_true' - , default=False - , help='Watch the input dashboard for changes on disk, and re-upload it, when changed.') - - parser.add_argument('-p', '--pretty' - , action='store_true' - , help='use JSON indentation when exporting or extraction of dashboards.') - - parser.add_argument('-v', '--verbose' - , action='store_true' - , help='verbose mode; display log message to stdout.') - - parser.add_argument('-V', '--version' - , action='version', version='{0} {1}'.format(PKG_NAME, PKG_VERSION) - , help='display program version and exit..') - - parser.add_argument('action', metavar='ACTION' - , nargs='?' - , choices=['import', 'export', 'remove'] - , default='import' - , help='action to perform. Is one of \'export\', \'import\' (default), or \'remove\'.\n' \ - 'export: lookup for dashboard name in Grafana and dump it to local file.\n' \ - 'import: import a local dashboard file (previously exported) to Grafana.\n' \ - 'remove: lookup for dashboard name in Grafana and remove it from Grafana server.' - ) - inArgs = myArgs() - args = parser.parse_args(namespace=inArgs) - - base_path = '.' -# base_path = os.path.dirname(os.path.abspath(__file__)) - if args.base_path is not None: - base_path = inArgs.base_path - - if args.config_file is None: - config = {"general": {"debug": False}} - else: - config_file = os.path.join(base_path, CONFIG_NAME) - if not re.search(r'^(\.|\/)?/', config_file): - config_file = os.path.join(base_path,args.config_file) - else: - config_file = args.config_file - config = load_yaml_config(config_file) - - if args.verbose is None: - if 'debug' in config['general']: - args.verbose = config['general']['debug'] - else: - args.verbose = False - - if args.allow_new is None: - args.allow_new = False - - if args.overwrite is None: - args.overwrite = False - - if args.pretty is None: - args.pretty = False - - #print( json.dumps(config, sort_keys=True, indent=4) ) - -#************ - if args.dashboard_name is not None: - config['general']['dashboard_name'] = args.dashboard_name - - if args.action == 'exporter' and ( not 'dashboard_name' in config['general'] or config['general']['dashboard_name'] is None) : - print("ERROR: no dashboard has been specified.") - sys.exit(1) - -#************ - config['check_folder'] = False - if args.grafana_folder is not None: - config['general']['grafana_folder'] = args.grafana_folder - config['check_folder'] = True - -#************ - if 'export_suffix' not in config['general'] or config['general']['export_suffix'] is None: - config['general']['export_suffix'] = "_%Y%m%d%H%M%S" - - params = grafana_settings(url=args.grafana_url, config=config, label=args.grafana_label) - params.update({ - 'overwrite': args.overwrite, - 'allow_new': args.allow_new, - }) - - try: - grafana_api = Grafana.Grafana( **params ) - except Exception as e: - print(f"ERROR: {e}") - sys.exit(1) - - #******************************************************************************* - if args.action == 'import': - if args.dashboard_file is None: - print('ERROR: no file to import provided!') - sys.exit(1) - - # Compute effective input file path. - import_path = '' - import_file = args.dashboard_file - if not re.search(r'^(?:(?:/)|(?:\.?\./))', import_file): - import_path = base_path - if 'imports_path' in config['general']: - import_path = os.path.join(import_path, config['general']['imports_path'] ) - import_file = os.path.join(import_path, import_file) - - def process_dashboard(): - try: - dash = read_dashboard_file(import_file) - except Exception as ex: - msg = f"Failed to load dashboard from: {import_file}. Reason: {ex}" - logger.exception(msg) - raise IOError(msg) - - try: - res = grafana_api.import_dashboard( dash ) - except GrafanaApi.GrafanaClientError as ex: - msg = f"Failed to upload dashboard to Grafana. Reason: {ex}" - logger.exception(msg) - raise IOError(msg) - - title = dash['title'] - folder_name = grafana_api.grafana_folder - if res: - logger.info(f"Dashboard '{title}' imported into folder '{folder_name}'") - else: - msg = f"Failed to import dashboard into Grafana. title={title}, folder={folder_name}" - logger.error(msg) - raise IOError(msg) - - try: - process_dashboard() - except: - sys.exit(1) - - if args.reload: - watchdog_service(import_file, process_dashboard) - - sys.exit(0) - - - #******************************************************************************* - elif args.action == 'remove': - dashboard_name = config['general']['dashboard_name'] - try: - res = grafana_api.remove_dashboard(dashboard_name) - print(f"OK: Dashboard removed: {dashboard_name}") - sys.exit(0) - except Grafana.GrafanaDashboardNotFoundError as exp: - print(f"KO: Dashboard not found in folder '{exp.folder}': {exp.dashboard}") - sys.exit(1) - except Grafana.GrafanaFolderNotFoundError as exp: - print(f"KO: Folder not found: {exp.folder}") - sys.exit(1) - except GrafanaApi.GrafanaBadInputError as exp: - print(f"KO: Removing dashboard failed: {dashboard_name}. Reason: {exp}") - sys.exit(1) - except Exception as exp: - print("ERROR: Dashboard '{0}' remove exception '{1}'".format(dashboard_name, traceback.format_exc())) - sys.exit(1) - - #******************************************************************************* - else: # export or - dashboard_name = config['general']['dashboard_name'] - try: - dash = grafana_api.export_dashboard(dashboard_name) - except (Grafana.GrafanaFolderNotFoundError, Grafana.GrafanaDashboardNotFoundError): - print("KO: Dashboard name not found: {0}".format(dashboard_name)) - sys.exit(1) - except Exception as exp: - print("ERROR: Dashboard '{0}' export exception '{1}'".format(dashboard_name, traceback.format_exc())) - sys.exit(1) - - if dash is not None: - save_dashboard(config, args, base_path, dashboard_name, dash, 'exported') - sys.exit(0) - -# end main... -#*********************************************************************************************** - -if __name__ == '__main__': - main() - -#*********************************************************************************************** -# over + setup_logging() + + # Get command line arguments. + parser = argparse.ArgumentParser(description="play with grafana dashboards json files.") + + parser.add_argument( + "-a", + "--allow_new", + action="store_true", + default=False, + help="If a dashboard with same name exists in an another folder, " + "allow to create a new dashboard with same name it that folder.", + ) + + parser.add_argument("-b", "--base_path", help="set base directory to find default files.") + parser.add_argument("-c", "--config_file", help="path to config files.") + + parser.add_argument("-d", "--dashboard_name", help="name of dashboard to export.") + + parser.add_argument("-u", "--grafana_url", help="Grafana URL to connect to.", required=False) + + parser.add_argument( + "-g", + "--grafana_label", + help="label in the config file that represents the grafana to connect to.", + default="default", + ) + + parser.add_argument("-f", "--grafana_folder", help="the folder name where to import into Grafana.") + + parser.add_argument("-i", "--dashboard_file", help="path to the dashboard file to import into Grafana.") + + parser.add_argument( + "-o", + "--overwrite", + action="store_true", + default=False, + help="if a dashboard with same name exists in same folder, overwrite it with this new one.", + ) + + parser.add_argument( + "-r", + "--reload", + action="store_true", + default=False, + help="Watch the input dashboard for changes on disk, and re-upload it, when changed.", + ) + + parser.add_argument( + "-p", "--pretty", action="store_true", help="use JSON indentation when exporting or extraction of dashboards." + ) + + parser.add_argument("-v", "--verbose", action="store_true", help="verbose mode; display log message to stdout.") + + parser.add_argument( + "-V", + "--version", + action="version", + version="{0} {1}".format(PKG_NAME, PKG_VERSION), + help="display program version and exit..", + ) + + parser.add_argument( + "action", + metavar="ACTION", + nargs="?", + choices=["import", "export", "remove"], + default="import", + help="action to perform. Is one of 'export', 'import' (default), or 'remove'.\n" + "export: lookup for dashboard name in Grafana and dump it to local file.\n" + "import: import a local dashboard file (previously exported) to Grafana.\n" + "remove: lookup for dashboard name in Grafana and remove it from Grafana server.", + ) + inArgs = myArgs() + args = parser.parse_args(namespace=inArgs) + + base_path = os.curdir + if args.base_path is not None: + base_path = inArgs.base_path + + if args.config_file is None: + config = {"general": {"debug": False}} + else: + config_file = os.path.join(base_path, CONFIG_NAME) + if not re.search(r"^(\.|\/)?/", config_file): + config_file = os.path.join(base_path, args.config_file) + else: + config_file = args.config_file + config = load_yaml_config(config_file) + + if args.verbose is None: + if "debug" in config["general"]: + args.verbose = config["general"]["debug"] + else: + args.verbose = False + + if args.allow_new is None: + args.allow_new = False + + if args.overwrite is None: + args.overwrite = False + + if args.pretty is None: + args.pretty = False + + if args.dashboard_name is not None: + config["general"]["dashboard_name"] = args.dashboard_name + + if args.action == "exporter" and ( + "dashboard_name" not in config["general"] or config["general"]["dashboard_name"] is None + ): + print("ERROR: no dashboard has been specified.") + sys.exit(1) + + config["check_folder"] = False + if args.grafana_folder is not None: + config["general"]["grafana_folder"] = args.grafana_folder + config["check_folder"] = True + + if "export_suffix" not in config["general"] or config["general"]["export_suffix"] is None: + config["general"]["export_suffix"] = "_%Y%m%d%H%M%S" + + params = grafana_settings(url=args.grafana_url, config=config, label=args.grafana_label) + params.update( + { + "overwrite": args.overwrite, + "allow_new": args.allow_new, + } + ) + + try: + grafana_api = Grafana.Grafana(**params) + except Exception as e: + print(f"ERROR: {e}") + sys.exit(1) + + # Import + if args.action == "import": + if args.dashboard_file is None: + print("ERROR: no file to import provided!") + sys.exit(1) + + # Compute effective input file path. + import_path = "" + import_file = args.dashboard_file + if not re.search(r"^(?:(?:/)|(?:\.?\./))", import_file): + import_path = base_path + if "imports_path" in config["general"]: + import_path = os.path.join(import_path, config["general"]["imports_path"]) + import_file = os.path.join(import_path, import_file) + + def process_dashboard(): + try: + dash = read_dashboard_file(import_file) + except Exception as ex: + msg = f"Failed to load dashboard from: {import_file}. Reason: {ex}" + logger.exception(msg) + raise IOError(msg) from ex + + try: + res = grafana_api.import_dashboard(dash) + except GrafanaApi.GrafanaClientError as ex: + msg = f"Failed to upload dashboard to Grafana. Reason: {ex}" + logger.exception(msg) + raise IOError(msg) from ex + + title = dash["title"] + folder_name = grafana_api.grafana_folder + if res: + logger.info(f"Dashboard '{title}' imported into folder '{folder_name}'") + else: + msg = f"Failed to import dashboard into Grafana. title={title}, folder={folder_name}" + logger.error(msg) + raise IOError(msg) + + try: + process_dashboard() + except Exception: + sys.exit(1) + + if args.reload: + watchdog_service(import_file, process_dashboard) + + sys.exit(0) + + # Remove + elif args.action == "remove": + dashboard_name = config["general"]["dashboard_name"] + try: + grafana_api.remove_dashboard(dashboard_name) + print(f"OK: Dashboard removed: {dashboard_name}") + sys.exit(0) + except Grafana.GrafanaDashboardNotFoundError as exp: + print(f"KO: Dashboard not found in folder '{exp.folder}': {exp.dashboard}") + sys.exit(1) + except Grafana.GrafanaFolderNotFoundError as exp: + print(f"KO: Folder not found: {exp.folder}") + sys.exit(1) + except GrafanaApi.GrafanaBadInputError as exp: + print(f"KO: Removing dashboard failed: {dashboard_name}. Reason: {exp}") + sys.exit(1) + except Exception: + print("ERROR: Dashboard '{0}' remove exception '{1}'".format(dashboard_name, traceback.format_exc())) + sys.exit(1) + + # Export + else: + dashboard_name = config["general"]["dashboard_name"] + try: + dash = grafana_api.export_dashboard(dashboard_name) + except (Grafana.GrafanaFolderNotFoundError, Grafana.GrafanaDashboardNotFoundError): + print("KO: Dashboard name not found: {0}".format(dashboard_name)) + sys.exit(1) + except Exception: + print("ERROR: Dashboard '{0}' export exception '{1}'".format(dashboard_name, traceback.format_exc())) + sys.exit(1) + + if dash is not None: + save_dashboard(config, args, base_path, dashboard_name, dash, "exported") + sys.exit(0) + + +if __name__ == "__main__": + main() diff --git a/grafana_import/constants.py b/grafana_import/constants.py index 19035b0..9616c9a 100644 --- a/grafana_import/constants.py +++ b/grafana_import/constants.py @@ -1,3 +1,3 @@ -PKG_NAME = 'grafana-import' -PKG_VERSION = '0.3.0dev1' -CONFIG_NAME = 'conf/grafana-import.yml' +PKG_NAME = "grafana-import" +PKG_VERSION = "0.3.0dev1" +CONFIG_NAME = "conf/grafana-import.yml" diff --git a/grafana_import/grafana.py b/grafana_import/grafana.py index 5574950..9208783 100644 --- a/grafana_import/grafana.py +++ b/grafana_import/grafana.py @@ -1,358 +1,328 @@ -#!/usr/bin/python3 -# -*- coding: utf-8 -*- - -#****************************************************************************************** - +import re +import traceback import typing as t +import unicodedata + import grafana_client.api as GrafanaApi import grafana_client.client as GrafanaClient -import re, traceback, unicodedata -from grafana_import.constants import (PKG_NAME) +from grafana_import.constants import PKG_NAME -#****************************************************************************************** -class GrafanaDashboardNotFoundError(Exception): - """ - input: - dashboard_name - folder - message - - """ - - def __init__(self, dashboard_name, folder, message): - self.dashboard = dashboard_name - self.folder = folder - self.message = message - # Backwards compatible with implementations that rely on just the message. - super(GrafanaDashboardNotFoundError, self).__init__(message) - -#****************************************************************************************** -class GrafanaFolderNotFoundError(Exception): - """ - input: - folder - message - - """ - - def __init__(self, folder, message): - self.folder = folder - self.message = message - # Backwards compatible with implementations that rely on just the message. - super(GrafanaFolderNotFoundError, self).__init__(message) - -#****************************************************************************************** -def remove_accents_and_space(input_str): - """ - build a valid file name from dashboard name. - - as mentioned in the function name remove .... - - input: a dashboard name - - :result: converted string - """ - nfkd_form = unicodedata.normalize('NFKD', input_str) - res = u"".join([c for c in nfkd_form if not unicodedata.combining(c)]) - res = re.sub(r'\s+', '_', res) - - return res - -#****************************************************************************************** -class Grafana(object): - #* to store the folders list, dashboards list (kind of cache) - folders: t.List[t.Any] = [] - dashboards: t.List[t.Any] = [] - - #*********************************************** - def __init__(self, **kwargs ): - - # Configure Grafana connectivity. - if "url" in kwargs: - self.grafana_api = GrafanaApi.GrafanaApi.from_url(kwargs["url"]) - else: - config = {} - config['protocol'] = kwargs.get('protocol', 'http') - config['host'] = kwargs.get('host', 'localhost') - config['port'] = kwargs.get('port', 3000) - config['token'] = kwargs.get('token', None) - if config['token'] is None: - raise GrafanaClient.GrafanaBadInputError('Grafana authentication token missing') - - config['verify_ssl'] = kwargs.get('verify_ssl', True) - - self.grafana_api = GrafanaApi.GrafanaApi( - auth=config['token'], - host=config['host'], - protocol=config['protocol'], - port=config['port'], - verify=config['verify_ssl'], - ) - - self.search_api_limit = kwargs.get('search_api_limit', 5000) - #* set the default destination folder for dash - self.grafana_folder = kwargs.get('folder', 'General') - - #* when importing dash, allow to overwrite dash if it already exists - self.overwrite = kwargs.get('overwrite', True) - - #* when importing dash, if dashboard name exists, but destination folder mismatch - #* allow to create new dashboard with same name in specified folder. - self.allow_new = kwargs.get('allow_new', False) - - #* try to connect to the API - try: - res = self.grafana_api.health.check() - if res['database'] != 'ok': - raise Exception('grafana is not UP') - except: - raise - - #*********************************************** - def find_dashboard(self, dashboard_name): - - #* use to retrive dashboards which name are matching the lookup named - #* some api version didn't return forlderTitle... require to lookup in two phases - found_dashs = [] - - #* init cache for dashboards. - if len(Grafana.dashboards) == 0: - #** collect all dashboard names. - try: - res = self.grafana_api.search.search_dashboards( - type_='dash-db', - limit=self.search_api_limit - ) - except Exception as e: - raise Exception("error: {}".format(traceback.format_exc()) ) - Grafana.dashboards = res - - dashboards = Grafana.dashboards - - folder = { - 'id': 0, - 'title': 'General', - } - if not re.match('general', self.grafana_folder, re.IGNORECASE): - found_folder = self.get_folder(folder_name=self.grafana_folder) - if found_folder is not None: - folder = found_folder - - board = None - #* find the board uid in the list - for cur_dash in dashboards: - if cur_dash['title'] == dashboard_name: - # set current dashbard as found candidate - board = cur_dash - # check the folder part - if ('folderTitle' in cur_dash and cur_dash['folderTitle'] == folder['title']) or \ - ('folderTitle' not in cur_dash and folder['id'] == 0 ): - # this is a requested folder or no folder ! - break - - return board - - #*********************************************** - def export_dashboard(self, dashboard_name): - """ - retrive the dashboard object from Grafana server. - params: - dashboard_name (str): name of the dashboard to retrieve - result: - dashboard (dict [json]) - """ - - try: - board = self.find_dashboard(dashboard_name) - - if board is None: - raise GrafanaClient.GrafanaClientError(response=None, message="Not Found", status_code=404) - - # Fetch the dashboard JSON representation by UID. - board = self.grafana_api.dashboard.get_dashboard(board['uid']) - return board - except Exception as ex: - if isinstance(ex, GrafanaClient.GrafanaClientError) and ex.status_code == 404: - raise GrafanaDashboardNotFoundError( - dashboard_name, - self.grafana_folder, - f"Dashboard not found: {dashboard_name}") - else: - raise - #*********************************************** - def remove_dashboard(self, dashboard_name): - """ - Retrieve the dashboard object from Grafana server and remove it. - params: - dashboard_name (str): name of the dashboard to retrieve - result: - True or Exception - """ - - res = False - folder = { - 'id' : 0, - 'title': self.grafana_folder, - } - #* check the destination folder is General - - if not re.match('general', self.grafana_folder, re.IGNORECASE): - #** check 'custom' folder existence (custom != General) - folder = self.get_folder( self.grafana_folder ) - if folder is None: - raise GrafanaFolderNotFoundError( - self.grafana_folder, - f"Folder not found: {self.grafana_folder}", - ) +class GrafanaDashboardNotFoundError(Exception): + """ + input: + dashboard_name + folder + message - #* collect the board object itself from it uid. - try: - board = self.find_dashboard(dashboard_name) - except Exception as e: - raise + """ - if board is None: - raise GrafanaDashboardNotFoundError(dashboard_name, folder['title'], 'dashboard not found') + def __init__(self, dashboard_name, folder, message): + self.dashboard = dashboard_name + self.folder = folder + self.message = message + # Backwards compatible with implementations that rely on just the message. + super(GrafanaDashboardNotFoundError, self).__init__(message) - if (folder['id'] == 0 and 'folderId' in board and board['folderId'] != folder['id'] ) \ - or (folder['id'] != 0 and not 'folderId' in board ): - raise GrafanaApi.GrafanaBadInputError("Dashboard name found but in folder '{0}'!".format(board['folderTitle'])) - if 'uid' in board: - try: - board = self.grafana_api.dashboard.delete_dashboard(board['uid']) - res = True - except Exception as exp: - raise - - return res - - #****************************************************************************************** - def get_folder(self, folder_name=None, folder_uid=None): - """ - try to find folder meta data (uid...) from folder name - params: - folder_name (str): name of the folder (case sensitive) into Grafana folders tree - return: - folder object (dict) - """ - if folder_name is None and folder_uid is None: - return None - - #* init cache for folders. - if len(Grafana.folders) == 0: - try: - res = self.grafana_api.folder.get_all_folders() - except Exception as e: - raise +class GrafanaFolderNotFoundError(Exception): + """ + input: + folder + message + + """ + + def __init__(self, folder, message): + self.folder = folder + self.message = message + # Backwards compatible with implementations that rely on just the message. + super(GrafanaFolderNotFoundError, self).__init__(message) + + +def remove_accents_and_space(input_str: str) -> str: + """ + build a valid file name from dashboard name. + + as mentioned in the function name remove .... + + input: a dashboard name + + :result: converted string + """ + nfkd_form = unicodedata.normalize("NFKD", input_str) + res = "".join([c for c in nfkd_form if not unicodedata.combining(c)]) + return re.sub(r"\s+", "_", res) + + +class Grafana: + # * to store the folders list, dashboards list (kind of cache) + folders: t.List[t.Any] = [] + dashboards: t.List[t.Any] = [] + + def __init__(self, **kwargs): + + # Configure Grafana connectivity. + if "url" in kwargs: + self.grafana_api = GrafanaApi.GrafanaApi.from_url(kwargs["url"]) + else: + config = {} + config["protocol"] = kwargs.get("protocol", "http") + config["host"] = kwargs.get("host", "localhost") + config["port"] = kwargs.get("port", 3000) + config["token"] = kwargs.get("token", None) + if config["token"] is None: + raise GrafanaClient.GrafanaBadInputError("Grafana authentication token missing") + + config["verify_ssl"] = kwargs.get("verify_ssl", True) + + self.grafana_api = GrafanaApi.GrafanaApi( + auth=config["token"], + host=config["host"], + protocol=config["protocol"], + port=config["port"], + verify=config["verify_ssl"], + ) - Grafana.folders = res + self.search_api_limit = kwargs.get("search_api_limit", 5000) + # * set the default destination folder for dash + self.grafana_folder = kwargs.get("folder", "General") - folders = Grafana.folders - folder = None + # * when importing dash, allow to overwrite dash if it already exists + self.overwrite = kwargs.get("overwrite", True) - for tmp_folder in folders: + # * when importing dash, if dashboard name exists, but destination folder mismatch + # * allow to create new dashboard with same name in specified folder. + self.allow_new = kwargs.get("allow_new", False) - if (folder_name is not None and folder_name == tmp_folder['title'] ) \ - or (folder_uid is not None and folder_uid == tmp_folder['folderId'] ): - folder = tmp_folder - break + # * try to connect to the API + try: + res = self.grafana_api.health.check() + if res["database"] != "ok": + raise Exception("grafana is not UP") + except: + raise - return folder + def find_dashboard(self, dashboard_name: str) -> t.Union[t.Dict[str, t.Any], None]: + """ + Retrieve dashboards which name are matching the lookup named. + Some api version didn't return folderTitle. Requires to lookup in two phases. + """ - #*********************************************** - def import_dashboard(self, dashboard): + # Init cache for dashboards. + if len(Grafana.dashboards) == 0: + # Collect all dashboard names. + try: + res = self.grafana_api.search.search_dashboards(type_="dash-db", limit=self.search_api_limit) + except Exception as ex: + raise Exception("error: {}".format(traceback.format_exc())) from ex + Grafana.dashboards = res + + dashboards = Grafana.dashboards + + folder = { + "id": 0, + "title": "General", + } + if not re.match("general", self.grafana_folder, re.IGNORECASE): + found_folder = self.get_folder(folder_name=self.grafana_folder) + if found_folder is not None: + folder = found_folder + + board = None + # * find the board uid in the list + for cur_dash in dashboards: + if cur_dash["title"] == dashboard_name: + # set current dashbard as found candidate + board = cur_dash + # check the folder part + if ("folderTitle" in cur_dash and cur_dash["folderTitle"] == folder["title"]) or ( + "folderTitle" not in cur_dash and folder["id"] == 0 + ): + # this is a requested folder or no folder ! + break + + return board + + def export_dashboard(self, dashboard_name: str) -> t.Dict[str, t.Any]: + """ + retrive the dashboard object from Grafana server. + params: + dashboard_name (str): name of the dashboard to retrieve + result: + dashboard (dict [json]) + """ + + try: + board = self.find_dashboard(dashboard_name) + + if board is None: + raise GrafanaClient.GrafanaClientError(response=None, message="Not Found", status_code=404) + + # Fetch the dashboard JSON representation by UID. + return self.grafana_api.dashboard.get_dashboard(board["uid"]) + except Exception as ex: + if isinstance(ex, GrafanaClient.GrafanaClientError) and ex.status_code == 404: + raise GrafanaDashboardNotFoundError( + dashboard_name, self.grafana_folder, f"Dashboard not found: {dashboard_name}" + ) from ex + raise - #** build a temporary meta dashboard struct to store info - #** by default dashboard will be overwritten - new_dash = { - 'dashboard': dashboard, - 'overwrite': True, - } + def remove_dashboard(self, dashboard_name: str) -> t.Dict[str, t.Any]: + """ + Retrieve the dashboard object from Grafana server and remove it. + params: + dashboard_name (str): name of the dashboard to retrieve + result: + True or Exception + """ + + res = {} + folder = { + "id": 0, + "title": self.grafana_folder, + } + + # Check if the destination folder is `General`. + if not re.match("general", self.grafana_folder, re.IGNORECASE): + # ** check 'custom' folder existence (custom != General) + folder = self.get_folder(self.grafana_folder) + if folder is None: + raise GrafanaFolderNotFoundError( + self.grafana_folder, + f"Folder not found: {self.grafana_folder}", + ) + + # Collect the board object itself from it uid. + board = self.find_dashboard(dashboard_name) + + if board is None: + raise GrafanaDashboardNotFoundError(dashboard_name, folder["title"], "dashboard not found") + + if (folder["id"] == 0 and "folderId" in board and board["folderId"] != folder["id"]) or ( + folder["id"] != 0 and "folderId" not in board + ): + raise GrafanaApi.GrafanaBadInputError( + "Dashboard name found but in folder '{0}'!".format(board["folderTitle"]) + ) - old_dash = self.find_dashboard(dashboard['title']) + if "uid" in board: + res = self.grafana_api.dashboard.delete_dashboard(board["uid"]) - #** check a previous dashboard existence (same folder, same uid) - if old_dash is None: - new_dash['overwrite'] = self.overwrite - dashboard['version'] = 1 + return res - #* check the destination folder is General - if re.match('general', self.grafana_folder, re.IGNORECASE): - new_dash['folderId'] = 0 - else: - #** check 'custom' folder existence (custom != General) - folder = self.get_folder( self.grafana_folder ) - if folder is None: - try: - folder = self.grafana_api.folder.create_folder( self.grafana_folder ) - except Exception as e: - raise + def get_folder(self, folder_name: str = None, folder_uid: str = None): + """ + try to find folder meta data (uid...) from folder name + params: + folder_name (str): name of the folder (case sensitive) into Grafana folders tree + return: + folder object (dict) + """ + if folder_name is None and folder_uid is None: + return None - if folder: - new_dash['folderId'] = folder['id'] - else: - raise Exception("KO: grafana folder '{0}' creation failed.".format(self.grafana_folder)) - else: - new_dash['folderId'] = folder['id'] - - #** several case - # read new folder1/dash1(uid1) => old folder1/dash1(uid1): classic update - # b) read new folder_new/dash1(uid1) => old folder1/dash1(uid1): create new dash in folder_new - # => new folder_new/dash1(uid_new) if allow_new - # c) read new folder_new/dash1(uid_new) => old folder1/dash1(uid1): create new in new folder folder_new - # => classic create (update) - # d) read new folder1/dash1(uid_new) => old folder1/dash1(uid1) - # => new folder1/dash1(uid1) if overwrite - if old_dash is not None: - if 'meta' in old_dash and 'folderUrl' in old_dash['meta']: - old_dash['folderId'] = old_dash['meta']['folderId'] - elif not 'folderId' in old_dash: - old_dash['folderId'] = 0 - - # case b) get a copy of an existing dash to a folder where dash is not present - if new_dash['folderId'] != old_dash['folderId']: - # if new_dash['dashboard']['uid'] == old_dash['uid']: - if self.allow_new: - new_dash['overwrite'] = False - #force the creation of a new dashboard - new_dash['dashboard']['uid'] = None - new_dash['dashboard']['id'] = None + # * init cache for folders. + if len(Grafana.folders) == 0: + res = self.grafana_api.folder.get_all_folders() + Grafana.folders = res + + folders = Grafana.folders + folder = None + + for tmp_folder in folders: + + if (folder_name is not None and folder_name == tmp_folder["title"]) or ( + folder_uid is not None and folder_uid == tmp_folder["folderId"] + ): + folder = tmp_folder + break + + return folder + + def import_dashboard(self, dashboard: t.Dict[str, t.Any]) -> bool: + + # ** build a temporary meta dashboard struct to store info + # ** by default dashboard will be overwritten + new_dash: t.Dict[str, t.Any] = { + "dashboard": dashboard, + "overwrite": True, + } + + old_dash = self.find_dashboard(dashboard["title"]) + + # ** check a previous dashboard existence (same folder, same uid) + if old_dash is None: + new_dash["overwrite"] = self.overwrite + dashboard["version"] = 1 + + # * check the destination folder is General + if re.match("general", self.grafana_folder, re.IGNORECASE): + new_dash["folderId"] = 0 + else: + # ** check 'custom' folder existence (custom != General) + folder = self.get_folder(self.grafana_folder) + if folder is None: + folder = self.grafana_api.folder.create_folder(self.grafana_folder) + + if folder: + new_dash["folderId"] = folder["id"] + else: + raise Exception("KO: grafana folder '{0}' creation failed.".format(self.grafana_folder)) else: - raise GrafanaClient.GrafanaBadInputError( - "Dashboard with the same title already exists in another folder. " - "Use `allow_new` to permit creation in a different folder.") - #** case d) send a copy to existing dash : update existing - elif new_dash['folderId'] == old_dash['folderId']: - if 'uid' not in new_dash['dashboard'] or new_dash['dashboard']['uid'] != old_dash['uid']: - if self.overwrite: - new_dash['dashboard']['uid'] = old_dash['uid'] - new_dash['dashboard']['id'] = old_dash['id'] - else: - raise GrafanaClient.GrafanaBadInputError( - "Dashboard with the same title already exists in this folder with another uid. " - "Use `overwrite` to permit overwriting it.") - else: - #force the creation of a new dashboard - new_dash['dashboard']['uid'] = None - new_dash['dashboard']['id'] = None - new_dash['overwrite'] = False - - new_dash['message'] = 'imported from {0}.'.format(PKG_NAME) - - try: - res = self.grafana_api.dashboard.update_dashboard(new_dash) - except Exception as e: - raise - - if res['status']: - res = True - else: - res = False - - return res - -#****************************************************************************************** -# over \ No newline at end of file + new_dash["folderId"] = folder["id"] + + # ** several case + # read new folder1/dash1(uid1) => old folder1/dash1(uid1): classic update + # b) read new folder_new/dash1(uid1) => old folder1/dash1(uid1): create new dash in folder_new + # => new folder_new/dash1(uid_new) if allow_new + # c) read new folder_new/dash1(uid_new) => old folder1/dash1(uid1): create new in new folder folder_new + # => classic create (update) + # d) read new folder1/dash1(uid_new) => old folder1/dash1(uid1) + # => new folder1/dash1(uid1) if overwrite + if old_dash is not None: + if "meta" in old_dash and "folderUrl" in old_dash["meta"]: + old_dash["folderId"] = old_dash["meta"]["folderId"] + elif "folderId" not in old_dash: + old_dash["folderId"] = 0 + + # case b) get a copy of an existing dash to a folder where dash is not present + if new_dash["folderId"] != old_dash["folderId"]: + # if new_dash['dashboard']['uid'] == old_dash['uid']: + if self.allow_new: + new_dash["overwrite"] = False + # force the creation of a new dashboard + new_dash["dashboard"]["uid"] = None + new_dash["dashboard"]["id"] = None + else: + raise GrafanaClient.GrafanaBadInputError( + "Dashboard with the same title already exists in another folder. " + "Use `allow_new` to permit creation in a different folder." + ) + # ** case d) send a copy to existing dash : update existing + elif new_dash["folderId"] == old_dash["folderId"]: + if "uid" not in new_dash["dashboard"] or new_dash["dashboard"]["uid"] != old_dash["uid"]: + if self.overwrite: + new_dash["dashboard"]["uid"] = old_dash["uid"] + new_dash["dashboard"]["id"] = old_dash["id"] + else: + raise GrafanaClient.GrafanaBadInputError( + "Dashboard with the same title already exists in this folder with another uid. " + "Use `overwrite` to permit overwriting it." + ) + else: + # force the creation of a new dashboard + new_dash["dashboard"]["uid"] = None + new_dash["dashboard"]["id"] = None + new_dash["overwrite"] = False + + new_dash["message"] = "imported from {0}.".format(PKG_NAME) + + res = self.grafana_api.dashboard.update_dashboard(new_dash) + if res["status"]: + res = True + else: + res = False + + return res diff --git a/grafana_import/service.py b/grafana_import/service.py index ff6208c..73f0690 100644 --- a/grafana_import/service.py +++ b/grafana_import/service.py @@ -30,7 +30,7 @@ def on_modified(self, event: FileSystemEvent) -> None: logger.exception(f"Processing file failed: {event.src_path}") -def watchdog_service(path: Path, action: t.Union[t.Callable, None] = None): +def watchdog_service(path: Path, action: t.Union[t.Callable, None] = None) -> None: """ https://python-watchdog.readthedocs.io/en/stable/quickstart.html """ diff --git a/grafana_import/util.py b/grafana_import/util.py index 3fe75fc..b8c7890 100644 --- a/grafana_import/util.py +++ b/grafana_import/util.py @@ -40,9 +40,8 @@ def load_yaml_config(config_file: str) -> ConfigType: def grafana_settings( - url: t.Union[str, None], - config: t.Union[ConfigType, None], - label: t.Union[str, None]) -> SettingsType: + url: t.Union[str, None], config: t.Union[ConfigType, None], label: t.Union[str, None] +) -> SettingsType: """ Acquire Grafana connection profile settings, and application settings. """ @@ -51,17 +50,19 @@ def grafana_settings( # Grafana connectivity. if url or "GRAFANA_URL" in os.environ: - params = {"url": url or os.environ["GRAFANA_URL"]} - if "GRAFANA_TOKEN" in os.environ: - params.update({"token": os.environ["GRAFANA_TOKEN"]}) + params = {"url": url or os.environ["GRAFANA_URL"]} + if "GRAFANA_TOKEN" in os.environ: + params.update({"token": os.environ["GRAFANA_TOKEN"]}) elif config is not None: - params = grafana_settings_from_config_section(config=config, label=label) - - # Additional application parameters. - params.update({ - "search_api_limit": config.get("grafana", {}).get("search_api_limit", 5000), - "folder": config.get("general", {}).get("grafana_folder", "General"), - }) + params = grafana_settings_from_config_section(config=config, label=label) + + # Additional application parameters. + params.update( + { + "search_api_limit": config.get("grafana", {}).get("search_api_limit", 5000), + "folder": config.get("general", {}).get("grafana_folder", "General"), + } + ) return params @@ -85,7 +86,7 @@ def grafana_settings_from_config_section(config: ConfigType, label: t.Union[str, if "token" not in config["grafana"]: raise ValueError(f"Authentication token missing in Grafana configuration at: {label}") - params = { + return { "host": config["grafana"].get("host", "localhost"), "protocol": config["grafana"].get("protocol", "http"), "port": config["grafana"].get("port", "3000"), @@ -93,10 +94,8 @@ def grafana_settings_from_config_section(config: ConfigType, label: t.Union[str, "verify_ssl": config["grafana"].get("verify_ssl", True), } - return params - -def file_is_executable(path: t.Union[str, Path]): +def file_is_executable(path: t.Union[str, Path]) -> bool: """ Is this file executable? @@ -114,7 +113,7 @@ def read_dashboard_file(path: t.Union[str, Path]) -> t.Dict[str, t.Any]: if path.suffix == ".json": try: - with open(path, 'r') as f: + with open(path, "r") as f: payload = f.read() except OSError as ex: raise IOError(f"Reading file failed: {path}. Reason: {ex.strerror}") from ex diff --git a/pyproject.toml b/pyproject.toml index fe095ca..07ec61b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,38 +1,6 @@ [tool.black] line-length = 120 -[tool.coverage.run] -branch = false -omit = [ - "tests/*", -] -source = ["grafana_import"] - -[tool.coverage.report] -fail_under = 0 -show_missing = true - -[tool.mypy] -packages = ["grafana_import"] -install_types = true -ignore_missing_imports = true -implicit_optional = true -non_interactive = true - -[tool.pytest.ini_options] -addopts = "-rA --verbosity=3 --cov --cov-report=term-missing --cov-report=xml" -minversion = "2.0" -log_level = "DEBUG" -log_cli_level = "DEBUG" -log_format = "%(asctime)-15s [%(name)-24s] %(levelname)-8s: %(message)s" -testpaths = [ - "grafana_import", - "tests", -] -xfail_strict = true -markers = [ -] - [tool.ruff] line-length = 120 @@ -65,26 +33,55 @@ lint.select = [ ] lint.extend-ignore = [ - # Unnecessary variable assignment before `return` statement - "RET504", - # Unnecessary `elif` after `return` statement - "RET505", -] - -# Intermediary ignores, until the code base has been improved further. -lint.exclude = [ - "grafana_import/cli.py", - "grafana_import/grafana.py", ] [tool.ruff.lint.per-file-ignores] -"tests/*" = ["S101"] # Use of `assert` detected +"tests/*" = [ + # Use of `assert` detected + "S101", +] +"grafana_import/cli.py" = [ + # `print` found + "T201", +] # =================== # Tasks configuration # =================== +[tool.pytest.ini_options] +addopts = "-rA --verbosity=3 --cov --cov-report=term-missing --cov-report=xml" +minversion = "2.0" +log_level = "DEBUG" +log_cli_level = "DEBUG" +log_format = "%(asctime)-15s [%(name)-24s] %(levelname)-8s: %(message)s" +testpaths = [ + "grafana_import", + "tests", +] +xfail_strict = true +markers = [ +] + +[tool.coverage.run] +branch = false +omit = [ + "tests/*", +] +source = ["grafana_import"] + +[tool.coverage.report] +fail_under = 0 +show_missing = true + +[tool.mypy] +packages = ["grafana_import"] +install_types = true +ignore_missing_imports = true +implicit_optional = true +non_interactive = true + [tool.poe.tasks] check = [ @@ -100,8 +97,8 @@ format = [ ] lint = [ - { shell = "ruff check . || true" }, - # { cmd = "black --check ." }, + { cmd = "ruff check ." }, + { cmd = "black --check ." }, { cmd = "validate-pyproject pyproject.toml" }, { cmd = "mypy grafana_import" }, ] diff --git a/setup.py b/setup.py index e70fd7b..bea570e 100644 --- a/setup.py +++ b/setup.py @@ -6,10 +6,10 @@ # Global variables requires = [ - 'grafana-client<5', - 'jinja2<4', - 'pyyaml<7', - 'watchdog<5', + "grafana-client<5", + "jinja2<4", + "pyyaml<7", + "watchdog<5", ] extras = { @@ -43,22 +43,18 @@ setup( name=PKG_NAME, version=PKG_VERSION, - description='Export and import Grafana dashboards using the Grafana HTTP API.', - long_description_content_type='text/markdown', + description="Export and import Grafana dashboards using the Grafana HTTP API.", + long_description_content_type="text/markdown", long_description=README, license="Apache 2.0", author="Jean-Francois Pik", author_email="jfpik78@gmail.com", url="https://github.com/grafana-toolbox/grafana-import", - entry_points={ - 'console_scripts': [ - 'grafana-import = grafana_import.cli:main' - ] - }, + entry_points={"console_scripts": ["grafana-import = grafana_import.cli:main"]}, packages=find_packages(), install_requires=requires, extras_require=extras, - package_data={'': ['conf/*']}, + package_data={"": ["conf/*"]}, classifiers=[ "Programming Language :: Python", "License :: OSI Approved :: Apache Software License", @@ -95,5 +91,5 @@ "Topic :: System :: Networking :: Monitoring", ], keywords="grafana http api grafana-client grafana-api http-client " - "grafana-utils grafana-automation grafana-toolbox dashboard", + "grafana-utils grafana-automation grafana-toolbox dashboard grafana-dashboard grafanalib grafonnet", ) diff --git a/tests/conftest.py b/tests/conftest.py index caab7df..91ab999 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -85,6 +85,6 @@ def gio_factory(settings) -> t.Callable: def mkgrafana(use_settings: bool = True) -> Grafana: if use_settings: return Grafana(**settings) - else: - return Grafana(url="http://localhost:3000") + return Grafana(url="http://localhost:3000") + return mkgrafana diff --git a/tests/test_cli.py b/tests/test_cli.py index 06ffe36..ba9bd07 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -16,8 +16,7 @@ def get_settings_arg(use_settings: bool = True): if use_settings: return f"--config_file {CONFIG_FILE}" - else: - return "--grafana_url http://localhost:3000" + return "--grafana_url http://localhost:3000" @pytest.mark.parametrize("use_settings", [True, False], ids=["config-yes", "config-no"]) diff --git a/tests/test_core.py b/tests/test_core.py index 6b1972d..49f3ad6 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -77,7 +77,7 @@ def test_remove_dashboard_success(mocked_grafana, mocked_responses, settings): gio = Grafana(**settings) outcome = gio.remove_dashboard("foobar") - assert outcome is True + assert outcome == {"status": "ok"} def test_remove_dashboard_folder_not_found(mocked_responses, settings): diff --git a/tests/util.py b/tests/util.py index 0e80ed2..3166b12 100644 --- a/tests/util.py +++ b/tests/util.py @@ -10,7 +10,7 @@ from mypy.typeshed.stdlib._typeshed import FileDescriptorOrPath, OpenTextMode -def mock_grafana_health(responses: RequestsMock): +def mock_grafana_health(responses: RequestsMock) -> None: """ Baseline mock for each Grafana conversation. """ @@ -22,7 +22,7 @@ def mock_grafana_health(responses: RequestsMock): ) -def mock_grafana_search(responses: RequestsMock): +def mock_grafana_search(responses: RequestsMock) -> None: responses.get( "http://localhost:3000/api/search?type=dash-db&limit=5000", json=[{"title": "foobar", "uid": "618f7589-7e3d-4399-a585-372df9fa5e85"}], @@ -31,31 +31,31 @@ def mock_grafana_search(responses: RequestsMock): ) -def mkdashboard(): +def mkdashboard() -> t.Dict[str, t.Any]: """ Example Grafana dashboard, generated using the `grafana-dashboard` package. https://github.com/fzyzcjy/grafana_dashboard_python/blob/master/examples/python_to_json/input_python/dashboard-one.py """ pytest.importorskip( - "grafana_dashboard", - reason="Skipping dashboard generation because `grafana-dashboard` is not available") + "grafana_dashboard", reason="Skipping dashboard generation because `grafana-dashboard` is not available" + ) from grafana_dashboard.manual_models import TimeSeries from grafana_dashboard.model.dashboard_types_gen import Dashboard, GridPos from grafana_dashboard.model.prometheusdataquery_types_gen import PrometheusDataQuery dashboard = Dashboard( - title='Dashboard One', + title="Dashboard One", panels=[ TimeSeries( - title='Panel Title', + title="Panel Title", gridPos=GridPos(x=0, y=0, w=12, h=9), targets=[ PrometheusDataQuery( - datasource='Prometheus', + datasource="Prometheus", expr='avg(1 - rate(node_cpu_seconds_total{mode="idle"}[$__rate_interval])) by (instance, job)', - legendFormat='{{instance}}' + legendFormat="{{instance}}", ) ], ) @@ -68,14 +68,12 @@ def mkdashboard(): real_open = builtins.open -def open_write_noop(file: "FileDescriptorOrPath", mode: "OpenTextMode" = "r", **kwargs): +def open_write_noop(file: "FileDescriptorOrPath", mode: "OpenTextMode" = "r", **kwargs) -> t.IO: """ A replacement for `builtins.open`, masking all write operations. """ if mode and mode.startswith("w"): if "b" in mode: return io.BytesIO() - else: - return io.StringIO() - else: - return real_open(file=file, mode=mode, **kwargs) + return io.StringIO() + return real_open(file=file, mode=mode, **kwargs)