NetworkX backend

cvcdocdb.networkx_graph provides the in-memory backend used for tests, examples, and lightweight local experimentation. It mirrors the public graph store API so tutorials can switch between backends with minimal code changes.

NetworkX-backed graph store — in-memory, no real Neo4j required.

class cvcdocdb.networkx_graph.NetworkXGraph(persistence_path=None)[source]

Bases: GraphStore

In-memory graph store using NetworkX.

Provides the same public interface as Neo4jGraph but stores nodes and edges in a NetworkX MultiDiGraph so tests can run without a real database.

Parameters:: persistence_path (Optional[str])

__init__(persistence_path=None)[source]

Parameters:: persistence_path (str | None)
Return type:: None

insertNode(node, insert_parent=True, update=False, replace=False, **kwargs)[source]

Insert a node into the graph.

Returns the internal node id assigned by the graph.

Parameters:

node (Node)
insert_parent (bool)
update (bool)
replace (bool)
kwargs (Any)

Return type:

int

insertRelation(rel, update=False, replace=False, **kwargs)[source]

Insert a relation (directed edge) between two nodes.

Mirrors Neo4jGraph.insertRelation semantics:

update=True: MERGE + SET — adds/updates attributes without deleting the relation.
update=False + replace=True: deletes the existing relation and creates a fresh one.
update=False + replace=False: creates a new relation. If one with the same src/dst/type already exists, raises RuntimeError (duplicate key).

Returns the internal edge identifier (u, v, key).

Parameters:

rel (Relation)
update (bool)
replace (bool)
kwargs (Any)

Return type:

int

deleteNode(node, propagation=False, detach=False, on_delete='cascade')[source]

Delete a node from the graph.

on_delete controls FK behavior:

"cascade" (default): ON DELETE CASCADE — delete connected edges
"restrict": ON DELETE RESTRICT — refuse if edges exist
"set_null": ON DELETE SET NULL — delete node, keep neighbors

Parameters:

node (Node)
propagation (bool)
detach (bool)
on_delete (str)

Return type:

bool

checkNode(node, **kwargs)[source]

Check if a node exists in the graph.

Looks up the node by its neo4j_id if set, otherwise searches by main_label and primary key.

Parameters:

node (Node) – The node to look up.
kwargs (Any)

Returns:

The internal node id if found, None otherwise.

Return type:

int | None

create(migration, update=False, replace=False)[source]

Bulk import nodes and relations.

Iterates over the node and relation lists, inserting each with the given update and replace flags.

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

get_node(node_id)[source]

Retrieve a node by its internal id.

Parameters:: node_id (int) – The internal node id.
Returns:: A Node instance with the stored attributes, or None if no node with that id exists.
Return type:: Node | None

get_node_ids()[source]

Return all internal node ids in the graph.

Return type:: List[int]

get_nodes()[source]

Alias for get_node_ids() for backward compatibility.

Return type:: List[int]

get_node_pks()[source]

Return all primary keys of nodes in the graph.

Return type:: List[Dict[str, Any]]

get_edges()[source]

Return all edges as (src_id, dst_id, rel_type) tuples.

Return type:: List[Tuple[int, int, str]]

get_node_attrs(node_id)[source]

Return attributes stored for a node.

Parameters:: node_id (int) – The internal node id.
Returns:: A dict of node attributes, or None if the node does not exist.
Return type:: Dict[str, Any] | None

get_edge_attrs(u, v, key)[source]

Return attributes stored for an edge.

Parameters:

u (int) – Source node id.
v (int) – Destination node id.
key (str) – Edge type / key.

Returns:

A dict of edge attributes, or None if the edge does not exist.

Return type:

Dict[str, Any] | None

get_dependency_value(node_id, relation_type)[source]

Return the name pk of the node reachable via a single outgoing relation_type edge from node_id.

Single hop over self._graph[node_id] — never scans the whole graph, regardless of graph size.

Parameters:

node_id (int)
relation_type (str)

Return type:

str | None

find_nodes_by_property(prop_name, value)[source]

Return node ids indexed by an exact property value match.

Parameters:

prop_name (str)
value (Any)

Return type:

List[int]

find_nodes(filters, match='all')[source]

Find node ids by indexed property filters.

Parameters:

filters (Dict[str, Any]) – Mapping of property name -> exact value.
match (str) – "all" (intersection) or "any" (union).

Return type:

List[int]

debug()[source]

Return a human-readable snapshot of the graph state.

Returns a dict with keys: - nodes: list of (id, label, pk) - edges: list of (src_id, dst_id, rel_type, attrs) - fk_index: dict mapping node_id -> list of (other_id, rel_type, direction)

Return type:: Dict[str, Any]

print_debug()[source]

Print a formatted snapshot of the graph state to stdout.

Return type:: None

query(filter_dict=None, projection=None, sort=None, limit_val=None, params=None)[source]

Query nodes using MongoDB-style filters or execute Cypher.

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.
sort (Tuple[str, int] | None) – Tuple of (field_name, direction) where direction is 1 (ascending) or -1 (descending). Ignored for Cypher queries.
limit_val (int | None) – Maximum number of results to return. Ignored for Cypher queries.
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).

Return type:

List[Dict[str, Any]]

Example

>>> # MongoDB-style
>>> graph.query({"age": {"$gt": 25}})
>>> # Cypher-style
>>> graph.query("MATCH (n:Person) RETURN n")

query_nodes(filter_dict=None)[source]

Return an NxQuery fluent builder.

Unlike query(), this returns a lazy, chainable query object instead of a list. Call .where(...), .limit(...), .count(), .to_list(), or .ids() to materialise.

Parameters:: filter_dict (Optional[Dict[str, Any]]) – Optional initial MongoDB-style filter.
Returns:: An NxQuery instance over all (or filtered) nodes.
Return type:: NxQuery

Example

>>> (graph.query_nodes()
...     .where({"main_label": "Person"})
...     .where({"age": {"$gte": 30}})
...     .limit(10)
...     .to_list())

schema_yaml(db_name)[source]

Introspect the database and return a YAML schema description.

Scans _node_attrs and _edge_attrs to collect all labels, properties, relationship types, and counts. Generates Python class names from labels and relationship types.

WeakNode detection: labels whose PK fields are a proper superset of another label’s PK fields are inferred as children (WeakNode). The child-specific PK fields are the set difference.

Parameters:: db_name (str) – Human-readable database name (e.g. "got").
Returns:: A YAML string suitable for code generation.
Return type:: str

count(filter_dict=None)[source]

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

create_group(strong_node, weak_nodes=None, weak_relations=None, **kwargs)[source]

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

Takes a snapshot of the graph state before any modifications. If any insertion fails, the snapshot is restored so the graph is left unchanged.

Parameters:

strong_node (Node) – The root (non-weak) node of the group.
weak_nodes (List[Node] | None) – Optional list of WeakNode instances.
weak_relations (List[Relation] | None) – Optional list of WeakRelation instances.
kwargs (Any)

Returns:

The internal id of the strong_node.

Raises:

RuntimeError – If any part of the group creation fails.

Return type:

int

init_propagation(background=False, progress_callback=None)[source]

Scan the backend graph and initialize propagation properties.

See GraphStore.init_propagation for details.

Parameters:

background (bool)
progress_callback (Callable[[int, int], None] | None)

Return type:

bool

close()[source]

Release resources and clear the graph.

Does NOT persist anything itself — every mutating method (insertNode/insertRelation/deleteNode/init_propagation/ create_group/enable_vector_index) already reloads the latest on-disk state and saves as part of a cross-process file-locked load-mutate-save cycle (see _guarded_write), so by the time close() runs the on-disk state already reflects every change made through this instance — including any other process’s writes that happened in between.

This used to call _save_state() unconditionally, which re-wrote the entire file with whatever was loaded into memory when this instance was opened — including for an instance that only ever did reads. Two overlapping opens of the same persistence path (e.g. a slow read in one process/request and a write in another) would race: whichever instance closed last silently overwrote the other’s writes with its own (possibly stale) snapshot, even though it never mutated anything itself. Removing the call here means a read-only instance can never clobber someone else’s concurrent write on close. The remaining “two writers” race (both mutate, neither just reads) is handled separately by _guarded_write’s reload-before-mutate under a cross-process lock.

Return type:: None

enable_vector_index(property_name, dimensions, space='cosine', ef_construction=200, m=16)[source]

Enable ANN vector indexing for a given node property.

The property values must be 1D vectors with exactly dimensions items.

Parameters:

property_name (str)
dimensions (int)
space (str)
ef_construction (int)
m (int)

Return type:

None

get_subdocuments(strong_node)[source]

Return all subdocuments (WeakNodes) reachable from strong_node through _propagate edges.

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.
Returns:: A list of dicts, one per subdocument.
Return type:: List[Dict[str, Any]]

query_vector_index(property_name, vector, top_k=10)[source]

Query nearest neighbors by vector similarity for one indexed property.

Parameters:

property_name (str)
vector (List[float] | Tuple[float, ...] | ndarray)
top_k (int)

Return type:

List[Tuple[int, float]]

list_vector_indexes()[source]

Return metadata for every enabled vector index.

Return type:: List[Dict[str, Any]]

batch(write=True)[source]

Hold the cross-process write lock for the whole with block.

Groups multiple mutating calls (or a mix of reads and mutating calls, as in cvcdocdb.migration.migrate()) under a single acquisition of the file lock instead of one per call — mirrors Neo4jGraph.batch()’s transaction-grouping contract on this backend. It’s a write lock: other processes/instances calling any mutating method (which all go through _guarded_write()) block until this block exits, while reads (which never touch the lock) are never blocked, including reads made by this instance. Reentrant — nesting is a no-op, same as _guarded_write.

Parameters:: write (bool) – If False, still acquires the lock (blocking concurrent writers for the whole block, so reads made inside see a consistent snapshot) but skips the resave on exit, since nothing was mutated — used when this instance is only being read from within the block (e.g. as the source of cvcdocdb.migration.migrate()), to avoid rewriting the whole persisted graph to disk for nothing.
Return type:: Iterator[None]