Skip to main content

Python REST API Quickstart

Query Nexatron from Python using the REST API directly with httpx or requests.

Authentication​

All API requests require an API key passed in the Authorization header:

Authorization: Bearer nxa_your_api_key_here

Generate an API key from Settings > API Keys in the Nexatron dashboard.

import httpx

BASE_URL = "https://api.nexatron.io"
API_KEY = "nxa_your_api_key_here"

client = httpx.Client(
    base_url=BASE_URL,
    headers={"Authorization": f"Bearer {API_KEY}"},
    timeout=60.0,
)

Execute a Query​

response = client.post("/api/v1/query", json={
    "question": "What were total sales last quarter?",
})
response.raise_for_status()

result = response.json()

print(result["generated_sql"])
# SELECT SUM(amount) AS total_sales
# FROM orders
# WHERE order_date >= '2026-01-01' AND order_date < '2026-04-01'

print(result["result_data"])
# [{"total_sales": 1284350.00}]

print(result["content"])
# "Total sales last quarter were $1,284,350.00."

Query with Conversation Context​

Pass a conversation_id to maintain context across follow-up questions:

# First question
r1 = client.post("/api/v1/query", json={
    "question": "Show revenue by region",
})
conversation_id = r1.json()["conversation_id"]

# Follow-up question (references previous results)
r2 = client.post("/api/v1/query", json={
    "question": "Which region grew the fastest?",
    "conversation_id": conversation_id,
})
print(r2.json()["content"])

Stream Results​

Use server-sent events for real-time streaming:

import json

with client.stream(
    "POST",
    "/api/v1/query/stream",
    json={"question": "Monthly revenue for 2025"},
) as response:
    # The server emits `id:`, `event:` and `data:` lines per SSE frame. The event
    # NAME is in the `event:` field — it is not duplicated inside the JSON — so
    # track it as the lines stream past, then act when its `data:` arrives.
    event_type = None
    for line in response.iter_lines():
        if not line:
            continue
        if line.startswith("event: "):
            event_type = line[7:].strip()
            continue
        if not line.startswith("data: "):
            continue

        payload = json.loads(line[6:])

        if event_type == "stage":
            print(f"  [{payload['stage']}] {payload['message']}")
        elif event_type == "sql":
            print(f"  SQL: {payload['sql']}")
        elif event_type == "data":
            print(f"  Rows: {len(payload['rows'])}")
        elif event_type == "content":
            print(f"  Summary: {payload['text']}")
        elif event_type == "done":
            print(f"  Conversation: {payload['conversation_id']}")
        elif event_type == "error":
            print(f"  Error: {payload['message']}")

Using requests​

import requests

BASE_URL = "https://api.nexatron.io"
headers = {"Authorization": "Bearer nxa_your_api_key_here"}

# Execute a query
response = requests.post(
    f"{BASE_URL}/api/v1/query",
    headers=headers,
    json={"question": "Top 10 customers by lifetime value"},
    timeout=60,
)
response.raise_for_status()

result = response.json()
for row in result["result_data"]["rows"]:
    print(f"  {row['customer_name']}: ${row['lifetime_value']:,.2f}")

Export Results​

Export as CSV / Excel / PDF​

Run the question first, then export the answer message by ID:

result = client.post("/api/v1/query", json={
    "question": "All orders from last month",
}).json()

export = client.get(
    f"/api/v1/query/messages/{result['message_id']}/export",
    params={"format": "csv"},  # csv, xlsx, or pdf
)

with open("orders.csv", "wb") as f:
    f.write(export.content)

Exporting re-runs the answer's query rather than downloading the rows shown on screen, so the file is the whole result and not the first page of it. It runs under your own access (the same row policies and column masking you would get by asking the question again) and is bounded by your workspace's query row limit. A PDF renders at most 5,000 rows.

Every file carries a commented governance header, and its Rows: line tells you how many rows the file contains and whether that is the complete result. Read that line rather than assuming: when your workspace's row limit is what stopped the read, the header says so (source row cap reached; true total not counted) instead of reporting the bound as the total. If you asked for a specific number ("top 10 regions"), the export honours it and returns ten rows.

If the answer could not be re-run (for example it combined several sources, or its warehouse is disconnected), the header says the rows are the stored snapshot saved with the answer and why they could not be refreshed.

Exporting a saved answer (GET /api/v1/reports/{id}/export) is different: it serves the rows stored with that answer rather than re-reading the source. Its header says so too, and it states the 5,000-row PDF limit on the occasions that limit is what shortened the document.

Export as JSON​

response = client.post("/api/v1/query", json={
    "question": "All orders from last month",
})

import json
with open("orders.json", "w") as f:
    json.dump(response.json()["data"], f, indent=2)

Load into pandas​

import pandas as pd

response = client.post("/api/v1/query", json={
    "question": "Daily active users for the past 90 days",
})
result = response.json()

df = pd.DataFrame(result["result_data"]["rows"])
print(df.describe())

Async Usage​

import httpx
import asyncio

async def main():
    async with httpx.AsyncClient(
        base_url="https://api.nexatron.io",
        headers={"Authorization": "Bearer nxa_your_api_key_here"},
        timeout=60.0,
    ) as client:
        response = await client.post("/api/v1/query", json={
            "question": "Revenue by product category",
        })
        result = response.json()
        print(result["content"])

asyncio.run(main())

Error Handling​

try:
    response = client.post("/api/v1/query", json={
        "question": "Show me the data",
    })
    response.raise_for_status()
except httpx.HTTPStatusError as e:
    if e.response.status_code == 401:
        print("Invalid or expired API key")
    elif e.response.status_code == 422:
        error = e.response.json()
        print(f"Validation error: {error['detail']}")
    elif e.response.status_code == 429:
        # Retry-After is sent only when the server knows the wait; otherwise back off and retry.
        print("Rate limited. Retry after:", e.response.headers.get("Retry-After") or "not sent")
    else:
        print(f"API error {e.response.status_code}: {e.response.text}")
except httpx.TimeoutException:
    print("Request timed out. Try a simpler query or increase the timeout.")

Next Steps​

Last updated on