sapixdb 0.4.0

Official Python SDK for SapixDB — the agent-native living database

SapixDB Python SDK

Official Python SDK for SapixDB — the agent-native living database.

Supports both sync and async (asyncio / FastAPI / Django Async). Python 3.9+.

Installation

pip install sapixdb
# or
uv add sapixdb
# or
poetry add sapixdb

Quick Start

from sapixdb import SapixClient

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

# 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..."

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

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

# Time travel — what did the DB look like yesterday?
from datetime import datetime, timedelta, timezone
yesterday = (datetime.now(timezone.utc) - timedelta(days=1)).isoformat()
snapshot = db.collection("orders").as_of(yesterday).latest()

Async Usage

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()

# Or as a context manager
async def main():
    async with AsyncSapixClient(url="http://localhost:7475", agent="my-app") as db:
        record = await db.collection("products").write({"name": "T-Shirt"})

asyncio.run(main())

FastAPI Integration

from fastapi import FastAPI
from sapixdb import AsyncSapixClient

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

@app.post("/products")
async def create_product(name: str, price: float):
    record = await db.collection("products").write({"name": name, "price": price})
    return {"id": record.id, "hash": record.hash}

@app.get("/products")
async def list_products():
    items = await db.collection("products").latest()
    return [{"id": r.id, **r.data} for r in items]

API Reference

SapixClient / AsyncSapixClient

Parameter	Type	Default	Description
url	str	—	SapixDB agent URL
agent	str	—	Agent ID (matches SAPIX_AGENT_ID)
headers	dict	{}	Extra HTTP headers
timeout	float	10.0	Request timeout in seconds

---

db.collection(name)

Returns a CollectionClient (or AsyncCollectionClient).

.write(data) → WriteResult

Append a new record. Nothing is ever overwritten.

.write_batch(records) → list[WriteResult]

Write multiple records. Async version runs them concurrently.

.get(record_id) → NucleotideRecord

Fetch a record by ID. Raises SapixNotFoundError if missing.

.latest(*, filter?, limit?) → list[NucleotideRecord]

Current version of every record.

.history(*, filter?, limit?) → list[NucleotideRecord]

Full append-only history — every version ever written.

.find(filter, *, limit?) → list[NucleotideRecord]

Filter records (latest version only).

.find_one(filter) → NucleotideRecord | None

First match, or None.

.as_of(timestamp) → CollectionQuery

Scope reads to a point in time. Returns a query object with .latest(), .find(), .find_one(), .all().

---

db.graph

.relate(src, dst, edge_type, weight=1.0)

Create a typed directed edge between two records.

.add_edge(src, dst, edge_type, weight=1.0)

Full edge creation.

.remove_edge(src, dst, edge_type)

Delete an edge.

.traverse(from_id, *, depth=1, direction="outbound") → TraverseResult

Walk the graph. direction: "outbound" | "inbound" | "both".

.neighbors(node_id, direction="outbound") → list[NucleotideRecord]

Direct neighbours (depth=1 shortcut).

.edges(node_id) → list[GraphEdge]

All outbound edges from a node.

---

db.ingest(collection, data) → WriteResult

Write via the ingest endpoint — for AI agents, webhooks, and pipelines.

db.ingest("ai_decisions", {
    "model": "gpt-4o",
    "action": "approve_loan",
    "confidence": 0.94,
    "reasoning": "Credit score 780, DTI 28%",
})

---

Error Handling

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}")
except SapixNetworkError:
    print("SapixDB is unreachable — is it running?")
except SapixError as e:
    print(f"Error {e.status}: {e}")

---

Full Example: Online Store

from sapixdb import SapixClient

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

# 1. Add a 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": "[e-mail]",
})

# 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 (appends — "placed" version is preserved forever)
db.collection("orders").write({
    "customer_id": customer.id,
    "status": "shipped",
    "tracking": "UPS-1Z999AA10123456784",
})

# 6. Audit: what was the order 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"