Python SDK

The official Python SDK for OnlyMetrix. Sync and async clients, full API coverage.

Install

pip install onlymetrix

Quick start

from onlymetrix import OnlyMetrix

om = OnlyMetrix("https://api.onlymetrix.com", api_key="omx_sk_...")

# List metrics
for m in om.metrics.list():
    print(f"{m.name}: {m.description}")

# Query a metric
result = om.metrics.query("total_revenue")
print(result.rows)  # [{"revenue_gbp": 17743429.16}]

# With filters and dimensions
result = om.metrics.query("revenue_by_country",
    filters={"country": "United Kingdom"},
    period="last_quarter"
)

Client resources

Resource	Description
om.metrics	List, query, and get metrics
om.tables	List tables, describe columns
om.analysis	Run analysis primitives (correlate, etc.)
om.reliability	Check metric health, alerts, subscribe to recovery
om.autoresearch	Run autoresearch jobs
om.setup	Connect warehouses, manage datasources, create metrics, manage keys
om.compiler	Check compiler status, import dbt/LookML
om.custom_analysis	Register and run custom multi-step analyses
om.auth	Signup, login, demo mode
om.admin	Cache invalidation, catalog sync

Metrics

# List all metrics
metrics = om.metrics.list()

# Search by intent
churn = om.metrics.list(search="churn")

# Filter by tag
finance = om.metrics.list(tag="finance")

# Query with options
result = om.metrics.query("total_revenue",
    filters={"country": "UK"},
    dimension="product",
    period="last_month",
    limit=10,
)
print(result.rows)
print(result.columns)
print(result.row_count)

# Get a single metric definition
m = om.metrics.get("total_revenue")
print(m.name, m.description, m.tags)

Time periods

The period parameter accepts semantic date keywords:

# Simple periods
result = om.metrics.query("revenue", period="last_month")
result = om.metrics.query("revenue", period="ytd")
result = om.metrics.query("revenue", period="last_30d")

# Comparison periods (returns current + previous + change)
result = om.metrics.query("revenue", period="mom")  # month-over-month
result = om.metrics.query("revenue", period="yoy")  # year-over-year

Tables

# List accessible tables
for t in om.tables.list():
    print(f"{t.schema}.{t.table} ({t.estimated_rows} rows)")

# Describe columns
desc = om.tables.describe("customers")
for col in desc.columns:
    pii = " [PII]" if col.is_pii else ""
    print(f"  {col.name} ({col.type}){pii}")

Raw SQL

result = om.query("SELECT country, COUNT(*) AS cnt FROM customers GROUP BY country")
for row in result.rows:
    print(row)

PII columns are automatically masked in results.

Analysis

# Correlate two metrics
result = om.analysis.correlate("churned_customers", "high_spenders")
print(result["explanation"])
# → "Jaccard similarity: 0.049 — populations are independent"

Reliability

# Check all metrics
status = om.reliability.status()
print(status["summary"])  # {total: 15, healthy: 12, degraded: 2, unreliable: 1}

# Check a specific metric
info = om.reliability.status_metric("total_revenue", detail=True)
print(info["status"])  # "healthy" / "degraded" / "unreliable"

# Get active alerts
alerts = om.reliability.alerts()

# Subscribe to recovery notification
om.reliability.subscribe("revenue", channel="webhook", target="https://hooks.slack.com/...")

Setup & admin

# Connect a warehouse
om.setup.connect_warehouse("postgres",
    host="localhost", port=5432,
    user="admin", password="secret",
    database="analytics"
)

# Create a metric
om.setup.create_metric(
    name="total_revenue",
    sql="SELECT SUM(amount) FROM orders",
    description="Total revenue",
    tags=["revenue", "finance"]
)

# Import from dbt
import json
with open("target/manifest.json") as f:
    om.setup.dbt_sync(json.load(f))

# Manage API keys
keys = om.setup.list_keys()
new_key = om.setup.generate_key(name="agent", scopes=["read", "query"])

Autoresearch

# Run autoresearch on a metric
job = om.autoresearch.run(
    metric="total_revenue",
    ground_truth_sql="SELECT SUM(total_amount) FROM invoices WHERE NOT is_cancellation"
)
print(job["job_id"])

Async client

from onlymetrix import AsyncOnlyMetrix

async with AsyncOnlyMetrix("https://api.onlymetrix.com", api_key="omx_sk_...") as om:
    metrics = await om.metrics.list()
    result = await om.metrics.query("total_revenue")

The async client has the same API as the sync client, with all methods as async.

LangChain integration

from onlymetrix.integrations.langchain import OnlyMetrixToolkit

toolkit = OnlyMetrixToolkit(
    base_url="https://api.onlymetrix.com",
    api_key="omx_sk_..."
)

# Returns LangChain tools for use with agents
tools = toolkit.get_tools()

See LangChain integration for details.

Error handling

from onlymetrix import OnlyMetrix, OnlyMetrixError

try:
    result = om.metrics.query("nonexistent_metric")
except OnlyMetrixError as e:
    print(f"Error {e.status_code}: {e.message}")