Python API Reference¶

pip install uni-db

cd bindings/uni-db
pip install maturin
maturin develop --release

import uni_db

# Open or create a database
db = uni_db.Uni.open("/path/to/db")

# Define schema
db.schema() \
  .label("Person").property("name", "string").property("age", "int").done() \
  .edge_type("KNOWS", ["Person"], ["Person"]).done() \
  .apply()

# Create a session for reads
session = db.session()

# Write data in a transaction
with session.tx() as tx:
    tx.execute("CREATE (a:Person {name: 'Alice', age: 30})")
    tx.execute("CREATE (b:Person {name: 'Bob', age: 25})")
    tx.execute("""
        MATCH (a:Person {name: 'Alice'}), (b:Person {name: 'Bob'})
        CREATE (a)-[:KNOWS {since: 2020}]->(b)
    """)
    tx.commit()

# Query through the session
result = session.query("MATCH (n:Person) WHERE n.age > 20 RETURN n.name AS name, n.age AS age")
for row in result:
    print(row["name"], row["age"])

# Parameterized query
result = session.query(
    "MATCH (n:Person) WHERE n.name = $name RETURN n",
    {"name": "Alice"}
)

import uni_db
import asyncio

async def main():
    async with uni_db.AsyncUni.open("/path/to/db") as db:
        session = db.session()

        # Reads work on the session
        result = await session.query("MATCH (n:Person) RETURN n.name AS name")
        for row in result:
            print(row["name"])

        # Writes go through a transaction
        tx = await session.tx()
        await tx.execute("CREATE (:Person {name: 'Carol', age: 28})")
        await tx.commit()

asyncio.run(main())

# Simplest usage
db = uni_db.Uni.open("./my_graph")

# In-memory for tests
db = uni_db.Uni.temporary()

# Context manager (flushes on exit)
with uni_db.Uni.open("./my_graph") as db:
    session = db.session()
    # ...

db = (
    uni_db.UniBuilder.open("/path/to/db")
    .cache_size(1024 * 1024 * 100)   # 100 MB cache
    .parallelism(4)
    .wal_enabled(True)
    .build()
)

session = db.session()

# Simple query
result = session.query("MATCH (n:Person) RETURN n.name AS name")

# Parameterized
result = session.query(
    "MATCH (n:Person) WHERE n.age > $min RETURN n",
    {"min": 21}
)

# Builder pattern with timeout
result = (
    session.query_with("MATCH (n:Person) WHERE n.age > $min RETURN n")
    .param("min", 21)
    .timeout(5.0)
    .max_memory(50_000_000)
    .fetch_all()
)

# Default transaction
tx = session.tx()

# Configured transaction
tx = session.tx_with().timeout(30.0).isolation("serialized").start()

# Watch all commits
with session.watch() as stream:
    for notification in stream:
        print(f"Version {notification.version}: {notification.labels_affected}")
        break  # just one

# Watch specific labels
stream = session.watch_with().labels(["Person"]).build()

class AuditHook:
    def before_query(self, ctx):
        print(f"Query: {ctx['query_text']}")

    def after_commit(self, ctx, result):
        print(f"Committed version {result.version}")

session.add_hook("audit", AuditHook())

with db.session() as session:
    result = session.query("MATCH (n) RETURN n LIMIT 10")

session = db.session()
with session.tx() as tx:
    tx.execute("CREATE (n:Person {name: 'Alice', age: 30})")
    tx.execute("CREATE (n:Person {name: 'Bob', age: 25})")
    result = tx.query("MATCH (n:Person) RETURN count(n) AS cnt")
    tx.commit()
# If an exception occurs before commit(), the transaction is rolled back automatically.

db.schema() \
  .label("Person") \
      .property("name", "string") \
      .property("age", "int") \
      .property_nullable("email", "string") \
      .vector("embedding", 128) \
      .index("name", "btree") \
      .done() \
  .label("Company") \
      .property("name", "string") \
      .done() \
  .edge_type("WORKS_AT", ["Person"], ["Company"]) \
      .property("since", "int") \
      .done() \
  .apply()

# Simple index
builder.index("name", "btree")

# Vector index with HNSW configuration
builder.index("embedding", {
    "type": "vector",
    "algorithm": "hnsw",
    "m": 32,
    "ef_construction": 400,
    "metric": "cosine"
})

# Full-text index with n-gram tokenizer
builder.index("content", {
    "type": "fulltext",
    "tokenizer": "ngram",
    "ngram_min": 2,
    "ngram_max": 4
})

session = db.session()
with session.tx() as tx:
    writer = tx.bulk_writer().batch_size(10000).build()

    # Insert vertices — returns list of allocated VIDs
    vids = writer.insert_vertices("Person", [
        {"name": "Alice", "age": 30},
        {"name": "Bob", "age": 25},
    ])

    # Insert edges — (source_vid, target_vid, properties)
    writer.insert_edges("KNOWS", [
        (vids[0], vids[1], {"since": 2020}),
    ])

    # Commit bulk data and rebuild indexes
    stats = writer.commit()
    print(f"Inserted {stats.vertices_inserted} vertices, {stats.edges_inserted} edges")

    tx.commit()

with tx.bulk_writer().build() as writer:
    writer.insert_vertices("Person", [...])
    writer.commit()

with session.tx() as tx:
    appender = tx.appender("Person")
    for record in data_source:
        appender.append({"name": record["name"], "age": record["age"]})
    stats = appender.finish()
    tx.commit()

db.schema() \
  .label("Document") \
      .property("title", "string") \
      .vector("embedding", 128) \
      .index("embedding", {"type": "vector", "metric": "cosine"}) \
      .done() \
  .apply()

session = db.session()
query_vec = [0.1, 0.2, ...]  # 128 dimensions
result = session.query(
    "CALL uni.vector.query('Document', 'embedding', $vec, 10) "
    "YIELD vid, distance RETURN vid, distance",
    {"vec": query_vec}
)
for row in result:
    print(row["vid"], row["distance"])

session = db.session()
result = session.locy("""
    reachable(x, y) :- MATCH (a)-[:KNOWS]->(b) WHERE a.name = x AND b.name = y.
    reachable(x, z) :- reachable(x, y), reachable(y, z).
    QUERY reachable(x, y) INTO result.
""")

# Access command results (QUERY output)
rows = result.rows()
if rows:
    for row in rows:
        print(row)

# Access derived relations
derived = result.derived  # dict: rule_name -> list[dict]

# Access evaluation statistics
print(result.stats)  # LocyStats(strata=..., iterations=..., time=...)

result = (
    session.locy_with(program)
    .param("start_node", "Alice")
    .timeout(30.0)
    .max_iterations(500)
    .with_config({
        "strict_probability_domain": True,
        "exact_probability": True,
        "max_bdd_variables": 1000,
    })
    .run()
)

result = session.locy(program)

with session.tx() as tx:
    apply_result = tx.apply(result.derived_fact_set)
    print(f"Applied {apply_result.facts_applied} facts")
    tx.commit()

with session.tx() as tx:
    apply_result = (
        tx.apply_with(result.derived_fact_set)
        .require_fresh(True)
        .max_version_gap(5)
        .run()
    )
    tx.commit()

db = (
    uni_db.UniBuilder.open("./graph")
    .xervo_catalog_from_file("./models.json")
    .build()
)

xervo = db.xervo()

# Check availability
if xervo.is_available():
    # Embed text -> list[list[float]]
    vectors = xervo.embed("embed/default", ["graph databases", "vector search"])

    # Generate with Message objects
    from uni_db import Message
    result = xervo.generate(
        "llm/default",
        [
            Message.system("You are a concise technical assistant."),
            Message.user("What is snapshot isolation?"),
        ],
        max_tokens=256,
        temperature=0.7,
    )
    print(result.text)    # Generated string
    print(result.usage)   # TokenUsage | None

    # Convenience wrapper -- single prompt string
    result = xervo.generate_text("llm/default", "Explain hybrid search in one sentence.")

db = await uni_db.AsyncUni.builder() \
    .open("./graph") \
    .xervo_catalog_from_file("./models.json") \
    .build()

xervo = db.xervo()
vectors = await xervo.embed("embed/default", ["hello"])
result = await xervo.generate_text("llm/default", "Hello!")

Message.user("content")        # role = "user"
Message.assistant("content")   # role = "assistant"
Message.system("content")      # role = "system"
Message(role, content)         # explicit

session = db.session()

# Prepare once
prepared = session.prepare("MATCH (n:Person) WHERE n.name = $name RETURN n")

# Execute many times
result1 = prepared.execute({"name": "Alice"})
result2 = prepared.execute({"name": "Bob"})

# Fluent binder pattern
result = prepared.bind().param("name", "Carol").execute()

# Get the original query text
print(prepared.query_text())

prepared = session.prepare_locy("""
    reachable(x, y) :- MATCH (a)-[:KNOWS]->(b) WHERE a.name = x AND b.name = y.
    QUERY reachable($start, y) INTO result.
""")

result = prepared.execute({"start": "Alice"})

# Fluent binder
result = prepared.bind().param("start", "Bob").execute()

session = db.session()

# Watch all commits
with session.watch() as stream:
    for notification in stream:
        print(f"v{notification.version}: {notification.labels_affected}")

# Watch with filters
stream = (
    session.watch_with()
    .labels(["Person", "Company"])
    .edge_types(["WORKS_AT"])
    .build()
)

class MyHook:
    def before_query(self, ctx):
        """Called before each query. Raise to reject."""
        if "DROP" in ctx["query_text"]:
            raise ValueError("DROP queries are forbidden")

    def after_query(self, ctx, metrics):
        """Called after each query with metrics."""
        if metrics.total_time_ms > 1000:
            print(f"Slow query: {ctx['query_text']}")

    def before_commit(self, ctx):
        """Called before commit. Raise to reject."""
        pass

    def after_commit(self, ctx, result):
        """Called after successful commit."""
        print(f"Committed v{result.version}: {result.mutations_committed} mutations")

session.add_hook("my_hook", MyHook())

from uni_db import DataType

schema.label("Person") \
    .property("name", DataType.STRING()) \
    .property("age", DataType.INT64()) \
    .property("score", DataType.FLOAT64()) \
    .property("active", DataType.BOOL()) \
    .property("embedding", DataType.vector(128)) \
    .property("tags", DataType.list(DataType.STRING())) \
    .property("metadata", DataType.map(DataType.STRING(), DataType.JSON())) \
    .done()

result = session.query("MATCH (n:Person) RETURN n.name AS name, n.age AS age")

# Iterate rows (each row is a dict)
for row in result:
    print(row["name"], row["age"])

# Index access
first = result[0]
last = result[-1]
slice = result[1:3]

# Length
print(len(result))

# Boolean (true if non-empty)
if result:
    print("Got results")

with session.query_with("MATCH (n) RETURN n").cursor() as cursor:
    print(cursor.columns)           # Column names
    row = cursor.fetch_one()        # Single row or None
    batch = cursor.fetch_many(100)  # Up to 100 rows
    rest = cursor.fetch_all()       # All remaining rows

for row in cursor:
    print(row)

explain = session.query_with("MATCH (n:Person) RETURN n").explain()

print(explain.plan_text)       # Human-readable plan
print(explain.warnings)        # Planner warnings
print(explain.cost_estimates)  # Estimated rows and cost
print(explain.index_usage)     # Index usage details
print(explain.suggestions)     # Index suggestions

result, profile = session.query_with("MATCH (n:Person) RETURN n").profile()

print(f"Total time: {profile.total_time_ms}ms")
print(f"Peak memory: {profile.peak_memory_bytes} bytes")
print(profile.plan_text)   # Plan with actual row counts
print(profile.operators)   # Per-operator statistics

explain = session.locy_with(program).explain()

print(explain.plan_text)
print(f"Strata: {explain.strata_count}")
print(f"Rules: {explain.rule_names}")
print(f"Recursive: {explain.has_recursive_strata}")
print(f"Commands: {explain.command_count}")

# Create a named snapshot
snapshot_id = db.create_snapshot("before-migration")

# List all snapshots
for snap in db.list_snapshots():
    print(snap.snapshot_id, snap.name, snap.created_at, snap.version_hwm)

# Restore to a snapshot
db.restore_snapshot(snapshot_id)

session = db.session()
session.pin_to_version(snapshot_id)

# All queries now read from the snapshot
result = session.query("MATCH (n:Person) RETURN n")

# Or pin to a timestamp (seconds since epoch)
session.pin_to_timestamp(1700000000.0)

# Unpin and return to latest
session.refresh()

snapshot_id = await db.create_snapshot("pre-migration")
snapshots = await db.list_snapshots()
await db.restore_snapshot(snapshot_id)

import uni_db

try:
    session = db.session()
    result = session.query("INVALID CYPHER")
except uni_db.UniParseError as e:
    print(f"Parse error: {e}")
except uni_db.UniQueryError as e:
    print(f"Query error: {e}")
except uni_db.UniError as e:
    print(f"Database error: {e}")

async with uni_db.AsyncUni.open("./graph") as db:
    session = db.session()

    # Reads
    result = await session.query("MATCH (n:Person) RETURN n.name AS name")

    # Writes
    tx = await session.tx()
    await tx.execute("CREATE (:Person {name: 'Dave', age: 35})")
    await tx.commit()

    # Locy
    result = await session.locy(program)

    # Xervo
    xervo = db.xervo()
    vectors = await xervo.embed("embed/default", ["hello world"])