Redash Polling Example

poll.py

import os
import requests
import time
from pprint import pprint

REDASH_HOST = os.environ.get('REDASH_HOST', 'https://demo.redash.io')

def poll_job(s, job):
    # TODO: add timeout
    while job['status'] not in (3,4):
        response = s.get('{}/api/jobs/{}'.format(REDASH_HOST, job['id']))
        job = response.json()['job']
        time.sleep(1)

    return job['status'] == 3

def get_fresh_query_result(query_id, api_key):
    s  = requests.Session()
    s.headers.update({'Authorization': 'Key {}'.format(api_key)})
    response = s.post('{}/api/queries/{}/refresh'.format(REDASH_HOST, query_id))
    if response.status_code != 200:
        raise Exception('Refresh failed.')

    # TODO: make sure it's true.
    poll_job(s, response.json()['job'])

    response = s.get('{}/api/queries/{}/results.json'.format(REDASH_HOST, query_id))
    if response.status_code != 200:
        raise Exception('Failed getting results.')

    return response.json()['query_result']['data']['rows']

if __name__ == '__main__':
    # Need to use a *user API key* here (and not a query API key).
    pprint(get_fresh_query_result(2305, 'TCX2SwMQwDzdY4OESdbcTNAyQrCykoPrD5wWWQTB'))