NetworkX backend

drm.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 drm.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

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.

Persists the current graph state and clears in-memory data.

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]]