Simple example of creating a query and fetching its results with app.redash.io

redash_create_and_run_query.py

import json, time
from redash_toolbelt import Redash

URL = "https://app.redash.io/<org slug>"
KEY = "<your api key>"
DS_ID = 1234  # enter an integer data source ID here

client = Redash(URL, KEY)

# new queries need query text, a name, and a target data source id
# parameters are optional but included for this example
# for a non-parameterized query set options = {}

options = {
    "parameters": [{"title": "param", "name": "param", "type": "number", "value": 1}]
}

query_data = {
    "query": "SELECT 1 foo where {{ param }} = {{ param }}",
    "name": "Some parameterized query Name",
    "data_source_id": DS_ID,
    "options": options,
}


# posting a new query returns a fresh query object
query = client._post("api/queries", json=query_data).json()

# the refresh endpoint accepts max_age and parameters objects
# it returns a new job object
refresh_payload = dict(max_age=0, parameters={"param": 2})
response = client._post(f"api/queries/{query.get('id')}/results", json=refresh_payload)
if response.status_code != 200:
    raise Exception("Refresh failed.")
job = response.json()["job"]

# jobs are asynchronous. ping the api for a status once per second until
# code 3 (success) or code 4 (failure) is received.
# code 3 indicates a query result is now available from the API
while job["status"] not in (3, 4):
    response = client._get(f"api/jobs/{job.get('id')}")
    job = response.json()["job"]
    time.sleep(1)

if job["status"] != 3:
    raise Exception("Refresh failed.")

# fetch the query result using its ID from the job object.
query_result_id = job["query_result_id"]
result = client._get(
    f"api/queries/{query.get('id')}/results/{query_result_id}.json"
).json()["query_result"]["data"]["rows"]

# print to screen (or write to a CSV). really whatever you need.
print(result)