Graph store interface

cvcdocdb.graph_store defines GraphStore, the abstract interface both cvcdocdb.networkx_graph and cvcdocdb.neo4j_graph implement. Code written against this interface (insert/update/delete, FK validation, cascade-delete strategies, bulk import, schema introspection, vector indexing) works unchanged against either backend.

Note

Neo4jGraph does not currently subclass GraphStore directly (it duck-types the same interface) — see the abstract methods below for the full contract either backend is expected to satisfy.

See also cvcdocdb.networkx_graph and cvcdocdb.neo4j_graph for the concrete backends, and cvcdocdb.migration for a generic tool that copies an entire graph from one GraphStore implementation to another using only this interface.

Abstract graph store interface for the DRM model.

Both Neo4jGraph and NetworkXGraph implement this ABC so that code in entities.py and elsewhere can accept any graph store without knowing the backing technology.

Usage:

from cvcdocdb.graph_store import GraphStore
from cvcdocdb.networkx_graph import NetworkXGraph

def process(store: GraphStore) -> None:
    node = Node(pk={"id": 1}, main_label="Test")
    store.insertNode(node)
    store.close()

process(NetworkXGraph())

class cvcdocdb.graph_store.GraphStore

Bases: ABC

Abstract interface for graph-backed stores (Neo4j, NetworkX, etc.).

Every concrete implementation must support the same core operations: insert, update, replace, delete nodes and relations, bulk import, and existence checks.

Subclasses must implement all abstract methods.

abstractmethod insertNode(node, insert_parent=True, update=False, replace=False, **kwargs)

Insert a node into the graph store.

Parameters:

• node (Node) – The node to insert.

• insert_parent (bool) – If the node is a WeakNode, insert its parent first. Defaults to True.

• update (bool) – If True, MERGE the node and update attributes without deleting it. Defaults to False.

• replace (bool) – If True and the node already exists, delete it (with detach) and create a fresh one. Defaults to False.

• **kwargs (Any) – Additional implementation-specific parameters.

Returns:

The internal node identifier assigned by the store.

Return type:

int | str

abstractmethod insertRelation(rel, update=False, replace=False, **kwargs)

Insert a directed relation (edge) between two nodes.

Parameters:

• rel (Relation) – The relation to insert.

• update (bool) – If True, MERGE the relation and update attributes.

• replace (bool) – If True and the relation already exists, delete it and create a fresh one.

• **kwargs (Any) – Additional implementation-specific parameters.

Returns:

The internal relation identifier.

Return type:

int | Tuple[int, int, str]

abstractmethod deleteNode(node, propagation=False, detach=False, on_delete='cascade')

Delete a node from the graph store.

Parameters:

• node (Node) – The node to delete.

• propagation (bool) – If True, recursively delete child nodes linked via edges with _propagate=TRUE.

• detach (bool) – If True, delete the node and all connected edges.

• on_delete (str) – Deletion strategy — "cascade", "restrict", or "set_null". Defaults to "cascade".

Returns:

True if the node was deleted, False otherwise.

Return type:

bool

abstractmethod checkNode(node, **kwargs)

Check if a node exists in the graph store.

Parameters:

• node (Node) – The node to look up.

• **kwargs (Any) – Additional implementation-specific parameters.

Returns:

The internal node id if found, None otherwise.

Return type:

int | None

abstractmethod create(migration, update=False, replace=False)

Bulk import nodes and relations.

Parameters:

• migration (Tuple[List, List]) – A tuple (node_list, relation_list).

• update (bool) – Passed to insertNode / insertRelation.

• replace (bool) – Passed to insertNode / insertRelation.

Return type:

None

abstractmethod close()

Release resources and clear the graph store.

Return type:

None

query(filter_dict=None, projection=None, sort=None, limit_val=None, params=None)

Query nodes using MongoDB-style dictionary filters (NetworkX) or execute raw Cypher (Neo4j).

Hybrid API — the first argument accepts either:

• dict — MongoDB-style filter matching against main_label and node attributes. Supports operators $eq, $ne, $gt, $gte, $lt, $lte, $in, $nin, $exists, $regex.

• str — a Cypher query string (MATCH, CREATE, DELETE, SET, RETURN, ORDER BY, LIMIT, aggregations, parameter substitution via $name).

Parameters:

• filter_dict (Dict[str, Any] | str | None) – Filter dict for MongoDB-style queries, or Cypher string for Cypher-style queries.

• projection (Dict[str, int] | None) – Dict of fields to include (1) or exclude (0). Ignored for Cypher queries (use RETURN aliases).

• sort (Tuple[str, int] | None) – Tuple of (field_name, direction) where direction is 1 (ascending) or -1 (descending). Ignored for Cypher queries (use ORDER BY).

• limit_val (int | None) – Maximum number of results to return. Ignored for Cypher queries (use LIMIT).

• params (Dict[str, Any] | None) – Optional parameter dict for $name substitution in Cypher queries. Ignored for MongoDB-style queries.

Returns:

A list of dicts, one per matching node (MongoDB-style) or one per result record (Cypher-style).

Raises:

NotImplementedError – If the backend does not support queries.

Return type:

List[Dict[str, Any]]

count(filter_dict=None)

Count nodes matching the filter dict.

Parameters:

filter_dict (Dict[str, Any] | None) – Dict of field/value pairs to match.

Returns:

Integer count of matching nodes.

Return type:

int

enable_vector_index(property_name, dimensions, space='cosine', **kwargs)

Enable vector indexing for one property if the backend supports it.

Backends without ANN/vector support should keep the default behavior and raise NotImplementedError.

Parameters:

• property_name (str)

• dimensions (int)

• space (str)

• kwargs (Any)

Return type:

None

query_vector_index(property_name, vector, top_k=10)

Query nearest nodes for a vector if the backend supports it.

Parameters:

• property_name (str)

• vector (List[float] | Tuple[float, ...])

• top_k (int)

Return type:

List[Tuple[int, float]]

list_vector_indexes()

Return metadata for every enabled vector index, as [{"property_name": ..., "dimensions": ..., "space": ...}, ...].

Used by cvcdocdb.migration.migrate() to recreate vector indexes on a target backend. Backends without vector index support (the default) return an empty list.

Return type:

List[Dict[str, Any]]

get_node(node_id)

Retrieve a node by its internal id.

Default implementation returns None. Subclasses may override.

Parameters:

node_id (int) – The internal node id.

Returns:

A Node instance, or None.

Return type:

Node | None

get_node_ids()

Return all internal node ids in the graph.

Returns the backend-specific internal ids (Neo4j id(n), NetworkX node keys). These are opaque identifiers and must not be compared with primary keys.

Default implementation returns an empty list.

Return type:

List[int]

get_node_pks()

Return all primary keys of nodes in the graph.

Each element is a dict with main_label and pk keys matching the Node interface.

Default implementation returns an empty list.

Return type:

List[Dict[str, Any]]

get_edges()

Return all edges as (src_node_id, dst_node_id, rel_type) tuples.

The node ids match what get_node_ids() returns (backend- specific internal ids).

Default implementation returns an empty list.

Return type:

List[Tuple[int, int, str]]

get_node_attrs(node_id)

Return attributes stored for a node.

Default implementation returns None.

Parameters:

node_id (int)

Return type:

Dict[str, Any] | None

get_edge_attrs(u, v, key)

Return attributes stored for an edge.

Default implementation returns None.

Parameters:

• u (int)

• v (int)

• key (str)

Return type:

Dict[str, Any] | None

get_dependency_value(node_id, relation_type)

Return the name primary key of the single node reachable from node_id via an outgoing relation_type edge.

Used to resolve be_value_properties (see Individu): a property such as nom is materialised at insert time as a separate Atribut/Valor node connected via a NOM edge instead of being stored directly on the node. This is a targeted, single-hop lookup — it must not scan the whole graph.

Parameters:

• node_id (int) – Internal id of the source node.

• relation_type (str) – Relation type to follow (e.g. "NOM").

Returns:

The connected node’s name pk value, or None if there is no such edge.

Return type:

str | None

Default implementation returns None.

debug()

Return a human-readable snapshot of the graph state.

Default implementation returns an empty dict.

Return type:

Dict[str, Any]

print_debug()

Print a formatted snapshot of the graph state to stdout.

Return type:

None

create_group(strong_node, weak_nodes=None, weak_relations=None, **kwargs)

Create a strong node together with its WeakNodes and WeakRelations atomically.

All nodes and relations are inserted in a single isolated transaction. If any step fails the entire group is rolled back so the graph is never left in a partially-created state.

Parameters:

• strong_node (Node) – The root (non-weak) node of the group.

• weak_nodes (List[Node] | None) – Optional list of WeakNode instances belonging to this group.

• weak_relations (List[Relation] | None) – Optional list of WeakRelation instances that connect the strong node (or other nodes) to the WeakNodes.

• kwargs (Any)

Returns:

The internal id of the strong_node.

Raises:

RuntimeError – If any part of the group creation fails — the transaction is rolled back automatically.

Return type:

int

init_propagation(background=False, progress_callback=None)

Scan the backend graph and initialize propagation properties on nodes and edges that are missing them.

This method inspects every node and edge, determines whether it participates in a WeakNode / WeakRelation hierarchy, and sets the corresponding _propagate, is_weak, parent_relation, and _dependencies properties.

Lazy + background approach: the first call runs synchronously and marks the graph as initialized. Subsequent calls return False immediately. For large graphs set background=True to run the scan in a background thread (the method still returns True once initialization completes).

Parameters:

• background (bool) – If True, run the scan in a background thread. The method returns immediately after starting the thread.

• progress_callback (callable | None) – Optional callable (processed, total) called periodically during the scan.

Returns:

True if initialization was performed (or is running in the background), False if already initialized.

Return type:

bool

abstractmethod schema_yaml(db_name)

Introspect the database and return a YAML schema description.

The YAML contains labels, properties, relationship types, counts, and Python class names derived from the schema.

Parameters:

db_name (str) – Human-readable database name (e.g. "got").

Returns:

A YAML string suitable for code generation.

Return type:

str

get_subdocuments(strong_node)

Return all subdocuments (WeakNodes) reachable from a strong node through _propagate edges.

This method follows edges in their declared direction (parent → child) and returns every descendant WeakNode as a dict with label, pk, and properties keys.

Parameters:

strong_node (Node) – The root (non-weak) node whose subdocuments should be retrieved.

Returns:

{
    "label": "Section",
    "pk": {"section": "intro"},
    "properties": {"title": "Introduction", ...},
}

Return type:

A list of dicts, one per subdocument

Raises:

NotImplementedError – If the backend does not support subdocument queries.