Metadata-Version: 2.5
Name: drivine
Version: 0.1.0
Summary: Graph database client for Python: Neo4j, Memgraph, FalkorDB, Neptune and Oracle.
Project-URL: Homepage, https://drivine.org
Project-URL: Repository, https://github.com/liberation-data/drivine-python
Author: Liberation Data
License-Expression: Apache-2.0
License-File: LICENSE
Keywords: cypher,database,falkordb,graph,memgraph,neo4j,neptune,ogm
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Topic :: Database
Classifier: Typing :: Typed
Requires-Python: >=3.11
Requires-Dist: pydantic>=2.7
Provides-Extra: falkordb
Requires-Dist: falkordb>=1.2; extra == 'falkordb'
Provides-Extra: neo4j
Requires-Dist: neo4j>=5.20; extra == 'neo4j'
Provides-Extra: oracle
Requires-Dist: oracledb>=3.0; extra == 'oracle'
Provides-Extra: pilgrim
Requires-Dist: pi-pilgrim>=0.2.0; extra == 'pilgrim'
Provides-Extra: testing
Requires-Dist: pytest>=8; extra == 'testing'
Requires-Dist: testcontainers>=4.8; extra == 'testing'
Description-Content-Type: text/markdown

# Drivine for Python

**A graph database client for Python**, beside [Drivine](https://github.com/liberation-data/drivine) for TypeScript
and [drivine4j](https://github.com/liberation-data/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](docs/design.md) 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:

```python
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:

```python
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](#statements) for Cypher of your own, [transactions](#transactions),
[queries](#queries) and [typed references](#typed-references), [the sync API](#sync), and
[DrivineAssembly](#with-pi-pilgrim) 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:

```python
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:

```python
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:

```python
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:

```python
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

```python
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

```python
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:

```python
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:

```python
@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:

```python
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")] = []
```

```python
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:

```python
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:

```python
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:

```python
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

```python
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:

```python
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:

```python
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:

```python
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](https://github.com/liberation-data/pi-pilgrim) assembly holding the registry
and the managers, for your own assembly to use. It needs `pip install "drivine[pilgrim]"`:

```python
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/`](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

```toml
[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.
