De-duplicate Redash dashboard widgets

README.md

De-Duplicate Redash Dashboard Widgets

Overview

When updating Redash dashboards, it's possible to add widgets rather than replacing them. That leads to a number of duplicate widgets that have the same visualization type and query details. This is a script built on top of redash-toolbelt that removes duplicate widgets, leaving only the most recently updated widget per set.

Usage

1. Install redash-toolbelt.
2. (Optional) Set the REDASH_URL and REDASH_API_KEY environment variables.
3. Run the de-duplication script from your Python environment with redash-toolbelt installed: python dedup_redash_dashboard_widgets.py --help.

Sample Script Invocations

Run in the default dry run mode, taking the URL and API key from environment variables:

python dedup_redash_dashboard_widgets.py

Let the script remove widgets:

python dedup_redash_dashboard_widgets.py --no-dry-run

Specify the URL and API key at runtime:

python dedup_redash_dashboard_widgets.py \
  --url https://redash.example.com \
  --api-key 'asdflkj123098'

dedup_redash_dashboard_widgets.py

import logging

import click
from itertools import groupby
from operator import itemgetter
from time import sleep

from redash_toolbelt import Redash


def widget_key(w):
    """Key function to sort and group widgets

    If two widgets in the same dashboard use the same
    visualization type and query details, we'll
    call them duplicates.
    """
    return (
        w["visualization"]["type"],
        w["visualization"]["query"]["name"],
        w["visualization"]["query"]["query_hash"],
    )


@click.pass_context
def dashboard_details(ctx):
    dashboards = ctx.obj["client"].dashboards()["results"]

    # Need get_dashboard to pull in widget details
    return [ctx.obj["client"].get_dashboard(d["id"]) for d in dashboards]


@click.pass_context
def delete_widget(ctx, widget):
    if ctx.obj["dry_run"]:
        print(f'DRY RUN: Would delete widget ID {widget["id"]}...')
        return
    print(f'Deleting widget ID {widget["id"]}...')
    try:
        ctx.obj["client"]._delete(f'/api/widgets/{widget["id"]}')
    except Exception as exc:
        logging.exception(exc)


def dedup_widget_group(group):
    if not len(group) > 1:
        print("No duplicate widgets")
        return

    # Keep the most recently updated widget per group, delete
    # the rest
    keep, *delete = sorted(group, key=itemgetter("updated_at"), reverse=True)
    print(f'Keep: {keep["id"]}, {keep["visualization"]["query"]["name"]}')
    print(f'Remove: {[(w["id"], w["visualization"]["query"]["name"]) for w in delete]}')
    for w in delete:
        delete_widget(w)
        sleep(0.1)
    print("Pausing after widget group")
    sleep(1)


@click.command()
@click.option(
    "--url",
    required=True,
    envvar="REDASH_URL",
    help="Redash root URL (or from env: REDASH_URL)",
)
@click.option(
    "--api-key",
    required=True,
    envvar="REDASH_API_KEY",
    help="Redash API key (or from env: REDASH_API_KEY)",
)
@click.option(
    "--dry-run/--no-dry-run",
    default=True,
    help="Check for duplicate widgets but don't delete them [default: dry-run]",
)
@click.pass_context
def cli(ctx, url, api_key, dry_run):
    """Remove duplicate widgets from Redash dashboards"""

    # ensure that ctx.obj exists and is a dict (in case `cli()` is called
    # by means other than the `if` block below)
    ctx.ensure_object(dict)
    ctx.obj["client"] = Redash(url, api_key)
    ctx.obj["dry_run"] = dry_run

    for d in dashboard_details():
        print(f'Checking dashboard: {d["name"]}')
        groups = {
            k: list(v)
            for k, v in groupby(sorted(d["widgets"], key=widget_key), key=widget_key)
        }
        for k, g in groups.items():
            print(f"Deduping widget group: {k}")
            dedup_widget_group(g)


if __name__ == "__main__":
    cli(obj={})