Skip to content

Instantly share code, notes, and snippets.

@mars2cbh

mars2cbh/refresh.py

Forked from arikfr/refresh.py
Created February 1, 2023 11:24
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save mars2cbh/6fbb21c90bccabf5f78085b68bbd45f5 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save mars2cbh/6fbb21c90bccabf5f78085b68bbd45f5 to your computer and use it in GitHub Desktop.
Download ZIP

Revisions

  1. @arikfr arikfr revised this gist Mar 24, 2021. 1 changed file with 6 additions and 3 deletions.
    9 changes: 6 additions & 3 deletions refresh.py
    View file
    Original file line number Diff line number Diff line change
    @@ -16,11 +16,14 @@ def poll_job(s, redash_url, job):

    return None

    def get_fresh_query_result(redash_url, query_id, api_key, params):
    def get_fresh_query_result(redash_url, query_id, api_key, parameters):
    s = requests.Session()
    s.headers.update({'Authorization': 'Key {}'.format(api_key)})

    response = s.post('{}/api/queries/{}/refresh'.format(redash_url, query_id), params=params)
    body = {
    "parameters": parameters
    }
    response = s.post('{}/api/queries/{}/results'.format(redash_url, query_id), json=body)

    if response.status_code != 200:
    raise Exception('Refresh failed.')
    @@ -37,7 +40,7 @@ def get_fresh_query_result(redash_url, query_id, api_key, params):
    return response.json()['query_result']['data']['rows']

    if __name__ == '__main__':
    params = {'p_param': 1243}
    params = {'param': 1243}
    query_id = 1234
    # Need to use a *user API key* here (and not a query API key).
    api_key = '...'
  2. @arikfr arikfr created this gist Apr 24, 2017.
    44 changes: 44 additions & 0 deletions refresh.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,44 @@
    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))