Skip to main content

agrag.graphdb.base.GraphStore

Bases: ABC

A graph database backend: schema, writes, and native vector search.

Functions:

close​

close() -> None

Release the backend connection.

connect​

connect() -> None

Open the backend connection and verify connectivity.

ensure_vector_index​

ensure_vector_index(*, label:str, vector_property:str, dimensions:int, distance:Distance) -> None

Create a native vector index if it does not exist.

Parameters:

  • label (str) – The node label to index.
  • vector_property (str) – The embedding property name.
  • dimensions (int) – The embedding dimension. If the index already exists with a different dimension, this raises.
  • distance (Distance) – The distance metric.

Raises:

  • EmbeddingDimensionMismatchError – The index already exists with a different dimension than dimensions.

execute_read​

execute_read(query:str, parameters:Mapping[str, Any] | None = None, *, timeout:float | None = None) -> list[dict[str, Any]]

Run a read transaction.

Parameters:

  • query (str) – The Cypher query to run.
  • parameters (Mapping[str, Any] | None) – The query parameters.
  • timeout (float | None) – Server-side transaction timeout in seconds. The database terminates the transaction when it runs longer. None uses the server's default timeout. Backends that cannot enforce a timeout ignore it.

Returns:

  • list[dict[str, Any]] – The result rows as dicts.

execute_write​

execute_write(query:str, parameters:Mapping[str, Any] | None = None) -> list[dict[str, Any]]

Run a write transaction.

Parameters:

  • query (str) – The Cypher query to run.
  • parameters (Mapping[str, Any] | None) – The query parameters.

Returns:

  • list[dict[str, Any]] – The result rows as dicts.

register_labels​

register_labels(labels:Sequence[str]) -> None

Mark labels as known, without writing anything.

setup_constraints()/setup_indexes() only cover labels this instance has already written (or that already exist live in the database) — both empty on a brand-new database. register_labels lets a caller holding a GraphSchema (Graph.open()) provision a fresh database fully before its first write.

Parameters:

  • labels (Sequence[str]) – The labels to register. Each must be a safe Cypher identifier.

Raises:

  • ValueError – Any label is not a safe identifier.

register_relation_types​

register_relation_types(types:Sequence[str]) -> None

Mark relationship types as known, without writing anything.

The relationship-type counterpart to register_labels — see its docstring for why this exists.

Parameters:

  • types (Sequence[str]) – The relationship types to register. Each must be a safe Cypher identifier.

Raises:

  • ValueError – Any type is not a safe identifier.

session​

session() -> AbstractAsyncContextManager[Any]

Open a session as an async context manager.

Returns:

  • AbstractAsyncContextManager[Any] – A context manager yielding a backend session.

setup_constraints​

setup_constraints() -> None

Create per-label and per-relation-type uniqueness constraints.

Constraints cover every node label and relationship type written through this instance or already present in the database, so a fresh instance can set up an existing database without first rewriting every record.

setup_indexes​

setup_indexes() -> None

Create per-label property indexes.

Covers every node label written through this instance or already present in the database, so a fresh instance can set up an existing database without first rewriting every record.

transaction​

transaction() -> AsyncIterator[GraphStoreTransaction]

Start an explicit transaction spanning multiple writes.

Every call through the yielded handle should join one backend transaction, committing as a whole on clean exit from the async with block and rolling back as a whole if the block raises. The default here simply yields self and gives no atomicity beyond what each individual call already provides; a backend that can offer real atomicity, such as Neo4jGraphStore, overrides this with a driver transaction.

Use this when a caller must guarantee several writes either all apply or none do, such as apply_merge's survivor upsert and alias claim.

Returns:

upsert_nodes​

upsert_nodes(label:str, nodes:Sequence[NodeRecord], *, batch_size:int = 256, pending_job_id:UUID | None = None) -> UpsertResult

Write or merge nodes, honoring each record's full label set.

Parameters:

  • label (str) – The label this batch is tracked under for constraint and index bookkeeping.
  • nodes (Sequence[NodeRecord]) – The node records to upsert. Each node's NodeRecord.labels names the full label set actually written to it, which may include labels beyond label.
  • batch_size (int) – Records per backend write call, applied within each distinct label set when nodes mixes more than one. Must be positive.
  • pending_job_id (UUID | None) – The in-flight Cutover Job's id. The store tags each node it creates, so retrieval skips it until the job commits. A node that already exists keeps its current pending state: a node tagged by another in-flight job stays tagged, and None does not commit it. None writes committed data.

Returns:

  • UpsertResult – The number written and one failure entry for each isolated record.

Raises:

  • ValueError – batch_size is not positive.

upsert_relations​

upsert_relations(relations:Sequence[RelationRecord], *, batch_size:int = 256, pending_job_id:UUID | None = None) -> UpsertResult

Write or merge relationships between existing nodes.

Parameters:

  • relations (Sequence[RelationRecord]) – The relation records to upsert.
  • batch_size (int) – Records per backend write call. Must be positive.
  • pending_job_id (UUID | None) – The in-flight Cutover Job's id. The store tags each relationship it creates, so retrieval skips it until the job commits. A relationship that already exists keeps its current pending state: one tagged by another in-flight job stays tagged, and None does not commit it. None writes committed data.

Returns:

  • UpsertResult – The number written and one failure entry for each isolated record.

Raises:

  • ValueError – batch_size is not positive.
vector_search(*, label:str, vector_property:str, query_vector:Sequence[float], limit:int = 10, filters:dict[str, Any] | None = None) -> list[VectorHit]

Search nodes by dense vector.

Parameters:

  • label (str) – The node label to search.
  • vector_property (str) – The embedding property name.
  • query_vector (Sequence[float]) – The dense query embedding.
  • limit (int) – Maximum number of hits.
  • filters (dict[str, Any] | None) – An optional flat-dict filter on node properties.

Returns:

  • list[VectorHit] – The matched hits, highest score first.