Python SDK

oneanalytics is the official Python SDK, generated from the OpenAPI spec. Pydantic models for every resource, both sync and async clients, type hints throughout.

Install

pip install oneanalytics

Python 3.10+ required.

Sync client

from oneanalytics import OneAnalytics

oa = OneAnalytics(
    base_url="https://api.analytics.rstglobal.in",
    token=os.environ["OA_TOKEN"],
)

# list workspaces
for ws in oa.workspaces.list_all():        # auto-paginates
    print(ws.name, ws.id)

# create one
finance = oa.workspaces.create(
    name="Finance",
    description="GL, AP, AR, cash",
)

# query a dataset
result = oa.datasets.query(
    dataset_id,
    dimensions=[{"id": "order_date", "grain": "month"}],
    measures=["revenue"],
    filters=[{"field": "region", "op": "in", "value": ["North", "South"]}],
    limit=1000,
)

import pandas as pd
df = pd.DataFrame(result.rows, columns=[c.name for c in result.columns])

Async client

from oneanalytics import AsyncOneAnalytics

async def main():
    async with AsyncOneAnalytics(token=os.environ["OA_TOKEN"]) as oa:
        result = await oa.datasets.query(dataset_id, ...)

All sync methods have an async counterpart on AsyncOneAnalytics with the same signature.

Errors

from oneanalytics.errors import OAError, QueryTimeout, RateLimited

try:
    oa.datasets.query(id, ...)
except QueryTimeout as e:
    print(f"Query timed out; trace_id={e.trace_id}")
except RateLimited as e:
    print(f"Retry after {e.retry_after}s")
except OAError as e:
    print(f"{e.code}: {e.detail}")

Every code in API errors has a corresponding subclass — or use OAError as a catch-all.

Idempotency

oa.sharing.create(
    subject_type="report",
    subject_id=rpt_id,
    principal_type="link",
    actions=["view"],
    idempotency_key="export-2026-04-16-rpt-123",
)

Omitted → we auto-generate a UUID.

Retries

Configured per-client. Defaults: 5 attempts, exponential backoff with jitter, max 30 s total wall time. 5xx + 429 are retried; 4xx are raised immediately.

oa = OneAnalytics(token=..., retry_max_attempts=3, retry_max_seconds=10)

DataFrame helper

oa.datasets.query_df(id, ...) returns a pandas.DataFrame directly, typed per the QueryResult.columns. Requires pip install oneanalytics[pandas].

Arrow stream

oa.datasets.query_arrow(id, ...) returns a pyarrow.RecordBatchReader. 20× faster than JSON for large queries. Requires pip install oneanalytics[arrow].

Logging

Structured logs via structlog. Set OA_DEBUG=1 to dump every request/response (headers, latency, body size — no body content) for debugging.