A high-performance graph library for Python implemented in Rust, providing fast graph algorithms and data structures for computational tasks
—
The foundational graph data structures in rustworkx, providing comprehensive node and edge management with support for arbitrary Python objects as node and edge weights. These classes form the basis for all graph operations and algorithms in the library.
Generic undirected multigraph supporting parallel edges and arbitrary node/edge data payloads. Provides comprehensive methods for graph construction, manipulation, and querying.
class PyGraph:
"""
Undirected multigraph with arbitrary node and edge data.
Parameters:
- multigraph (bool): Allow parallel edges between nodes (default: True)
"""
def __init__(self, multigraph: bool = True): ...
# Node operations
def add_node(self, node_weight) -> int:
"""Add node with data payload, returns node index."""
def add_nodes_from(self, node_list) -> list[int]:
"""Add multiple nodes, returns list of node indices."""
def remove_node(self, node_index: int):
"""Remove node and all incident edges."""
def nodes(self):
"""Get all node data payloads."""
def node_indices(self) -> list[int]:
"""Get all node indices."""
# Edge operations
def add_edge(self, node_a: int, node_b: int, edge_weight) -> int:
"""Add edge between nodes, returns edge index."""
def add_edges_from(self, edge_list):
"""Add multiple edges from list of (node_a, node_b, weight) tuples."""
def add_edges_from_no_data(self, edge_list):
"""Add edges without data from list of (node_a, node_b) tuples."""
def remove_edge(self, source: int, target: int):
"""Remove edge between nodes."""
def remove_edges_from(self, edge_list):
"""Remove multiple edges."""
def edges(self):
"""Get all edge data payloads."""
def edge_list(self) -> list[tuple[int, int]]:
"""Get edge list as (source, target) tuples."""
def weighted_edge_list(self) -> list[tuple[int, int, Any]]:
"""Get weighted edge list as (source, target, weight) tuples."""
# Graph properties and queries
def num_nodes(self) -> int:
"""Number of nodes in graph."""
def num_edges(self) -> int:
"""Number of edges in graph."""
def degree(self, node_index: int) -> int:
"""Get degree of specified node."""
def neighbors(self, node_index: int):
"""Get neighboring nodes."""
def find_node_by_weight(self, weight) -> int:
"""Find first node with matching weight."""
# Graph operations
def copy(self):
"""Create deep copy of graph."""
def clear(self):
"""Remove all nodes and edges."""
def subgraph(self, node_list: list[int]):
"""Create subgraph from specified nodes."""
def to_directed(self):
"""Convert to directed graph."""
def compose(self, other_graph, node_map: dict):
"""Compose with another graph using node mapping."""
def contract_nodes(self, node_list: list[int], new_weight):
"""Contract specified nodes into single node."""
# Attributes
attrs: Any # Graph-level attributes
multigraph: bool # Whether parallel edges are allowedGeneric directed multigraph with additional directed-specific functionality including cycle checking, predecessor/successor queries, and directed graph operations.
class PyDiGraph:
"""
Directed multigraph with cycle checking capabilities.
Parameters:
- check_cycle (bool): Enable runtime cycle detection (default: False)
- multigraph (bool): Allow parallel edges between nodes (default: True)
"""
def __init__(self, check_cycle: bool = False, multigraph: bool = True): ...
# Inherits all PyGraph methods plus:
# Directed node operations
def add_child(self, parent_node: int, child_weight, edge_weight) -> int:
"""Add child node connected to parent, returns child index."""
def add_parent(self, child_node: int, parent_weight, edge_weight) -> int:
"""Add parent node connected to child, returns parent index."""
# Directed edge queries
def in_edges(self, node_index: int):
"""Get incoming edges for node."""
def out_edges(self, node_index: int):
"""Get outgoing edges for node."""
def in_degree(self, node_index: int) -> int:
"""Get in-degree of node."""
def out_degree(self, node_index: int) -> int:
"""Get out-degree of node."""
# Directed neighbors
def predecessors(self, node_index: int):
"""Get predecessor nodes."""
def successors(self, node_index: int):
"""Get successor nodes."""
# Directed operations
def reverse(self):
"""Reverse direction of all edges."""
def to_undirected(self):
"""Convert to undirected graph."""
def is_symmetric(self) -> bool:
"""Check if graph is symmetric (bidirectional edges)."""
def make_symmetric(self):
"""Make graph symmetric by adding reverse edges."""
def find_successor_node_by_edge(self, source: int, edge_weight) -> int:
"""Find successor node by edge weight."""
# Attributes
check_cycle: bool # Runtime cycle detection flag
attrs: Any # Graph-level attributes
multigraph: bool # Whether parallel edges are allowedSpecialized directed graph class with enhanced cycle prevention. Functionally identical to PyDiGraph but conceptually represents a DAG.
class PyDAG(PyDiGraph):
"""
Directed Acyclic Graph - alias for PyDiGraph with cycle checking.
Identical functionality to PyDiGraph, exists for semantic clarity
and backward compatibility. Supports runtime cycle detection.
"""
passimport rustworkx as rx
# Create undirected graph
graph = rx.PyGraph()
nodes = graph.add_nodes_from(['A', 'B', 'C', 'D'])
graph.add_edges_from([
(nodes[0], nodes[1], 1.0),
(nodes[1], nodes[2], 2.0),
(nodes[2], nodes[3], 3.0)
])
# Create directed graph with cycle checking
digraph = rx.PyDiGraph(check_cycle=True)
try:
n1 = digraph.add_node("Node 1")
n2 = digraph.add_child(n1, "Node 2", "edge_1_to_2")
n3 = digraph.add_child(n2, "Node 3", "edge_2_to_3")
# This would raise DAGWouldCycle if cycle checking is enabled
# digraph.add_edge(n3, n1, "would_create_cycle")
except rx.DAGWouldCycle:
print("Cycle detected!")# Multigraph allows parallel edges
multigraph = rx.PyGraph(multigraph=True)
n1 = multigraph.add_node("A")
n2 = multigraph.add_node("B")
multigraph.add_edge(n1, n2, "edge1")
multigraph.add_edge(n1, n2, "edge2") # Parallel edge allowed
# Simple graph updates existing edge
simple_graph = rx.PyGraph(multigraph=False)
n1 = simple_graph.add_node("A")
n2 = simple_graph.add_node("B")
simple_graph.add_edge(n1, n2, "edge1")
simple_graph.add_edge(n1, n2, "edge2") # Updates existing edge# Node and edge removal
graph.remove_node(node_index) # Removes node and incident edges
graph.remove_edge(source, target) # Removes specific edge
graph.remove_edges_from([(0, 1), (1, 2)]) # Batch removal
# Graph querying
neighbors = graph.neighbors(node_index)
degree = graph.degree(node_index)
node_data = graph.nodes()[node_index]
edge_data = graph.edges()[edge_index]
# Graph transformation
directed = graph.to_directed() # Convert undirected to directed
undirected = digraph.to_undirected() # Convert directed to undirected
subgraph = graph.subgraph([0, 1, 2]) # Extract subgraph
copy = graph.copy() # Deep copyInstall with Tessl CLI
npx tessl i tessl/pypi-rustworkx