Advanced Features
Undo/redo, computing flows, sub-flows, viewport portal, accessibility, and more.
Overview
The advanced feature set: history with enableUndoRedo / undoRedoAction, topological traversal with computeAction / computeResult, collapsible sub-flows (toggleCollapseNode / collapsedGroups), floating annotations via viewportOverlays, resize constraints, and full ARIA/keyboard accessibility.
Live demo
Drag the nodes, pan, and zoom — the canvas below is a real, running DashFlows component (no callbacks, just the rendered graph):
# File: docs/advanced/demo.py
"""Live, callback-free demo for the docs page. Rendered via `.. exec::docs.advanced.demo`."""
import dash_flows
nodes = [{'id': 'src', 'type': 'input', 'data': {'label': 'Input'}, 'position': {'x': 60, 'y': 40}},
{'id': 'm1', 'type': 'default', 'data': {'label': 'Step 1'}, 'position': {'x': 60, 'y': 170}},
{'id': 'm2', 'type': 'default', 'data': {'label': 'Step 2'}, 'position': {'x': 260, 'y': 170}},
{'id': 'agg', 'type': 'default', 'data': {'label': 'Aggregate'}, 'position': {'x': 160, 'y': 300}},
{'id': 'snk', 'type': 'output', 'data': {'label': 'Output'}, 'position': {'x': 160, 'y': 430}}]
edges = [{'id': 'a1', 'source': 'src', 'target': 'm1'},
{'id': 'a2', 'source': 'src', 'target': 'm2'},
{'id': 'a3', 'source': 'm1', 'target': 'agg', 'animated': True},
{'id': 'a4', 'source': 'm2', 'target': 'agg'},
{'id': 'a5', 'source': 'agg', 'target': 'snk'}]
component = dash_flows.DashFlows(
id="advanced-demo",
nodes=nodes,
edges=edges,
style={'border': '1px solid var(--mantine-color-default-border)',
'borderRadius': '8px',
'height': '440px'},
showControls=True,
showMiniMap=True,
enableUndoRedo=True,
helperLines=True,
fitView=True,
colorMode='system',
):defaultExpanded: false :withExpandedButton: true
Examples
Each example below is a complete, runnable Dash app from the examples/ folder. Run any of them with python examples/<file>.
Complete Showcase
A large example combining multiple node types, edge types, a group node, ELK layout switching, and live selection/count panels in one app. The full app (examples/13_complete_showcase.py) also includes a light/dark theme toggle; the embedded twin below drops that wrapper since the docs page already supplies a theme provider.
# File: examples/13_complete_showcase.py
"""
Example 13: Complete Feature Showcase
=====================================
A comprehensive example combining multiple DashFlows features:
- Multiple node types with custom styling
- Various edge types
- Interactive toolbar
- Dark mode toggle
- Layout controls
- DevTools panel
"""
import dash
from dash import html, Input, Output, State, callback, clientside_callback, dcc
import dash_flows
import dash_mantine_components as dmc
import json
app = dash.Dash(__name__, suppress_callback_exceptions=True)
# Comprehensive node configuration
initial_nodes = [
# Input nodes
{
"id": "input-api",
"type": "input",
"data": {"label": "REST API", "sublabel": "External Data"},
"position": {"x": 50, "y": 50},
},
{
"id": "input-db",
"type": "input",
"data": {"label": "Database", "sublabel": "PostgreSQL"},
"position": {"x": 50, "y": 180},
},
# Processing nodes
{
"id": "process-validate",
"type": "default",
"data": {"label": "Validate", "sublabel": "Schema Check"},
"position": {"x": 250, "y": 50},
},
{
"id": "process-transform",
"type": "default",
"data": {"label": "Transform", "sublabel": "ETL Pipeline"},
"position": {"x": 250, "y": 180},
},
{
"id": "process-merge",
"type": "toolbar",
"data": {
"label": "Merge Data",
"sublabel": "Click for actions",
"toolbarPosition": "top",
},
"position": {"x": 450, "y": 115},
},
# Group node with children
{
"id": "group-ml",
"type": "group",
"data": {"label": "ML Pipeline"},
"position": {"x": 650, "y": 30},
"style": {"width": 280, "height": 220},
},
{
"id": "ml-train",
"type": "default",
"data": {"label": "Train Model"},
"position": {"x": 30, "y": 50},
"parentId": "group-ml",
"extent": "parent",
},
{
"id": "ml-evaluate",
"type": "default",
"data": {"label": "Evaluate"},
"position": {"x": 30, "y": 130},
"parentId": "group-ml",
"extent": "parent",
},
# Output nodes
{
"id": "output-dashboard",
"type": "output",
"data": {"label": "Dashboard", "sublabel": "Visualization"},
"position": {"x": 450, "y": 300},
},
{
"id": "output-export",
"type": "output",
"data": {"label": "Export", "sublabel": "CSV/JSON"},
"position": {"x": 650, "y": 300},
},
]
initial_edges = [
{"id": "e1", "source": "input-api", "target": "process-validate", "animated": True},
{"id": "e2", "source": "input-db", "target": "process-transform", "type": "smoothstep"},
{"id": "e3", "source": "process-validate", "target": "process-merge"},
{"id": "e4", "source": "process-transform", "target": "process-merge", "type": "step"},
{"id": "e5", "source": "process-merge", "target": "ml-train", "type": "simplebezier"},
{"id": "e6", "source": "ml-train", "target": "ml-evaluate"},
{"id": "e7", "source": "process-merge", "target": "output-dashboard", "type": "button", "data": {"label": "Live", "showButton": True}},
{"id": "e8", "source": "ml-evaluate", "target": "output-export"},
]
# Layout presets
layouts = {
"horizontal": json.dumps({
"elk.algorithm": "layered",
"elk.direction": "RIGHT",
"elk.spacing.nodeNode": 60,
"elk.layered.spacing.nodeNodeBetweenLayers": 100,
}),
"vertical": json.dumps({
"elk.algorithm": "layered",
"elk.direction": "DOWN",
"elk.spacing.nodeNode": 60,
}),
"radial": json.dumps({
"elk.algorithm": "org.eclipse.elk.radial",
"elk.radial.radius": 180,
}),
}
# App layout
app.layout = dmc.MantineProvider(
id="theme-provider",
forceColorScheme="light",
children=[
# Header
dmc.Paper([
dmc.Group([
dmc.Title("DashFlows Complete Showcase", order=2),
dmc.Group([
dmc.ActionIcon(
id="theme-toggle",
variant="subtle",
size="lg",
children=dmc.Text(id="theme-icon", size="lg"),
),
]),
], justify="space-between"),
], p="md", mb="md", withBorder=True),
# Controls
dmc.Paper([
dmc.Group([
dmc.SegmentedControl(
id="layout-control",
data=[
{"value": "none", "label": "Manual"},
{"value": "horizontal", "label": "Horizontal"},
{"value": "vertical", "label": "Vertical"},
{"value": "radial", "label": "Radial"},
],
value="none",
),
dmc.Switch(id="show-minimap", label="MiniMap", checked=True),
dmc.Switch(id="show-controls", label="Controls", checked=True),
dmc.Switch(id="show-devtools", label="DevTools", checked=False),
], gap="lg"),
], p="sm", mb="md", withBorder=True),
# Main flow area
dmc.Grid([
dmc.GridCol([
html.Div(id="flow-container"),
], span=12 if True else 9),
]),
# Info panels
dmc.Grid([
dmc.GridCol([
dmc.Paper([
dmc.Text("Selected Items", fw=600, mb="xs"),
html.Div(id="selection-display"),
], p="md", withBorder=True),
], span=4),
dmc.GridCol([
dmc.Paper([
dmc.Text("Node Count", fw=600, mb="xs"),
html.Div(id="node-count"),
], p="md", withBorder=True),
], span=4),
dmc.GridCol([
dmc.Paper([
dmc.Text("Edge Count", fw=600, mb="xs"),
html.Div(id="edge-count"),
], p="md", withBorder=True),
], span=4),
], mt="md"),
],
)
# Theme toggle
clientside_callback(
"""
function(n, scheme) {
if (!n) return [scheme, scheme === 'dark' ? 'O' : '*'];
const newScheme = scheme === 'light' ? 'dark' : 'light';
return [newScheme, newScheme === 'dark' ? 'O' : '*'];
}
""",
Output("theme-provider", "forceColorScheme"),
Output("theme-icon", "children"),
Input("theme-toggle", "n_clicks"),
State("theme-provider", "forceColorScheme"),
prevent_initial_call=False,
)
# Render flow with settings
@callback(
Output("flow-container", "children"),
Input("layout-control", "value"),
Input("show-minimap", "checked"),
Input("show-controls", "checked"),
Input("show-devtools", "checked"),
)
def render_flow(layout, show_mm, show_ctrl, show_dev):
layout_opts = layouts.get(layout) if layout != "none" else None
return dash_flows.DashFlows(
id="showcase-flow",
nodes=initial_nodes,
edges=initial_edges,
style={"height": "500px", "borderRadius": "8px"},
fitView=True,
showControls=show_ctrl,
showMiniMap=show_mm,
showDevTools=show_dev,
layoutOptions=layout_opts,
backgroundVariant="dots",
elementsSelectable=True,
nodesConnectable=True,
nodesDraggable=True,
)
# Selection display
@callback(
Output("selection-display", "children"),
Input("showcase-flow", "selectedNodes"),
Input("showcase-flow", "selectedEdges"),
prevent_initial_call=True,
)
def show_selection(nodes, edges):
# Handle various data formats
node_ids = []
edge_ids = []
if nodes:
if isinstance(nodes, list):
for n in nodes:
if isinstance(n, dict) and "id" in n:
node_ids.append(n["id"])
elif isinstance(n, str):
node_ids.append(n)
if edges:
if isinstance(edges, list):
for e in edges:
if isinstance(e, dict) and "id" in e:
edge_ids.append(e["id"])
elif isinstance(e, str):
edge_ids.append(e)
if not node_ids and not edge_ids:
return dmc.Text("Nothing selected", c="dimmed", size="sm")
items = []
if node_ids:
items.append(dmc.Group([dmc.Badge(nid, color="blue", size="sm") for nid in node_ids], gap="xs"))
if edge_ids:
items.append(dmc.Group([dmc.Badge(eid, color="grape", size="sm") for eid in edge_ids], gap="xs"))
return dmc.Stack(items, gap="xs")
# Node count
@callback(
Output("node-count", "children"),
Input("showcase-flow", "nodes"),
)
def show_node_count(nodes):
count = len(nodes) if nodes else 0
return dmc.Text(f"{count} nodes", size="xl", fw=700)
# Edge count
@callback(
Output("edge-count", "children"),
Input("showcase-flow", "edges"),
)
def show_edge_count(edges):
count = len(edges) if edges else 0
return dmc.Text(f"{count} edges", size="xl", fw=700)
if __name__ == "__main__":
app.run(debug=True, port=8062):defaultExpanded: false :withExpandedButton: true
How it works
layoutOptions— set from aSegmentedControlto switch between horizontal, vertical, and radial ELK layouts on the flyshowMiniMap/showControls/showDevTools— toggled by switches, demonstrating that chrome can be turned on/off per renderselectedNodes/selectedEdges— read in a callback to render badges for the current selectionnodes/edges(as callback Inputs) — used purely to display live counts, showing that the props stay in sync with canvas statetype: "group"withparentId/extent: "parent"children — nests the ML Pipeline nodes inside a group nodetype: "toolbar"andtype: "button"edge — shows the ToolbarNode and ButtonEdge working alongside default/input/output nodes
Callback Stress Test
A performance harness that hammers the callback chain: a 5x10 node/edge grid, batch add/remove/shuffle/connect operations (up to hundreds of elements at once), and half a dozen simultaneous listeners on click/drag/selection/connection/deletion events, all while tracking a live metrics dashboard.
This example is intentionally not embedded as a live demo — running it inline would fire the same high-frequency callback storm against the whole docs page and degrade every other embedded example on it. Run it standalone with python examples/23_callback_stress_test.py instead.
# File: examples/23_callback_stress_test.py
"""
Example 23: Callback Stress Test
================================
This example stress tests Dash callbacks when interacting with DashFlows:
- Large number of nodes and edges
- Multiple simultaneous callbacks monitoring different events
- Rapid node manipulation (add, remove, update)
- Stress testing drag, selection, and connection events
- Performance metrics tracking
- Batch operations on nodes/edges
"""
import dash
from dash import html, Input, Output, State, callback, ctx, dcc, ALL, MATCH
from dash.exceptions import PreventUpdate
import dash_flows
import dash_mantine_components as dmc
import json
import time
import random
import uuid
app = dash.Dash(__name__, suppress_callback_exceptions=True)
# Configuration
GRID_ROWS = 5
GRID_COLS = 10
NODE_SPACING_X = 180
NODE_SPACING_Y = 120
def generate_grid_nodes(rows, cols):
"""Generate a grid of nodes for stress testing."""
nodes = []
for row in range(rows):
for col in range(cols):
node_id = f"node-{row}-{col}"
node_type = "default"
if col == 0:
node_type = "input"
elif col == cols - 1:
node_type = "output"
nodes.append({
"id": node_id,
"type": node_type,
"data": {
"label": f"N{row},{col}",
"sublabel": f"idx: {row * cols + col}"
},
"position": {
"x": 50 + col * NODE_SPACING_X,
"y": 50 + row * NODE_SPACING_Y
},
})
return nodes
def generate_grid_edges(rows, cols):
"""Generate edges connecting adjacent nodes in the grid."""
edges = []
for row in range(rows):
for col in range(cols - 1):
source = f"node-{row}-{col}"
target = f"node-{row}-{col + 1}"
edges.append({
"id": f"edge-{source}-{target}",
"source": source,
"target": target,
"type": random.choice(["default", "smoothstep", "step"]),
})
return edges
initial_nodes = generate_grid_nodes(GRID_ROWS, GRID_COLS)
initial_edges = generate_grid_edges(GRID_ROWS, GRID_COLS)
# Stress test metrics store
metrics_store = {
"callback_counts": {},
"last_callback_times": {},
"total_callbacks": 0,
}
app.layout = dmc.MantineProvider([
dcc.Store(id="metrics-store", data={
"callbacks_fired": 0,
"nodes_added": 0,
"nodes_removed": 0,
"edges_added": 0,
"edges_removed": 0,
"drags_completed": 0,
"selections_made": 0,
"connections_made": 0,
"start_time": time.time(),
}),
dcc.Interval(id="metrics-interval", interval=500, n_intervals=0),
html.H1("DashFlows Callback Stress Test", style={"marginBottom": "10px"}),
# Control Panel
dmc.Paper([
dmc.Group([
dmc.Text("Stress Test Controls:", fw=700),
dmc.Button("Add 10 Random Nodes", id="btn-add-nodes", color="green", size="xs"),
dmc.Button("Remove 10 Random Nodes", id="btn-remove-nodes", color="red", size="xs"),
dmc.Button("Shuffle All Positions", id="btn-shuffle", color="blue", size="xs"),
dmc.Button("Connect Random Pairs", id="btn-connect", color="violet", size="xs"),
dmc.Button("Select All", id="btn-select-all", color="cyan", size="xs"),
dmc.Button("Clear Selection", id="btn-clear-selection", color="gray", size="xs"),
dmc.Button("Reset Grid", id="btn-reset", color="orange", size="xs"),
], gap="xs"),
], p="sm", mb="sm", withBorder=True),
# Batch Operations Panel
dmc.Paper([
dmc.Group([
dmc.Text("Batch Operations:", fw=700),
dmc.NumberInput(id="batch-size", value=50, min=10, max=500, step=10,
label="Batch Size", w=120, size="xs"),
dmc.Button("Add Batch Nodes", id="btn-batch-add", color="teal", size="xs"),
dmc.Button("Batch Update Labels", id="btn-batch-update", color="indigo", size="xs"),
dmc.Button("Stress Connect All", id="btn-stress-connect", color="pink", size="xs"),
], gap="xs"),
], p="sm", mb="sm", withBorder=True),
# Main Layout
dmc.Grid([
# Flow Canvas
dmc.GridCol([
dash_flows.DashFlows(
id="stress-flow",
nodes=initial_nodes,
edges=initial_edges,
style={"height": "500px", "border": "1px solid #ddd"},
fitView=True,
showControls=True,
showMiniMap=True,
miniMapPosition="bottom-left",
nodesDraggable=True,
nodesConnectable=True,
elementsSelectable=True,
selectNodesOnDrag=False,
backgroundVariant="dots",
connectionMode="loose",
),
], span=9),
# Metrics Panel
dmc.GridCol([
dmc.Stack([
dmc.Paper([
dmc.Text("Live Metrics", fw=700, size="lg", mb="xs"),
html.Div(id="live-metrics"),
], p="sm", withBorder=True),
dmc.Paper([
dmc.Text("Callback Activity", fw=700, size="lg", mb="xs"),
html.Div(id="callback-activity"),
], p="sm", withBorder=True),
dmc.Paper([
dmc.Text("Current State", fw=700, size="lg", mb="xs"),
html.Div(id="current-state"),
], p="sm", withBorder=True),
dmc.Paper([
dmc.Text("Last Events", fw=700, size="lg", mb="xs"),
html.Div(id="last-events", style={"maxHeight": "150px", "overflow": "auto"}),
], p="sm", withBorder=True),
], gap="xs"),
], span=3),
]),
# Event Log
dmc.Paper([
dmc.Group([
dmc.Text("Event Log (last 20):", fw=700),
dmc.Button("Clear Log", id="btn-clear-log", size="xs", variant="subtle"),
], justify="space-between", mb="xs"),
html.Div(id="event-log", style={
"maxHeight": "150px",
"overflow": "auto",
"fontFamily": "monospace",
"fontSize": "11px",
}),
], p="sm", mt="sm", withBorder=True),
# Hidden stores for event tracking
dcc.Store(id="event-log-store", data=[]),
dcc.Store(id="callback-counter", data=0),
])
# ============================================================================
# STRESS TEST CALLBACKS - Node Manipulation
# ============================================================================
@callback(
Output("stress-flow", "nodes", allow_duplicate=True),
Output("event-log-store", "data", allow_duplicate=True),
Output("metrics-store", "data", allow_duplicate=True),
Input("btn-add-nodes", "n_clicks"),
State("stress-flow", "nodes"),
State("event-log-store", "data"),
State("metrics-store", "data"),
prevent_initial_call=True,
)
def add_random_nodes(n_clicks, nodes, log, metrics):
"""Add 10 random nodes to stress test node addition."""
if not n_clicks:
raise PreventUpdate
nodes = nodes or []
new_nodes = []
for i in range(10):
node_id = f"rand-{uuid.uuid4().hex[:8]}"
new_nodes.append({
"id": node_id,
"type": random.choice(["default", "input", "output"]),
"data": {
"label": f"New-{i}",
"sublabel": f"Added #{n_clicks}"
},
"position": {
"x": random.randint(50, 1500),
"y": random.randint(50, 500)
},
})
log = log or []
log.insert(0, f"[ADD] Added 10 nodes (total: {len(nodes) + 10})")
log = log[:20]
metrics["nodes_added"] = metrics.get("nodes_added", 0) + 10
metrics["callbacks_fired"] = metrics.get("callbacks_fired", 0) + 1
return nodes + new_nodes, log, metrics
@callback(
Output("stress-flow", "nodes", allow_duplicate=True),
Output("stress-flow", "edges", allow_duplicate=True),
Output("event-log-store", "data", allow_duplicate=True),
Output("metrics-store", "data", allow_duplicate=True),
Input("btn-remove-nodes", "n_clicks"),
State("stress-flow", "nodes"),
State("stress-flow", "edges"),
State("event-log-store", "data"),
State("metrics-store", "data"),
prevent_initial_call=True,
)
def remove_random_nodes(n_clicks, nodes, edges, log, metrics):
"""Remove 10 random nodes to stress test node removal."""
if not n_clicks or not nodes or len(nodes) <= 10:
raise PreventUpdate
nodes_to_remove = set(random.sample([n["id"] for n in nodes], min(10, len(nodes))))
remaining_nodes = [n for n in nodes if n["id"] not in nodes_to_remove]
remaining_edges = [e for e in (edges or [])
if e["source"] not in nodes_to_remove
and e["target"] not in nodes_to_remove]
log = log or []
log.insert(0, f"[REMOVE] Removed 10 nodes (remaining: {len(remaining_nodes)})")
log = log[:20]
metrics["nodes_removed"] = metrics.get("nodes_removed", 0) + 10
metrics["callbacks_fired"] = metrics.get("callbacks_fired", 0) + 1
return remaining_nodes, remaining_edges, log, metrics
@callback(
Output("stress-flow", "nodes", allow_duplicate=True),
Output("event-log-store", "data", allow_duplicate=True),
Output("metrics-store", "data", allow_duplicate=True),
Input("btn-shuffle", "n_clicks"),
State("stress-flow", "nodes"),
State("event-log-store", "data"),
State("metrics-store", "data"),
prevent_initial_call=True,
)
def shuffle_positions(n_clicks, nodes, log, metrics):
"""Shuffle all node positions to stress test batch position updates."""
if not n_clicks or not nodes:
raise PreventUpdate
updated_nodes = []
for node in nodes:
updated_node = {**node}
updated_node["position"] = {
"x": random.randint(50, 1500),
"y": random.randint(50, 500)
}
updated_nodes.append(updated_node)
log = log or []
log.insert(0, f"[SHUFFLE] Shuffled {len(nodes)} node positions")
log = log[:20]
metrics["callbacks_fired"] = metrics.get("callbacks_fired", 0) + 1
return updated_nodes, log, metrics
@callback(
Output("stress-flow", "edges", allow_duplicate=True),
Output("event-log-store", "data", allow_duplicate=True),
Output("metrics-store", "data", allow_duplicate=True),
Input("btn-connect", "n_clicks"),
State("stress-flow", "nodes"),
State("stress-flow", "edges"),
State("event-log-store", "data"),
State("metrics-store", "data"),
prevent_initial_call=True,
)
def connect_random_pairs(n_clicks, nodes, edges, log, metrics):
"""Create 10 random connections to stress test edge creation."""
if not n_clicks or not nodes or len(nodes) < 2:
raise PreventUpdate
edges = edges or []
existing_connections = {(e["source"], e["target"]) for e in edges}
node_ids = [n["id"] for n in nodes]
new_edges = []
attempts = 0
while len(new_edges) < 10 and attempts < 50:
source = random.choice(node_ids)
target = random.choice(node_ids)
if source != target and (source, target) not in existing_connections:
new_edges.append({
"id": f"edge-{uuid.uuid4().hex[:8]}",
"source": source,
"target": target,
"type": random.choice(["default", "smoothstep", "step"]),
"animated": random.choice([True, False]),
})
existing_connections.add((source, target))
attempts += 1
log = log or []
log.insert(0, f"[CONNECT] Created {len(new_edges)} new edges (total: {len(edges) + len(new_edges)})")
log = log[:20]
metrics["edges_added"] = metrics.get("edges_added", 0) + len(new_edges)
metrics["callbacks_fired"] = metrics.get("callbacks_fired", 0) + 1
return edges + new_edges, log, metrics
# ============================================================================
# BATCH OPERATIONS
# ============================================================================
@callback(
Output("stress-flow", "nodes", allow_duplicate=True),
Output("event-log-store", "data", allow_duplicate=True),
Output("metrics-store", "data", allow_duplicate=True),
Input("btn-batch-add", "n_clicks"),
State("batch-size", "value"),
State("stress-flow", "nodes"),
State("event-log-store", "data"),
State("metrics-store", "data"),
prevent_initial_call=True,
)
def batch_add_nodes(n_clicks, batch_size, nodes, log, metrics):
"""Add a large batch of nodes at once."""
if not n_clicks:
raise PreventUpdate
batch_size = batch_size or 50
nodes = nodes or []
new_nodes = []
for i in range(batch_size):
node_id = f"batch-{uuid.uuid4().hex[:8]}"
new_nodes.append({
"id": node_id,
"type": random.choice(["default", "input", "output"]),
"data": {
"label": f"B{i}",
"sublabel": f"Batch #{n_clicks}"
},
"position": {
"x": random.randint(50, 2000),
"y": random.randint(50, 800)
},
})
log = log or []
log.insert(0, f"[BATCH ADD] Added {batch_size} nodes (total: {len(nodes) + batch_size})")
log = log[:20]
metrics["nodes_added"] = metrics.get("nodes_added", 0) + batch_size
metrics["callbacks_fired"] = metrics.get("callbacks_fired", 0) + 1
return nodes + new_nodes, log, metrics
@callback(
Output("stress-flow", "nodes", allow_duplicate=True),
Output("event-log-store", "data", allow_duplicate=True),
Output("metrics-store", "data", allow_duplicate=True),
Input("btn-batch-update", "n_clicks"),
State("stress-flow", "nodes"),
State("event-log-store", "data"),
State("metrics-store", "data"),
prevent_initial_call=True,
)
def batch_update_labels(n_clicks, nodes, log, metrics):
"""Update all node labels to stress test batch updates."""
if not n_clicks or not nodes:
raise PreventUpdate
updated_nodes = []
for i, node in enumerate(nodes):
updated_node = {**node}
updated_node["data"] = {
**node.get("data", {}),
"label": f"U{i}-{n_clicks}",
"sublabel": f"Updated #{n_clicks}"
}
updated_nodes.append(updated_node)
log = log or []
log.insert(0, f"[BATCH UPDATE] Updated {len(nodes)} node labels")
log = log[:20]
metrics["callbacks_fired"] = metrics.get("callbacks_fired", 0) + 1
return updated_nodes, log, metrics
@callback(
Output("stress-flow", "edges", allow_duplicate=True),
Output("event-log-store", "data", allow_duplicate=True),
Output("metrics-store", "data", allow_duplicate=True),
Input("btn-stress-connect", "n_clicks"),
State("stress-flow", "nodes"),
State("stress-flow", "edges"),
State("event-log-store", "data"),
State("metrics-store", "data"),
prevent_initial_call=True,
)
def stress_connect_all(n_clicks, nodes, edges, log, metrics):
"""Create many random connections to stress test edge handling."""
if not n_clicks or not nodes or len(nodes) < 2:
raise PreventUpdate
edges = edges or []
existing_connections = {(e["source"], e["target"]) for e in edges}
node_ids = [n["id"] for n in nodes]
# Create up to 100 new edges or 2x current nodes, whichever is smaller
target_new_edges = min(100, len(nodes) * 2)
new_edges = []
attempts = 0
while len(new_edges) < target_new_edges and attempts < target_new_edges * 5:
source = random.choice(node_ids)
target = random.choice(node_ids)
if source != target and (source, target) not in existing_connections:
new_edges.append({
"id": f"stress-{uuid.uuid4().hex[:8]}",
"source": source,
"target": target,
"type": random.choice(["default", "smoothstep", "step", "straight"]),
"animated": random.choice([True, False, False, False]),
})
existing_connections.add((source, target))
attempts += 1
log = log or []
log.insert(0, f"[STRESS CONNECT] Created {len(new_edges)} edges (total: {len(edges) + len(new_edges)})")
log = log[:20]
metrics["edges_added"] = metrics.get("edges_added", 0) + len(new_edges)
metrics["callbacks_fired"] = metrics.get("callbacks_fired", 0) + 1
return edges + new_edges, log, metrics
# ============================================================================
# SELECTION CALLBACKS
# ============================================================================
@callback(
Output("stress-flow", "nodes", allow_duplicate=True),
Output("event-log-store", "data", allow_duplicate=True),
Output("metrics-store", "data", allow_duplicate=True),
Input("btn-select-all", "n_clicks"),
State("stress-flow", "nodes"),
State("event-log-store", "data"),
State("metrics-store", "data"),
prevent_initial_call=True,
)
def select_all_nodes(n_clicks, nodes, log, metrics):
"""Select all nodes to stress test selection handling."""
if not n_clicks or not nodes:
raise PreventUpdate
selected_nodes = [{**n, "selected": True} for n in nodes]
log = log or []
log.insert(0, f"[SELECT ALL] Selected {len(nodes)} nodes")
log = log[:20]
metrics["selections_made"] = metrics.get("selections_made", 0) + 1
metrics["callbacks_fired"] = metrics.get("callbacks_fired", 0) + 1
return selected_nodes, log, metrics
@callback(
Output("stress-flow", "nodes", allow_duplicate=True),
Output("event-log-store", "data", allow_duplicate=True),
Output("metrics-store", "data", allow_duplicate=True),
Input("btn-clear-selection", "n_clicks"),
State("stress-flow", "nodes"),
State("event-log-store", "data"),
State("metrics-store", "data"),
prevent_initial_call=True,
)
def clear_selection(n_clicks, nodes, log, metrics):
"""Clear selection on all nodes."""
if not n_clicks or not nodes:
raise PreventUpdate
cleared_nodes = [{**n, "selected": False} for n in nodes]
log = log or []
log.insert(0, f"[CLEAR] Cleared selection on {len(nodes)} nodes")
log = log[:20]
metrics["callbacks_fired"] = metrics.get("callbacks_fired", 0) + 1
return cleared_nodes, log, metrics
@callback(
Output("stress-flow", "nodes", allow_duplicate=True),
Output("stress-flow", "edges", allow_duplicate=True),
Output("event-log-store", "data", allow_duplicate=True),
Output("metrics-store", "data", allow_duplicate=True),
Input("btn-reset", "n_clicks"),
State("event-log-store", "data"),
State("metrics-store", "data"),
prevent_initial_call=True,
)
def reset_grid(n_clicks, log, metrics):
"""Reset to initial grid configuration."""
if not n_clicks:
raise PreventUpdate
log = log or []
log.insert(0, f"[RESET] Reset to initial {GRID_ROWS}x{GRID_COLS} grid")
log = log[:20]
metrics["callbacks_fired"] = metrics.get("callbacks_fired", 0) + 1
return (
generate_grid_nodes(GRID_ROWS, GRID_COLS),
generate_grid_edges(GRID_ROWS, GRID_COLS),
log,
metrics
)
# ============================================================================
# EVENT MONITORING CALLBACKS
# ============================================================================
@callback(
Output("event-log-store", "data", allow_duplicate=True),
Output("metrics-store", "data", allow_duplicate=True),
Input("stress-flow", "clickedNode"),
State("event-log-store", "data"),
State("metrics-store", "data"),
prevent_initial_call=True,
)
def on_node_click(clicked_node, log, metrics):
"""Monitor node click events."""
if not clicked_node:
raise PreventUpdate
node_id = clicked_node.get("id", "unknown") if isinstance(clicked_node, dict) else clicked_node
log = log or []
log.insert(0, f"[CLICK] Node: {node_id}")
log = log[:20]
metrics["callbacks_fired"] = metrics.get("callbacks_fired", 0) + 1
return log, metrics
@callback(
Output("event-log-store", "data", allow_duplicate=True),
Output("metrics-store", "data", allow_duplicate=True),
Input("stress-flow", "draggedNode"),
State("event-log-store", "data"),
State("metrics-store", "data"),
prevent_initial_call=True,
)
def on_node_drag(dragged_node, log, metrics):
"""Monitor node drag events."""
if not dragged_node:
raise PreventUpdate
node_id = dragged_node.get("id", "unknown")
is_dragging = dragged_node.get("isDragging", False)
if not is_dragging: # Only log when drag completes
log = log or []
end_pos = dragged_node.get("endPosition", {})
log.insert(0, f"[DRAG] Node {node_id} to ({end_pos.get('x', 0):.0f}, {end_pos.get('y', 0):.0f})")
log = log[:20]
metrics["drags_completed"] = metrics.get("drags_completed", 0) + 1
metrics["callbacks_fired"] = metrics.get("callbacks_fired", 0) + 1
return log, metrics
raise PreventUpdate
@callback(
Output("event-log-store", "data", allow_duplicate=True),
Output("metrics-store", "data", allow_duplicate=True),
Input("stress-flow", "lastConnection"),
State("event-log-store", "data"),
State("metrics-store", "data"),
prevent_initial_call=True,
)
def on_connection(last_connection, log, metrics):
"""Monitor new connection events."""
if not last_connection:
raise PreventUpdate
source = last_connection.get("source", "?")
target = last_connection.get("target", "?")
log = log or []
log.insert(0, f"[CONNECTION] {source} -> {target}")
log = log[:20]
metrics["connections_made"] = metrics.get("connections_made", 0) + 1
metrics["callbacks_fired"] = metrics.get("callbacks_fired", 0) + 1
return log, metrics
@callback(
Output("event-log-store", "data", allow_duplicate=True),
Output("metrics-store", "data", allow_duplicate=True),
Input("stress-flow", "selectedNodes"),
State("event-log-store", "data"),
State("metrics-store", "data"),
prevent_initial_call=True,
)
def on_selection_change(selected_nodes, log, metrics):
"""Monitor selection changes."""
count = len(selected_nodes) if selected_nodes else 0
log = log or []
log.insert(0, f"[SELECTION] {count} node(s) selected")
log = log[:20]
metrics["selections_made"] = metrics.get("selections_made", 0) + 1
metrics["callbacks_fired"] = metrics.get("callbacks_fired", 0) + 1
return log, metrics
@callback(
Output("event-log-store", "data", allow_duplicate=True),
Output("metrics-store", "data", allow_duplicate=True),
Input("stress-flow", "deletedNodes"),
Input("stress-flow", "deletedEdges"),
State("event-log-store", "data"),
State("metrics-store", "data"),
prevent_initial_call=True,
)
def on_delete(deleted_nodes, deleted_edges, log, metrics):
"""Monitor deletion events."""
log = log or []
if deleted_nodes:
count = len(deleted_nodes)
log.insert(0, f"[DELETE] {count} node(s) deleted")
metrics["nodes_removed"] = metrics.get("nodes_removed", 0) + count
if deleted_edges:
count = len(deleted_edges)
log.insert(0, f"[DELETE] {count} edge(s) deleted")
metrics["edges_removed"] = metrics.get("edges_removed", 0) + count
if not deleted_nodes and not deleted_edges:
raise PreventUpdate
log = log[:20]
metrics["callbacks_fired"] = metrics.get("callbacks_fired", 0) + 1
return log, metrics
# ============================================================================
# DISPLAY CALLBACKS
# ============================================================================
@callback(
Output("live-metrics", "children"),
Input("metrics-interval", "n_intervals"),
State("metrics-store", "data"),
)
def update_live_metrics(n_intervals, metrics):
"""Update live metrics display."""
elapsed = time.time() - metrics.get("start_time", time.time())
callbacks_fired = metrics.get("callbacks_fired", 0)
rate = callbacks_fired / elapsed if elapsed > 0 else 0
return dmc.Stack([
dmc.Group([
dmc.Text("Elapsed:", size="sm"),
dmc.Text(f"{elapsed:.1f}s", fw=600, size="sm"),
], justify="space-between"),
dmc.Group([
dmc.Text("Callbacks:", size="sm"),
dmc.Text(f"{callbacks_fired}", fw=600, size="sm"),
], justify="space-between"),
dmc.Group([
dmc.Text("Rate:", size="sm"),
dmc.Text(f"{rate:.2f}/s", fw=600, size="sm"),
], justify="space-between"),
], gap=4)
@callback(
Output("callback-activity", "children"),
Input("metrics-store", "data"),
)
def update_callback_activity(metrics):
"""Update callback activity display."""
return dmc.Stack([
dmc.Group([
dmc.Text("Nodes Added:", size="sm"),
dmc.Badge(str(metrics.get("nodes_added", 0)), color="green", size="sm"),
], justify="space-between"),
dmc.Group([
dmc.Text("Nodes Removed:", size="sm"),
dmc.Badge(str(metrics.get("nodes_removed", 0)), color="red", size="sm"),
], justify="space-between"),
dmc.Group([
dmc.Text("Edges Added:", size="sm"),
dmc.Badge(str(metrics.get("edges_added", 0)), color="blue", size="sm"),
], justify="space-between"),
dmc.Group([
dmc.Text("Edges Removed:", size="sm"),
dmc.Badge(str(metrics.get("edges_removed", 0)), color="orange", size="sm"),
], justify="space-between"),
dmc.Group([
dmc.Text("Drags:", size="sm"),
dmc.Badge(str(metrics.get("drags_completed", 0)), color="violet", size="sm"),
], justify="space-between"),
dmc.Group([
dmc.Text("Connections:", size="sm"),
dmc.Badge(str(metrics.get("connections_made", 0)), color="cyan", size="sm"),
], justify="space-between"),
], gap=4)
@callback(
Output("current-state", "children"),
Input("stress-flow", "nodes"),
Input("stress-flow", "edges"),
)
def update_current_state(nodes, edges):
"""Display current node and edge counts."""
node_count = len(nodes) if nodes else 0
edge_count = len(edges) if edges else 0
return dmc.Stack([
dmc.Group([
dmc.Text("Total Nodes:", size="sm"),
dmc.Text(f"{node_count}", fw=700, size="lg", c="blue"),
], justify="space-between"),
dmc.Group([
dmc.Text("Total Edges:", size="sm"),
dmc.Text(f"{edge_count}", fw=700, size="lg", c="grape"),
], justify="space-between"),
], gap=4)
@callback(
Output("last-events", "children"),
Input("stress-flow", "clickedNode"),
Input("stress-flow", "hoveredNode"),
Input("stress-flow", "draggedNode"),
)
def update_last_events(clicked, hovered, dragged):
"""Display last event information."""
events = []
if clicked:
node_id = clicked.get("id", clicked) if isinstance(clicked, dict) else clicked
events.append(dmc.Text(f"Clicked: {node_id}", size="xs"))
if hovered:
node_id = hovered.get("id", hovered) if isinstance(hovered, dict) else hovered
events.append(dmc.Text(f"Hovered: {node_id}", size="xs", c="dimmed"))
if dragged:
node_id = dragged.get("id", "?")
is_dragging = dragged.get("isDragging", False)
status = "dragging..." if is_dragging else "dropped"
events.append(dmc.Text(f"Drag: {node_id} ({status})", size="xs"))
if not events:
events.append(dmc.Text("No recent events", size="xs", c="dimmed"))
return dmc.Stack(events, gap=2)
@callback(
Output("event-log", "children"),
Input("event-log-store", "data"),
)
def update_event_log(log):
"""Display the event log."""
if not log:
return dmc.Text("No events yet...", c="dimmed", size="sm")
return html.Div([
html.Div(entry, style={"padding": "2px 0"}) for entry in log
])
@callback(
Output("event-log-store", "data", allow_duplicate=True),
Input("btn-clear-log", "n_clicks"),
prevent_initial_call=True,
)
def clear_log(n_clicks):
"""Clear the event log."""
return []
if __name__ == "__main__":
app.run(debug=True, port=8093):defaultExpanded: false :withExpandedButton: true
How it works
generate_grid_nodes/generate_grid_edges— build a 50-node grid up front so the stress test starts with a non-trivial graph- Batch buttons (
Add Batch Nodes,Stress Connect All, etc.) — each dispatches one callback that mutatesnodes/edgeswithallow_duplicate=Trueoutputs so multiple buttons can target the same props clickedNode,draggedNode,lastConnection,selectedNodes,deletedNodes/deletedEdges— five independent listener callbacks log every interaction type into a shared event log storedcc.Interval— polls a metrics store every 500ms to compute a live callbacks-per-second rate, which is what the "stress test" is actually measuring
Phase 1 Features
A showcase of the React Flow 12.10.1 upgrade props: a drag threshold before a connection starts, z-index elevation for selected elements, auto-panning when tabbing between nodes, and an edge toolbar that appears on selection.
# File: examples/25_phase1_features.py
"""
Example 25: Phase 1 Feature Showcase
=====================================
Demonstrates new features from React Flow 12.10.1 upgrade:
- connectionDragThreshold: drag distance before connection starts
- zIndexMode: selected elements elevate above others
- autoPanOnNodeFocus: viewport pans when tabbing to nodes
- Glass connection line: styled preview during edge creation
- Type-colored MiniMap: node types shown as distinct colors
- EdgeToolbar on ButtonEdge: toolbar appears on selected edges
"""
import dash
from dash import html, dcc, callback, Input, Output, State
import dash_flows
app = dash.Dash(__name__)
nodes = [
# Input nodes (green in minimap)
{
"id": "source-1",
"type": "input",
"data": {"label": "API Source", "sublabel": "REST Endpoint"},
"position": {"x": 50, "y": 50},
},
{
"id": "source-2",
"type": "input",
"data": {"label": "DB Source", "sublabel": "PostgreSQL"},
"position": {"x": 50, "y": 200},
},
# Default nodes (blue in minimap)
{
"id": "validate",
"type": "default",
"data": {"label": "Validate", "sublabel": "Schema Check"},
"position": {"x": 300, "y": 50},
},
{
"id": "transform",
"type": "default",
"data": {"label": "Transform", "sublabel": "Map & Filter"},
"position": {"x": 300, "y": 200},
},
{
"id": "merge",
"type": "default",
"data": {"label": "Merge", "sublabel": "Join Data"},
"position": {"x": 550, "y": 125},
},
# Output nodes (purple in minimap)
{
"id": "output-1",
"type": "output",
"data": {"label": "Dashboard", "sublabel": "Visualization"},
"position": {"x": 800, "y": 50},
},
{
"id": "output-2",
"type": "output",
"data": {"label": "Export", "sublabel": "CSV File"},
"position": {"x": 800, "y": 200},
},
# Toolbar node (amber in minimap)
{
"id": "config",
"type": "toolbar",
"data": {"label": "Config", "sublabel": "Settings"},
"position": {"x": 550, "y": 300},
},
]
edges = [
# Standard edges
{"id": "e1", "source": "source-1", "target": "validate", "type": "smoothstep", "animated": True},
{"id": "e2", "source": "source-2", "target": "transform", "type": "smoothstep", "animated": True},
{"id": "e3", "source": "validate", "target": "merge", "type": "smoothstep"},
{"id": "e4", "source": "transform", "target": "merge", "type": "smoothstep"},
# ButtonEdge with toolbar — select this edge to see the EdgeToolbar
{
"id": "e5",
"source": "merge",
"target": "output-1",
"type": "button",
"data": {
"label": "Primary",
"showButton": False, # Hide inline delete button
"showToolbar": True, # EdgeToolbar appears when selected instead
},
},
{
"id": "e6",
"source": "merge",
"target": "output-2",
"type": "button",
"data": {
"label": "Secondary",
"showButton": False,
"showToolbar": True,
},
},
# Edge to config
{"id": "e7", "source": "merge", "target": "config", "type": "smoothstep", "style": {"strokeDasharray": "5 5"}},
]
app.layout = html.Div([
html.H1("Phase 1 Feature Showcase"),
html.Div([
html.H3("New Features Demonstrated"),
html.Ul([
html.Li([
html.Strong("connectionDragThreshold=10"),
" — Drag 10px from a handle before the connection line appears (prevents accidental connections)",
]),
html.Li([
html.Strong("zIndexMode='elevate'"),
" — Selected nodes and their connected edges elevate above all other elements",
]),
html.Li([
html.Strong("autoPanOnNodeFocus=True"),
" — Click inside the flow canvas first, then press Tab to cycle through nodes; viewport auto-pans to keep the focused node visible",
]),
html.Li([
html.Strong("Glass Connection Line"),
" — Drag from any handle to see the new blue dashed connection preview with glow effect",
]),
html.Li([
html.Strong("Type-Colored MiniMap"),
" — Check the minimap (bottom-right): green=input, blue=default, purple=output, amber=toolbar",
]),
html.Li([
html.Strong("EdgeToolbar"),
" — Click the 'Primary' or 'Secondary' edge line (not a node) to select it — a glass toolbar with edit/delete buttons appears above the edge",
]),
]),
], style={"padding": "0 16px", "fontSize": "14px"}),
dash_flows.DashFlows(
id="feature-flow",
nodes=nodes,
edges=edges,
# New Phase 1 props
connectionDragThreshold=10,
zIndexMode="elevate",
autoPanOnNodeFocus=True,
# Standard display options
style={"height": "600px", "border": "1px solid #ddd"},
fitView=True,
fitViewOptions={"padding": 0.15},
showControls=True,
showMiniMap=True,
showBackground=True,
colorScheme="default",
),
])
if __name__ == "__main__":
app.run(debug=True, port=8025):defaultExpanded: false :withExpandedButton: true
How it works
connectionDragThreshold=10— the connection line only appears after dragging 10px from a handle, preventing accidental connections on a light clickzIndexMode="elevate"— selected nodes and their connected edges render above everything elseautoPanOnNodeFocus=True— tabbing through nodes (after clicking into the canvas) pans the viewport to keep the focused node visibledata.showToolbar=Trueon atype="button"edge — replaces the inline delete button with a floatingEdgeToolbarthat appears when the edge is selected- MiniMap node coloring is automatic by node
type(green=input, blue=default, purple=output, amber=toolbar) — no extra prop needed
Accessibility
Custom ARIA labels for the diagram, minimap, and controls, per-node/per-edge ariaLabel overrides, and keyboard-only navigation via focusable nodes and edges.
# File: examples/29_accessibility.py
"""
Example 29: Accessibility / ARIA Pass-through
==============================================
Demonstrates ARIA label customization, keyboard navigation, and focus
management for screen-reader and keyboard-only users.
Key features:
- ariaLabelConfig: custom labels for the diagram, minimap, and controls
- Per-node / per-edge ariaLabel: individual element descriptions
- nodesFocusable / edgesFocusable: Tab key navigates between elements
- disableKeyboardA11y=False: arrow-key panning and keyboard shortcuts active
"""
import dash
from dash import html, dcc, callback, Input, Output
import dash_flows
app = dash.Dash(__name__)
nodes = [
{
"id": "start",
"type": "input",
"data": {"label": "Start", "sublabel": "Entry point"},
"position": {"x": 50, "y": 120},
"ariaLabel": "Start process node — entry point of the workflow",
},
{
"id": "validate",
"type": "default",
"data": {"label": "Validate", "sublabel": "Check data"},
"position": {"x": 280, "y": 50},
"ariaLabel": "Validation node — checks input data for correctness",
},
{
"id": "process",
"type": "default",
"data": {"label": "Process", "sublabel": "Transform"},
"position": {"x": 280, "y": 200},
"ariaLabel": "Processing node — transforms validated data",
},
{
"id": "end",
"type": "output",
"data": {"label": "Complete", "sublabel": "Output"},
"position": {"x": 510, "y": 120},
"ariaLabel": "Completion node — final output of the workflow",
},
]
edges = [
{"id": "e1", "source": "start", "target": "validate", "ariaLabel": "Flow from start to validation"},
{"id": "e2", "source": "start", "target": "process", "ariaLabel": "Flow from start to processing"},
{"id": "e3", "source": "validate", "target": "end", "ariaLabel": "Flow from validation to completion"},
{"id": "e4", "source": "process", "target": "end", "ariaLabel": "Flow from processing to completion"},
]
aria_config = {
"rfDiagram": "Interactive workflow diagram with 4 nodes and 4 connections",
"miniMap": "Minimap navigation — overview of the workflow diagram",
"controls": "Zoom and pan controls for the workflow diagram",
}
# Keyboard shortcut reference table
shortcuts = [
("Tab / Shift+Tab", "Move focus between nodes and edges"),
("Enter / Space", "Select the focused element"),
("Escape", "Clear selection / close any overlay"),
("Arrow keys", "Pan the viewport (when focus is on the canvas)"),
("+ / −", "Zoom in / zoom out"),
("Delete / Backspace","Delete selected node or edge"),
]
app.layout = html.Div([
html.H2("Accessibility / ARIA Labels"),
html.P(
"Tab through nodes and edges using the keyboard. "
"Each element has an ARIA label that screen readers will announce. "
"Inspect the DOM to see all applied aria-* attributes.",
style={"color": "#666", "marginBottom": "12px"},
),
# Keyboard shortcut reference
html.Details([
html.Summary("⌨ Keyboard shortcuts (click to expand)",
style={"cursor": "pointer", "fontWeight": "500", "fontSize": "13px", "color": "#374151"}),
html.Table([
html.Tbody([
html.Tr([
html.Td(k, style={"padding": "3px 10px 3px 0", "fontFamily": "monospace",
"fontSize": "12px", "whiteSpace": "nowrap", "color": "#1d4ed8"}),
html.Td(v, style={"padding": "3px 0", "fontSize": "12px", "color": "#374151"}),
])
for k, v in shortcuts
]),
], style={"borderCollapse": "collapse", "marginTop": "6px"}),
], style={
"background": "#f0f7ff", "border": "1px solid #bfdbfe",
"borderRadius": "8px", "padding": "10px 14px", "marginBottom": "12px",
}),
dash_flows.DashFlows(
id="accessible-flow",
nodes=nodes,
edges=edges,
ariaLabelConfig=aria_config,
fitView=True,
showControls=True,
showMiniMap=True,
showBackground=True,
disableKeyboardA11y=False,
nodesFocusable=True,
edgesFocusable=True,
style={"height": "420px"},
),
# Live info panel
html.Div(id="a11y-info", style={
"marginTop": "12px", "padding": "12px 14px",
"background": "#f9fafb", "border": "1px solid #e5e7eb",
"borderRadius": "8px", "minHeight": "56px",
}),
], style={"padding": "20px"})
@callback(
Output("a11y-info", "children"),
Input("accessible-flow", "clickedNode"),
Input("accessible-flow", "clickedEdge"),
prevent_initial_call=True,
)
def show_element_info(clicked_node, clicked_edge):
from dash import ctx
if ctx.triggered_id == "accessible-flow" and clicked_node:
nid = clicked_node["id"]
label = clicked_node["data"].get("label", nid)
# Find the matching ariaLabel
aria = next((n["ariaLabel"] for n in nodes if n["id"] == nid), "No ARIA label set")
return html.Div([
html.Strong(f"Node clicked: {label}"),
html.Br(),
html.Span(f"ARIA label: \"{aria}\"",
style={"fontSize": "12px", "color": "#6b7280", "fontStyle": "italic"}),
html.Br(),
html.Span("Tip: Tab to next node, Delete/Backspace to remove selection",
style={"fontSize": "11px", "color": "#9ca3af"}),
])
if ctx.triggered_id == "accessible-flow" and clicked_edge:
eid = clicked_edge["id"]
aria = next((e["ariaLabel"] for e in edges if e["id"] == eid), "No ARIA label set")
return html.Div([
html.Strong(f"Edge clicked: {eid}"),
html.Br(),
html.Span(f"ARIA label: \"{aria}\"",
style={"fontSize": "12px", "color": "#6b7280", "fontStyle": "italic"}),
])
return html.Span("Click a node or edge to see its ARIA label.",
style={"color": "#9ca3af", "fontSize": "13px"})
if __name__ == "__main__":
app.run(debug=True, port=8029):defaultExpanded: false :withExpandedButton: true
How it works
ariaLabelConfig— supplies custom labels for therfDiagram,miniMap, andcontrolsregions, read by screen readersariaLabelon individual node/edge dicts — gives each element its own accessible description, shown here in the info panel below the canvasnodesFocusable/edgesFocusable— makes every node and edge reachable via Tab / Shift+TabdisableKeyboardA11y=False— keeps arrow-key panning, Enter/Space selection, and Delete/Backspace shortcuts activeclickedNode/clickedEdge— read in a callback to display which element was activated and its ARIA label
Resize Constraints
Min/max width and height bounds, plus locked aspect ratios, applied to both ResizableNode and GroupNode.
# File: examples/30_resize_constraints.py
"""
Example 30: Node Resize Constraints
====================================
Demonstrates resize constraints on ResizableNode and GroupNode:
- keepAspectRatio: Lock aspect ratio during resize
- maxWidth / maxHeight: Maximum size limits
"""
import dash
from dash import html
import dash_flows
app = dash.Dash(__name__)
nodes = [
# Resizable node with aspect ratio lock
{
"id": "aspect-locked",
"type": "resizable",
"data": {
"label": "Aspect Ratio Locked (2:1)",
"handles": [
{"id": "source-right", "type": "source", "position": "right"},
{"id": "target-left", "type": "target", "position": "left"},
],
"keepAspectRatio": True,
"initialWidth": 300,
"initialHeight": 150,
"minWidth": 200,
"minHeight": 100,
},
"position": {"x": 50, "y": 50},
"style": {"width": 300, "height": 150},
},
# Resizable node with max dimensions
{
"id": "max-constrained",
"type": "resizable",
"data": {
"label": "Max 400x200",
"handles": [
{"id": "source-right", "type": "source", "position": "right"},
{"id": "target-left", "type": "target", "position": "left"},
],
"maxWidth": 400,
"maxHeight": 200,
"minWidth": 150,
"minHeight": 80,
},
"position": {"x": 50, "y": 280},
"style": {"width": 250, "height": 120},
},
# Group node with resize constraints
{
"id": "constrained-group",
"type": "group",
"data": {
"label": "Constrained Group (max 500x400)",
"maxWidth": 500,
"maxHeight": 400,
"minWidth": 200,
"minHeight": 150,
},
"position": {"x": 450, "y": 50},
"style": {"width": 350, "height": 300},
},
# Child inside group
{
"id": "child-1",
"type": "default",
"data": {"label": "Child Node A"},
"position": {"x": 30, "y": 50},
"parentId": "constrained-group",
"extent": "parent",
},
{
"id": "child-2",
"type": "default",
"data": {"label": "Child Node B"},
"position": {"x": 30, "y": 150},
"parentId": "constrained-group",
"extent": "parent",
},
# Group with aspect ratio lock
{
"id": "aspect-group",
"type": "group",
"data": {
"label": "Aspect Locked Group",
"keepAspectRatio": True,
"minWidth": 150,
"minHeight": 150,
},
"position": {"x": 450, "y": 380},
"style": {"width": 250, "height": 250},
},
]
edges = [
{"id": "e1", "source": "aspect-locked", "target": "max-constrained"},
{"id": "e2", "source": "child-1", "target": "child-2"},
]
app.layout = html.Div([
html.H2("Node Resize Constraints"),
# Constraint summary
html.Div([
html.Strong("How to resize: ", style={"fontSize": "13px"}),
html.Span("Click a node to select it, then drag the blue corner / edge handles that appear.",
style={"fontSize": "13px", "color": "#555"}),
html.Br(),
html.Div([
html.Span("■ Top-left ResizableNode", style={"color": "#3b82f6", "fontWeight": "600", "fontSize": "12px"}),
html.Span(" — aspect ratio locked 2:1, min 200×100", style={"fontSize": "12px", "color": "#555", "marginRight": "18px"}),
html.Span("■ Bottom-left ResizableNode", style={"color": "#8b5cf6", "fontWeight": "600", "fontSize": "12px"}),
html.Span(" — max 400×200, min 150×80", style={"fontSize": "12px", "color": "#555"}),
], style={"marginTop": "4px"}),
html.Div([
html.Span("■ Right GroupNode", style={"color": "#10b981", "fontWeight": "600", "fontSize": "12px"}),
html.Span(" — max 500×400, min 200×150", style={"fontSize": "12px", "color": "#555", "marginRight": "18px"}),
html.Span("■ Bottom-right GroupNode", style={"color": "#f59e0b", "fontWeight": "600", "fontSize": "12px"}),
html.Span(" — aspect ratio locked, min 150×150", style={"fontSize": "12px", "color": "#555"}),
], style={"marginTop": "2px"}),
], style={
"background": "#f9fafb", "border": "1px solid #e5e7eb",
"borderRadius": "8px", "padding": "10px 14px", "marginBottom": "12px",
}),
dash_flows.DashFlows(
id="resize-flow",
nodes=nodes,
edges=edges,
fitView=True,
showControls=True,
showMiniMap=True,
showBackground=True,
style={"height": "700px"},
),
], style={"padding": "20px"})
if __name__ == "__main__":
app.run(debug=True, port=8030):defaultExpanded: false :withExpandedButton: true
How it works
data.minWidth/data.maxWidth/data.minHeight/data.maxHeight— clamp how far aresizablenode orgroupnode can be dragged in each dimensiondata.keepAspectRatio=True— locks the width:height ratio while resizing, regardless of which handle is draggeddata.handles— customizes which side(s) of aresizablenode expose source/target connection points- Group resize constraints apply to the container only — child nodes with
parentId/extent: "parent"stay clipped to whatever size the group currently is
Undo / Redo
History tracking for node moves, connections, and deletions, with dedicated Undo/Redo buttons and a live history counter.
# File: examples/32_undo_redo.py
"""
Example 32: Undo/Redo System
=============================
Demonstrates the undo/redo history system. Move nodes, create connections,
delete elements — then undo and redo your changes.
Features demonstrated:
- Node movement undo/redo (drag a node, then undo)
- Node/edge deletion undo/redo (select + Delete key, or use button)
- Connection creation undo (draw an edge between nodes)
- History counter shows pending undo/redo steps
Uses experimental_useOnNodesChangeMiddleware and
experimental_useOnEdgesChangeMiddleware from React Flow 12.10.1.
"""
import dash
from dash import html, dcc, callback, Input, Output, State
import dash_flows
app = dash.Dash(__name__)
initial_nodes = [
{"id": "a", "type": "input", "data": {"label": "Node A", "sublabel": "Source"}, "position": {"x": 50, "y": 150}},
{"id": "b", "type": "default", "data": {"label": "Node B", "sublabel": "Process 1"}, "position": {"x": 280, "y": 50}},
{"id": "c", "type": "default", "data": {"label": "Node C", "sublabel": "Process 2"}, "position": {"x": 280, "y": 250}},
{"id": "d", "type": "output", "data": {"label": "Node D", "sublabel": "Output"}, "position": {"x": 510, "y": 150}},
]
initial_edges = [
{"id": "e-ab", "source": "a", "target": "b"},
{"id": "e-ac", "source": "a", "target": "c"},
{"id": "e-bd", "source": "b", "target": "d"},
]
btn_style = {
"padding": "8px 16px",
"border": "1px solid #ddd",
"borderRadius": "6px",
"cursor": "pointer",
"fontSize": "13px",
"fontWeight": "500",
"minWidth": "80px",
}
app.layout = html.Div([
html.H2("Undo/Redo System"),
# Instruction panel
html.Div([
html.Div([
html.Strong("Try these actions:"),
html.Ul([
html.Li("Drag a node to a new position → then Undo"),
html.Li("Select a node (click) → press Delete or Backspace → then Undo"),
html.Li("Draw a new edge by dragging from a handle → then Undo"),
html.Li("Use the Delete Selected button below to remove selected nodes"),
], style={"margin": "6px 0 0 0", "paddingLeft": "18px", "fontSize": "13px", "color": "#555"}),
]),
], style={
"background": "#f0f7ff",
"border": "1px solid #bfdbfe",
"borderRadius": "8px",
"padding": "10px 14px",
"marginBottom": "10px",
"fontSize": "13px",
}),
# Controls row
html.Div([
html.Button(
"⟲ Undo", id="btn-undo",
style={**btn_style, "background": "#fee2e2", "borderColor": "#fca5a5"},
disabled=True,
),
html.Button(
"⟳ Redo", id="btn-redo",
style={**btn_style, "background": "#dbeafe", "borderColor": "#93c5fd"},
disabled=True,
),
html.Button(
"🗑 Delete Selected", id="btn-delete",
style={**btn_style, "background": "#fef9c3", "borderColor": "#fde047"},
),
html.Span(id="history-info", style={
"padding": "8px 14px",
"background": "#f0f0f0",
"borderRadius": "6px",
"fontSize": "12px",
"color": "#555",
"marginLeft": "4px",
}),
], style={"display": "flex", "gap": "8px", "alignItems": "center", "marginBottom": "10px"}),
# Store to hold selected node IDs for delete action
dcc.Store(id="selected-ids-store"),
dash_flows.DashFlows(
id="undo-redo-flow",
nodes=initial_nodes,
edges=initial_edges,
fitView=True,
enableUndoRedo=True,
undoRedoMaxHistory=50,
showControls=True,
showMiniMap=True,
showBackground=True,
style={"height": "500px"},
),
html.Div(id="action-log", style={
"marginTop": "10px",
"padding": "8px 12px",
"background": "#fafafa",
"border": "1px solid #e5e7eb",
"borderRadius": "6px",
"fontSize": "12px",
"color": "#6b7280",
"minHeight": "32px",
}),
], style={"padding": "20px"})
@callback(
Output("undo-redo-flow", "undoRedoAction"),
Input("btn-undo", "n_clicks"),
Input("btn-redo", "n_clicks"),
prevent_initial_call=True,
)
def handle_undo_redo(undo_clicks, redo_clicks):
from dash import ctx
if ctx.triggered_id == "btn-undo":
return {"action": "undo"}
elif ctx.triggered_id == "btn-redo":
return {"action": "redo"}
return dash.no_update
@callback(
Output("btn-undo", "disabled"),
Output("btn-redo", "disabled"),
Output("history-info", "children"),
Input("undo-redo-flow", "undoRedoState"),
)
def update_buttons(state):
if not state:
return True, True, "History: 0 undo / 0 redo"
undo_count = state.get("undoCount", 0)
redo_count = state.get("redoCount", 0)
label = f"History: {undo_count} undo / {redo_count} redo"
return (
not state.get("canUndo", False),
not state.get("canRedo", False),
label,
)
@callback(
Output("selected-ids-store", "data"),
Input("undo-redo-flow", "selectedNodes"),
Input("undo-redo-flow", "selectedEdges"),
)
def track_selected(sel_nodes, sel_edges):
# selectedNodes / selectedEdges are already arrays of string IDs
return {"nodes": sel_nodes or [], "edges": sel_edges or []}
@callback(
Output("undo-redo-flow", "deleteElementsAction"),
Output("action-log", "children"),
Input("btn-delete", "n_clicks"),
State("selected-ids-store", "data"),
prevent_initial_call=True,
)
def delete_selected(n_clicks, selected):
if not selected:
return dash.no_update, "Nothing selected to delete."
node_ids = selected.get("nodes", [])
edge_ids = selected.get("edges", [])
if not node_ids and not edge_ids:
return dash.no_update, "Nothing selected to delete."
parts = []
if node_ids:
parts.append(f"{len(node_ids)} node(s)")
if edge_ids:
parts.append(f"{len(edge_ids)} edge(s)")
msg = f"Deleted {' and '.join(parts)}. Press ⟲ Undo to restore."
# Route through deleteElementsAction so React Flow's normal deletion
# flow fires — the undo/redo middleware can then capture the snapshot.
return {"nodeIds": node_ids, "edgeIds": edge_ids}, msg
if __name__ == "__main__":
app.run(debug=True, port=8032):defaultExpanded: false :withExpandedButton: true
How it works
enableUndoRedo=TrueandundoRedoMaxHistory=50— turns on history tracking and caps how many snapshots are keptundoRedoAction— set to{"action": "undo"}or{"action": "redo"}from the button callbacks to step through historyundoRedoState— an output prop withcanUndo/canRedo/undoCount/redoCount, used here to enable/disable the buttons and render the history badgedeleteElementsAction— routes the "Delete Selected" button through React Flow's normal deletion pipeline (rather than mutatingnodes/edgesdirectly) so the undo/redo middleware can capture the changeselectedNodes/selectedEdges— tracked in a store so the delete button knows what to remove
Computing Flows
Topological sort and value propagation across a small pipeline graph: input nodes hold values, operation nodes combine them, and an output node shows the final result.
# File: examples/33_computing_flows.py
"""
Example 33: Computing Flows
============================
Demonstrates graph traversal and computation propagation.
Input nodes have numbers, operation nodes perform math,
output nodes show results. Click "Compute" to propagate values.
JS does topological sort only — Python handles all business logic.
Pipeline: (A + B) × C = (10 + 5) × 3 = 45
"""
import dash
from dash import html, callback, Input, Output, State
import dash_flows
app = dash.Dash(__name__)
initial_nodes = [
{
"id": "input-a",
"type": "input",
"data": {"label": "Input A", "sublabel": "Value: 10", "computedValue": 10},
"position": {"x": 50, "y": 50},
},
{
"id": "input-b",
"type": "input",
"data": {"label": "Input B", "sublabel": "Value: 5", "computedValue": 5},
"position": {"x": 50, "y": 200},
},
{
"id": "input-c",
"type": "input",
"data": {"label": "Input C", "sublabel": "Value: 3", "computedValue": 3},
"position": {"x": 50, "y": 350},
},
{
"id": "add",
"type": "default",
"data": {"label": "Add (+)", "sublabel": "A + B", "operation": "add", "computedValue": None},
"position": {"x": 300, "y": 100},
},
{
"id": "multiply",
"type": "default",
"data": {"label": "Multiply (×)", "sublabel": "result × C", "operation": "multiply", "computedValue": None},
"position": {"x": 550, "y": 200},
},
{
"id": "output",
"type": "output",
"data": {"label": "Result", "sublabel": "Waiting...", "computedValue": None},
"position": {"x": 800, "y": 200},
},
]
initial_edges = [
{"id": "e1", "source": "input-a", "target": "add"},
{"id": "e2", "source": "input-b", "target": "add"},
{"id": "e3", "source": "add", "target": "multiply"},
{"id": "e4", "source": "input-c", "target": "multiply"},
{"id": "e5", "source": "multiply", "target": "output"},
]
btn_style = {
"padding": "10px 24px",
"border": "none",
"borderRadius": "8px",
"background": "linear-gradient(135deg, #3b82f6, #8b5cf6)",
"color": "white",
"cursor": "pointer",
"fontSize": "14px",
"fontWeight": "600",
}
reset_btn_style = {
"padding": "10px 18px",
"border": "1px solid #ddd",
"borderRadius": "8px",
"background": "#f9fafb",
"color": "#374151",
"cursor": "pointer",
"fontSize": "14px",
}
app.layout = html.Div([
html.H2("Computing Flows"),
html.P(
"Pipeline: (A + B) × C → Result. Click ⚡ Compute to propagate values through the graph.",
style={"color": "#666"},
),
html.Div([
html.Button("⚡ Compute", id="btn-compute", style=btn_style),
html.Button("↺ Reset", id="btn-reset", style=reset_btn_style),
html.Span(id="compute-status", style={
"padding": "8px 16px", "background": "#f0f0f0",
"borderRadius": "6px", "fontSize": "13px",
}),
], style={"display": "flex", "gap": "10px", "alignItems": "center", "marginBottom": "10px"}),
dash_flows.DashFlows(
id="compute-flow",
nodes=initial_nodes,
edges=initial_edges,
fitView=True,
showControls=True,
showMiniMap=True,
showBackground=True,
smartHandles=True,
style={"height": "500px"},
),
], style={"padding": "20px"})
@callback(
Output("compute-flow", "computeAction"),
Input("btn-compute", "n_clicks"),
prevent_initial_call=True,
)
def trigger_compute(n):
return {"action": "compute"}
@callback(
Output("compute-flow", "nodes"),
Output("compute-status", "children"),
Input("compute-flow", "computeResult"),
State("compute-flow", "nodes"),
prevent_initial_call=True,
)
def process_computation(result, nodes):
if not result:
return dash.no_update, dash.no_update
traversal = result.get("traversalOrder", [])
node_inputs_meta = result.get("nodeInputs", {})
node_map = {n["id"]: n for n in nodes}
# Seed computed_values from nodes that already have a value (source nodes).
# We maintain our own dict so intermediate results propagate correctly
# through the topological order — the JS snapshot only knows values that
# existed BEFORE this compute call.
computed_values = {}
for n in nodes:
val = n["data"].get("computedValue")
if val is not None:
computed_values[n["id"]] = val
steps = []
for node_id in traversal:
node = node_map.get(node_id)
if not node:
continue
meta = node_inputs_meta.get(node_id, {})
incoming = meta.get("inputs", []) # [{nodeId, value, data}, ...]
operation = node["data"].get("operation")
if not incoming:
# Source node — keep its seed value, nothing to compute
continue
# Use our locally propagated computed_values for inputs, not the
# stale JS snapshot values (which are None for intermediate nodes).
input_values = [
computed_values[inp["nodeId"]]
for inp in incoming
if inp["nodeId"] in computed_values
]
if not input_values:
continue
if operation == "add":
computed = sum(input_values)
steps.append(f"{node_id}: {' + '.join(str(v) for v in input_values)} = {computed}")
elif operation == "multiply":
computed = 1
for v in input_values:
computed *= v
steps.append(f"{node_id}: {' × '.join(str(v) for v in input_values)} = {computed}")
else:
# Output node or unknown: pass through
computed = sum(input_values)
steps.append(f"{node_id}: result = {computed}")
computed_values[node_id] = computed
node["data"] = dict(node["data"])
node["data"]["computedValue"] = computed
node["data"]["sublabel"] = f"= {computed}"
updated_nodes = list(node_map.values())
final_value = computed_values.get("output", "?")
status = f"✓ Result = {final_value} ({' → '.join(steps)})"
return updated_nodes, status
@callback(
Output("compute-flow", "nodes", allow_duplicate=True),
Output("compute-status", "children", allow_duplicate=True),
Input("btn-reset", "n_clicks"),
prevent_initial_call=True,
)
def reset_flow(n):
import copy
return copy.deepcopy(initial_nodes), "Reset — click ⚡ Compute to run again."
if __name__ == "__main__":
app.run(debug=True, port=8033):defaultExpanded: false :withExpandedButton: true
How it works
computeAction={"action": "compute"}— triggers a client-side topological sort of the graphcomputeResult— an output prop withtraversalOrder(node IDs in dependency order) andnodeInputs(each node's incoming values); JS only does the sort, Python does all the math- The callback walks
traversalOrder, applies each node'sdata.operation(add/multiply) to its inputs, and writes the running value back intodata.computedValue/data.sublabel smartHandles=True— lets edges route to whichever side of a node is closest, useful once the pipeline fans out- The "Reset" button restores
initial_nodesviacopy.deepcopyso a fresh compute run starts from the original seed values
Viewport Portal
Floating annotations rendered at specific flow coordinates that pan and zoom together with the canvas, with a small editor to add, edit, and remove them.
# File: examples/34_viewport_portal.py
"""
Example 34: ViewportPortal — Floating Annotations
===================================================
Demonstrates floating annotations rendered at specific flow coordinates
via ViewportPortal. Overlays move with pan/zoom.
Features:
- Add / remove annotations
- Select an annotation from the list and edit its x, y position and content
- Changes update immediately in the flow
"""
import dash
from dash import html, dcc, callback, Input, Output, State
import dash_flows
app = dash.Dash(__name__)
nodes = [
{"id": "server", "type": "default", "data": {"label": "API Server", "sublabel": "Port 8080"}, "position": {"x": 100, "y": 100}},
{"id": "db", "type": "default", "data": {"label": "Database", "sublabel": "PostgreSQL"}, "position": {"x": 400, "y": 100}},
{"id": "cache", "type": "default", "data": {"label": "Cache", "sublabel": "Redis"}, "position": {"x": 400, "y": 280}},
{"id": "client", "type": "input", "data": {"label": "Client App"}, "position": {"x": -150, "y": 100}},
{"id": "cdn", "type": "output", "data": {"label": "CDN"}, "position": {"x": 700, "y": 100}},
]
edges = [
{"id": "e1", "source": "client", "target": "server"},
{"id": "e2", "source": "server", "target": "db"},
{"id": "e3", "source": "server", "target": "cache"},
{"id": "e4", "source": "db", "target": "cdn"},
]
initial_overlays = [
{
"x": -150, "y": 58,
"content": "🌐 External Traffic",
"style": {
"background": "rgba(59,130,246,0.1)", "border": "1px dashed rgba(59,130,246,0.4)",
"padding": "4px 10px", "borderRadius": "6px",
"fontSize": "11px", "color": "#3b82f6", "fontWeight": "600", "whiteSpace": "nowrap",
},
},
{
"x": 200, "y": 68,
"content": "← REST API →",
"style": {
"fontSize": "10px", "color": "#888", "fontStyle": "italic", "whiteSpace": "nowrap",
},
},
{
"x": 380, "y": 200,
"content": "⚡ Hot path",
"style": {
"background": "rgba(245,158,11,0.15)", "border": "1px solid rgba(245,158,11,0.3)",
"padding": "3px 8px", "borderRadius": "4px",
"fontSize": "10px", "color": "#d97706", "fontWeight": "500", "whiteSpace": "nowrap",
},
},
{
"x": 100, "y": -2,
"content": "── Internal Network ──────────────────────",
"style": {
"fontSize": "10px", "color": "#aaa", "letterSpacing": "1px", "whiteSpace": "nowrap",
},
},
]
def _build_options(overlays):
return [
{"label": f"[{i}] {o['content'][:28]}{'…' if len(o['content']) > 28 else ''}",
"value": i}
for i, o in enumerate(overlays)
]
# Shared button style
_btn = {
"padding": "7px 14px", "border": "1px solid #ddd",
"borderRadius": "6px", "background": "#fff", "cursor": "pointer", "fontSize": "13px",
}
app.layout = html.Div([
html.H2("ViewportPortal — Floating Annotations"),
html.P(
"Annotations are anchored to flow coordinates and move with pan/zoom. "
"Select one from the list below to edit its position or content.",
style={"color": "#666"},
),
# Top controls
html.Div([
html.Button("＋ Add Annotation", id="btn-add-overlay", style=_btn),
html.Button("✕ Clear All", id="btn-clear-overlays", style=_btn),
], style={"display": "flex", "gap": "8px", "marginBottom": "10px"}),
# Flow
dash_flows.DashFlows(
id="portal-flow",
nodes=nodes,
edges=edges,
viewportOverlays=initial_overlays,
fitView=True,
showControls=True,
showMiniMap=True,
showBackground=True,
smartHandles=True,
style={"height": "460px"},
),
# Editor panel
html.Div([
html.H4("Edit Annotations", style={"margin": "0 0 10px 0", "fontSize": "14px"}),
html.Div(style={"display": "flex", "gap": "16px", "alignItems": "flex-start"}, children=[
# Annotation list (left side)
html.Div([
html.Div("Select:", style={"fontSize": "12px", "color": "#888", "marginBottom": "4px"}),
dcc.RadioItems(
id="annotation-selector",
options=_build_options(initial_overlays),
value=None,
labelStyle={"display": "block", "fontSize": "13px", "padding": "3px 0", "cursor": "pointer"},
),
], style={"minWidth": "220px"}),
# Edit form (right side)
html.Div(id="edit-form", style={"flex": "1"}, children=[
html.P("Select an annotation to edit it.", style={"color": "#aaa", "fontSize": "13px"}),
]),
]),
], style={
"marginTop": "14px", "padding": "14px",
"border": "1px solid #e5e7eb", "borderRadius": "8px",
"background": "#fafafa",
}),
# Hidden stores
dcc.Store(id="overlays-store", data=initial_overlays),
], style={"padding": "20px"})
# ── Build / update the annotation list & sync overlay store ──────────────────
@callback(
Output("portal-flow", "viewportOverlays"),
Output("overlays-store", "data"),
Output("annotation-selector", "options"),
Input("btn-add-overlay", "n_clicks"),
Input("btn-clear-overlays","n_clicks"),
State("overlays-store", "data"),
prevent_initial_call=True,
)
def add_or_clear(add_clicks, clear_clicks, current):
from dash import ctx
import random
if ctx.triggered_id == "btn-clear-overlays":
return [], [], []
current = current or []
idx = len(current) + 1
new_overlay = {
"x": random.randint(-100, 500),
"y": random.randint(-50, 320),
"content": f"📌 Note #{idx}",
"style": {
"background": "rgba(139,92,246,0.1)", "border": "1px solid rgba(139,92,246,0.3)",
"padding": "4px 10px", "borderRadius": "6px",
"fontSize": "11px", "color": "#7c3aed", "fontWeight": "500", "whiteSpace": "nowrap",
},
}
updated = current + [new_overlay]
opts = _build_options(updated)
return updated, updated, opts
# ── Populate the edit form when an annotation is selected ────────────────────
@callback(
Output("edit-form", "children"),
Input("annotation-selector", "value"),
State("overlays-store", "data"),
)
def show_edit_form(selected_idx, overlays):
if selected_idx is None or not overlays:
return html.P("Select an annotation to edit it.", style={"color": "#aaa", "fontSize": "13px"})
o = overlays[selected_idx]
field = {"width": "80px", "padding": "5px 8px", "border": "1px solid #ddd",
"borderRadius": "5px", "fontSize": "13px"}
content_field = {**field, "width": "260px"}
return html.Div([
html.Div([
html.Label("X position:", style={"fontSize": "12px", "color": "#666", "marginRight": "6px"}),
dcc.Input(id="edit-x", type="number", value=o["x"], style=field, debounce=True),
html.Label("Y position:", style={"fontSize": "12px", "color": "#666", "margin": "0 6px 0 14px"}),
dcc.Input(id="edit-y", type="number", value=o["y"], style=field, debounce=True),
], style={"display": "flex", "alignItems": "center", "marginBottom": "8px"}),
html.Div([
html.Label("Content:", style={"fontSize": "12px", "color": "#666", "marginRight": "6px"}),
dcc.Input(id="edit-content", type="text", value=o["content"], style=content_field, debounce=True),
], style={"display": "flex", "alignItems": "center", "marginBottom": "8px"}),
html.Button("💾 Apply", id="btn-apply-edit", style={
"padding": "7px 18px", "border": "none", "borderRadius": "6px",
"background": "#3b82f6", "color": "white", "cursor": "pointer", "fontSize": "13px",
}),
html.Div(id="edit-feedback", style={"marginTop": "6px", "fontSize": "12px", "color": "#16a34a"}),
])
# Placeholder outputs so Dash doesn't complain about missing IDs on initial load
app.layout.children.append(html.Div([
dcc.Input(id="edit-x", style={"display": "none"}),
dcc.Input(id="edit-y", style={"display": "none"}),
dcc.Input(id="edit-content", style={"display": "none"}),
html.Button(id="btn-apply-edit", style={"display": "none"}),
html.Div(id="edit-feedback"),
], id="_hidden-edit-ids", style={"display": "none"}))
# ── Apply edits ───────────────────────────────────────────────────────────────
@callback(
Output("portal-flow", "viewportOverlays", allow_duplicate=True),
Output("overlays-store", "data", allow_duplicate=True),
Output("annotation-selector", "options", allow_duplicate=True),
Output("edit-feedback", "children"),
Input("btn-apply-edit", "n_clicks"),
State("annotation-selector", "value"),
State("edit-x", "value"),
State("edit-y", "value"),
State("edit-content", "value"),
State("overlays-store", "data"),
prevent_initial_call=True,
)
def apply_edit(n, selected_idx, x, y, content, overlays):
if selected_idx is None or not overlays:
return dash.no_update, dash.no_update, dash.no_update, "Nothing to update."
updated = [dict(o) for o in overlays]
updated[selected_idx] = {
**updated[selected_idx],
"x": x if x is not None else updated[selected_idx]["x"],
"y": y if y is not None else updated[selected_idx]["y"],
"content": content if content is not None else updated[selected_idx]["content"],
}
opts = _build_options(updated)
return updated, updated, opts, f"✓ Annotation [{selected_idx}] updated."
# ── Sync selector options when store resets to initial overlays ──────────────
@callback(
Output("annotation-selector", "options", allow_duplicate=True),
Input("overlays-store", "data"),
prevent_initial_call=True,
)
def sync_options(overlays):
return _build_options(overlays or [])
if __name__ == "__main__":
app.run(debug=True, port=8034):defaultExpanded: false :withExpandedButton: true
How it works
viewportOverlays— a list of{x, y, content, style}dicts rendered viaViewportPortal, anchored to flow coordinates rather than screen pixels- Adding an annotation appends a new overlay dict with randomized coordinates and re-renders the list
- Selecting an annotation from the
RadioItemslist populates an inline edit form (x,y,contentinputs) sourced from adcc.Storeholding the current overlay list - Hidden placeholder inputs mirror the edit-form IDs so the "Apply" callback's
Statereferences resolve even before an annotation has ever been selected - Because overlays are plain data, they can be generated, persisted, or restored exactly like nodes and edges
Sub-flows
Collapsible group nodes: double-click a group (or use the toggle buttons) to collapse it to a compact box and hide its children, while edges to/from the group stay connected.
# File: examples/35_subflows.py
"""
Example 35: Sub-flows (Collapsible Groups)
===========================================
Demonstrates expandable/collapsible group nodes.
Architecture pattern for React Flow sub-flows:
- Edges that cross group boundaries connect to the GROUP node, not to children.
- Edges inside a group connect child-to-child (these are intra-group and safe).
- When a group collapses: children + their internal edges are hidden; the group
shrinks to a compact box; external edges remain connected at the group level.
Double-click a group or use the buttons to toggle collapse/expand.
"""
import dash
from dash import html, callback, Input, Output, State
import dash_flows
app = dash.Dash(__name__)
# ── Initial graph ─────────────────────────────────────────────────────────────
#
# source ──► [group-ingest: parse ──► validate] ──► [group-process: transform ──► enrich] ──► sink
#
# External edges connect to GROUP nodes (not directly to children).
# Internal edges connect children within the same group (safe pattern).
initial_nodes = [
# External nodes
{
"id": "source",
"type": "input",
"data": {"label": "Data Source"},
"position": {"x": 50, "y": 190},
},
# ── Ingestion group ────────────────────────────────────────────────────────
{
"id": "group-ingest",
"type": "group",
"data": {
"label": "Ingestion Pipeline",
"collapsedWidth": 200,
"collapsedHeight": 52,
},
"position": {"x": 260, "y": 60},
"style": {"width": 280, "height": 270},
},
{
"id": "parse",
"type": "default",
"data": {"label": "Parse", "sublabel": "Deserialize"},
"position": {"x": 50, "y": 55},
"parentId": "group-ingest",
"extent": "parent",
},
{
"id": "validate",
"type": "default",
"data": {"label": "Validate", "sublabel": "Schema check"},
"position": {"x": 50, "y": 170},
"parentId": "group-ingest",
"extent": "parent",
},
# ── Processing group ───────────────────────────────────────────────────────
{
"id": "group-process",
"type": "group",
"data": {
"label": "Processing",
"collapsedWidth": 200,
"collapsedHeight": 52,
},
"position": {"x": 680, "y": 60},
"style": {"width": 280, "height": 270},
},
{
"id": "transform",
"type": "default",
"data": {"label": "Transform", "sublabel": "Normalize"},
"position": {"x": 50, "y": 55},
"parentId": "group-process",
"extent": "parent",
},
{
"id": "enrich",
"type": "default",
"data": {"label": "Enrich", "sublabel": "Add metadata"},
"position": {"x": 50, "y": 170},
"parentId": "group-process",
"extent": "parent",
},
# External nodes
{
"id": "sink",
"type": "output",
"data": {"label": "Data Sink"},
"position": {"x": 1100, "y": 190},
},
]
initial_edges = [
# External → Group (NOT external → child inside group — avoids handle timing issues)
{"id": "e-src-ingest", "source": "source", "target": "group-ingest", "animated": True},
# Intra-group edges (child → child within the SAME group — always safe)
{"id": "e-parse-validate", "source": "parse", "target": "validate"},
{"id": "e-transform-enrich", "source": "transform", "target": "enrich"},
# Group → Group (cross-group at group level — safe)
{"id": "e-ingest-process", "source": "group-ingest", "target": "group-process", "animated": True},
# Group → External
{"id": "e-process-sink", "source": "group-process", "target": "sink", "animated": True},
]
# ── Layout ────────────────────────────────────────────────────────────────────
btn_style = {
"padding": "8px 16px",
"border": "1px solid #ddd",
"borderRadius": "6px",
"background": "#fff",
"cursor": "pointer",
"fontSize": "13px",
}
app.layout = html.Div([
html.H2("Sub-flows — Collapsible Groups"),
html.Div([
html.Strong("Pattern: ", style={"fontSize": "13px"}),
html.Span(
"External edges connect to the group container. "
"Internal edges connect children within the same group. "
"Collapsing hides children and shrinks the group; external connections stay intact.",
style={"fontSize": "13px", "color": "#555"},
),
], style={
"background": "#f0f7ff", "border": "1px solid #bfdbfe",
"borderRadius": "8px", "padding": "10px 14px", "marginBottom": "10px",
}),
html.Div([
html.Button("Toggle Ingestion", id="btn-toggle-ingest", style=btn_style),
html.Button("Toggle Processing", id="btn-toggle-process", style=btn_style),
html.Button("Expand All", id="btn-expand-all", style=btn_style),
html.Span(id="collapse-info", style={
"padding": "8px 16px", "background": "#f0f0f0",
"borderRadius": "6px", "fontSize": "12px", "color": "#666",
}),
], style={"display": "flex", "gap": "8px", "alignItems": "center", "marginBottom": "10px"}),
dash_flows.DashFlows(
id="subflow",
nodes=initial_nodes,
edges=initial_edges,
fitView=True,
showControls=True,
showMiniMap=True,
showBackground=True,
style={"height": "520px"},
),
], style={"padding": "20px"})
# ── Callbacks ─────────────────────────────────────────────────────────────────
@callback(
Output("subflow", "toggleCollapseNode"),
Input("btn-toggle-ingest", "n_clicks"),
Input("btn-toggle-process", "n_clicks"),
Input("subflow", "doubleClickedNode"),
prevent_initial_call=True,
)
def toggle_group(ingest, process, dbl_clicked):
from dash import ctx
if ctx.triggered_id == "btn-toggle-ingest":
return "group-ingest"
if ctx.triggered_id == "btn-toggle-process":
return "group-process"
if ctx.triggered_id == "subflow" and dbl_clicked:
node_id = dbl_clicked.get("id", "")
if node_id.startswith("group-"):
return node_id
return dash.no_update
@callback(
Output("subflow", "nodes"),
Input("btn-expand-all", "n_clicks"),
State("subflow", "nodes"),
prevent_initial_call=True,
)
def expand_all(n_clicks, nodes):
"""Force-expand all groups by clearing collapsed state."""
if not nodes:
return dash.no_update
updated = []
for n in nodes:
if n.get("type") == "group" and n.get("data", {}).get("collapsed"):
n = dict(n)
n["data"] = {**n["data"], "collapsed": False}
# Restore original size from initial_nodes
orig = next((x for x in initial_nodes if x["id"] == n["id"]), None)
if orig and orig.get("style"):
n["style"] = orig["style"]
n["hidden"] = False
elif n.get("hidden"):
n = dict(n)
n["hidden"] = False
updated.append(n)
return updated
@callback(
Output("collapse-info", "children"),
Input("subflow", "collapsedGroups"),
)
def show_collapsed(groups):
if not groups:
return "All groups expanded"
return f"Collapsed: {', '.join(groups)}"
if __name__ == "__main__":
app.run(debug=True, port=8035):defaultExpanded: false :withExpandedButton: true
How it works
toggleCollapseNode— set to a group node'sidto flip its collapsed/expanded statecollapsedGroups— an output prop listing the IDs of currently-collapsed groups, shown in the status linedata.collapsedWidth/data.collapsedHeight— the size a group shrinks to once collapsed- Architecture pattern: edges that cross a group boundary connect to the group node, not to a child inside it; edges between two children of the same group connect child-to-child directly
doubleClickedNode— lets users collapse/expand by double-clicking the group header, in addition to the toggle buttons
Source: /advanced
Note for AI agents: This is the static, prerendered view of an interactive Dash application served because we detected a non-JS user agent. Full prose docs:
- /advanced/llms.txt — LLM-friendly documentation
- /sitemap.xml
- /robots.txt