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.