Python SDK Guide

SDK version: 0.5.9 (pure-Python ctypes SDK) Native engine version: 2.0.3 (PyO3 module) License: Apache-2.0 (SDK); the core engine is AGPL-3.0-or-later Prerequisites: Python 3.9+

Complete guide to SochDB's Python SDK covering the embedded Database, transactions, key-value and SQL operations, namespaces and collections, vector search, priority queues, agent memory, temporal graph, semantic cache, and the native HNSW engine.

---

⚠️ Two packages both named sochdb

There are two distinct Python packages that both publish under the PyPI name sochdb, with different versions and mostly disjoint APIs. Knowing which one you imported matters.

	Pure-Python SDK	Native PyO3 engine
Version	0.5.9	2.0.3
Mechanism	ctypes FFI + thin gRPC/IPC clients	Compiled Rust extension (sochdb._native)
Scope	Broad embedded + server SDK	Focused HNSW / BM25 / RRF / TableDatabase engine
Key classes	Database, Namespace, Collection, Queue, AgentMemory, VectorIndex, StudioClient	HnswIndex, BM25Index, RRFFusion, ThreeLaneHybridIndex, MultiShardHnswIndex, TableDatabase

This guide is primarily about the 0.5.9 pure-Python SDK — it is what you want for general application use. The Native HNSW engine section covers the 2.0.3 module, which exposes the low-level indexing primitives. Throughout this page each code block notes which package a class comes from when there is any ambiguity.

note

Both packages set __version__ and ship Database/Transaction classes, but those classes are not the same — the 0.5.9 Database is the high-level KV/SQL/namespace database; the 2.0.3 Database is a low-level native KV store. Install the one that matches the API you need.

---

Installation

pip install sochdb

from sochdb import Database

with Database.open("./my_database") as db:
    db.put(b"user:123", b'{"name":"Alice","age":30}')
    value = db.get(b"user:123")
    print(value.decode())
    # {"name":"Alice","age":30}

Pre-built wheels target Linux (x86_64, aarch64), macOS (Intel, Apple Silicon), and Windows (x64). The package requires Python 3.9 or newer.

---

Quick Start

Embedded mode (FFI)

Direct FFI bindings to the native library. No server required — ideal for local development, notebooks, simple apps, and edge deployments.

from sochdb import Database

with Database.open("./mydb") as db:
    db.put(b"key", b"value")
    value = db.get(b"key")

Configuration presets

Database.open() accepts an optional config dict. There is also a concurrent-access entry point for multi-process scenarios backed by MVCC.

from sochdb import Database

# Multi-process access to the same database
db = Database.open_concurrent("./shared_db")

For a quick in-memory database (uses a temporary directory), the high-level helpers are handy:

from sochdb import Client, open_collection

client = Client(path=":memory:")            # temp dir under the hood
collection = open_collection("docs", dimension=384)

Server mode (gRPC / IPC)

Thin clients connect to a running SochDB server. Embedded uses filesystem paths; gRPC uses host:port strings. There is no sochdb:// URI scheme.

from sochdb import SochDBClient  # alias: GrpcClient

client = SochDBClient("localhost:50051")
# Signature: put(key, value, namespace="default", ttl_seconds=0)
client.put(b"user:123", b'{"name": "Alice"}', namespace="my_namespace")
value = client.get(b"user:123", namespace="my_namespace")
client.close()

---

Transactions

Transaction.commit() returns an HLC-backed monotonic commit timestamp (an int), not a log sequence number. SochDB uses Serializable Snapshot Isolation (SSI); a serialization conflict raises TransactionError on commit.

Automatic transactions

with db.transaction() as txn:
    txn.put(b"account:1:balance", b"1000")
    txn.put(b"account:2:balance", b"500")
    # Commits on clean exit, aborts on exception

Manual control

from sochdb.errors import TransactionError

txn = db.begin_transaction()   # alias for db.transaction()
try:
    txn.put(b"key1", b"value1")
    txn.put(b"key2", b"value2")

    for key, value in txn.scan(b"key", b"key~"):
        print(f"{key.decode()}: {value.decode()}")

    commit_ts = txn.commit()   # int: HLC commit timestamp
    print(f"Committed at HLC {commit_ts}")
except TransactionError:
    txn.abort()                # idempotent
    raise

You can also pass a function with db.with_transaction(fn). Per-transaction operations include put/get/delete, put_path/get_path/delete_path, exists, scan, scan_prefix, scan_prefix_unchecked, scan_batched, and execute(sql).

note

Isolation behaviour is exposed through the IsolationLevel enum and tracked client-side. A serialization conflict surfaces as TransactionConflictError (a subclass of TransactionError).

---

Key-Value Operations

Basic operations

db.put(b"key", b"value", ttl_seconds=0)   # ttl_seconds=0 = no expiry
value = db.get(b"key")                     # bytes | None
db.delete(b"key")

Path API

db.put_path("users/alice/email", b"alice@example.com")
email = db.get_path("users/alice/email")
db.delete_path("users/alice/email")

Batch operations

db.put_batch([(b"k1", b"v1"), (b"k2", b"v2")])
values = db.get_batch([b"k1", b"k2"])
db.delete_batch([b"k1", b"k2"])
exists = db.exists(b"k1")

---

Prefix Scanning

There are two range/prefix iteration APIs. scan(start, end) walks an explicit byte range, while scan_prefix(prefix) walks all keys under a prefix.

# Multi-tenant data
db.put(b"tenants/acme/users/1", b'{"name":"Alice"}')
db.put(b"tenants/acme/users/2", b'{"name":"Bob"}')
db.put(b"tenants/globex/users/1", b'{"name":"Charlie"}')

# Scan only ACME data
for key, value in db.scan_prefix(b"tenants/acme/"):
    print(f"{key.decode()}: {value.decode()}")

Minimum 2-byte prefix

scan_prefix(prefix) enforces a minimum 2-byte prefix and raises ValueError for shorter prefixes. This is a safety guard that prevents an accidental full-database scan and guarantees the scan never crosses into another prefix or tenant. If you genuinely need an empty or 1-byte prefix (a full scan), use scan_prefix_unchecked(prefix) — it skips the guard and is intended for internal/administrative use.

# Explicit full scan (bypasses the 2-byte guard)
for key, value in db.scan_prefix_unchecked(b""):
    ...

---

SQL Database

The 0.5.9 SDK ships a KV-backed SQL engine exposed via db.execute(sql) (with the alias db.execute_sql). It supports a practical subset of SQL.

from sochdb import Database

with Database.open("./sql_db") as db:
    db.execute("""
        CREATE TABLE users (
            id INT,
            name TEXT,
            email TEXT,
            age INT
        )
    """)

    db.execute("INSERT INTO users (id, name, email, age) VALUES (1, 'Alice', 'alice@example.com', 30)")
    db.execute("INSERT INTO users (id, name, email, age) VALUES (2, 'Bob', 'bob@example.com', 25)")

    result = db.execute("SELECT name, age FROM users WHERE age > 26 ORDER BY age DESC LIMIT 10")
    for row in result.rows:
        print(row)

db.execute() returns a SQLQueryResult with rows, columns, and rows_affected.

Supported statements: CREATE TABLE, DROP TABLE, INSERT INTO, SELECT ... [WHERE] [ORDER BY] [LIMIT], UPDATE ... SET, DELETE FROM. Column types: INT, TEXT, FLOAT, BOOL, BLOB.

db.execute("UPDATE users SET age = 31 WHERE name = 'Alice'")
db.execute("DELETE FROM users WHERE age < 26")

No CREATE INDEX in SQL

The SQL engine in the 0.5.9 SDK does not support CREATE INDEX / DROP INDEX. Vector indexing is configured through the separate db.create_index(...) method (see Vector indexing on Database), not via SQL DDL. The richer SQL feature set (joins, aggregates, CREATE INDEX) lives in the server-side engine — see the SQL guide.

You can also run SQL inside a transaction together with KV writes for atomicity:

with db.transaction() as txn:
    txn.execute("INSERT INTO users (id, name) VALUES (3, 'Carol')")
    txn.put(b"user:3:metadata", b'{"verified": true}')
    # Atomic SQL + KV commit

---

Namespaces & Collections

Type-safe multi-tenant isolation with vector collections.

Creating namespaces

from sochdb import Database, NamespaceConfig

with Database.open("./multi_tenant") as db:
    config = NamespaceConfig(
        name="tenant_123",
        display_name="Acme Corp",
        labels={"tier": "enterprise"},
    )
    ns = db.create_namespace(config)

    # Or fetch an existing one
    ns = db.namespace("tenant_123")
    # Or get-or-create
    ns = db.get_or_create_namespace("tenant_123")

Other namespace methods on Database: use_namespace(name), list_namespaces(), and delete_namespace(name, force=False).

Creating collections

from sochdb import CollectionConfig, DistanceMetric

config = CollectionConfig(
    name="documents",
    dimension=384,                 # None = auto-infer from first vector
    metric=DistanceMetric.COSINE,
)
collection = ns.create_collection(config)

# Or the simple form
collection = ns.create_collection("embeddings", dimension=768)

Vector operations

# Insert a single vector with metadata
collection.insert(
    vector=[0.1, 0.2, 0.3],            # 384-dim in practice
    metadata={"source": "web", "url": "https://example.com"},
    id="doc_001",                       # optional; auto-generated if omitted
)

# Batch insert
collection.insert_batch(
    vectors=[[0.1, 0.2], [0.3, 0.4], [0.5, 0.6]],
    metadatas=[{"type": "a"}, {"type": "b"}, {"type": "c"}],
    ids=["doc_1", "doc_2", "doc_3"],
)

Search

The unified search(SearchRequest) API returns an iterable SearchResults, and there are convenience wrappers for the common cases.

from sochdb import SearchRequest

# Unified request
results = collection.search(
    SearchRequest(vector=query_embedding, k=10, filter={"source": "web"})
)
for result in results:
    print(f"ID: {result.id}, Score: {result.score:.4f}")

# Convenience methods
results = collection.vector_search(query_embedding, k=10)
results = collection.keyword_search("neural networks", k=10)
results = collection.hybrid_search(query_embedding, "deep learning", k=10)

# Exact (brute-force) search variants
results = collection.vector_search_exact(query_embedding, k=10)

# Tune recall/latency at query time
collection.set_ef_search(128)

Additional Collection methods include insert_multi, add, upsert, query, get(id), delete(id), count(), and info().

---

Vector Search

VectorIndex / BatchAccumulator (embedded HNSW via ctypes)

VectorIndex is the ctypes-backed embedded HNSW index in the 0.5.9 SDK.

from sochdb import VectorIndex
import numpy as np

index = VectorIndex(dimension=768, metric="cosine")   # cosine | euclidean | dot_product

embeddings = np.random.randn(10000, 768).astype(np.float32)
ids = list(range(len(embeddings)))

# Batch insert (much faster than per-vector insert)
index.insert_batch(ids, embeddings)

query = np.random.randn(768).astype(np.float32)
for vec_id, distance in index.search(query, k=10):
    print(f"{vec_id}: {distance:.4f}")

# Tune recall/latency
index.ef_search = 128

VectorIndex also offers insert(id, vector), insert_batch_fast, search_fast, search_ultra, search_exact, search_exact_f64, build_flat_cache(), and dimension().

For large ingest, BatchAccumulator buffers inserts and flushes them efficiently:

from sochdb import VectorIndex, BatchAccumulator

index = VectorIndex(dimension=768)
acc = BatchAccumulator(index, estimated_size=1_000_000)
for vec_id, vec in stream_vectors():
    acc.add_single(vec_id, vec)
acc.flush()
acc.save("./index_dir")

Vector indexing on Database

Database also exposes helper methods to create and query a named HNSW index directly:

# ef_construction defaults to 256, max_connections (M) defaults to 32
db.create_index("docs", dimension=768, max_connections=32, ef_construction=256)
db.insert_vectors("docs", ids=[1, 2, 3], vectors=[[...], [...], [...]])
results = db.search("docs", query=[...], k=10)

---

Native HNSW engine (2.0.3)

These classes come from the native PyO3 package (sochdb 2.0.3), not the 0.5.9 SDK. They are the lowest-level, highest-throughput indexing primitives. If the compiled extension is missing, importing them raises ImportError.

HnswIndex

from sochdb import HnswIndex, recommended_hnsw_params
import numpy as np

# Defaults: m=32, ef_construction=200, metric="cosine", precision="f32"
index = HnswIndex(dimension=768, m=32, ef_construction=200, metric="cosine")

vectors = np.random.randn(100_000, 768).astype(np.float32)  # must be C-contiguous
index.insert_batch(vectors)                                  # auto-generates sequential IDs

query = np.random.randn(768).astype(np.float32)
ids, distances = index.search(query, k=10)                   # returns two numpy arrays

Supported metric values: "cosine", "euclidean"/"l2", "dot"/"dot_product"/"inner_product". Supported precision: "f32"/"float32", "f16"/"float16", "bf16"/"bfloat16".

Key methods: insert_batch_with_ids(ids, vectors), search_batch(queries, k, ef_search), search_filtered(query, k, filter, ef_search) (AND-semantics metadata filter), set_metadata/set_metadata_batch, optimize(), refine_graph(), refine_graph_additive(), repair(), diagnose(), save(path)/HnswIndex.load(path), and stats().

recommended_hnsw_params and builders

from sochdb import recommended_hnsw_params, build_index_from_numpy

params = recommended_hnsw_params(dimension=768, n_vectors=1_000_000, target_recall=0.95)
# -> {"m": ..., "ef_construction": ..., "ef_search": ..., "note": ...}

# Build an index, auto-selecting params when m/ef_construction are omitted
index = build_index_from_numpy(vectors, ids=ids)

recommended_hnsw_params picks M by dimension (≤128 → 16, 129–512 → 24, 513+ → 32), sets ef_construction = max(200, M*8), and scales ef_search to the requested recall.

There is also build_index(embeddings, m=32, ef_construction=200, metric="cosine", ids=None) and build_index_from_file(...). bulk_build_index(...) is deprecated and now warns.

MultiShardHnswIndex

Python-only wrapper

MultiShardHnswIndex is a pure-Python scatter-gather wrapper in the 2.0.3 package, built on top of HnswIndex for very large (100M–1B) vector sets. It uses Python threads and per-shard locks and routes by id % n_shards. It is not a core-engine type — the Rust core has no multi-shard HNSW. Do not expect it server-side.

from sochdb import MultiShardHnswIndex

idx = MultiShardHnswIndex(dimension=768, n_shards=8, target_recall=0.95)
idx.insert_batch_with_ids(ids, vectors)
result_ids, distances = idx.search(query, k=10, failure_policy="raise")
paths = idx.save("./big_index")   # writes ./big_index_shard_{i}.hnsw

Hybrid indexes (BM25 + RRF)

The native package also exposes BM25Index, RRFFusion, the pure-Python HybridSearchIndex (composes HNSW + BM25 + RRF), and the native ThreeLaneHybridIndex (grep + BM25 + HNSW fused with RRF).

from sochdb import HybridSearchIndex

# adaptive_rrf_k is a Python-SDK feature (the core RRF-k is fixed at 60.0)
hybrid = HybridSearchIndex(dimension=768, bm25_weight=0.4, vector_weight=0.6, adaptive_rrf_k=True)
hybrid.build(doc_ids, texts, embeddings)
hits = hybrid.search(query_text, query_embedding, k=10)

note

adaptive_rrf_k=True only adjusts the RRF constant inside this Python HybridSearchIndex. The Rust/core fusion uses a fixed RRF k of 60.0 — there is no adaptive RRF-k in the engine.

---

Priority Queue

First-class queue API with ordered-key task entries and a configurable visibility timeout.

from sochdb import Database
from sochdb.queue import create_queue, QueueConfig

db = Database.open("./queue_db")

# Convenience factory
queue = create_queue(db, queue_id="tasks", visibility_timeout_ms=30000, max_attempts=3)

You can also build a PriorityQueue directly with PriorityQueue.from_database(db, ...), from_client(...), or from_backend(...), and configure it via QueueConfig().with_visibility_timeout(...).with_max_attempts(...).with_dead_letter_queue(...).

Enqueue, dequeue, acknowledge

task_id = queue.enqueue(
    priority=1,                                  # lower = higher priority
    payload=b'{"action": "process_order", "order_id": 123}',
    metadata={"source": "api"},
)

task = queue.dequeue()
if task:
    try:
        process(task.payload)
        queue.ack(task.task_id)
    except Exception:
        queue.nack(task.task_id)                 # retry or dead-letter

Additional methods: enqueue_batch, extend_visibility(task_id, ms), peek(), stats() (returns QueueStats), and list_tasks(limit=100).

stats = queue.stats()
print(stats)

StreamingTopK is available for efficient ordered top-k accumulation:

from sochdb.queue import StreamingTopK

topk = StreamingTopK()
for item in items:
    topk.push(item)
ordered = topk.get_sorted()

Backends include FFIQueueBackend, GrpcQueueBackend, and InMemoryQueueBackend.

---

Agent Memory

Added in v0.5.8/0.5.9, AgentMemory provides episodic write + bi-temporal retrieval for agent workloads.

from sochdb import SochDBClient
from sochdb.memory import AgentMemory, QueryLanes, create_agent_memory

# AgentMemory talks to a running SochDB server (ContextService), not an
# embedded Database — pass a SochDBClient, not Database.open(...).
client = SochDBClient("localhost:50051")
memory = AgentMemory(client, namespace="default", session_id="sess-1", token_limit=4096)
# Or: memory = create_agent_memory(client, namespace="default", token_limit=4096)

# Write an episode (t_valid_from is optional; bi-temporal)
memory.write_episode("User asked about pricing tiers.", metadata={"role": "user"})

# Retrieve context, optionally as-of a point in time (unix ms)
ctx = memory.search(
    "pricing",
    lanes=QueryLanes.HYBRID,     # lexical | three_lane | hybrid | bm25 | trigram
    token_limit=2048,
    as_of=None,                  # int unix-ms for point-in-time recall
)
print(ctx)

QueryLanes constants: LEXICAL="lexical", THREE_LANE="three_lane", HYBRID="hybrid", BM25="bm25", TRIGRAM="trigram". Other AgentMemory methods: get_episode(doc_id), compile_context(sections, ...), estimate_tokens(content), and format_context(content).

Lower-level building blocks are also importable from sochdb.memory: ExtractionPipeline, Consolidator, HybridRetriever, NamespaceManager, and dataclasses Entity, Relation, Assertion, CanonicalFact, RetrievalResult.

Example patterns, not SDK classes

Higher-level agent scaffolding — context builders, policy hooks, tool routing, and graph-overlay helpers — are example patterns that live in the separate sochdb-python-examples repository (e.g. ContextQueryBuilder/ContextComponent, the validate_user/redact_pii policy hooks, the tool-routing demo). They are demonstration scripts, not importable classes in the sochdb package. Build them yourself using the primitives above, or copy the examples.

---

Temporal Graph

Time-aware relationships support point-in-time and range queries.

from sochdb import Database

with Database.open("./temporal_db") as db:
    db.add_temporal_edge(
        namespace="org",
        from_id="alice",
        edge_type="WORKS_AT",
        to_id="acme_corp",
        valid_from=1704067200000,   # 2024-01-01 (unix ms)
        valid_until=1735689600000,  # 2025-01-01
        properties={"role": "Engineer"},
    )

    db.add_temporal_edge(
        namespace="org",
        from_id="alice",
        edge_type="WORKS_AT",
        to_id="globex_inc",
        valid_from=1735689600000,
        valid_until=0,              # 0 = no end (current)
        properties={"role": "Senior Engineer"},
    )

Querying

# Point in time: "Where did Alice work on 2024-06-15?"
results = db.query_temporal_graph(
    namespace="org",
    node_id="alice",
    mode="POINT_IN_TIME",          # CURRENT | POINT_IN_TIME | RANGE
    timestamp=1718409600000,
    edge_type="WORKS_AT",
)
for edge in results:
    print(edge)

End an edge with db.end_temporal_edge(...). A plain (non-temporal) graph API is also available: add_node, add_edge, traverse, delete_node, delete_edge, get_neighbors(direction="outgoing"), and find_path(from, to, max_depth=10).

---

Semantic Cache

An embedding-aware cache. Entries are stored under a cache name + key with an associated embedding; lookups match by embedding similarity (cosine) above a threshold.

# cache_put(cache_name, key, value, embedding, ttl_seconds=0) -> bool
db.cache_put(
    "answers",
    key="q:pricing",
    value="Our pricing has three tiers...",   # str, not bytes
    embedding=[0.1, 0.2, 0.3],
    ttl_seconds=3600,
)

# cache_get matches by embedding similarity, not by key:
# cache_get(cache_name, query_embedding, threshold=0.85) -> str | None
hit = db.cache_get("answers", [0.1, 0.2, 0.3], threshold=0.85)

db.cache_delete("answers", "q:pricing")
db.cache_clear("answers")
stats = db.cache_stats("answers")

---

Hosted Studio

StudioClient talks to the hosted SochDB Studio service for event ingestion and health checks.

from sochdb import StudioClient

studio = StudioClient(base_url="https://studio.example.com", api_key="...", timeout=30.0)
print(studio.health())
result = studio.ingest_events(events)

Errors surface as StudioAPIError(status_code, message); ingest_events returns a StudioEventIngestResult.

---

TOON Format

SochDB's native, token-efficient output notation. TOON typically uses 40-66% fewer tokens than equivalent JSON for tabular data, which makes it well suited for LLM context.

from sochdb import Database

records = [
    {"id": 1, "name": "Alice", "email": "alice@example.com"},
    {"id": 2, "name": "Bob", "email": "bob@example.com"},
]

toon_str = db.to_toon("users", records, ["name", "email"])
print(toon_str)

# Round-trip back. from_toon returns a (table_name, fields, records) tuple.
table_name, fields, records_again = db.from_toon(toon_str)

Database also exposes to_json and from_json. The wire/context format enums (WireFormat, ContextFormat, CanonicalFormat) live in sochdb.format.

---

CLI Tools

Three console scripts are installed with the package:

sochdb-server        # embedded/IPC server
sochdb-bulk          # bulk vector index build/query
sochdb-grpc-server   # gRPC server

Bulk index build example:

sochdb-bulk build-index \
    --input embeddings.npy \
    --output index.hnsw \
    --dimension 768 \
    --max-connections 32 \
    --ef-construction 256 \
    --metric cosine

---

Error Handling

The SDK provides a structured error hierarchy. All errors derive from SochDBError, which carries an optional code and context and offers .to_dict().

from sochdb.errors import (
    SochDBError,            # base
    ConnectionError,
    DatabaseError,
    ProtocolError,

    # Transactions
    TransactionError,
    TransactionConflictError,

    # Namespaces
    NamespaceError,
    NamespaceNotFoundError,
    NamespaceExistsError,
    NamespaceAccessError,

    # Collections
    CollectionError,
    CollectionNotFoundError,
    CollectionExistsError,
    CollectionConfigError,

    # Validation
    ValidationError,
    DimensionMismatchError,
    InvalidMetadataError,
    ScopeViolationError,

    # Queries
    QueryError,
    QueryTimeoutError,
    EmbeddingError,

    # Locks
    LockError,
    DatabaseLockedError,
    LockTimeoutError,
    EpochMismatchError,
    SplitBrainError,
)

Handling lock errors

from sochdb import Database
from sochdb.errors import DatabaseLockedError, LockTimeoutError

try:
    db = Database.open("./shared_db")
except DatabaseLockedError as e:
    print(f"Locked by another process: {e}")
    db = Database.open_concurrent("./shared_db")   # retry concurrently
except LockTimeoutError as e:
    print(f"Timed out waiting for lock: {e}")

Handling dimension errors

from sochdb.errors import DimensionMismatchError

try:
    collection.insert([1.0, 2.0, 3.0])   # 3-dim into a 384-dim collection
except DimensionMismatchError as e:
    print(f"Expected {e.expected} dimensions, got {e.actual}")

---

Best Practices

Use context managers

# Automatic cleanup
with Database.open("./db") as db:
    db.put(b"key", b"value")

Use prefix scans for multi-tenancy

for key, value in db.scan_prefix(f"tenants/{tenant_id}/".encode()):
    process(key, value)

Batch your writes

# Fast: one batched call
db.put_batch(items)

# For vectors, accumulate then flush
acc = BatchAccumulator(index)
for vid, vec in stream():
    acc.add_single(vid, vec)
acc.flush()

Use TOON for LLM context

records = [dict(zip(result.columns, r)) for r in result.rows]
toon_context = db.to_toon("users", records, ["name", "email"])
# Send TOON to the LLM instead of JSON — saves 40-66% tokens

---

Complete Example: Multi-Tenant SaaS with SQL + KV

from sochdb import Database
import json

def main():
    with Database.open("./saas_db") as db:
        db.execute("""
            CREATE TABLE tenants (
                id INT,
                name TEXT,
                created_at TEXT
            )
        """)
        db.execute("INSERT INTO tenants (id, name, created_at) VALUES (1, 'ACME Corp', '2026-01-01')")
        db.execute("INSERT INTO tenants (id, name, created_at) VALUES (2, 'Globex Inc', '2026-01-01')")

        # Per-tenant KV data
        db.put(b"tenants/1/users/alice", b'{"role":"admin","email":"alice@acme.com"}')
        db.put(b"tenants/1/users/bob", b'{"role":"user","email":"bob@acme.com"}')
        db.put(b"tenants/2/users/charlie", b'{"role":"admin","email":"charlie@globex.com"}')

        result = db.execute("SELECT id, name FROM tenants ORDER BY name")
        for row in result.rows:
            tenant_id, tenant_name = row[0], row[1]
            prefix = f"tenants/{tenant_id}/".encode()
            users = list(db.scan_prefix(prefix))
            print(f"\n{tenant_name} ({len(users)} users):")
            for key, value in users:
                u = json.loads(value.decode())
                print(f"  {key.decode()}: {u['email']} ({u['role']})")

if __name__ == "__main__":
    main()

---

Resources

• Python SDK GitHub
• PyPI Package
• SQL Guide
• Go SDK
• Node.js SDK
• Rust SDK

---

Last updated: June 2026 — SDK 0.5.9, native engine 2.0.3.