Skip to content

Instantly share code, notes, and snippets.

@susodapop
Forked from arikfr/redash.py
Last active February 26, 2020 22:23
Show Gist options
  • Save susodapop/3868a973108ad1032393a1da03632846 to your computer and use it in GitHub Desktop.
Save susodapop/3868a973108ad1032393a1da03632846 to your computer and use it in GitHub Desktop.
Edited to allow dashboard clone that forks the underlying queries
from redash import Redash
# Enter your user API key, Redash URL, and the URL slug for your target dashboard
API_KEY = ''
REDASH_URL = 'https://app.redash.io/<your organization slug>'
DASHBOARD_SLUG = ''
redash = Redash(REDASH_URL, API_KEY)
dash = redash.fork_dashboard_and_queries(DASHBOARD_SLUG)
import requests
import os
class Redash(object):
def __init__(self, redash_url, api_key):
self.redash_url = redash_url
self.session = requests.Session()
self.session.headers.update({'Authorization': 'Key {}'.format(api_key)})
def test_credentials(self):
try:
response = self._get('api/session')
return True
except requests.exceptions.HTTPError:
return False
def queries(self, page=1, page_size=25):
"""GET api/queries"""
return self._get('api/queries', params=dict(page=page, page_size=page_size)).json()
def dashboards(self, page=1, page_size=25):
"""GET api/dashboards"""
return self._get('api/dashboards', params=dict(page=page, page_size=page_size)).json()
def dashboard(self, slug):
"""GET api/dashboards/{slug}"""
return self._get('api/dashboards/{}'.format(slug)).json()
def create_dashboard(self, name):
return self._post('api/dashboards', json={'name': name}).json()
def update_dashboard(self, dashboard_id, properties):
return self._post('api/dashboards/{}'.format(dashboard_id), json=properties).json()
def create_widget(self, dashboard_id, visualization_id, text, options):
data = {
'dashboard_id': dashboard_id,
'visualization_id': visualization_id,
'text': text,
'options': options,
'width': 1,
}
return self._post('api/widgets', json=data)
def duplicate_dashboard(self, slug, new_name=None):
current_dashboard = self.dashboard(slug)
if new_name is None:
new_name = u'Copy of: {}'.format(current_dashboard['name'])
new_dashboard = self.create_dashboard(new_name)
if current_dashboard['tags']:
self.update_dashboard(new_dashboard['id'], {'tags': current_dashboard['tags']})
for widget in current_dashboard['widgets']:
visualization_id = None
if 'visualization' in widget:
visualization_id = widget['visualization']['id']
self.create_widget(new_dashboard['id'], visualization_id, widget['text'], widget['options'])
return new_dashboard
def scheduled_queries(self):
"""Loads all queries and returns only the scheduled ones."""
queries = self.paginate(self.queries)
return filter(lambda query: query['schedule'] is not None, queries)
def update_query(self, query_id, data):
"""POST /api/queries/{query_id} with the provided data object."""
path = 'api/queries/{}'.format(query_id)
return self._post(path, json=data)
def paginate(self, resource):
"""Load all items of a paginated resource"""
stop_loading = False
page = 1
page_size = 100
items = []
while not stop_loading:
response = resource(page=page, page_size=page_size)
items += response['results']
page += 1
stop_loading = response['page'] * response['page_size'] >= response['count']
return items
def fork_query(self, query_id):
"""Call the fork API. Returns reference to a new query object.
"""
path = 'api/queries/{}/fork'.format(query_id)
return self._post(path)
def map_orig_visualization_ids(self, old_query_id, new_query_id):
"""Compares visualization details between two queries that have been forked.
Returns an association table of old_viz_id:new_viz_id key-value-pairs
"""
association_table = {}
orig = self._get('api/queries/{}'.format(old_query_id)).json()
dup = self._get('api/queries/{}'.format(new_query_id)).json()
for viz in orig['visualizations']:
this_viz_id = viz['id']
for cloned_viz in dup['visualizations']:
if viz['name'] == cloned_viz['name'] and viz['options'] == cloned_viz['options'] and viz['type'] == cloned_viz['type']:
association_table[this_viz_id] = cloned_viz['id']
if this_viz_id not in association_table.keys():
print('Viz id {} could not be found in cloned query!'.format(this_viz_id))
return association_table
def fork_dashboard_and_queries(self, slug, new_name=None):
current_dashboard = self.dashboard(slug)
queries_to_duplicate = set([widget['visualization']['query']['id'] for widget in current_dashboard['widgets'] if 'visualization' in widget])
viz_id_assoc_table = {}
for orig in queries_to_duplicate:
dup = self.fork_query(orig).json()['id']
assoc = self.map_orig_visualization_ids(orig, dup)
viz_id_assoc_table.update(assoc)
if new_name is None:
new_name = u'Copy of: {}'.format(current_dashboard['name'])
new_dashboard = self.create_dashboard(new_name)
if current_dashboard['tags']:
self.update_dashboard(new_dashboard['id'], {'tags': current_dashboard['tags']})
for widget in current_dashboard['widgets']:
visualization_id = None
if 'visualization' in widget:
visualization_id = viz_id_assoc_table[widget['visualization']['id']]
self.create_widget(new_dashboard['id'], visualization_id, widget['text'], widget['options'])
return new_dashboard
def _get(self, path, **kwargs):
return self._request('GET', path, **kwargs)
def _post(self, path, **kwargs):
return self._request('POST', path, **kwargs)
def _request(self, method, path, **kwargs):
url = '{}/{}'.format(self.redash_url, path)
response = self.session.request(method, url, **kwargs)
response.raise_for_status()
return response
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment