Skip to main content
When you consume Explorer query results programmatically, run the query yourself and track the run_id it returns. This gives you a run you triggered and can poll for, so you always fetch results from a run whose status you know is success.
Scheduled query runs are best suited for viewing data in the Allium App. Get latest run returns the most recent run regardless of status, so a failed scheduled refresh can shadow the last good dataset. Triggering your own run avoids this — you hold the run_id and decide when the data is valid.

​
The workflow

1

Run the query

Call Run query with your query_id. The response returns a run_id for the run you just triggered.
curl -X POST "https://api.allium.so/api/v1/explorer/queries/{query_id}/run-async" \
  -H "X-API-KEY: your-api-key-here" \
  -H "Content-Type: application/json" \
  -d '{}'

{ "run_id": "abc123..." }
2

Poll for status

Poll Fetch query run status with the run_id until the status is terminal. Treat success as ready, and failed or canceled as a retry signal.
curl "https://api.allium.so/api/v1/explorer/query-runs/{run_id}/status" \
  -H "X-API-KEY: your-api-key-here"
Statuses created, queued, and running mean the run is still in progress — keep polling.
3

Fetch results

Once the status is success, call Fetch query run results with the same run_id to retrieve the dataset for the run you triggered.
curl "https://api.allium.so/api/v1/explorer/query-runs/{run_id}/results" \
  -H "X-API-KEY: your-api-key-here"

​
Example

import time
import requests

API_KEY = "your-api-key-here"
QUERY_ID = "your-query-id"
BASE = "https://api.allium.so/api/v1/explorer"
HEADERS = {"X-API-KEY": API_KEY}

# 1. Run the query and capture the run_id you triggered
run = requests.post(f"{BASE}/queries/{QUERY_ID}/run-async", headers=HEADERS, json={})
run_id = run.json()["run_id"]

# 2. Poll status until the run reaches a terminal state
while True:
    status = requests.get(f"{BASE}/query-runs/{run_id}/status", headers=HEADERS).json()
    if status in ("success", "failed", "canceled"):
        break
    time.sleep(5)

# 3. Fetch results only when the run you triggered succeeded
if status == "success":
    results = requests.get(f"{BASE}/query-runs/{run_id}/results", headers=HEADERS).json()
    print(results["data"])
else:
    raise RuntimeError(f"Query run {run_id} ended with status: {status}")
Because you hold the run_id for the entire loop, you never depend on the latest run being the one you want, and you don’t need a separate store to track run statuses or preserve the last successful run.