Redash Refresh API usage example with parameters Raw
import os | |
import requests | |
import time | |
from pprint import pprint | |
def poll_job(s, redash_url, job): | |
# TODO: add timeout | |
while job['status'] not in (3,4): | |
response = s.get('{}/api/jobs/{}'.format(redash_url, job['id'])) | |
job = response.json()['job'] | |
time.sleep(1) | |
if job['status'] == 3: | |
return job['query_result_id'] | |
return None | |
def get_fresh_query_result(redash_url, query_id, api_key, params): | |
s = requests.Session() | |
s.headers.update({'Authorization': 'Key {}'.format(api_key)}) | |
response = s.post('{}/api/queries/{}/refresh'.format(redash_url, query_id), params=params) | |
if response.status_code != 200: | |
raise Exception('Refresh failed.') | |
result_id = poll_job(s, redash_url, response.json()['job']) | |
if result_id: | |
response = s.get('{}/api/queries/{}/results/{}.json'.format(redash_url, query_id, result_id)) | |
if response.status_code != 200: | |
raise Exception('Failed getting results.') | |
else: | |
raise Exception('Query execution failed.') | |
return response.json()['query_result']['data']['rows'] | |
if __name__ == '__main__': | |
params = {'p_param': 1243} | |
query_id = 1234 | |
# Need to use a *user API key* here (and not a query API key). | |
api_key = '...' | |
pprint(get_fresh_query_result('https://app.redash.io/acme', query_id, api_key, params)) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment