Programmatic query workflow

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

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..." }

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.

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"

Polling 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.

Example: fetch transactions for multiple wallets

A common pattern is fetching data for a list of addresses. The naive approach fires one query per address — each run costs an execution unit and a cold-start overhead. Pass all addresses in a single parameterized query instead and split the flat results in code.

Step 1 — Create the query

Use {{param_name}} syntax to define parameters. The {{addresses}} parameter will receive a SQL-safe, comma-separated list of quoted address strings.

SELECT
    block_timestamp,
    hash,
    from_address,
    to_address,
    value / 1e18       AS value_eth,
    gas_used,
    gas_price / 1e9    AS gas_price_gwei,
    status
FROM ethereum.raw.transactions
WHERE (
    from_address IN ({{addresses}})
    OR to_address  IN ({{addresses}})
)
AND block_timestamp >= CURRENT_TIMESTAMP - INTERVAL '{{lookback_days}} days'
ORDER BY block_timestamp DESC

Save this in the Allium App or via Create query and copy the returned query_id.

Step 2 — Run once for all wallets

Format your address list as SQL string literals before substituting into the parameter, then trigger a single run.

import time
import requests

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

WALLETS = [
    "0xd8da6bf26964af9d7eed9e03e53415d37aa96045",
    "0xab5801a7d398351b8be11c439e05c5b3259aec9b",
    "0x1db3439a222c519ab44bb1144fc28167b4fa6ee6",
]

# Format as SQL string literals: 'addr1', 'addr2', ...
addresses_sql = ", ".join(f"'{w.lower()}'" for w in WALLETS)

# One run for all wallets — one execution unit
run = requests.post(
    f"{BASE}/queries/{QUERY_ID}/run-async",
    headers=HEADERS,
    json={
        "parameters": {
            "addresses":     addresses_sql,
            "lookback_days": "30",
        }
    },
)
run_id = run.json()["run_id"]

while True:
    status = requests.get(f"{BASE}/query-runs/{run_id}/status", headers=HEADERS).json()
    if status in ("success", "failed", "canceled"):
        break
    time.sleep(3)

if status != "success":
    raise RuntimeError(f"Run {run_id} ended with status: {status}")

transactions = requests.get(f"{BASE}/query-runs/{run_id}/results", headers=HEADERS).json()["data"]

Step 3 — Split by wallet in code

The query returns a flat list. Group it by wallet on the application side.

from collections import defaultdict

wallet_set = {w.lower() for w in WALLETS}
by_wallet  = defaultdict(list)

for tx in transactions:
    sender   = tx["from_address"].lower()
    receiver = tx["to_address"].lower()

    # A tx between two tracked wallets appears in both — each wallet gets its full view
    if sender in wallet_set:
        by_wallet[sender].append({**tx, "role": "sender"})
    if receiver in wallet_set:
        by_wallet[receiver].append({**tx, "role": "receiver"})

for wallet, txs in by_wallet.items():
    print(f"{wallet}: {len(txs)} transactions")

If a tracked wallet sends to another tracked wallet, that transaction appears in both lists. Deduplicate on hash before grouping if you need a unique-transaction count.

Why this matters

Batching into a single run eliminates per-query startup cost. The result set is identical — the split is just done in code instead of across API calls.

Realtime API

Stream Transformation API

Beam API

Explorer API

Account API

Deep Dives

The workflow

Polling example

Example: fetch transactions for multiple wallets

Step 1 — Create the query

Step 2 — Run once for all wallets

Step 3 — Split by wallet in code

Why this matters

​The workflow

​Polling example

​Example: fetch transactions for multiple wallets

​Step 1 — Create the query

​Step 2 — Run once for all wallets

​Step 3 — Split by wallet in code

​Why this matters

The workflow

Polling example

Example: fetch transactions for multiple wallets

Step 1 — Create the query

Step 2 — Run once for all wallets

Step 3 — Split by wallet in code

Why this matters