Python SDK Reference

from onyx_database import onyx

db = onyx.init()

account = {
    "id": "acct_1",
    "name": "Checking",
    "balance": 1250,
}

db.save("Account", account)

fetched = db.from_table("Account").find("acct_1")

db.save("Account", { **account, "balance": 1400 })

db.delete("Account", "acct_1")

transactions = [
    { "id": "txn_batch_1", "accountId": "acct_batch", "amount": 1.99, "currency": "USD" },
    { "id": "txn_batch_2", "accountId": "acct_batch", "amount": 12.5, "currency": "USD" },
    { "id": "txn_batch_3", "accountId": "acct_batch", "amount": 45.0, "currency": "USD" },
    { "id": "txn_batch_4", "accountId": "acct_batch", "amount": 7.25, "currency": "USD" },
]

db.save("Transaction", transactions)

from onyx_database import onyx, eq, gt, gte, contains, desc

db = onyx.init()

transactions = (
    db.from_table("Transaction")
      .where(
          eq("status", "posted")
          .and_(gt("amount", 100))
          .and_(gte("createdAt", "2025-01-01"))
          .and_(contains("merchant", "aws"))
      )
      .order_by(desc("createdAt"))
      .limit(25)
      .list()
)

rows = (
    db.select("id", "accountId", "amount", "createdAt")
      .from_table("Transaction")
      .list()
)

from onyx_database import onyx, eq

db = onyx.init()

tx = (
    db.from_table("Transaction")
      .where(eq("id", "txn_123"))
      .first_or_none()
)

tx_alt = tx  # first_or_none() returns the record or None

from onyx_database import onyx, within, gt

db = onyx.init()

accounts_with_large_tx = (
    db.from_table("Account")
      .where(
          within(
              "id",
              db
                .select("accountId")
                .from_table("Transaction")
                .where(gt("amount", 1000))
          )
      )
      .list()
)

from onyx_database import onyx, eq, gt

db = onyx.init()

updated = (
    db.from_table("Transaction")
      .where(eq("status", "pending").and_(gt("amount", 100)))
      .set_updates({ "status": "posted" })
      .update()
)

from onyx_database import onyx, eq

db = onyx.init()

deleted_count = (
    db.from_table("Transaction")
      .where(eq("status", "archived"))
      .delete()
)

from onyx_database import onyx, eq

db = onyx.init()

by_merchant = (
    db.select("merchant", "currency")
      .from_table("Transaction")
      .where(eq("status", "posted"))
      .group_by("merchant", "currency")
      .list()
)

from onyx_database import onyx, sum, eq

db = onyx.init()

totals = (
    db.select(sum("amount"), "merchant")
      .from_table("Transaction")
      .where(eq("status", "posted"))
      .list()
)

from onyx_database import onyx, eq
import time

db = onyx.init()
events = []

handle = (
    db.from_table("Transaction")
      .where(eq("status", "posted"))
      .page_size(100)
      .on_item(lambda tx, action: events.append((action, tx)))
      .stream(include_query_results=True, keep_alive=False)
)

time.sleep(1)
handle["cancel"]()

from onyx_database import onyx, asc, eq

db = onyx.init()

page1 = (
    db.from_table("Transaction")
      .order_by(asc("createdAt"))
      .limit(50)
      .page()
)

records = page1.get("records", [])
next_page = page1.get("nextPage") or page1.get("next_page")

if next_page:
    page2 = (
        db.from_table("Transaction")
          .order_by(asc("createdAt"))
          .page(next_page=next_page)
    )
    print("page2 size", len(page2.get("records", [])), "next token", page2.get("nextPage"))

import json
import os
from urllib.parse import quote
from urllib.request import Request, urlopen

base_url = os.getenv("ONYX_DATABASE_BASE_URL", "https://api.onyx.dev").rstrip("/")
database_id = quote(os.environ["ONYX_DATABASE_ID"], safe="")
model_id = quote("churn-model", safe="")
endpoint = f"{base_url}/data/{database_id}/model-builder/published-model/{model_id}/predict"

def predict_request(body, suffix=""):
    request = Request(endpoint + suffix, method="POST", headers={
        "Content-Type": "application/json",
        "x-onyx-key": os.environ["ONYX_DATABASE_API_KEY"],
        "x-onyx-secret": os.environ["ONYX_DATABASE_API_SECRET"],
    }, data=json.dumps(body).encode("utf-8"))
    # urlopen raises HTTPError for unsuccessful HTTP responses.
    with urlopen(request, timeout=60) as response:
        return json.load(response)

result = predict_request({"inputs": [
    {"accountAgeDays": 420, "usageScore": 0.87},
    {"accountAgeDays": 90, "usageScore": 0.42},
]})
for record in result["records"]:
    print(record["input"], record["prediction"])

script_result = predict_request({
    "scriptId": "score-active-customers",
    "scriptParameters": {"segment": "enterprise", "limit": "50"},
}, "/script")
print(script_result["predictions"])

from onyx_database import onyx

db = onyx.init()

current = db.get_schema() or {}
entities = current.get("entities", [])

temp_table = {
    "name": "TempTable",
    "identifier": { "name": "id", "generator": "UUID", "type": "String" },
    "attributes": [
        { "name": "id", "type": "String", "isNullable": False },
        { "name": "name", "type": "String", "isNullable": False },
    ],
}

next_schema = {
    **current,
    "entities": [*entities, temp_table],
    "revisionDescription": "Add TempTable via SDK",
}

diff = db.diff_schema(next_schema)
validation = db.validate_schema(next_schema)

if validation.get("valid"):
    db.update_schema(next_schema, publish=True)