Drivine for Python

A graph database client for Python, beside Drivine for TypeScript and drivine4j for the JVM. Async first, typed, and fast: one statement per round trip, no session cache, no lazy loading.

Status: early. On Neo4j, Memgraph, FalkorDB and Oracle: the low-level PersistenceManager, transactions, and the object manager's loading, saving, batches, patches, edges and deletes of fragments and views, the query DSL with typed references, grouped reads, schema management, and vector and full-text search, async and sync. Neptune connects the same way, for the PersistenceManager. See the design for how it came about.

pip install "drivine[neo4j]"      # Neo4j, Memgraph, Neptune
pip install "drivine[falkordb]"   # FalkorDB
pip install "drivine[oracle]"     # Oracle 23ai and 26ai, through python-oracledb in thin mode
pip install "drivine[pilgrim]"    # DrivineAssembly, for pi-pilgrim; optional

Python 3.11 or newer.

Quick start

Describe the database in the environment:

DATABASE_TYPE=NEO4J
DATABASE_HOST=localhost
DATABASE_USER=neo4j
DATABASE_PASSWORD=…

Declare a fragment for each node a use case reads, and a view that composes them:

from typing import Annotated

from drivine import GraphView, NodeFragment, NodeId, Relationship, RelationshipFragment, Root, TargetNode


class Person(NodeFragment, labels=["Person"]):
    uuid: Annotated[str, NodeId()]
    name: str
    bio: str | None = None


class Organization(NodeFragment, labels=["Organization"]):
    uuid: Annotated[str, NodeId()]
    name: str


class WorkHistory(RelationshipFragment):
    role: str
    organization: Annotated[Organization, TargetNode()]


class PersonCareer(GraphView):
    person: Annotated[Person, Root()]
    employment_history: Annotated[list[WorkHistory], Relationship("WORKS_FOR")] = []

Then load and save, each in one statement:

import asyncio

from drivine import DatabaseRegistry, PersistenceManagerFactory, StatelessGraphObjectManager, refs


async def main() -> None:
    registry = DatabaseRegistry()
    registry.build_or_resolve_from_env()
    gom = StatelessGraphObjectManager(PersistenceManagerFactory(registry).get())
    try:
        await gom.save(
            PersonCareer(
                person=Person(uuid="p1", name="Ada", bio="Design Lead"),
                employment_history=[WorkHistory(role="Lead", organization=Organization(uuid="o1", name="Studio"))],
            )
        )
        q = refs(PersonCareer)
        for career in await gom.load_all(PersonCareer, where=q.person.bio.contains("Lead")):
            print(career.person.name, [it.organization.name for it in career.employment_history])
    finally:
        await registry.close()


asyncio.run(main())

From here: the persistence manager for Cypher of your own, transactions, queries and typed references, the sync API, and DrivineAssembly to wire it into an application.

Connecting

Describe a database in the environment, with the same variables as TypeScript's Drivine and drivine4j, so one .env serves an application in any of them:

DATABASE_TYPE=NEO4J
DATABASE_HOST=localhost
DATABASE_PORT=7687
DATABASE_USER=neo4j
DATABASE_PASSWORD=…

A second database takes a prefix, ANALYTICS_DATABASE_TYPE and so on. Then register them:

from drivine import DatabaseRegistry, PersistenceManagerFactory

registry = DatabaseRegistry()
registry.build_or_resolve_from_env()              # "default"
registry.build_or_resolve_from_env("analytics")   # ANALYTICS_DATABASE_*

pm = PersistenceManagerFactory(registry).get()    # the default database
...
await registry.close()                            # when the application stops

On FalkorDB, DATABASE_NAME names the graph, graph by default.

On Oracle, DATABASE_NAME is the service (FREEPDB1, say), and DATABASE_DEFAULT_GRAPH_PATH names the graph, drivine by default. A graph is drivine4j's tables, <graph>_node, <graph>_edge and <graph>_schema, and a SQL property graph over them, made on the first connection when they aren't there, so one store serves the JVM, Node and Python. The object manager writes SQL for it, and the query DSL, grouped reads, search and schema management work as on the graph engines. A raw where is SQL over the node's row, n: its labels, props (a JSON object keyed by stored name) and stamp, say JSON_VALUE(n.props, '$.name') = $name. Your own statements are SQL too.

Or in code:

registry.builder().database_type(DatabaseType.MEMGRAPH).host("localhost").register("graph")

Drivine reads os.environ, and doesn't load .env files itself: use python-dotenv, pydantic-settings, or your platform's way.

Statements

The PersistenceManager runs your statements and maps their results:

from drivine import QuerySpecification

spec = (QuerySpecification[Person]()
    .with_statement("MATCH (p:Person {city: $city}) RETURN properties(p)")
    .bind({"city": "Sydney"})    # named parameters, or a list for $1, $2, …
    .transform(Person)           # validated with Pydantic: a model, dataclass, TypedDict…
    .filter(lambda it: it.age >= 18)
    .limit(10))

people = await pm.query(spec)
person = await pm.get_one(spec)          # exactly one, or it raises
maybe = await pm.maybe_get_one(spec)     # one or None
await pm.execute(spec)                   # no results

• A node or relationship comes back as its properties, and Neo4j's dates and times as the standard library's.
• A row of one column is its value, and a row of several is a list.
• A failed statement raises DrivineError, which holds the specification, and the database's error as its cause.

Cursors

A cursor streams a large result, reading it from the driver batch_size rows at a time:

spec = CursorSpecification[Route]("MATCH (r:Route) RETURN properties(r)").batch_size(500).transform(Route)
async with await pm.open_cursor(spec) as cursor:
    async for route in cursor:
        ...

It works in a transaction or outside one, where it has a connection of its own until it is closed.

Statements in files

from drivine import cypher_file

routes_between = cypher_file(__package__, "routes_between.cypher")

Read with importlib.resources, so it works from an installed wheel as well as a source tree.

Transactions

from drivine import transactional


class IssueService:
    def __init__(self, issues: IssueRepository, audit: AuditLog):
        self.issues = issues
        self.audit = audit

    @transactional()           # join the current transaction, or start one
    async def raise_issue(self, issue: Issue) -> Issue:
        saved = await self.issues.save(issue)
        await self.audit.record(saved)   # same transaction
        return saved

Or for a block:

from drivine import transaction

async with transaction():
    await pm.execute(first)
    await pm.execute(second)

async with transaction(rollback=True):   # a test that leaves nothing behind
    ...

• The outermost call or block owns the transaction. It commits when it ends, and rolls back when an exception leaves it. Inner ones join it.
• A failure anywhere rolls the whole transaction back. An exception that leaves any @transactional call, even one the caller catches, marks the transaction rollback-only. It then rolls back, and its end raises TransactionRolledBackError rather than committing what the inner call half did. no_rollback_for=(NotFound,) exempts an exception.
• One transaction spans every database it touches. It opens a connection to each, and begins its transaction, the first time it runs a statement there. Commits are one after another, not two-phase.
• A @transactional method that never reaches Drivine costs nothing: no connection is opened. Code with fakes in its tests, or with Drivine taken out, runs as written.
• Outside a transaction, each statement runs on its own and commits. Drivine never starts a transaction you didn't ask for.
• It works on calls to self, since it wraps the function itself, not a proxy.

FalkorDB

FalkorDB has no multi-statement transactions: each statement commits as it runs. With DATABASE_TRANSACTION_MODE=WARN, the default, a transaction runs statement by statement and a rollback logs that nothing was rolled back; a save_all of several statements does the same. With STRICT, a transaction is refused at its first statement, and such a batch raises NonAtomicBatchError, so no code path can rely on atomicity it doesn't have.

Tasks share their transaction, one statement at a time

The current transaction lives in a context variable, so asyncio carries it across every await and into the tasks started inside it:

@transactional()
async def import_all(self, people: list[Person]) -> None:
    async with asyncio.TaskGroup() as tasks:
        for person in people:
            tasks.create_task(self.people.add(person))   # all in this transaction

A database transaction runs one statement at a time, so Drivine queues them, in the order they were issued. A cursor takes its turn for each row it reads, so other statements run between its rows. A statement that fails dooms the transaction, and the statements after it raise TransactionRolledBackError at once.

The queue keeps concurrency correct, not parallel. For a parallel import, give each task its own transaction.

Fragments and views

Map a node with a NodeFragment, and compose fragments and relationships into a GraphView, which loads in one statement:

from typing import Annotated

from drivine import GraphView, NodeFragment, NodeId, Relationship, RelationshipFragment, Root, TargetNode


class Person(NodeFragment, labels=["Person"]):
    uuid: Annotated[str, NodeId()]
    name: str
    bio: str | None = None


class Organization(NodeFragment, labels=["Organization"]):
    uuid: Annotated[str, NodeId()]
    name: str


class WorkHistory(RelationshipFragment):     # a relationship's properties, and the node it leads to
    start_date: str
    role: str
    target: Annotated[Organization, TargetNode()]


class PersonCareer(GraphView):
    person: Annotated[Person, Root()]
    employment_history: Annotated[list[WorkHistory], Relationship("WORKS_FOR")] = []

gom = drivine.graph_object_manager()        # from DrivineAssembly, or StatelessGraphObjectManager(pm)

career = await gom.load(PersonCareer, uuid)
leads = await gom.load_all(PersonCareer, "n.bio CONTAINS $text", {"text": "Lead"})   # Cypher over the root, n
total = await gom.count(Person)

Queries

A query is built from a class's references, which mirror its fields:

from drivine import any_of, not_, refs

q = refs(PersonCareer)
leads = await gom.load_all(
    PersonCareer,
    where=q.person.bio.contains("Lead"),
    order_by=q.person.name.asc(),
    limit=20,
)

await gom.load_all(PersonCareer, where=[                                 # a list: all must hold
    q.employment_history.target.name.eq("Acme"),                         # across a relationship
    any_of(q.person.name.eq("Ada"), q.person.name.eq("Grace")),
    not_(q.person.bio.is_null()),
])

# one relationship that meets both, rather than each met by any
q.employment_history.any(lambda it: [it.role.eq("Engineer"), it.target.name.eq("Acme")])
q.employment_history.none(lambda it: it.role.eq("Lead"))

• Operators: eq, neq, gt, gte, lt, lte, in_, not_in, contains, starts_with, ends_with, matches, contains_ignore_case, eq_ignore_case, has_item, is_null, is_not_null. On a node: instance_of(Type), has_any_label(...), and property(name) or predicate(name, operator, value) for a property named at run time by its stored name.
• Fields go by their Python names, q.start_date, and Drivine writes the stored names. A field the class doesn't have fails as the query is written, naming it and the class. q.x == 1 raises: write q.x.eq(1).
• Order and pages: order_by takes terms, asc() or desc(), then nulls_first() or nulls_last(); limit and skip apply in the database. An order on a list relationship's property sorts that list in each object.
• Keyset pages: seek=[q.released.after(last.released), q.uuid.after(last.uuid)], one value for each term of order_by, ending with a unique key.
• An order by an expression: plus, minus, times, div, negate, exp, floor, coalesce, case_of with above, below, at_least, at_most, equal_to, has_value and the rest, all computed in the database.
• count, delete, delete_all and patch take conditions too.

Typed references

refs is untyped: Python can't derive a reference type for each field from a class, as TypeScript does. A misspelt field still fails, but when the query is written. To have pyright or mypy check each query, and an editor complete its fields, write typed references for your models:

python -m drivine.stubs myapp.models                 # writes myapp/models_refs.py and models_refs.pyi
python -m drivine.stubs myapp.models --check         # in CI: fails when the models changed since

Then import refs from the module written:

from myapp.models_refs import refs

q = refs(PersonCareer)                  # PersonCareerRefs
q.person.bio.contains("Lead")           # StringRef
q.person.born.gt("1815")                # error: born is ComparableRef[int]
q.person.nmae                           # error: PersonNodes has no attribute nmae

Each field's reference follows from its type: text is a StringRef; numbers, dates and times are a ComparableRef; booleans and enums a ValueRef; lists and sets a ListRef; a property bag gives any entry; and a relationship leads to the fields of the nodes it reaches, with any and none. At run time the module is drivine's refs, so nothing changes but what the checker knows. Several modules can share one: python -m drivine.stubs myapp.models myapp.billing --out myapp/query_refs.py.

Grouped reads

Group a fragment's nodes by keys, with numbers about each group, in one statement:

from drivine import avg, count, hop, key, path_key, refs

q = refs(Review)
top = await gom.groups.load(
    Review,
    keys={"page": key(q.page)},                                  # or path_key(Page, [hop("REVIEWS")]): the node
    aggregates={"votes": count(q.helpful.eq(True)), "rating": avg(q.rating)},
    where=q.published.is_not_null(),
    having=[("votes", "gte", 3)],
    order_by=[("votes", "desc")],
    limit=10,
)
# [{"page": "/pricing", "votes": 12, "rating": 4.5}, ...]

await gom.groups.count_by(Review, "page")                       # {"/pricing": 14, ...}

Aggregates are count, sum_, avg, min_ and max_, each with an optional filter. Ties in an order are broken by the keys, so a page of groups is the same every time.

• Models are Pydantic. A loaded fragment is validated, so a bad value fails at the load.
• The annotation says what a relationship holds: a fragment, a relationship fragment or a view, one or a list. A single relationship typed without None is required, so only roots that have it load; Person | None is optional. A list is required with Relationship(..., required=True).
• Properties are stored in camel case, as TypeScript and JVM models store them: start_date as startDate. GraphProperty("name") names one, and property_names="as_written" keeps a class's fields as they are written.
• A subclass is loaded as itself. A fragment's labels are inherited, and a node loads as the most specific subclass whose labels it carries.
• Recursion: Relationship("HAS_LOCATION", max_depth=3) on a list of the view's own class expands three levels; depth={"sub_locations": 5} overrides it for one load.
• More markers: GraphPath([hop(...), ...]) for the nodes at the end of several hops, Count and Aggregate computed in the query, SortedBy, ReadOnly, Transient, NodeStamp, NodeLabels, PropertyBag, JsonPacked, NullAsDefault and EmptyWhenAbsent, as in TypeScript's Drivine.

The statements are TypeScript's, character for character: the tests check them against those TypeScript's Drivine writes for the same models.

Saving

saved = await gom.save(career)                                       # fields that aren't None; relationships added
await gom.save(career, replace=["employment_history"])               # this list is the whole list
await gom.save(person, only=["name"])                                # or except_=["embedding"]
await gom.save(person, null_policy=NullPolicy.CLEAR)                 # a field assigned None clears its property
created = await gom.create(person)                                   # only if absent; None when it was there
updated = await gom.update(Person, uuid, lambda it: it.model_copy(update={"name": "Ada"}))
await gom.delete(PersonCareer, uuid, cascade=CascadeType.DELETE_ORPHAN)
await gom.delete_all(Person, "n.name = $name", {"name": "Ada"})

await gom.save_all(chunks)                                           # fragments in one statement per 1,000
result = await gom.patch(Session, session_id, set={"last_activity_at": now},
                         where="n.lastActivityAt IS NULL OR n.lastActivityAt < $now", bindings={"now": now})
results = await gom.patch_all(chunks, only=["embedding"])            # named fields, where each node is there

await gom.edges.relate(node_ref(Issue, a), node_ref(Person, b), "ASSIGNED_TO")
await gom.edges.unrelate(node_ref(Issue, a), node_ref(Person, b), "ASSIGNED_TO")
people = await gom.edges.load_related(node_ref(Issue, a), "ASSIGNED_TO", Person)

• A save is one statement, applied whole or not at all: the root, the relationships it drops, each related node and the relationship to it.
• A save adds and removes nothing, unless a field is named in replace. removed_targets=DELETE_UNREFERENCED also deletes a removed node that nothing else refers to.
• Stamps refuse a stale save. On a class with a NodeStamp field, a save of an object loaded before another writer changed the node raises StaleObjectError and writes nothing. Use the object a save returns: it carries the new stamp. update loads, applies your change and saves what it altered, again if the node changed meanwhile.
• save_all runs its statements in one transaction, or in yours, so a failure leaves nothing written.
• patch writes named fields of one node by id, if it meets a condition, without loading it, and says whether it found the node, applied the patch and changed anything. It never makes a node.
• Edges relate and unrelate stored nodes by id, marking both ends' stamps.
• Outside a transaction, a save the engine turns away because another writer held the node is run again after a pause. Inside one, it joins the transaction, and fails for its owner to decide.

The store format, stamps included, is TypeScript's and drivine4j's, so the three can share a store.

Schema

Declare indexes and constraints on the fragments, and hold them on every database at start-up:

class Chunk(NodeFragment, labels=["Chunk"], schema=[RangeIndex(["section_id", "position"])]):
    uuid: Annotated[str, NodeId(), Unique()]
    section_id: str
    position: int
    text: Annotated[str, FullTextIndex()]
    embedding: Annotated[list[float] | None, VectorIndex(similarity="COSINE")] = None


class AppDrivine(DrivineAssembly):
    def schema_options(self) -> SchemaOptions:
        return SchemaOptions([SchemaCatalog.from_fragments(Chunk, dimensions=lambda label, prop: 1536)])

await AssemblyHolder.shared(AppDrivine).start()      # makes what is missing; idempotent

Without pi-pilgrim, enforce it yourself: await SchemaManager(registry, factory, options).enforce().

• ensure is idempotent. An item that is there is left; one there with another shape (a vector index's dimensions, say) is reported as DRIFT and left, unless recreate_on_drift. A uniqueness constraint the data already breaks is a VIOLATION, with a sample of the values. mode="FAIL_FAST", the default, raises on either; "WARN" logs.
• Versions. catalog.with_version("2") recreates the catalog's items once, the first time it is enforced. Items an owner applied before and no longer declares are reported, never dropped. Each owner's version and inventory are kept in a _DrivineSchema node, as drivine4j and TypeScript keep them.
• By hand: gom.indexes.ensure(range_index("Session", ["lastActivityAt", "sessionId"])), gom.indexes.recreate, gom.indexes.drop, gom.constraints.ensure(uniqueness_constraint("Person", "email")), and list.
• Each engine's DDL is written for it: named items on Neo4j, Memgraph's text and vector indexes, FalkorDB's index per property and its constraint commands. Vector index tuning (hnsw_m, hnsw_ef_construction, and an engine's own options) is passed where the engine takes it, and reported where it doesn't.

Search

A fragment's VectorIndex field drives load_nearest, and its FullTextIndex drives load_matching. Either takes a fragment or a view, searched by its root, and gives each hit with its score, best first:

hits = await gom.load_nearest(Chunk, embedding, top_k=5, where=refs(Chunk).section_id.eq(section))
for hit in hits:
    print(hit.score, hit.value.text)

found = await gom.load_matching(Chunk, "graph AND database", top_k=10, threshold=0.5)

• One scale on every engine. A nearest score is a similarity, higher nearer: (1 + cos) / 2 for cosine, 1 / (1 + d²) for euclidean. A matching score is scaled so that the best hit is 1.
• Filters apply after the search: threshold, where, and a view's required relationships. So a search can return fewer than top_k; search_k asks the index for more, and top_k trims after the filter.
• property names the field to search when a fragment declares more than one index; partition_label searches the vector index of another label. Full-text text is the engine's own query language, passed as it is.

Index advice

An ordered query with no range index over exactly its ordered properties scans and sorts. Drivine says so, once for each label and properties, naming the index to make. gom.index_advice = IndexAdvicePolicy.FAIL raises instead, for development and CI (or override index_advice() on the assembly); OFF says nothing. The index list is read once: gom.index_advisor.refresh() after making indexes as the application runs.

Store identity

await gom.store_identity() gives the store's stable id, minted in the store on first sight: two connections see the same id when, and only when, they look at the same data.

Sync

drivine.sync is the same API without async and await, for scripts, notebooks, Django and Flask:

from drivine.sync import DatabaseRegistry, PersistenceManagerFactory, StatelessGraphObjectManager, transactional

registry = DatabaseRegistry()
registry.build_or_resolve_from_env()
gom = StatelessGraphObjectManager(PersistenceManagerFactory(registry).get())


@transactional
def promote(uuid: str) -> None:
    person = gom.load(Person, uuid)
    gom.save(person.model_copy(update={"bio": "Lead"}))

• Each driver's own sync client, not an event loop behind the scenes: it costs nothing extra, and works where a loop is already running, as in Jupyter.
• The same models, query DSL and backends. A class is one class in both APIs; only what does I/O is sync.
• Each thread has its own transaction, as asyncio gives each task's context its own.
• @transactional from drivine.sync takes a plain function, and drivine's takes an async one.

It is written from the async source by scripts/unasync.py, as the Neo4j driver writes its sync API, and tested by the same tests, written alike.

With pi-pilgrim

DrivineAssembly is a pi-pilgrim assembly holding the registry and the managers, for your own assembly to use. It needs pip install "drivine[pilgrim]":

from pipilgrim import AssemblyHolder, PilgrimAssembly
from drivine.drivine_assembly import DrivineAssembly      # drivine.sync.drivine_assembly for the sync API


class AppAssembly(PilgrimAssembly):
    def __init__(self):
        self.drivine = AssemblyHolder.shared(DrivineAssembly)
        super().__init__()

    def route_repository(self) -> RouteRepository:
        return self.shared(lambda: RouteRepository(self.drivine.persistence_manager()))

    def report_repository(self) -> ReportRepository:
        return self.shared(lambda: ReportRepository(self.drivine.persistence_manager("analytics")))

• It registers the database the environment describes. To register yours in code, subclass it and override register_databases(registry).
• Assembled injects a manager into a class something else makes: pm: PersistenceManager = Assembled(assembly=DrivineAssembly).
• Call await assembly.drivine.close() when the application stops, in FastAPI's lifespan, say.

Nothing else needs pi-pilgrim: import drivine doesn't load it, and the managers are plain classes.

The sample app

sample/ is the port of drivine4j's sample, as TypeScript's Drivine has it: an issue tracker and a guide's users, with fragments, views, relationship fragments, polymorphic users, property bags, stamps and search, and its tests on every engine. It is a Poetry project of its own, whose build writes its typed references.

Testing

pip install "drivine[testing]" adds a pytest plugin. With

[tool.pytest.ini_options]
drivine_test_containers = true

it starts a container for each datasource the environment configures, Neo4j, Memgraph, FalkorDB or Oracle, before the tests run, and points the datasource at it. Datasources of one type share a container. USE_LOCAL_NEO4J=true (or _MEMGRAPH, _FALKORDB, _ORACLE) leaves that type's datasources as they are, to test against your own instance, and NEO4J_TEST_IMAGE picks another image.

Drivine's own tests run on Oracle only with ORACLE_TESTS=true: its image is large, and the database wants two gigabytes.

Logging

Drivine logs through the standard logging module: drivine.statements logs each statement at DEBUG, with its session and how long it took. DRIVINE_LOG_LEVEL, as TypeScript's Drivine reads it, sets the level of the drivine logger when the application hasn't.

Building

uv sync
uv run pytest            # starts Neo4j, Memgraph and FalkorDB in Docker
uv run ruff check . && uv run pyright

drivine/sync and tests/sync_integration are written from the async source: change that, and run uv run python scripts/unasync.py. A unit test fails while they are out of date. drivine/sync/concurrency.py, the counterpart of drivine/concurrency.py, is the one sync module written by hand, and a test marked async_only (one that runs tasks concurrently) is left out of the sync tests.

License

Apache 2.0.