Chore: Format code using Ruff

Makefile

@@ -14,8 +14,7 @@ $(eval pytest       := $(venvpath)/bin/pytest)
 $(eval bumpversion  := $(venvpath)/bin/bumpversion)
 $(eval twine        := $(venvpath)/bin/twine)
 $(eval sphinx       := $(venvpath)/bin/sphinx-build)
-$(eval black        := $(venvpath)/bin/black)
-$(eval isort        := $(venvpath)/bin/isort)
+$(eval ruff         := $(venvpath)/bin/ruff)
 
 
 # Setup Python virtualenv
@@ -54,8 +53,8 @@ test-coverage: install-tests
 # Formatting
 # ----------
 format: install-releasetools
-	$(isort) grafana_wtf test
-	$(black) .
+	$(ruff) format
+	$(ruff) check --fix --ignore=ERA --ignore=F401 --ignore=F841 --ignore=T20 --ignore=ERA001 .
 
 
 # -------

grafana_wtf/commands.py

@@ -180,7 +180,7 @@ def run():
       export CACHE_TTL=infinite
       grafana-wtf find geohash
 
-    """
+    """  # noqa: E501
 
     # Parse command line arguments
     options = normalize_options(docopt(run.__doc__, version=f"{__appname__} {__version__}"))
@@ -225,7 +225,8 @@ def run():
     # Sanity checks
     if grafana_url is None:
         raise DocoptExit(
-            'No Grafana URL given. Please use "--grafana-url" option or environment variable "GRAFANA_URL".'
+            'No Grafana URL given. Please use "--grafana-url" option '
+            'or environment variable "GRAFANA_URL".'
         )
 
     log.info(f"Grafana location: {grafana_url}")
@@ -273,7 +274,8 @@ def run():
         # Sanity checks.
         if output_format.startswith("tab") and options.sql:
             raise DocoptExit(
-                f"Options --format={output_format} and --sql can not be used together, only data output is supported."
+                f"Options --format={output_format} and --sql can not be used together, "
+                f"only data output is supported."
             )
 
         entries = engine.log(dashboard_uid=options.dashboard_uid)
@@ -319,7 +321,9 @@ def run():
         output_results(output_format, results)
 
     if options.explore and options.dashboards:
-        results = engine.explore_dashboards(with_data_details=options.data_details, queries_only=options.queries_only)
+        results = engine.explore_dashboards(
+            with_data_details=options.data_details, queries_only=options.queries_only
+        )
         output_results(output_format, results)
 
     if options.explore and options.permissions:

grafana_wtf/core.py

@@ -36,9 +36,9 @@
 
 
 class GrafanaEngine:
-
     # Configure a larger HTTP request pool.
-    # TODO: Review the pool settings and eventually adjust according to concurrency level or other parameters.
+    # TODO: Review the pool settings and eventually adjust according
+    #       to concurrency level or other parameters.
     # https://urllib3.readthedocs.io/en/latest/advanced-usage.html#customizing-pool-behavior
     # https://laike9m.com/blog/requests-secret-pool_connections-and-pool_maxsize,89/
     session_args = dict(pool_connections=100, pool_maxsize=100, retries=5)
@@ -52,7 +52,9 @@
 
         self.concurrency = 5
 
-        self.grafana = self.grafana_client_factory(self.grafana_url, grafana_token=self.grafana_token)
+        self.grafana = self.grafana_client_factory(
+            self.grafana_url, grafana_token=self.grafana_token
+        )
         self.set_user_agent()
         self.data = GrafanaDataModel()
         self.finder = JsonPathFinder()
@@ -66,14 +68,18 @@
 
     def enable_cache(self, expire_after=60, drop_cache=False):
         if expire_after is None:
-            log.info(f"Response cache will never expire (infinite caching)")
+            log.info("Response cache will never expire (infinite caching)")
         elif expire_after == 0:
-            log.info(f"Response cache will expire immediately (expire_after=0)")
+            log.info("Response cache will expire immediately (expire_after=0)")
         else:
             log.info(f"Response cache will expire after {expire_after} seconds")
 
         session = CachedSession(
-            cache_name=__appname__, expire_after=expire_after, use_cache_dir=True, wal=True, **self.session_args
+            cache_name=__appname__,
+            expire_after=expire_after,
+            use_cache_dir=True,
+            wal=True,
+            **self.session_args,
         )
         self.set_session(session)
         self.set_user_agent()
@@ -86,7 +92,7 @@
         return self
 
     def clear_cache(self):
-        log.info(f"Clearing cache")
+        log.info("Clearing cache")
         requests_cache.clear()
 
     def enable_concurrency(self, concurrency):
@@ -171,7 +177,11 @@
         if Version(self.grafana.version) < Version("11"):
             self.data.notifications = self.grafana.notifications.lookup_channels()
         else:
-            warnings.warn("Notification channel scanning support for Grafana 11 is not implemented yet", UserWarning)
+            warnings.warn(
+                "Notification channel scanning support for Grafana 11 is not implemented yet",
+                UserWarning,
+                stacklevel=2,
+            )
 
     def scan_datasources(self):
         log.info("Scanning datasources")
@@ -185,7 +195,8 @@
             if isinstance(ex, GrafanaUnauthorizedError):
                 log.error(
                     self.get_red_message(
-                        "Please use --grafana-token or GRAFANA_TOKEN " "for authenticating with Grafana"
+                        "Please use --grafana-token or GRAFANA_TOKEN "
+                        "for authenticating with Grafana"
                     )
                 )
 
@@ -207,7 +218,7 @@
 
         except GrafanaClientError as ex:
             self.handle_grafana_error(ex)
-            return
+            return None
 
         if self.progressbar:
             self.start_progressbar(len(self.data.dashboard_list))
@@ -221,7 +232,9 @@
             self.taqadum.close()
 
         # Improve determinism by returning stable sort order.
-        self.data.dashboards = munchify(sorted(self.data.dashboards, key=lambda x: x["dashboard"]["uid"]))
+        self.data.dashboards = munchify(
+            sorted(self.data.dashboards, key=lambda x: x["dashboard"]["uid"])
+        )
 
         return self.data.dashboards
 
@@ -231,7 +244,9 @@
         log.error(self.get_red_message(message))
         if isinstance(ex, GrafanaUnauthorizedError):
             log.error(
-                self.get_red_message("Please use --grafana-token or GRAFANA_TOKEN " "for authenticating with Grafana")
+                self.get_red_message(
+                    "Please use --grafana-token or GRAFANA_TOKEN for authenticating with Grafana"
+                )
             )
 
     def fetch_dashboard(self, dashboard_info):
@@ -270,10 +285,13 @@
             # for response in await asyncio.gather(*tasks):
             #    pass
 
+    @staticmethod
+    def get_red_message(message):
+        return colored.stylize(message, colored.fg("red") + colored.attr("bold"))
+
 
 class GrafanaWtf(GrafanaEngine):
     def info(self):
-
         response = OrderedDict(
             grafana=OrderedDict(
                 version=self.version,
@@ -308,7 +326,9 @@
 
         # Count numbers of panels, annotations and variables for all dashboards.
         try:
-            dashboard_summary = OrderedDict(dashboard_panels=0, dashboard_annotations=0, dashboard_templating=0)
+            dashboard_summary = OrderedDict(
+                dashboard_panels=0, dashboard_annotations=0, dashboard_templating=0
+            )
             for dbdetails in self.dashboard_details():
                 # TODO: Should there any deduplication be applied when counting those entities?
                 dashboard_summary["dashboard_panels"] += len(dbdetails.panels)
@@ -324,7 +344,9 @@
     def build_info(self):
         response = None
         error = None
-        error_template = f"The request to {self.grafana_url.rstrip('/')}/api/frontend/settings failed"
+        error_template = (
+            f"The request to {self.grafana_url.rstrip('/')}/api/frontend/settings failed"
+        )
         try:
             response = self.grafana.client.GET("/frontend/settings")
             if not isinstance(response, dict):
@@ -353,7 +375,9 @@
             yield DashboardDetails(dashboard=dashboard)
 
     def search(self, expression):
-        log.info('Searching Grafana at "{}" for expression "{}"'.format(self.grafana_url, expression))
+        log.info(
+            'Searching Grafana at "{}" for expression "{}"'.format(self.grafana_url, expression)
+        )
 
         results = Munch(datasources=[], dashboard_list=[], dashboards=[])
 
@@ -370,7 +394,9 @@
     def replace(self, expression, replacement, dry_run: bool = False):
         if dry_run:
             log.info("Dry-run mode enabled, skipping any actions")
-        log.info(f'Replacing "{expression}" by "{replacement}" within Grafana at "{self.grafana_url}"')
+        log.info(
+            f'Replacing "{expression}" by "{replacement}" within Grafana at "{self.grafana_url}"'
+        )
         for dashboard in self.data.dashboards:
             payload_before = json.dumps(dashboard)
             payload_after = payload_before.replace(expression, replacement)
@@ -433,29 +459,27 @@
             if effective_item:
                 results.append(effective_item)
 
-    @staticmethod
-    def get_red_message(message):
-        return colored.stylize(message, colored.fg("red") + colored.attr("bold"))
-
     def get_dashboard_versions(self, dashboard_id):
         # https://grafana.com/docs/http_api/dashboard_versions/
         get_dashboard_versions_path = "/dashboards/id/%s/versions" % dashboard_id
-        r = self.grafana.dashboard.client.GET(get_dashboard_versions_path)
-        return r
+        return self.grafana.dashboard.client.GET(get_dashboard_versions_path)
 
     def explore_datasources(self):
         # Prepare indexes, mapping dashboards by uid, datasources by name
         # as well as dashboards to datasources and vice versa.
         ix = Indexer(engine=self)
 
-        # Compute list of exploration items, associating datasources with the dashboards that use them.
+        # Compute list of exploration items, associating
+        # datasources with the dashboards that use them.
         results_used = []
         results_unused = []
         for datasource in ix.datasources:
             ds_identifier = datasource.get("uid", datasource.get("name"))
             dashboard_uids = ix.datasource_dashboard_index.get(ds_identifier, [])
             dashboards = list(map(ix.dashboard_by_uid.get, dashboard_uids))
-            item = DatasourceExplorationItem(datasource=datasource, used_in=dashboards, grafana_url=self.grafana_url)
+            item = DatasourceExplorationItem(
+                datasource=datasource, used_in=dashboards, grafana_url=self.grafana_url
+            )
 
             # Format results in a more compact form, using only a subset of all the attributes.
             result = item.format_compact()
@@ -466,16 +490,18 @@
                 if result not in results_unused:
                     results_unused.append(result)
 
-        results_used = sorted(results_used, key=lambda x: x["datasource"]["name"] or x["datasource"]["uid"])
-        results_unused = sorted(results_unused, key=lambda x: x["datasource"]["name"] or x["datasource"]["uid"])
+        results_used = sorted(
+            results_used, key=lambda x: x["datasource"]["name"] or x["datasource"]["uid"]
+        )
+        results_unused = sorted(
+            results_unused, key=lambda x: x["datasource"]["name"] or x["datasource"]["uid"]
+        )
 
-        response = OrderedDict(
+        return OrderedDict(
             used=results_used,
             unused=results_unused,
         )
 
-        return response
-
     def explore_dashboards(self, with_data_details: bool = False, queries_only: bool = False):
         # Prepare indexes, mapping dashboards by uid, datasources by name
         # as well as dashboards to datasources and vice versa.
@@ -484,7 +510,8 @@
         # Those dashboard names or uids will be ignored.
         ignore_dashboards = ["-- Grafana --", "-- Mixed --", "grafana", "-- Dashboard --"]
 
-        # Compute list of exploration items, looking for dashboards with missing data sources.
+        # Compute list of exploration items, looking
+        # for dashboards with missing data sources.
         results = []
         for uid in sorted(ix.dashboard_by_uid):
             dashboard = ix.dashboard_by_uid[uid]
@@ -597,13 +624,17 @@
         for dashboard in dashboards:
             for panel in dashboard["dashboard"].get("panels", []):
                 if "alert" in panel and panel["alert"]["notifications"]:
-                    related_panels += self.extract_channel_related_information(channel_uid, dashboard, panel)
+                    related_panels += self.extract_channel_related_information(
+                        channel_uid, dashboard, panel
+                    )
 
                 # Some dashboards have a deeper nested structure
                 elif "panels" in panel:
                     for subpanel in panel["panels"]:
                         if "alert" in subpanel and subpanel["alert"]["notifications"]:
-                            related_panels += self.extract_channel_related_information(channel_uid, dashboard, subpanel)
+                            related_panels += self.extract_channel_related_information(
+                                channel_uid, dashboard, subpanel
+                            )
         if related_panels:
             channel["related_panels"] = related_panels
         return channel
@@ -613,7 +644,9 @@
         related_information = []
         for notification in panel["alert"]["notifications"]:
             if "uid" in notification and notification["uid"] == channel_uid:
-                related_information.append({"dashboard": dashboard["dashboard"]["title"], "panel": panel["title"]})
+                related_information.append(
+                    {"dashboard": dashboard["dashboard"]["title"], "panel": panel["title"]}
+                )
         return related_information
 
 

grafana_wtf/model.py

@@ -120,7 +120,9 @@
 
     def queries_only(self):
         """
-        Return a representation of data details information, only where query expressions are present.
+        Return a representation of data details information.
+
+        Only where query expressions are present.
         """
         # All attributes containing query-likes.
         attributes_query_likes = ["expr", "jql", "query", "rawSql", "target"]
@@ -145,7 +147,11 @@
                             continue
                     # Unnest items with nested "query" slot.
                     for slot in ["query", "target"]:
-                        if slot in new_item and isinstance(new_item[slot], dict) and "query" in new_item[slot]:
+                        if (
+                            slot in new_item
+                            and isinstance(new_item[slot], dict)
+                            and "query" in new_item[slot]
+                        ):
                             new_item["query"] = new_item[slot]["query"]
                     new_items.append(new_item)
             return new_items
@@ -196,6 +202,7 @@
 {data}
 """.strip(),
                 UserWarning,
+                stacklevel=2,
             )
             del data["datasource"]
 

grafana_wtf/report/data.py

@@ -28,7 +28,7 @@ def serialize_results(output_format: str, results: List):
 
 
 class DataSearchReport(TabularSearchReport):
-    def __init__(self, grafana_url, verbose=False, format=None):
+    def __init__(self, grafana_url, verbose=False, format=None):  # noqa: A002
         self.grafana_url = grafana_url
         self.verbose = verbose
         self.format = format
@@ -42,7 +42,11 @@ def display(self, expression, result):
                 grafana=self.grafana_url,
                 expression=expression,
             ),
-            datasources=self.get_output_items("Datasource", result.datasources, self.compute_url_datasource),
-            dashboards=self.get_output_items("Dashboard", result.dashboards, self.compute_url_dashboard),
+            datasources=self.get_output_items(
+                "Datasource", result.datasources, self.compute_url_datasource
+            ),
+            dashboards=self.get_output_items(
+                "Dashboard", result.dashboards, self.compute_url_dashboard
+            ),
         )
         output_results(self.format, output)