Python SDK

pip install sapixdb

uv add sapixdb

poetry add sapixdb

from sapixdb import SapixClient

db = SapixClient(url="http://localhost:7475", agent="my-app")

# Check connection
print(db.ping())  # True

# Write a record
record = db.collection("products").write({
    "name":  "Classic T-Shirt",
    "price": 29.99,
    "stock": 100,
})
print(record.id)    # "nuc_abc123"
print(record.hash)  # "sha3:e7f2a1..."  — cryptographic proof

# Read latest records
products = db.collection("products").latest()

# Filter
shirts = db.collection("products").find({"category": "apparel"})

# Time travel
from datetime import datetime, timedelta, timezone
yesterday = (datetime.now(timezone.utc) - timedelta(days=1)).isoformat()
snapshot = db.collection("orders").as_of(yesterday).latest()

import asyncio
from sapixdb import AsyncSapixClient

async def main():
    db = AsyncSapixClient(url="http://localhost:7475", agent="my-app")

    record = await db.collection("products").write({"name": "T-Shirt", "price": 29.99})
    products = await db.collection("products").latest()

    # Async context manager
    async with AsyncSapixClient(url="http://localhost:7475", agent="my-app") as db:
        shirts = await db.collection("products").find({"category": "apparel"})

asyncio.run(main())

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from sapixdb import AsyncSapixClient, SapixNotFoundError

app = FastAPI()
db  = AsyncSapixClient(url="http://localhost:7475", agent="store")


class ProductIn(BaseModel):
    name:     str
    price:    float
    stock:    int = 0
    category: str | None = None


@app.post("/products", status_code=201)
async def create_product(body: ProductIn):
    record = await db.collection("products").write(body.model_dump())
    return {"id": record.id, "hash": record.hash, "timestamp": record.timestamp}


@app.get("/products")
async def list_products(category: str | None = None):
    filter_ = {"category": category} if category else None
    items = await db.collection("products").latest(filter=filter_)
    return [{"id": r.id, **r.data} for r in items]


@app.get("/products/{record_id}")
async def get_product(record_id: str):
    try:
        r = await db.collection("products").get(record_id)
        return {"id": r.id, **r.data}
    except SapixNotFoundError:
        raise HTTPException(status_code=404, detail="Product not found")

from datetime import datetime, timedelta, timezone

# 30 minutes ago
ts = (datetime.now(timezone.utc) - timedelta(minutes=30)).isoformat()

snapshot = db.collection("orders").as_of(ts).find({"customer_id": "cust_001"})
# Returns orders exactly as they existed 30 minutes ago

# Link order → customer
db.graph.relate(order.id, customer.id, "placed_by")
db.graph.relate(order.id, product.id, "contains")

# Find everything connected to a customer (depth 2)
result = db.graph.traverse(customer.id, depth=2, direction="inbound")
print(result.nodes)  # list of NucleotideRecord
print(result.edges)  # list of GraphEdge

# Log AI agent decisions permanently and immutably
db.ingest("ai_decisions", {
    "model":      "gpt-4o",
    "action":     "approve_loan",
    "confidence": 0.94,
    "applicant":  "cust_001",
    "reasoning":  "Credit score 780, DTI 28%",
})
# Every decision is cryptographically signed — you can always prove
# what the AI decided, when, and why.

from sapixdb import SapixError, SapixNetworkError, SapixNotFoundError

try:
    record = db.collection("orders").get("nuc_missing")

except SapixNotFoundError as e:
    print(f"Not found: {e.record_id}")         # record ID that was missing

except SapixNetworkError:
    print("Cannot reach SapixDB — is it running?")

except SapixError as e:
    print(f"Error {e.status}: {e}")            # HTTP status + message

from sapixdb import SapixClient

db = SapixClient(url="http://localhost:7475", agent="store")

# 1. Add product
shirt = db.collection("products").write({
    "sku": "SHIRT-001", "name": "Classic T-Shirt",
    "price": 29.99, "stock": 200, "category": "apparel",
})

# 2. Register customer
customer = db.collection("customers").write({
    "name": "Alice Johnson", "email": "[email protected]",
})

# 3. Place order
order = db.collection("orders").write({
    "customer_id": customer.id,
    "items": [{"product_id": shirt.id, "qty": 2, "unit_price": 29.99}],
    "total": 59.98,
    "status": "placed",
})

# 4. Link in graph
db.graph.relate(order.id, customer.id, "placed_by")
db.graph.relate(order.id, shirt.id, "contains")

# 5. Ship (append — "placed" version is preserved forever)
db.collection("orders").write({
    "customer_id": customer.id,
    "status":      "shipped",
    "tracking":    "UPS-1Z999AA10123456784",
})

# 6. Audit: what was the status when it was placed?
original = db.collection("orders").as_of(order.timestamp).find_one(
    {"customer_id": customer.id}
)
print(original.data["status"])  # "placed" — not "shipped"

npm install sapixdb — zero dependencies, works in Node.js, Bun, Deno, and browsers.