Here is an example of such a component, just to give you a headstart (just use it as inspiration, don't reuse it!):
# vueflow_canvas.py
from __future__ import annotations
import asyncio
from typing import Any, Dict, Iterable, List, Optional, Union
from nicegui.element import Element
class VueFlowCanvas(Element, component="vueflow_canvas.vue"):
"""
NiceGUI wrapper for Vue Flow.
Features:
- set graph from Python
- add, update, remove nodes and edges
- request current graph from Python
- get events in Python
- define simple custom nodes using HTML content
- register external Vue components by global name for custom nodes
"""
def __init__(
self,
*,
height: int = 600,
show_controls: bool = True,
fit_on_init: bool = True,
snap_to_grid: bool = True,
connection_mode: str = "loose", # "loose" or "strict"
allow_user_edit: bool = True,
initial_nodes: Optional[Iterable[Dict[str, Any]]] = None,
initial_edges: Optional[Iterable[Dict[str, Any]]] = None,
core_version: str = "latest",
extras_version: str = "latest",
) -> None:
super().__init__()
self._props.update(
height=height,
showControls=show_controls,
fitOnInit=fit_on_init,
snapToGrid=snap_to_grid,
connectionMode=connection_mode,
allowUserEdit=allow_user_edit,
initialNodes=list(initial_nodes or []),
initialEdges=list(initial_edges or []),
coreVersion=core_version,
extrasVersion=extras_version,
)
# graph bulk ops
def set_graph(self, nodes: Iterable[Dict[str, Any]], edges: Iterable[Dict[str, Any]]) -> None:
self.run_method("setGraph", list(nodes), list(edges))
def clear(self) -> None:
self.run_method("clearGraph")
# node ops
def add_node(self, node: Dict[str, Any]) -> None:
self.run_method("addNode", dict(node))
def update_node(self, node_id: str, patch: Dict[str, Any]) -> None:
self.run_method("updateNode", str(node_id), dict(patch))
def remove_node(self, node_id: str) -> None:
self.run_method("removeNode", str(node_id))
# edge ops
def add_edge(self, edge: Dict[str, Any]) -> None:
self.run_method("addEdge", dict(edge))
def remove_edge(self, edge_id: str) -> None:
self.run_method("removeEdge", str(edge_id))
# view ops
def fit_view(self) -> None:
self.run_method("fitView")
def zoom_in(self) -> None:
self.run_method("zoomIn")
def zoom_out(self) -> None:
self.run_method("zoomOut")
# import and export
def import_graph(self, jsonish: Union[str, Dict[str, Any]]) -> None:
self.run_method("importGraph", jsonish)
def export_graph(self) -> None:
"""Triggers a 'graph' event you can listen to with .on('graph', handler)."""
self.run_method("exportGraph")
def request_graph(self) -> None:
"""Same as export_graph but named for request semantics."""
self.run_method("requestGraph")
async def get_graph(self) -> Dict[str, List[Dict[str, Any]]]:
"""
Awaitable helper that returns the current graph as a dict with 'nodes' and 'edges'.
"""
loop = asyncio.get_running_loop()
fut: asyncio.Future = loop.create_future()
def once(e):
if not fut.done():
fut.set_result(e.args if isinstance(e.args, dict) else e.args[0] if e.args else {"nodes": [], "edges": []})
self.off("graph", once)
self.on("graph", once)
self.run_method("requestGraph")
result = await fut
if not isinstance(result, dict):
return {"nodes": [], "edges": []}
return result
# custom node helpers
def register_html_node_type(self, name: str) -> None:
"""
Registers a node type that renders node.data["html"] inside the node.
Use it by creating nodes with type=name and data={"html": "<b>Hi</b>"}.
"""
self.run_method("registerHtmlNodeType", str(name))
def register_external_node_type(self, name: str, global_var: str) -> None:
"""
Registers a node type backed by a Vue component available at window[global_var].
This is useful if you add a script that defines window.MyCoolNode = {...}.
"""
self.run_method("registerExternalNodeType", str(name), str(global_var))
<!-- vueflow_canvas.vue -->
<template>
<div class="vf-wrap">
<div v-if="showControls" class="vf-bar">
<button class="vf-btn" @click="fitView">Fit</button>
<button class="vf-btn" @click="zoomIn">Zoom +</button>
<button class="vf-btn" @click="zoomOut">Zoom -</button>
<button class="vf-btn" @click="clearGraph">Clear</button>
<span class="vf-stat">nodes: {{ statNodes }} edges: {{ statEdges }}</span>
</div>
<div ref="mount" class="vf-canvas" :style="wrapperStyle"></div>
</div>
</template>
<script>
export default {
props: {
height: {type: Number, default: 600},
showControls: {type: Boolean, default: true},
fitOnInit: {type: Boolean, default: true},
snapToGrid: {type: Boolean, default: true},
connectionMode: {type: String, default: "loose"}, // "strict" or "loose"
allowUserEdit: {type: Boolean, default: true},
// initial data
initialNodes: {type: Array, default: () => []},
initialEdges: {type: Array, default: () => []},
// cdn versions (override if you need pinning)
coreVersion: {type: String, default: "latest"},
extrasVersion: {type: String, default: "latest"}
},
data() {
return {
// inner Vue app and bridge
innerApp: null,
bridge: null,
// live stats
statNodes: 0,
statEdges: 0
};
},
computed: {
wrapperStyle() {
return "height:" + String(this.height) + "px;";
}
},
async mounted() {
await this.ensureVueFlowLoaded();
this.mountInnerApp();
},
beforeUnmount() {
try {
if (this.innerApp) {
this.innerApp.unmount();
}
} catch (_) {
}
this.innerApp = null;
this.bridge = null;
},
methods: {
// public api available from Python via run_method
fitView() {
if (this.bridge && this.bridge.api && this.bridge.api.fitView) {
this.bridge.api.fitView();
}
},
zoomIn() {
if (this.bridge && this.bridge.api && this.bridge.api.zoomIn) {
this.bridge.api.zoomIn();
}
},
zoomOut() {
if (this.bridge && this.bridge.api && this.bridge.api.zoomOut) {
this.bridge.api.zoomOut();
}
},
clearGraph() {
if (!this.bridge) return;
this.setGraph([], []);
},
setGraph(nodes, edges) {
if (!this.bridge || !this.bridge.api) return;
this.bridge.api.setNodes(Array.isArray(nodes) ? nodes : []);
this.bridge.api.setEdges(Array.isArray(edges) ? edges : []);
this.updateStats();
},
addNode(node) {
if (!this.bridge || !this.bridge.api) return;
if (!node || typeof node !== "object") return;
this.bridge.api.addNodes([node]);
this.updateStatsSoon();
},
addEdge(edge) {
if (!this.bridge || !this.bridge.api) return;
if (!edge || typeof edge !== "object") return;
this.bridge.api.addEdges([edge]);
this.updateStatsSoon();
},
updateNode(id, patch) {
if (!this.bridge || !this.bridge.api) return;
if (!id) return;
this.bridge.api.updateNode(id, patch || {});
this.updateStatsSoon();
},
removeNode(id) {
if (!this.bridge || !this.bridge.api) return;
if (!id) return;
this.bridge.api.removeNodes([id]);
this.updateStatsSoon();
},
removeEdge(id) {
if (!this.bridge || !this.bridge.api) return;
if (!id) return;
this.bridge.api.removeEdges([id]);
this.updateStatsSoon();
},
requestGraph() {
if (!this.bridge || !this.bridge.api) return;
const payload = {
nodes: this.bridge.api.getNodes ? this.bridge.api.getNodes() : [],
edges: this.bridge.api.getEdges ? this.bridge.api.getEdges() : []
};
this.$emit("graph", payload);
},
importGraph(jsonish) {
if (!jsonish) return;
let data = null;
try {
data = typeof jsonish === "string" ? JSON.parse(jsonish) : jsonish;
} catch (_) {
}
if (!data || typeof data !== "object") return;
this.setGraph(data.nodes || [], data.edges || []);
},
exportGraph() {
if (!this.bridge || !this.bridge.api) return;
const payload = {
nodes: this.bridge.api.getNodes ? this.bridge.api.getNodes() : [],
edges: this.bridge.api.getEdges ? this.bridge.api.getEdges() : []
};
this.$emit("graph", payload);
},
// custom node helpers
// built in html node renders node.data.html as raw html
registerHtmlNodeType(name) {
if (!this.bridge) return;
const key = String(name || "").trim();
if (!key) return;
this.bridge.addNodeType(key, this.bridge.htmlNode);
},
// if you expose a Vue component on window[globalVar], you can register it
registerExternalNodeType(name, globalVar) {
if (!this.bridge) return;
const key = String(name || "").trim();
const gv = String(globalVar || "").trim();
if (!key || !gv) return;
const comp = window[gv];
if (!comp) return;
this.bridge.addNodeType(key, comp);
},
// internal
updateStats() {
if (!this.bridge || !this.bridge.api) return;
try {
this.statNodes = (this.bridge.api.getNodes && this.bridge.api.getNodes().length) || 0;
this.statEdges = (this.bridge.api.getEdges && this.bridge.api.getEdges().length) || 0;
} catch (_) {
}
},
updateStatsSoon() {
setTimeout(() => this.updateStats(), 0);
},
async ensureVueFlowLoaded() {
const coreVer = this.coreVersion || "1.47.0";
const extrasVer = this.extrasVersion || "1.7.0";
// CSS
await this.loadCssOnce(`https://cdn.jsdelivr.net/npm/@vue-flow/core@${coreVer}/dist/style.min.css`);
await this.loadCssOnce(`https://cdn.jsdelivr.net/npm/@vue-flow/core@${coreVer}/dist/theme-default.css`);
try {
await this.loadCssOnce(`https://cdn.jsdelivr.net/npm/@vue-flow/minimap@${extrasVer}/dist/style.css`);
} catch (_) {
}
try {
await this.loadCssOnce(`https://cdn.jsdelivr.net/npm/@vue-flow/controls@${extrasVer}/dist/style.css`);
} catch (_) {
}
// JS - IIFE globals
await this.loadScriptOnce(
`https://cdn.jsdelivr.net/npm/@vue-flow/core@${coreVer}/dist/vue-flow-core.iife.min.js`,
() => window.VueFlow || window.VueFlowCore
);
try {
await this.loadScriptOnce(
`https://cdn.jsdelivr.net/npm/@vue-flow/minimap@${extrasVer}/dist/vue-flow-minimap.iife.min.js`,
() => window.VueFlowMiniMap
);
} catch (_) {
}
try {
await this.loadScriptOnce(
`https://cdn.jsdelivr.net/npm/@vue-flow/controls@${extrasVer}/dist/vue-flow-controls.iife.min.js`,
() => window.VueFlowControls
);
} catch (_) {
}
try {
await this.loadScriptOnce(
`https://cdn.jsdelivr.net/npm/@vue-flow/background@${extrasVer}/dist/vue-flow-background.iife.min.js`,
() => window.VueFlowBackground
);
} catch (_) {
}
// cache a small handle so mountInnerApp can read from it
if (!window.__vf_modules__) window.__vf_modules__ = {};
const mod = window.__vf_modules__;
mod.core = window.VueFlow || window.VueFlowCore || {};
mod.minimap = window.VueFlowMiniMap || null;
mod.controls = window.VueFlowControls || null;
mod.background = window.VueFlowBackground || null;
},
mountInnerApp() {
const {createApp, h, reactive, shallowRef} = window.Vue;
const mod = window.__vf_modules__;
const VF = mod.core || {};
const VueFlow = VF.VueFlow || VF.default || VF;
const useVueFlow = VF.useVueFlow;
const Background = mod.background && (mod.background.Background || mod.background.default || mod.background);
const Controls = mod.controls && (mod.controls.Controls || mod.controls.default || mod.controls);
const MiniMap = mod.minimap && (mod.minimap.MiniMap || mod.minimap.default || mod.minimap);
const bridge = {
api: null,
nodeTypesRef: shallowRef({}),
htmlNode: {
name: "HtmlNode",
props: ["id", "data", "selected"],
setup(props) {
return () => h("div", {
class: "vf-html-node",
innerHTML: props.data && props.data.html ? String(props.data.html) : ""
});
}
},
addNodeType: (name, comp) => {
const next = Object.assign({}, bridge.nodeTypesRef.value);
next[name] = comp;
bridge.nodeTypesRef.value = next;
}
};
// expose default "html" type
bridge.addNodeType("html", bridge.htmlNode);
const Outer = {
name: "VueFlowOuter",
setup: () => {
const nodes = reactive([...this.initialNodes]);
const edges = reactive([...this.initialEdges]);
const flowProps = {
nodes,
edges,
fitViewOnInit: this.fitOnInit,
connectionMode: this.connectionMode,
snapToGrid: this.snapToGrid,
elevateEdgesOnSelect: true,
edgesUpdatable: this.allowUserEdit,
nodesDraggable: this.allowUserEdit,
nodesConnectable: this.allowUserEdit,
elementsSelectable: true,
panOnDrag: true,
zoomOnScroll: true,
nodeTypes: bridge.nodeTypesRef
};
// bridge component that taps into useVueFlow instance
const Bridge = {
name: "Bridge",
setup: () => {
const api = useVueFlow();
bridge.api = {
// getters
getNodes: () => api.getNodes(),
getEdges: () => api.getEdges(),
// setters
setNodes: ns => api.setNodes(ns || []),
setEdges: es => api.setEdges(es || []),
// mutations
addNodes: ns => api.addNodes(ns || []),
addEdges: es => api.addEdges(es || []),
removeNodes: ids => api.removeNodes(ids || []),
removeEdges: ids => api.removeEdges(ids || []),
updateNode: (id, patch) => api.updateNode(id, patch || {}),
// view
fitView: opts => api.fitView(opts || {}),
zoomIn: () => api.zoomIn(),
zoomOut: () => api.zoomOut(),
};
// forward vueflow events to the outer SFC so Python can subscribe
api.onNodesChange(changes => {
// high level adds and removes
const added = changes.filter(c => c.type === "add").map(c => c.item);
const removed = changes.filter(c => c.type === "remove").map(c => c.id);
if (added.length) {
// nodes added
try {
window.__nicegui_emit__(changes, "nodesAdded", added);
} catch (_) {
}
}
if (removed.length) {
try {
window.__nicegui_emit__(changes, "nodesRemoved", removed);
} catch (_) {
}
}
try {
window.__nicegui_emit__(changes, "nodesChange", changes);
} catch (_) {
}
});
api.onEdgesChange(changes => {
const added = changes.filter(c => c.type === "add").map(c => c.item);
const removed = changes.filter(c => c.type === "remove").map(c => c.id);
if (added.length) {
try {
window.__nicegui_emit__(changes, "edgesAdded", added);
} catch (_) {
}
}
if (removed.length) {
try {
window.__nicegui_emit__(changes, "edgesRemoved", removed);
} catch (_) {
}
}
try {
window.__nicegui_emit__(changes, "edgesChange", changes);
} catch (_) {
}
});
api.onConnect(params => {
try {
window.__nicegui_emit__(params, "connect", params);
} catch (_) {
}
});
api.onConnectStart(params => {
try {
window.__nicegui_emit__(params, "connectStart", params);
} catch (_) {
}
});
api.onConnectEnd(event => {
try {
window.__nicegui_emit__(event, "connectEnd", {});
} catch (_) {
}
});
api.onPaneClick(e => {
try {
window.__nicegui_emit__(e, "paneClick", {});
} catch (_) {
}
});
api.onMoveEnd(viewport => {
try {
window.__nicegui_emit__(viewport, "viewportChange", viewport);
} catch (_) {
}
});
// node click
api.onNodeClick(({node, event}) => {
try {
window.__nicegui_emit__({nodeId: node.id}, "nodeClick", {id: node.id});
} catch (_) {
}
});
// edge click
api.onEdgeClick(({edge, event}) => {
try {
window.__nicegui_emit__({edgeId: edge.id}, "edgeClick", {id: edge.id});
} catch (_) {
}
});
return () => null;
}
};
return () => h(
VueFlow,
flowProps,
{
default: () => [
h(Bridge),
Background ? h(Background) : null,
MiniMap ? h(MiniMap) : null,
Controls ? h(Controls) : null
]
}
);
}
};
// nicegui event bridge helper on window
const outer = this;
window.__nicegui_emit__ = function (source, name, payload) {
try {
outer.$emit(name, payload);
} catch (_) {
}
};
const app = createApp({
render() {
return h(Outer);
}
});
app.mount(this.$refs.mount);
this.innerApp = app;
this.bridge = bridge;
this.updateStats();
},
loadScriptOnce(url, readyPredicate) {
try {
if (readyPredicate && readyPredicate()) return Promise.resolve();
} catch (_) {
}
const already = Array.from(document.scripts).some(s => s.src === url);
if (already) {
return new Promise(resolve => {
const wait = () => {
try {
if (!readyPredicate || readyPredicate()) return resolve();
} catch (_) {
}
requestAnimationFrame(wait);
};
wait();
});
}
return new Promise((resolve, reject) => {
const s = document.createElement("script");
s.src = url;
s.async = true;
s.onload = () => {
const wait = () => {
try {
if (!readyPredicate || readyPredicate()) return resolve();
} catch (_) {
}
requestAnimationFrame(wait);
};
wait();
};
s.onerror = reject;
document.head.appendChild(s);
});
},
loadCssOnce(url) {
const sheets = Array.from(document.styleSheets).map(s => s.href).filter(Boolean);
if (sheets.includes(url)) return Promise.resolve();
return new Promise((resolve, reject) => {
const link = document.createElement("link");
link.rel = "stylesheet";
link.href = url;
link.onload = () => resolve();
link.onerror = reject;
document.head.appendChild(link);
});
}
}
};
</script>
<style scoped>
.vf-wrap {
display: flex;
flex-direction: column;
width: 100%;
height: 100%;
}
.vf-bar {
display: flex;
align-items: center;
gap: 8px;
padding: 8px;
font-size: 12px;
}
.vf-btn {
padding: 4px 8px;
border: 1px solid #ccc;
border-radius: 4px;
background: white;
cursor: pointer;
}
.vf-stat {
margin-left: auto;
opacity: 0.7;
}
.vf-canvas {
width: 100%;
height: 100%;
border: 1px solid #ddd;
border-radius: 4px;
}
/* optional styling for html nodes */
.vf-html-node {
padding: 8px 10px;
border: 1px solid #383838;
border-radius: 6px;
background: #fff;
font-family: ui-sans-serif, system-ui, Arial, sans-serif;
font-size: 12px;
}
</style>
We would like to use https://vueflow.dev/ inside our nicegui based application to create node based environments. For that, we need a nicegui component that allows us to interact with vueflow. The general idea is the following:
- We need a vue-flow-canvas that is displayed on the website (by default full width and full height by default), the component should also reflect the nicegui style (bright or dark mode), and feel like a native nicegui component (follow their rules)
- The events of the graph view canvas should be sent back to python to be able to interact with it (for example when nodes are moved, connections are made, nodes are added and so on)
- Also, it should be possible to programmatically create nodes and connections
- It should be possible to define new nodes types as new nicegui (vue) components. there should already be a base node which also supports events, connections and so on. It should be possible to define connection data types (text, image and so on) (or multi / single connectables) for a node (inputs and outputs) and of course the gui (as html). if possible, it would be also great that such a type can also include sub-elements from nicegui (like a textbox or and image maybe later, just as example). Such a node-type has a "process" event which is called when the graph is executing this node. Please merge my description of the requirements with the suppport vueflow or nicegui already provides
Please use python 3.12 standard to programm it, make a good oop architecture (one file per class), use dataclasses, typehints and also docstrings (:param: style). Write good code and reusable / extendable code. Always use enums instead of strings if possible (types and so on).
As an example how it should work, I'll add you some pseudo code how it could look like:
@dataclass
class DataType(Generics[T]):
name: str
type: Type[T]
@dataclass
class DataPort(ABC, Generics[DT]):
name: str
type: Type[DT]
value: DT.type | None = None
@dataclass
class InputDataPort(DataPort[DT])
accepts: int = 10 # (how many inputs are allowed) maybe rename this
@dataclass
class OutputDataPort(DataPort[DT])
state: DataPortState.Dirty
TextDataType = DataType("Text", str)
ImageDataType = DataType("Image", np.ndarray) # uses base64 string to exchange image data
@dataclass
class NodeBase(Generics)
name: str
auto_process: bool = False
@abstractmethod
def process()
pass
# post_init: create a list of all inputs and one of all outputs by analyzing the fields and check InputDataPort / OutputDataPort types
@dataclass
class TextDataNode(NodeBase):
text: OutputDataPort("Text", TextDataType)
@dataclass
class TextToImageNode(NodeBase):
text: InputDataPort("Prompt", TextDataType, 5)
image: OuputDataPort("Image", ImageDataType)
@abstractmethod
def process():
print(f"mocking generation process with {text.value}")
image.value = np.zeros((100, 100))
@dataclass
class Connection:
start: OutputDataPort
end: InputDataPort
class DataGraph:
nodes: list[NodeBase] = []
connections: list[Connection] = []
def execute(node: NodeBase | None):
# go through all node inputs, check which are connected and maybe dirty, recurse upwards the graph
# re-process this node
# send the data downstream to all nodes which auto_process = true
# maybe optomize how this work. it would be good to be able to set if a processing node should automatically run or not
pass
# ui relevant
@dataclass
class NodeElement(Generics[NodeBase])
pass
class TextDataNodeElement(NodeElement[TextDataNode], component="text_data_node_element.vue"):
# render the text data node elemnt
pass
class TextToImageNodeElement(NodeElement[TextToImageNode], component="text_to_image_node_element.vue"):
# render the text-to-image node elemnt
pass
graph = DataGraph()
# example gui
@ui.page('/')
def page():
canvas = DataFlowCanvas(graph, elements=[TextDataNodeElement, TextToImageNodeElement])
canvas.on_new_connection = ...
ui.run()