Skip to content
Drivine

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.
Section titled “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

Section titled “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

Section titled “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.