sombra 0.4.0

Python bindings for the Sombra graph database.

sombra

Python bindings for the Sombra graph database. The package exposes the Stage 8 fluent query builder together with lightweight CRUD helpers that forward to the Rust planner/executor through pyo3. Native failures are surfaced as typed exceptions (see wrap_native_error), and both Database and query streams provide context managers to ensure handles are released promptly.

Typed facade

For schema-aware CRUD helpers that mirror the TypeScript ergonomics, use the SombraDB wrapper under sombra.typed:

from typing import TypedDict
from typing_extensions import Literal

from sombra.typed import NodeSchema, SombraDB, TypedGraphSchema


class PersonProps(TypedDict):
    name: str
    age: int


class PersonNode(NodeSchema):
    properties: PersonProps


class GraphSchema(TypedGraphSchema):
    nodes: dict[str, PersonNode]
    edges: dict[str, TypedDict("KnowsEdge", {"from": Literal["Person"], "to": Literal["Person"], "properties": dict})]


schema: GraphSchema = {
    "nodes": {"Person": {"properties": {"name": "", "age": 0}}},
    "edges": {"KNOWS": {"from": "Person", "to": "Person", "properties": {}}},
}

db = SombraDB("/tmp/typed.db", schema=schema)
friend = db.add_node("Person", {"name": "Ada", "age": 33})

See examples/typed.py for a longer walkthrough that mirrors the Node.js demo.

Quick start

from sombra import Database
from sombra.query import eq

db = Database.open("/tmp/sombra.db")
db.seed_demo()

rows = db.query().nodes("User").where(eq("name", "Ada")).select("name").execute()
print(rows[0]["name"])

# Need request ids or feature flags? Ask for the metadata envelope:
payload = (
    db.query().nodes("User").request_id("example").select("name").execute(with_meta=True)
)
print(payload.request_id(), len(payload.rows()))

user_id = db.create_node("User", {"name": "New User"})
db.update_node(user_id, set_props={"bio": "updated"})
db.delete_node(user_id, cascade=True)

# Clean up native handles explicitly or via context managers
db.close()

async def stream_users() -> None:
    async with db.query().nodes("User").stream() as stream:
        async for row in stream:
            print(row["n0"])

Run the end-to-end example in examples/crud.py to see the workflow in action:

python examples/crud.py

Performance

The Python bindings have minimal overhead compared to the Rust core (~3-5%). Benchmark results on a typical developer machine:

Operation	Throughput
Node + edge creation	~9,000 ops/sec
Point reads	~20,000 reads/sec

Tips for optimal performance:

1. Use the builder for bulk operations – db.create() batches all nodes and edges into a single transaction, which is significantly faster than individual create_node/create_edge calls.

2. Use release builds – If building from source, always use maturin develop --release. Debug builds are significantly slower.

3. Tune synchronous mode – For write-heavy workloads where durability can be relaxed, set synchronous="normal" in options. The default "full" ensures every commit is fsync'd.

4. Use direct lookups when possible – db.get_node_record(id) is faster than running a query for single-node fetches.

Benchmarks

benchmarks/crud.py performs simple create/read/update/delete micro-benchmarks against a throwaway database:

python benchmarks/crud.py

benchmarks/realistic_bench.py runs a larger benchmark comparable to the Node.js and Rust benchmarks:

python benchmarks/realistic_bench.py

Predicate builders accept timezone-aware datetime objects directly and reject naive datetimes so callers do not need to juggle epochs. Property projections ({"var": "a", "prop": "name", "as": "alias"}) return scalar columns when you only need a subset of properties.

Development

Install the native module via maturin:

cd bindings/python
maturin develop --release
python -m pytest tests

Use maturin build to produce distributable wheels once you are ready to publish the bindings.

Release Workflow

1. Land a feat commit that touches bindings/python/** whenever you want the PyO3 wheel to bump its minor version. Release Please maps that commit to a new sombrapy release PR.
2. Before merging the PR, build and test the wheel locally: maturin develop, pytest -q, and maturin build --release (or maturin publish --dry-run).
3. Merge the release PR to tag the repo; the publish-python.yml workflow uploads the wheels to PyPI. Re-run maturin publish manually only if the workflow fails.