SochDB Python API Reference

Complete reference for the SochDB Python surface.

Two packages, one name

There are two distinct Python packages, both published as sochdb on PyPI, with different versions and mostly disjoint APIs:

	Pure-Python SDK	Native PyO3 engine
Version	0.5.9	2.0.3
Mechanism	ctypes FFI + gRPC/IPC thin clients	Rust/PyO3 compiled extension (sochdb._native)
Scope	Broad embedded + server SDK: Database, Namespace, Collection, Queue, AgentMemory, temporal graph, semantic cache, StudioClient	Focused vector/lexical engine: HnswIndex, BM25Index, RRFFusion, ThreeLaneHybridIndex, MultiShardHnswIndex, TableDatabase, build_index*, recommended_hnsw_params
Python	>=3.9	>=3.9
License	Apache-2.0 (language SDK)	AGPL-3.0-or-later (core engine; commercial licensing available)

This page is split into Part A — Pure-Python SDK (0.5.9) (recommended for general usage) and Part B — Native engine (2.0.3). Each class below states which package it comes from.

The native PyO3 engine (sochdb 2.0.3) is the core Rust engine compiled as a Python extension and is therefore licensed AGPL-3.0-or-later (commercial licensing available), the same as the core. The pure-Python language SDK (sochdb 0.5.9) is Apache-2.0.

Install:

pip install sochdb

---

Part A — Pure-Python SDK (0.5.9)

The broad embedded + server SDK. Use this for most application code.

import sochdb

print(sochdb.__version__)  # "0.5.9"

Opening a database

Embedded

from sochdb import Database

# Embedded FFI engine backed by a filesystem path
db = Database.open("./data", config=None)

# Concurrent-friendly open
db = Database.open_concurrent("./data")

Convenience

import sochdb

# High-level helpers; ":memory:" creates a temp dir
client = sochdb.Client(path=":memory:")
collection = sochdb.open_collection("docs", path=":memory:", dimension=384)

Server / gRPC

from sochdb import SochDBClient, IpcClient

# gRPC server mode (host:port string)
client = SochDBClient("localhost:50051")     # alias: GrpcClient

# Local IPC
ipc = IpcClient(...)

Connection model

There is no connect(uri) / URI-scheme parser in the SDK. Embedded mode uses filesystem paths; gRPC mode uses host:port strings. There is no sochdb:// scheme.

TableDatabase does not exist in the 0.5.9 package — it lives only in the native 2.0.3 engine.

Signatures

Database.open(path: str, config: dict | None = None) -> Database
Database.open_concurrent(path: str) -> Database

sochdb.Client(path: str = ":memory:")
sochdb.open_collection(name: str, path: str = ":memory:", dimension: int | None = None)

---

Database — key/value operations

db.put(key: bytes, value: bytes, ttl_seconds: int = 0)
db.get(key: bytes) -> bytes | None
db.delete(key: bytes)

db.put_path(path: str, value)
db.get_path(path: str)

db.scan(start: bytes = b"", end: bytes = b"") -> list[tuple[bytes, bytes]]
db.scan_prefix(prefix: bytes)            # see prefix-safety note below
db.scan_prefix_unchecked(prefix: bytes)  # allows empty/short prefix (full scan)

# Batch helpers
db.put_batch(items)
db.get_batch(keys)
db.delete_batch(keys)
db.scan_path(prefix)
db.exists(key)
db.exists_in_txn(txn, key)

Prefix safety

scan_prefix enforces a minimum 2-byte prefix and raises ValueError if the prefix is shorter than 2 bytes, preventing accidental full-database scans. It never returns keys from other prefixes/tenants. Use scan_prefix_unchecked only when you intentionally need a full or short-prefix scan.

---

Transactions

txn = db.transaction()        # Transaction (context manager)
txn = db.begin_transaction()  # alias
db.with_transaction(fn)       # run fn inside a transaction

Transaction supports put/get/delete, put_path/get_path/delete_path, exists, scan, scan_prefix, scan_prefix_unchecked, scan_batched, and execute(sql).

Transaction.commit() -> int   # HLC-backed monotonic commit timestamp
Transaction.abort() -> None    # idempotent
Transaction.execute(sql: str)

commit() raises TransactionError on a Serializable Snapshot Isolation (SSI) conflict (FFI code -2). SSI is tracked Python-side; an IsolationLevel enum is available.

with db.transaction() as txn:          # auto-commit on clean exit,
    txn.put(b"k", b"v")                 # auto-abort on exception
    ts = txn.commit()                   # explicit commit also allowed

---

SQL execution

db.execute(sql: str) -> SQLQueryResult
db.execute_sql = db.execute            # alias
# also: Transaction.execute(sql)

SQLQueryResult exposes rows, columns, and rows_affected.

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

No index DDL in SQL

The SQL engine does not support CREATE INDEX / DROP INDEX. Vector indexing is done through the dedicated db.create_index(...) method below, not SQL DDL.

---

Vector indexing helper methods (on Database)

These are convenience methods on Database (FFI/gRPC) for managing a named vector index:

db.create_index(name: str, dimension: int,
                max_connections: int = 32, ef_construction: int = 256)
db.insert_vectors(index_name: str, ids, vectors)
db.search(index_name: str, query, k: int = 10)

---

VectorIndex — embedded HNSW (ctypes)

A lower-level embedded HNSW index wrapper.

from sochdb import VectorIndex, BatchAccumulator

index = VectorIndex(...)            # .ef_search is a read/write property

index.insert(id, vector)
index.insert_batch(ids, vectors) -> int
index.insert_batch_fast(ids, vectors)

index.search(query, k=10) -> list[tuple[int, float]]
index.search_fast(...)
index.search_ultra(...)
index.search_exact(...)
index.search_exact_f64(...)

index.build_flat_cache()
index.dimension()
len(index)

BatchAccumulator buffers inserts for high-throughput building:

acc = BatchAccumulator(index, estimated_size=0)
acc.add(...)
acc.add_single(id, vector)
n = acc.flush() -> int
acc.save(dir)
acc.load(dir)
acc.count()

Profiling helpers: enable_profiling(), disable_profiling(), dump_profiling(). A PerformanceWarning is emitted for slow paths.

---

Namespaces & Collections

# Namespace management (on Database)
db.create_namespace(...)
db.namespace(name) -> Namespace
db.get_or_create_namespace(...)
db.use_namespace(name)
db.list_namespaces()
db.delete_namespace(name, force=False)

Namespace groups collections and supports KV ops:

ns = db.namespace("tenant_a")
ns.create_collection(...)
ns.get_collection(name) -> Collection
ns.collection(name)
ns.list_collections()
ns.delete_collection(name)
# plus put / get / delete / scan

Collection is the document + vector search surface:

col.insert(...)
col.insert_batch(...)
col.insert_multi(...)
col.add(...)
col.upsert(...)

col.query(...)
col.search(request: SearchRequest) -> SearchResults
col.vector_search(...)
col.keyword_search(...)
col.hybrid_search(...)
col.vector_search_exact(...)
col.vector_search_exact_f64(...)

col.set_ef_search(n)
col.get(id)
col.delete(id)
col.count()
col.info()
len(col)

Supporting types: NamespaceConfig, CollectionConfig, DistanceMetric (str enum), QuantizationType (str enum), SearchRequest, SearchResult, SearchResults (iterable).

---

Queue API

from sochdb import create_queue

q = create_queue(
    db_or_client,
    queue_id="default",
    visibility_timeout_ms=30000,
    max_attempts=3,
    namespace="default",
)  # -> PriorityQueue

PriorityQueue construction helpers and operations:

PriorityQueue.from_database(...)
PriorityQueue.from_client(...)
PriorityQueue.from_backend(...)

q.enqueue(...)
q.enqueue_batch(...)
q.dequeue(...)
q.ack(task_id)
q.nack(...)
q.extend_visibility(task_id, ms)
q.peek()
q.stats() -> QueueStats
q.list_tasks(limit=100)

Supporting types: Task, TaskState (enum), QueueKey, QueueConfig (with_visibility_timeout / with_max_attempts / with_dead_letter_queue), QueueStats, StreamingTopK (push, get_sorted).

Backends: QueueBackend / QueueTransaction (ABCs), FFIQueueBackend, GrpcQueueBackend, InMemoryQueueBackend.

---

Agent Memory

Added in v0.5.8/0.5.9. High-level memory for agents over SochDB.

from sochdb import AgentMemory, create_agent_memory, QueryLanes

mem = AgentMemory(
    client,
    namespace="default",
    session_id=None,
    token_limit=4096,
    output_format="markdown",
)

Methods

mem.write_episode(text, *, t_valid_from=None, metadata=None,
                  namespace=None) -> EpisodeWriteResult

mem.search(query, *, token_limit=None, lanes="lexical", format=None,
           namespace=None, as_of=None) -> ContextQueryResult   # as_of: bi-temporal unix ms

mem.get_episode(doc_id, *, namespace=None) -> str
mem.compile_context(sections, ...)
mem.estimate_tokens(content, *, model="")
mem.format_context(content, *, format=None)

QueryLanes constants: LEXICAL="lexical", THREE_LANE="three_lane", HYBRID="hybrid", BM25="bm25", TRIGRAM="trigram".

ContextSectionType: GET=0, LAST=1, SEARCH=2, SELECT=3.

Also exported: AgentMemoryConfig, build_search_section, build_ingest_section, create_agent_memory, and lower-level pipelines — ExtractionPipeline / create_extraction_pipeline, Consolidator / create_consolidator, HybridRetriever / create_retriever, NamespaceManager / create_namespace_manager — plus dataclasses Entity, Relation, Assertion, CanonicalFact, RetrievalResult.

Context builder is example-only

A ContextQueryBuilder / ContextComponent fluent context-builder, graph-overlay demos, policy hooks (validate_user, redact_pii), and tool routing are not importable SDK classes in this package — they are example patterns in the separate sochdb-python-examples repo. Use AgentMemory and the memory pipelines above for the supported in-SDK API. (In the Rust and Node SDKs, several of these are real modules.)

---

Temporal graph (on Database)

db.add_temporal_edge(namespace, from_id, edge_type, to_id,
                     valid_from, valid_until=0, properties=None)

db.query_temporal_graph(namespace, node_id, mode="CURRENT",
                        timestamp=None, edge_type=None) -> list[dict]
# mode: "CURRENT" | "POINT_IN_TIME" | "RANGE"

db.end_temporal_edge(...)

Plain (non-temporal) graph helpers:

db.add_node(...)
db.add_edge(...)
db.traverse(...)
db.delete_node(...)
db.delete_edge(...)
db.get_neighbors(..., direction="outgoing")
db.find_path(from_, to, max_depth=10)

---

Semantic cache (on Database)

db.cache_put(cache_name, key, value, embedding: list[float],
             ttl_seconds=0) -> bool
db.cache_get(...)
db.cache_delete(cache_name, key)
db.cache_clear(cache_name)
db.cache_stats(cache_name)

---

Hosted Studio

from sochdb import StudioClient

studio = StudioClient(base_url, api_key=None, timeout=30.0)
studio.health() -> dict
studio.ingest_events(...) -> StudioEventIngestResult

Errors/types: StudioAPIError(status_code, message), StudioEventIngestResult.

---

Server-mode client (SochDBClient)

The gRPC client (SochDBClient, alias GrpcClient) mirrors much of the embedded surface:

# vector / collection
create_index, insert_vectors, search,
create_collection, add_documents, search_collection,
# graph
add_node, add_edge, traverse,
add_temporal_edge, query_temporal_graph,
# cache
cache_get, cache_put,
# agent / context
query_context, write_episode, estimate_tokens, format_context,
# tracing
start_trace, start_span, end_span,
# kv
get, put, delete,
close

Result dataclasses: SearchResult, Document, GraphNode, GraphEdge, TemporalEdge, ContextSectionResult, ContextQueryResult, EpisodeWriteResult.

---

Format utilities

from sochdb import (
    WireFormat, ContextFormat, CanonicalFormat,
    FormatCapabilities, FormatConversionError,
)

# TOON helpers on Database
db.to_toon(...)
db.to_json(...)
db.from_json(...)
db.from_toon(...)

---

Errors & error codes

from sochdb import SochDBError, ErrorCode

class SochDBError(message, code=None, context=None):
    def to_dict(self) -> dict: ...

ErrorCode is an IntEnum. The helper from_rust_error(code, message, context) maps engine codes to Python exceptions.

Hierarchy

Base	Subclasses
ConnectionError	—
TransactionError	TransactionConflictError
ProtocolError	—
DatabaseError	—
NamespaceError	NamespaceNotFoundError, NamespaceExistsError, NamespaceAccessError
CollectionError	CollectionNotFoundError, CollectionExistsError, CollectionConfigError
ValidationError	DimensionMismatchError(expected, actual), InvalidMetadataError, ScopeViolationError
QueryError	QueryTimeoutError(timeout_seconds)
(standalone)	EmbeddingError
LockError	DatabaseLockedError(path, holder_pid), LockTimeoutError(path, timeout_secs), EpochMismatchError(expected, actual), SplitBrainError

---

CLI entry points

The 0.5.9 package installs three console scripts:

sochdb-server        # sochdb.cli_server:main
sochdb-bulk          # sochdb.cli_bulk:main
sochdb-grpc-server   # sochdb.cli_grpc:main

---

Part B — Native engine (2.0.3)

A focused HNSW / BM25 / RRF / relational engine compiled as a PyO3 extension (sochdb._native). If the compiled extension is missing, the exported classes raise ImportError when used.

import sochdb

print(sochdb.version())  # CARGO_PKG_VERSION, e.g. "2.0.3"
sochdb.is_safe_mode()    # bool

Exports: Database, HnswIndex, BM25Index, RRFFusion, TableDatabase, Transaction, build_index, build_index_from_numpy, build_index_from_file, recommended_hnsw_params, version, is_safe_mode, HybridSearchIndex, HybridSearchResult, ThreeLaneHybridIndex, MultiShardHnswIndex, bulk_build_index.

Same name, different package

This Database/Transaction is the native engine's KV store — it is not the same class as the pure-Python Database in Part A. Pick one package per project to avoid confusion.

HnswIndex

HnswIndex(dimension: int, m: int = 32, ef_construction: int = 200,
          metric: str = "cosine", precision: str = "f32")

• metric: "cosine" | "euclidean" / "l2" | "dot" / "dot_product" / "inner_product"
• precision: "f32" / "float32" | "f16" / "float16" | "bf16" / "bfloat16"
• Internally max_connections_layer0 = m * 2, level_multiplier = 1 / ln(m).

Methods

# Insert (numpy float32 arrays; C-contiguous required, zero-copy)
index.insert_batch(vectors: "float32[N,D]") -> int            # auto IDs
index.insert_batch_with_ids(ids: "uint64[N]",
                            vectors: "float32[N,D]") -> int

# Search
index.search(query: "float32[D]", k: int,
             ef_search: int | None = None) -> tuple[list, list]   # (uint64 ids, float32 dists)
index.search_batch(queries: "float32[Q,D]", k: int,
                   ef_search: int | None = None) -> tuple[list, list]  # pads with u64::MAX / inf
index.search_filtered(query, k, filter: "list[tuple[str, str]]",
                      ef_search: int | None = None)              # AND-semantics; ef default 200

# Metadata
index.set_metadata(node_id, metadata: "list[tuple[str, str]]")
index.set_metadata_batch(node_ids: "uint64[]", metadata_list: "list[dict]")

# Graph maintenance / repair
index.optimize() -> int            # rebuild layer-0 exact brute-force kNN (~0.3s / 10K)
index.refine_graph() -> int
index.refine_graph_additive() -> int   # fills empty slots, never removes edges
index.repair() -> int                  # reconnect orphans via BFS
index.diagnose() -> dict               # {reachable, total, orphan_count, avg_degree,
                                       #  zero_degree_nodes, target_degree}

# Persistence & stats
index.save(path)                       # compressed
HnswIndex.load(path)                   # @staticmethod
index.stats() -> dict                  # {num_vectors, dimension, max_layer, avg_connections}
index.len                              # property
index.dimension                        # property
index.is_empty()
len(index)

Non-contiguous arrays

insert_batch and insert_batch_with_ids require C-contiguous float32 arrays and raise ValueError otherwise. Set SOCHDB_BATCH_SAFE_MODE=1 to enable a slower, safer batch path; SOCHDB_DEBUG_INSERT logs the insert path.

Module functions

build_index(embeddings, m=32, ef_construction=200,
            metric="cosine", ids=None) -> HnswIndex

build_index_from_numpy(embeddings, *, m=None, ef_construction=None,
                       metric="cosine", ids=None) -> HnswIndex
# When m / ef_construction are None, defaults come from recommended_hnsw_params(D).

build_index_from_file(input_path, output_path, *, dimension=None,
                      m=16, ef_construction=100, batch_size=1000,
                      quiet=False) -> dict      # subprocess / mmap via sochdb._bulk

recommended_hnsw_params(dimension, n_vectors=None,
                        target_recall=0.95) -> dict   # {m, ef_construction, ef_search, note}

recommended_hnsw_params heuristics:

• M: dimension <= 128 → M=16; 129–512 → M=24; 513+ → M=32.
• ef_construction = max(200, m * 8).
• ef_search: recall >= 0.99 → 40·M; >= 0.95 → 20·M; >= 0.90 → 10·M; else 6·M.

bulk_build_index is deprecated

bulk_build_index(...) emits a DeprecationWarning and delegates to the in-process build. Prefer build_index_from_numpy or build_index_from_file.

MultiShardHnswIndex

A pure-Python threaded scatter-gather wrapper (part of the native package) for very large collections (100M–1B vectors). Routes by id % n_shards.

Python-only wrapper

MultiShardHnswIndex is not a core-engine or server type. It is a Python wrapper that fans out across several HnswIndex shards using Python threads and per-shard locks; it is not auto-tuned beyond recommended_hnsw_params.

MultiShardHnswIndex(dimension, n_shards=8, m=None, ef_construction=None,
                    ef_search=None, metric="cosine", target_recall=0.95)

idx.insert_batch_with_ids(ids, vectors) -> int
idx.search(query, k=10, ef_search=None,
           failure_policy="raise") -> tuple[list, list]   # "raise" | "partial" | "ignore"

idx.total_vectors            # property
idx.shard_sizes()
idx.save(prefix) -> list[str]   # writes {prefix}_shard_{i}.hnsw

MultiShardHnswIndex.load(prefix, n_shards, dimension, *, ef_search=None,
                         metric="cosine", target_recall=0.95)   # @classmethod

BM25Index

BM25Index(k1=1.2, b=0.75, min_idf=0.0)

bm25.add_document(doc_id: int, text: str)
bm25.add_auto(text: str) -> int      # returns assigned id
bm25.search(query: str, k=10) -> list[tuple[int, float]]
bm25.num_documents()
bm25.vocab_size()

RRFFusion

Reciprocal Rank Fusion of vector and lexical result lists.

RRFFusion(k=60.0, vector_weight=1.0, lexical_weight=1.0)
rrf.fuse(vector_results, lexical_results, limit=10) -> list[tuple[int, float]]

Fixed vs. adaptive k

In the native engine and the Rust core, the RRF k constant is fixed (default 60.0). Only the pure-Python HybridSearchIndex (below) accepts adaptive_rrf_k=True.

HybridSearchIndex

A pure-Python composer over HnswIndex + BM25Index + RRFFusion.

HybridSearchIndex(dimension, *, m=16, ef_construction=100, metric="cosine",
                  bm25_weight=0.4, vector_weight=0.6, rrf_k=60.0,
                  adaptive_rrf_k=True)

idx.build(doc_ids, texts, embeddings)
idx.search(query_text, query_embedding, *, k=10,
           candidate_k=None) -> list[HybridSearchResult]
idx.vector_search(...)
idx.size               # property

HybridSearchResult is a dataclass: doc_id, score, vector_score, bm25_score, vector_rank, bm25_rank.

ThreeLaneHybridIndex

A native three-lane index combining grep + BM25 + HNSW, fused with RRF and allowed-set filtering.

ThreeLaneHybridIndex(dimension, m=16, ef_construction=200,
                     ef_search=128, metric="cosine")
idx.build(...)
idx.search(...)

Database (native KV)

Database.open(path, config=None) -> Database   # @staticmethod
# config presets: "default" | "throughput"/"fast" | "latency"/"oltp" | "durable"/"safe"

db.put(key: bytes, value: bytes, txn=None)     # auto-commits if txn is None
db.get(key, txn=None) -> bytes | None
db.delete(key, txn=None)
db.scan(prefix, txn=None) -> list[tuple[bytes, bytes]]

db.begin() -> int          # txn id
db.commit(txn=None) -> int # commit timestamp
db.abort(txn=None)

db.fsync()
db.checkpoint() -> int
db.gc() -> int
db.transaction() -> Transaction   # context manager
db.close()                        # no-op

Transaction argument is advisory

On the native Database, the txn argument to put/get/scan/commit/abort is largely ignored — the transaction is tracked internally.

Transaction (native): .id, commit() -> int, abort(), and __enter__ / __exit__ (auto-commit on clean exit, auto-abort on exception).

TableDatabase

A native relational / columnar store. Builder + columnar scan only — there is no SQL execute on this class.

TableDatabase.open(path, config=None)   # @staticmethod
# config presets: "default" | "throughput"/"fast" | "durable"/"safe"

tdb.register_table(name, columns: "list[tuple[str, str]]")
# column types: int64 | uint64 | float64 | text | binary | bool

tdb.begin_write() -> tuple[int, int]
tdb.begin_read() -> tuple[int, int]
tdb.commit(txn) -> int
tdb.abort(txn)
tdb.abort_read(txn)

tdb.insert_row(txn, table, row_id: int, values: list)
tdb.load_csv(table, csv_path) -> int     # GIL released; commits every 100K rows

tdb.scan_columnar(txn, table, columns=None) -> dict[str, list]
# result also carries "__row_count__" and "__bytes_read__"

tdb.table_count() -> int
tdb.list_tables() -> list[str]

---

Where each thing lives

Feature	Package
Database, Transaction, IsolationLevel, namespaces, collections	Pure-Python 0.5.9
VectorIndex, BatchAccumulator	Pure-Python 0.5.9
Queue API, AgentMemory + memory pipelines	Pure-Python 0.5.9
SochDBClient, IpcClient, StudioClient	Pure-Python 0.5.9
Temporal graph + semantic cache methods	Pure-Python 0.5.9
Format utilities, error types	Pure-Python 0.5.9
HnswIndex, BM25Index, RRFFusion, ThreeLaneHybridIndex, MultiShardHnswIndex, HybridSearchIndex	Native 2.0.3
Native Database / Transaction / TableDatabase	Native 2.0.3
build_index*, recommended_hnsw_params	Native 2.0.3
Context builder, graph overlay, policy hooks, tool routing	Example-only (sochdb-python-examples)

See also

• Python SDK guide
• HNSW vector search guide
• SQL API reference

---

For the latest documentation, see sochdb.dev