Using LoraDB in Python

Overview

lora-python is a PyO3 binding built with maturin. It ships a synchronous Database plus an asyncio-friendly AsyncDatabase wrapper for normal query execution, snapshots, and WAL-backed opens. Switching the core execute path between them is a one-line import change.

Installation / Setup

Requirements

Python 3.8+
For building from source: Rust toolchain (rustup) + maturin

Install

pip install lora-python

Creating a Client / Connection

from lora_python import Database

db = Database.create()

Database.create() and Database() do the same thing — the factory exists for API symmetry with AsyncDatabase.

Asyncio equivalent:

from lora_python import AsyncDatabase

db = await AsyncDatabase.create()

Running Your First Query

from lora_python import Database

db = Database.create()

db.execute("CREATE (:Person {name: 'Ada', born: 1815})")

result = db.execute("MATCH (p:Person) RETURN p.name AS name, p.born AS born")

print(result["rows"])
# [{'name': 'Ada', 'born': 1815}]

Examples

Minimal working example

Already shown above.

Parameterised query

result = db.execute(
    "MATCH (p:Person) WHERE p.name = $name RETURN p.name AS name",
    {"name": "Ada"},
)

Python values map to engine values automatically: int/float/bool/str/None and list/dict pass through. For temporal and spatial values, use the tagged helpers below.

Explain and profile

explain and profile are binding methods on Database, not Cypher keywords that you prepend to the query string. db.explain(...) returns the compiled plan without invoking the executor:

plan = db.explain(
    "MATCH (p:Person) WHERE p.name = $name RETURN p",
    {"name": "Ada"},
)

print(plan["shape"])             # "readOnly" or "mutating"
print(plan["result_columns"])    # ["p"]
print(plan["tree"]["operator"])  # top-level physical operator

Python uses snake_case keys for the explain/profile envelopes: result_columns, estimated_rows, total_elapsed_ns, and per_operator. details values in the plan tree are opaque human-readable strings; don't parse them as a stable machine contract.

Use db.profile(...) when you want the same plan plus runtime metrics:

profile = db.profile(
    "MATCH (p:Person) WHERE p.name = $name RETURN p",
    {"name": "Ada"},
)

print(profile["metrics"]["total_elapsed_ns"])
print(profile["metrics"]["total_rows"])
print(profile["metrics"]["mutated"])
print(profile["metrics"]["per_operator"])

profile executes the query

Mutating queries passed to profile produce the same side effects as execute. Use explain to inspect a mutating CREATE, MERGE, SET, DELETE, or REMOVE plan without changing the graph.

Both methods accept the same parameter values as execute, including tagged helper dicts for temporal, spatial, vector, and binary values. Graph dicts such as returned nodes are result values; for input, pass property values or typed helper values:

from lora_python import date, wgs84

params = {
    "since": date("1800-01-01"),
    "near": wgs84(4.89, 52.37),
    "radius": 5000.0,
}

plan = db.explain(
    """
    MATCH (c:City)
    WHERE c.founded >= $since
      AND geo.distance(c.location, $near) < $radius
    RETURN c.name AS name
    """,
    params,
)

profile = db.profile(
    """
    MATCH (c:City)
    WHERE c.founded >= $since
      AND geo.distance(c.location, $near) < $radius
    RETURN c.name AS name
    """,
    params,
)

Structured result handling

from lora_python import Database, is_node

db = Database.create()
db.execute("CREATE (:Person {name: 'Ada'})")

result = db.execute("MATCH (n:Person) RETURN n")

for row in result["rows"]:
    n = row["n"]
    if is_node(n):
        print(n["id"], n["labels"], n["properties"])

Available guards: is_node, is_relationship, is_path, is_point, is_temporal.

FastAPI route handler

from fastapi import FastAPI, HTTPException
from lora_python import AsyncDatabase, LoraQueryError

app = FastAPI()
db: AsyncDatabase  # initialised at startup

@app.on_event("startup")
async def _bootstrap():
    global db
    db = await AsyncDatabase.create()

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    try:
        res = await db.execute(
            "MATCH (u:User {id: $id}) RETURN u {.id, .handle, .tier} AS user",
            {"id": user_id},
        )
    except LoraQueryError as exc:
        raise HTTPException(status_code=400, detail=str(exc))
    rows = res["rows"]
    if not rows:
        raise HTTPException(status_code=404)
    return rows[0]["user"]

Works unchanged under Flask/Django/Litestar — swap the framework but keep the AsyncDatabase instance at module scope.

Handle errors

from lora_python import Database, LoraQueryError, InvalidParamsError

db = Database.create()

try:
    db.execute("BAD QUERY")
except LoraQueryError as exc:
    print("query failed:", exc)
except InvalidParamsError as exc:
    print("bad params:", exc)

LoraError is the common base class — catch it if you don't need to distinguish.

Sync vs async

Sync:

from lora_python import Database

db = Database.create()
db.execute("CREATE (:Node)")

Async:

import asyncio
from lora_python import AsyncDatabase

async def main():
    db = await AsyncDatabase.create()
    await db.execute("CREATE (:Person {name: 'Ada'})")
    result = await db.execute("MATCH (n:Person) RETURN n.name AS name")
    return result["rows"]

asyncio.run(main())

AsyncDatabase delegates to asyncio.to_thread so long queries don't block the event loop. For the core execute path, switching between sync and async code is a one-line import change.

Persisting your graph

LoraDB can save the in-memory graph to a single file and restore it later. Python has three persistence shapes:

Database.create() / Database() => in-memory
Database.create("app", {"database_dir": "./data"}) / Database("app", {"database_dir": "./data"}) => container-backed

Async follows the same rule:

await AsyncDatabase.create() => in-memory
await AsyncDatabase.create("app", {"database_dir": "./data"}) => container-backed
Database.open_wal("./wal", options) / await AsyncDatabase.open_wal("./wal", options) => explicit WAL with optional managed snapshots

from lora_python import Database

db = Database.create()         # in-memory
# db = Database.create("app", {"database_dir": "./data"})  # archive: ./data/app.loradb
db.execute("CREATE (:Person {name: 'Ada'})")

# Save everything to disk.
meta = db.save_snapshot("graph.bin")
print(meta["nodeCount"], meta["relationshipCount"])

# Restore into a fresh handle (in a new process, for example).
db = Database.create()
db.load_snapshot("graph.bin")

durable = Database.open_wal(
    "./data/wal",
    {
        "snapshot_dir": "./data/snapshots",
        "snapshot_every_commits": 1000,
        "snapshot_keep_old": 2,
        "snapshot_options": {
            "compression": {"format": "gzip", "level": 1},
        },
    },
)

AsyncDatabase exposes the same two methods as coroutines — the sync call runs on a worker thread via asyncio.to_thread, so large saves do not block the event loop:

import asyncio
from lora_python import AsyncDatabase

async def main():
    db = await AsyncDatabase.create()  # in-memory
    # db = await AsyncDatabase.create("app", {"database_dir": "./data"})  # archive: ./data/app.loradb
    await db.execute("CREATE (:Person {name: 'Ada'})")
    await db.save_snapshot("graph.bin")

    db2 = await AsyncDatabase.create()
    await db2.load_snapshot("graph.bin")

    durable = await AsyncDatabase.open_wal(
        "./data/async-wal",
        {"snapshot_dir": "./data/async-snapshots", "snapshot_every_commits": 1000},
    )
    await durable.close()

asyncio.run(main())

Both save and load process the whole graph. A crash between saves loses every mutation since the last save unless you opened the database with WAL.

Passing a database name and directory opens or creates an container-backed persistent database at <database_dir>/<name>.loradb. Reopening the same path replays committed writes before the handle is returned. open_wal opens a raw WAL directory; when snapshot_dir and snapshot_every_commits are set, the database writes managed checkpoint snapshots after that many committed transactions. Python does not expose WAL status, truncate, or sync-mode controls; use Rust or lora-server for those operator knobs. Call db.close() / await db.close() before reopening the same archive or WAL directory inside one process.

See the canonical Snapshots guide for the full metadata shape, atomic-rename guarantees, and boundaries, and WAL and checkpoints for the recovery model.

Common Patterns

Bulk insert from a list

rows = [{"id": i, "name": f"user-{i}"} for i in range(100)]

db.execute(
    "UNWIND $rows AS row CREATE (:User {id: row.id, name: row.name})",
    {"rows": rows},
)

See UNWIND.

Typed helpers

from lora_python import Database, date, duration, wgs84

db = Database.create()

db.execute(
    "CREATE (:Trip {when: $when, span: $span, origin: $origin})",
    {
        "when":   date("2026-05-01"),
        "span":   duration("PT90M"),
        "origin": wgs84(4.89, 52.37),
    },
)

Available helpers: date, time, localtime, datetime, localdatetime, duration, cartesian, cartesian_3d, wgs84, wgs84_3d.

Repository pattern

from lora_python import Database, LoraQueryError

class UserRepo:
    def __init__(self, db: Database):
        self._db = db

    def upsert(self, user_id: int, handle: str):
        self._db.execute(
            """
            MERGE (u:User {id: $id})
              ON CREATE SET u.created = temporal.timestamp()
              SET u.handle = $handle, u.updated = temporal.timestamp()
            """,
            {"id": user_id, "handle": handle},
        )

    def find_by_handle(self, handle: str):
        res = self._db.execute(
            "MATCH (u:User {handle: $handle}) RETURN u {.*} AS user",
            {"handle": handle},
        )
        rows = res["rows"]
        return rows[0]["user"] if rows else None

Other methods

db.clear()                        # drop all nodes + relationships
db.close()                        # release the native handle
db.node_count                     # int — property, not a method
db.relationship_count             # int — property

node_count and relationship_count are read-only properties. AsyncDatabase exposes the same count properties and an async close() method.

Error Handling

Class	When
`LoraError`	Base — catch if you don't need to distinguish
`LoraQueryError`	Parse / semantic / runtime query error
`InvalidParamsError`	A parameter couldn't be mapped to a `LoraValue`

Engine-level causes live in Troubleshooting.

Performance / Best Practices

Thread-safety. Database is safe to share across threads. Auto-commit reads can overlap on snapshots; write commits and explicit read-write transactions serialize. No Python-level locking needed.
GIL. Database.execute releases the GIL while Rust code runs, so other Python threads / asyncio tasks can progress. This is the real non-blocking mechanism — AsyncDatabase is a thin wrapper that uses it.
Integer precision. Python integers are arbitrary precision, so i64 values round-trip cleanly (unlike the JS bindings).
No cancellation. Once a query is dispatched it runs to completion — bound traversals and UNWIND sizes.
Parameters, not f-strings. Never interpolate user input into a query string.

Overview​

Installation / Setup​

Requirements​

Install​

Creating a Client / Connection​

Running Your First Query​

Examples​

Minimal working example​

Parameterised query​

Explain and profile​

Structured result handling​

FastAPI route handler​

Handle errors​

Sync vs async​

Persisting your graph​

Common Patterns​

Bulk insert from a list​

Typed helpers​

Repository pattern​

Other methods​

Error Handling​

Performance / Best Practices​

See also​