-
-
Save susodapop/3868a973108ad1032393a1da03632846 to your computer and use it in GitHub Desktop.
Edited to allow dashboard clone that forks the underlying queries
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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