尧图网站开发 
Reflex 企业版 React Flow 实战示例动态连线建节点与自定义节点连接数限制【免费下载链接】reflex️ Web apps in pure Python 项目地址: https://gitcode.com/GitHub_Trending/re/reflex导读examples.md 是 Reflex Enterprise 中rxe.flow基于 React Flow 封装的流程图组件的实战示例集展示如何构建交互式流程图的两种典型能力在连线拖放结束时自动创建新节点与边以及为自定义节点的连接点Handle设置连接数量上限。读完本文你将掌握on_connect_end与screen_to_flow_position的协作方式、rx.memo自定义节点与get_node_connections的用法并能直接复用文中的两段可运行代码为工作流编辑器、DAG 编排器等场景落地基础能力。示例环境与核心概念两段示例都遵循 Reflex Flow 的统一编程模型使用reflex_enterprise as rxe命名空间流程图组件挂在rxe.flow之下类型定义Node、Edge、HandleType、Position、XYPosition、ConnectionInProgress、NoConnection等从reflex_enterprise.components.flow.types导入节点与边以 Python 字典类型别名Node/Edge表示存放在rx.State中画布渲染依赖rxe.flow组件配套辅助组件包括rxe.flow.controls()缩放/适配/锁定控制面板、rxe.flow.mini_map()缩略图、rxe.flow.background()背景网格变更应用逻辑rxe.flow.util.apply_node_changes/apply_edge_changes/add_edge必须写在组件的事件回调 lambda 中状态侧只保存结果列表——这一点在 interactivity.md 中有明确约定示例中set_nodes/set_edges均为纯 setter。关于流程图的基本概念节点、Handle、边、连接线、视口可参考 overview.md节点与边的完整字段定义见 nodes.md 与 edges.md。示例一连线拖放到空白处自动生成新节点Add Node on Edge Drop效果与适用场景当用户从一个节点的连接点拖出连线、并松开在画布空白区域连接未命中任何目标 Handle即isValid为假时系统会在鼠标松开的位置自动创建一个新节点同时补上一条从源节点指向新节点的边。这非常适合可视化编程、流程编排等需要随手拉出节点的交互场景。状态定义import reflex as rx import reflex_enterprise as rxe from reflex_enterprise.components.flow.types import ( ConnectionInProgress, Edge, NoConnection, Node, XYPosition, ) node_style { color: #000000, } initial_nodes: list[Node] [ { id: 0, type: input, data: {label: Node}, position: {x: 0, y: 50}, style: node_style, }, ] class AddNodesOnEdgeDropState(rx.State): nodes: rx.Field[list[Node]] rx.field(default_factorylambda: initial_nodes) edges: rx.Field[list[Edge]] rx.field(default_factorylist) node_id: int 1 rx.event def increment(self): self.node_id 1 rx.event def set_nodes(self, nodes: list[Node]): self.nodes nodes rx.event def set_edges(self, edges: list[Edge]): self.edges edges rx.event def handle_connect_end( self, connection_status: NoConnection | ConnectionInProgress, event: rx.event.PointerEventInfo, flow_position: XYPosition, ): if not connection_status[isValid]: node_id str(self.node_id) self.increment() self.nodes.append({ id: node_id, position: flow_position, data: {label: fNode {node_id}}, origin: (0.5, 0.0), style: node_style, }) self.edges.append({ id: node_id, source: connection_status[fromNode][id], target: node_id, style: node_style, })关键点拆解nodes/edges使用rx.Field[list[Node]]rx.field(default_factory...)声明保持与 Reflex 状态系统的响应式绑定node_id作为自增计数器保证每次新节点/新边的id唯一handle_connect_end接收三个参数连接状态对象NoConnection | ConnectionInProgress联合类型、指针事件信息rx.event.PointerEventInfo、以及由调用方换算好的流程坐标XYPosition。当connection_status[isValid]为假即连线未命中有效目标时读取connection_status[fromNode][id]作为边的sourceorigin: (0.5, 0.0)表示以节点顶部中心为定位原点与组件上node_origin(0.5, 0.0)的设置保持一致使新节点以鼠标松开点为中心生成。画布与事件接线def add_node_on_edge_drop(): return rx.box( rxe.flow.provider( rxe.flow( rxe.flow.controls(), rxe.flow.mini_map(), rxe.flow.background(), on_connectlambda connection: AddNodesOnEdgeDropState.set_edges( rxe.flow.util.add_edge(connection, AddNodesOnEdgeDropState.edges) ), on_connect_end( lambda connection_status, event: ( AddNodesOnEdgeDropState.handle_connect_end( connection_status, event, rxe.flow.api.screen_to_flow_position( xevent.client_x, yevent.client_y, ), ) ) ), nodesAddNodesOnEdgeDropState.nodes, edgesAddNodesOnEdgeDropState.edges, default_nodesAddNodesOnEdgeDropState.nodes, default_edgesAddNodesOnEdgeDropState.edges, on_nodes_changelambda changes: AddNodesOnEdgeDropState.set_nodes( rxe.flow.util.apply_node_changes( AddNodesOnEdgeDropState.nodes, changes ) ), on_edges_changelambda changes: AddNodesOnEdgeDropState.set_edges( rxe.flow.util.apply_edge_changes( AddNodesOnEdgeDropState.edges, changes ) ), fit_viewTrue, fit_view_options{padding: 2}, node_origin(0.5, 0.0), ) ), height100vh, width100vw, )逐条说明外层包裹rxe.flow.providerFlowProvider 是上下文提供器让rxe.flow.api下的 hooks如screen_to_flow_position可以在rxe.flow之外访问流程内部状态。provider 不接收任何事件处理器所有事件都要挂在内部的rxe.flow上详见 components.md。on_connect连线成功命中目标时触发rxe.flow.util.add_edge(connection, edges)根据连接对象生成新边并追加到当前边列表utils.md。on_connect_end每次连线结束无论成功与否都会触发。这里把指针的屏幕像素坐标event.client_x/event.client_y交给rxe.flow.api.screen_to_flow_position换算成流程画布坐标再连同连接状态一起传入状态事件。坐标换算发生在客户端最终XYPosition以参数形式进入后端事件处理器hooks.md 对此有专门说明并指向本示例作为完整实现。受控双绑同时传入nodes/edges与default_nodes/default_edges配合on_nodes_change/on_edges_change将画布拖拽、选中、删除等变更同步回状态——这正是 interactivity.md 中受控流程的标准接法apply_node_changes/apply_edge_changes在客户端求值产生新列表再由纯 setter 存回状态。视图适配fit_viewTrue让流程自动缩放适配视口fit_view_options{padding: 2}控制留白node_origin(0.5, 0.0)统一节点定位原点。示例二自定义节点的连接数限制Connection Limit on Custom Node效果与适用场景该示例创建了一个名为custom的自定义节点其左侧目标 Handle 通过is_connectable属性限制最多只允许 1 条连接——一旦已有连接Handle 即不可再接受新连线。适用于端口数量受限的组件建模如单一上游依赖、容量约束等场景。状态与节点布局import reflex as rx import reflex_enterprise as rxe from reflex_enterprise.components.flow.types import Edge, HandleType, Node, Position node_style { color: #000000, } class ConnectionLimitState(rx.State): nodes: rx.Field[list[Node]] rx.field( default_factorylambda: [ { id: 1, type: input, data: {label: Node 1}, position: {x: 0, y: 25}, sourcePosition: right, style: node_style, }, { id: 2, type: custom, data: {}, position: {x: 250, y: 50}, style: node_style, }, { id: 3, type: input, data: {label: Node 2}, position: {x: 0, y: 100}, sourcePosition: right, style: node_style, }, ] ) edges: rx.Field[list[Edge]] rx.field(default_factorylist) rx.event def set_nodes(self, nodes: list[Node]): self.nodes nodes rx.event def set_edges(self, edges: list[Edge]): self.edges edges三个节点中1和3是input类型仅含源 HandlesourcePosition指向右侧2是custom类型data为空字典渲染逻辑完全由自定义组件承担。状态侧仍是标准的受控接线set_nodes/set_edges两个纯 setter 配合apply_node_changes/apply_edge_changes使用。带连接数限制的 Handle 组件rx.memo def custom_handle( type: rx.Var[HandleType], position: rx.Var[Position], connection_count: rx.Var[int] ) - rxe.components.flow.Handle: connections rxe.flow.api.get_node_connections() return rxe.flow.handle( typetype, positionposition, connection_countconnection_count, is_connectableconnections.length() connection_count.guess_type(), ) rx.memo def custom_node() - rx.el.Div: return rx.el.div( custom_handle(typetarget, positionleft, connection_count1), rx.el.div(← Only one edge allowed), class_nameborder border-1 p-2 rounded-sm, border_colorrx.color_mode_cond(black, ), colorblack, bgwhite, )实现原理rx.memo装饰器与 React.memo 语义对应按声明式参数type、position、connection_count构建组件参数类型用rx.Var[...]标注是自定义节点的标准写法可对比 nodes.md 中颜色选择器节点示例rxe.flow.api.get_node_connections()返回当前节点上的连接数组支持按节点、Handle 类型、Handle ID 过滤见 hooks.md。这里无参调用取该节点全部连接is_connectable动态判定connections.length() connection_count.guess_type()在客户端求值——连接数已满时返回假Handle 自动禁用。connection_count作为rx.Var[int]传入通过.guess_type()得到比较用的数值调用custom_handle(typetarget, positionleft, connection_count1)即把上限设为 1rxe.flow.handle的connection_count属性将上限值同步给底层 Handle 实现与is_connectable配合完成达到上限即不可再连的限制逻辑。渲染画布def connection_limit(): return rx.box( rxe.flow( rxe.flow.background(), nodesConnectionLimitState.nodes, edgesConnectionLimitState.edges, default_nodesConnectionLimitState.nodes, default_edgesConnectionLimitState.edges, on_nodes_changelambda changes: ConnectionLimitState.set_nodes( rxe.flow.util.apply_node_changes(ConnectionLimitState.nodes, changes) ), on_edges_changelambda changes: ConnectionLimitState.set_edges( rxe.flow.util.apply_edge_changes(ConnectionLimitState.edges, changes) ), on_connectlambda connection: ConnectionLimitState.set_edges( rxe.flow.util.add_edge(connection, ConnectionLimitState.edges) ), node_types{ custom: rx.vars.function.ArgsFunctionOperation.create( (), custom_node() ) }, color_modelight, fit_viewTrue, ), height100vh, width100vw, )接线要点node_types注册自定义节点将custom类型映射到custom_node()渲染函数。此处通过rx.vars.function.ArgsFunctionOperation.create((), custom_node())将无参数组件封装为可序列化的函数操作对象供前端按节点类型调用on_connect成功连线时用add_edge追加边新连线一旦到达custom节点其目标 Handle 的连接数变为 1is_connectable随即为假后续拖线到该 Handle 将被拒绝color_modelight固定浅色主题fit_viewTrue自动适配视口。两示例背后的通用模式综合两个示例可以看出 Reflex Flow 交互式流程图的固定骨架状态层Node/Edge字典列表 自增 id 计数器 纯 setter 事件set_nodes/set_edges变更应用on_nodes_change/on_edges_change内用rxe.flow.util.apply_node_changes/apply_edge_changes在客户端求值on_connect内用rxe.flow.util.add_edge生成新边——这些工具函数只能在组件代码中使用严禁在rx.event内部调用interactivity.md 明确强调进阶交互需要画布级操作如按屏幕坐标建节点时用rxe.flow.api.screen_to_flow_position换算坐标用rxe.flow.api.get_node_connections读取连接状态这些 hooks 依赖外层rxe.flow.providerhooks.md自定义渲染rx.memorx.Var参数构建节点/Handle/边组件通过node_types/edge_types注册。若需进一步扩展可将示例一与示例二组合在handle_connect_end中同时检查目标节点的剩余容量实现容量不足则拒绝、容量有余则落点建节点的完整交互边与路径的更多自定义get_bezier_path、base_edge、edge_label_renderer可参考 edges.md 中的按钮边示例。【免费下载链接】reflex️ Web apps in pure Python 项目地址: https://gitcode.com/GitHub_Trending/re/reflex创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 
Reflex 企业版 React Flow 实战示例动态连线建节点与自定义节点连接数限制【免费下载链接】reflex️ Web apps in pure Python 项目地址: https://gitcode.com/GitHub_Trending/re/reflex导读examples.md 是 Reflex Enterprise 中rxe.flow基于 React Flow 封装的流程图组件的实战示例集展示如何构建交互式流程图的两种典型能力在连线拖放结束时自动创建新节点与边以及为自定义节点的连接点Handle设置连接数量上限。读完本文你将掌握on_connect_end与screen_to_flow_position的协作方式、rx.memo自定义节点与get_node_connections的用法并能直接复用文中的两段可运行代码为工作流编辑器、DAG 编排器等场景落地基础能力。示例环境与核心概念两段示例都遵循 Reflex Flow 的统一编程模型使用reflex_enterprise as rxe命名空间流程图组件挂在rxe.flow之下类型定义Node、Edge、HandleType、Position、XYPosition、ConnectionInProgress、NoConnection等从reflex_enterprise.components.flow.types导入节点与边以 Python 字典类型别名Node/Edge表示存放在rx.State中画布渲染依赖rxe.flow组件配套辅助组件包括rxe.flow.controls()缩放/适配/锁定控制面板、rxe.flow.mini_map()缩略图、rxe.flow.background()背景网格变更应用逻辑rxe.flow.util.apply_node_changes/apply_edge_changes/add_edge必须写在组件的事件回调 lambda 中状态侧只保存结果列表——这一点在 interactivity.md 中有明确约定示例中set_nodes/set_edges均为纯 setter。关于流程图的基本概念节点、Handle、边、连接线、视口可参考 overview.md节点与边的完整字段定义见 nodes.md 与 edges.md。示例一连线拖放到空白处自动生成新节点Add Node on Edge Drop效果与适用场景当用户从一个节点的连接点拖出连线、并松开在画布空白区域连接未命中任何目标 Handle即isValid为假时系统会在鼠标松开的位置自动创建一个新节点同时补上一条从源节点指向新节点的边。这非常适合可视化编程、流程编排等需要随手拉出节点的交互场景。状态定义import reflex as rx import reflex_enterprise as rxe from reflex_enterprise.components.flow.types import ( ConnectionInProgress, Edge, NoConnection, Node, XYPosition, ) node_style { color: #000000, } initial_nodes: list[Node] [ { id: 0, type: input, data: {label: Node}, position: {x: 0, y: 50}, style: node_style, }, ] class AddNodesOnEdgeDropState(rx.State): nodes: rx.Field[list[Node]] rx.field(default_factorylambda: initial_nodes) edges: rx.Field[list[Edge]] rx.field(default_factorylist) node_id: int 1 rx.event def increment(self): self.node_id 1 rx.event def set_nodes(self, nodes: list[Node]): self.nodes nodes rx.event def set_edges(self, edges: list[Edge]): self.edges edges rx.event def handle_connect_end( self, connection_status: NoConnection | ConnectionInProgress, event: rx.event.PointerEventInfo, flow_position: XYPosition, ): if not connection_status[isValid]: node_id str(self.node_id) self.increment() self.nodes.append({ id: node_id, position: flow_position, data: {label: fNode {node_id}}, origin: (0.5, 0.0), style: node_style, }) self.edges.append({ id: node_id, source: connection_status[fromNode][id], target: node_id, style: node_style, })关键点拆解nodes/edges使用rx.Field[list[Node]]rx.field(default_factory...)声明保持与 Reflex 状态系统的响应式绑定node_id作为自增计数器保证每次新节点/新边的id唯一handle_connect_end接收三个参数连接状态对象NoConnection | ConnectionInProgress联合类型、指针事件信息rx.event.PointerEventInfo、以及由调用方换算好的流程坐标XYPosition。当connection_status[isValid]为假即连线未命中有效目标时读取connection_status[fromNode][id]作为边的sourceorigin: (0.5, 0.0)表示以节点顶部中心为定位原点与组件上node_origin(0.5, 0.0)的设置保持一致使新节点以鼠标松开点为中心生成。画布与事件接线def add_node_on_edge_drop(): return rx.box( rxe.flow.provider( rxe.flow( rxe.flow.controls(), rxe.flow.mini_map(), rxe.flow.background(), on_connectlambda connection: AddNodesOnEdgeDropState.set_edges( rxe.flow.util.add_edge(connection, AddNodesOnEdgeDropState.edges) ), on_connect_end( lambda connection_status, event: ( AddNodesOnEdgeDropState.handle_connect_end( connection_status, event, rxe.flow.api.screen_to_flow_position( xevent.client_x, yevent.client_y, ), ) ) ), nodesAddNodesOnEdgeDropState.nodes, edgesAddNodesOnEdgeDropState.edges, default_nodesAddNodesOnEdgeDropState.nodes, default_edgesAddNodesOnEdgeDropState.edges, on_nodes_changelambda changes: AddNodesOnEdgeDropState.set_nodes( rxe.flow.util.apply_node_changes( AddNodesOnEdgeDropState.nodes, changes ) ), on_edges_changelambda changes: AddNodesOnEdgeDropState.set_edges( rxe.flow.util.apply_edge_changes( AddNodesOnEdgeDropState.edges, changes ) ), fit_viewTrue, fit_view_options{padding: 2}, node_origin(0.5, 0.0), ) ), height100vh, width100vw, )逐条说明外层包裹rxe.flow.providerFlowProvider 是上下文提供器让rxe.flow.api下的 hooks如screen_to_flow_position可以在rxe.flow之外访问流程内部状态。provider 不接收任何事件处理器所有事件都要挂在内部的rxe.flow上详见 components.md。on_connect连线成功命中目标时触发rxe.flow.util.add_edge(connection, edges)根据连接对象生成新边并追加到当前边列表utils.md。on_connect_end每次连线结束无论成功与否都会触发。这里把指针的屏幕像素坐标event.client_x/event.client_y交给rxe.flow.api.screen_to_flow_position换算成流程画布坐标再连同连接状态一起传入状态事件。坐标换算发生在客户端最终XYPosition以参数形式进入后端事件处理器hooks.md 对此有专门说明并指向本示例作为完整实现。受控双绑同时传入nodes/edges与default_nodes/default_edges配合on_nodes_change/on_edges_change将画布拖拽、选中、删除等变更同步回状态——这正是 interactivity.md 中受控流程的标准接法apply_node_changes/apply_edge_changes在客户端求值产生新列表再由纯 setter 存回状态。视图适配fit_viewTrue让流程自动缩放适配视口fit_view_options{padding: 2}控制留白node_origin(0.5, 0.0)统一节点定位原点。示例二自定义节点的连接数限制Connection Limit on Custom Node效果与适用场景该示例创建了一个名为custom的自定义节点其左侧目标 Handle 通过is_connectable属性限制最多只允许 1 条连接——一旦已有连接Handle 即不可再接受新连线。适用于端口数量受限的组件建模如单一上游依赖、容量约束等场景。状态与节点布局import reflex as rx import reflex_enterprise as rxe from reflex_enterprise.components.flow.types import Edge, HandleType, Node, Position node_style { color: #000000, } class ConnectionLimitState(rx.State): nodes: rx.Field[list[Node]] rx.field( default_factorylambda: [ { id: 1, type: input, data: {label: Node 1}, position: {x: 0, y: 25}, sourcePosition: right, style: node_style, }, { id: 2, type: custom, data: {}, position: {x: 250, y: 50}, style: node_style, }, { id: 3, type: input, data: {label: Node 2}, position: {x: 0, y: 100}, sourcePosition: right, style: node_style, }, ] ) edges: rx.Field[list[Edge]] rx.field(default_factorylist) rx.event def set_nodes(self, nodes: list[Node]): self.nodes nodes rx.event def set_edges(self, edges: list[Edge]): self.edges edges三个节点中1和3是input类型仅含源 HandlesourcePosition指向右侧2是custom类型data为空字典渲染逻辑完全由自定义组件承担。状态侧仍是标准的受控接线set_nodes/set_edges两个纯 setter 配合apply_node_changes/apply_edge_changes使用。带连接数限制的 Handle 组件rx.memo def custom_handle( type: rx.Var[HandleType], position: rx.Var[Position], connection_count: rx.Var[int] ) - rxe.components.flow.Handle: connections rxe.flow.api.get_node_connections() return rxe.flow.handle( typetype, positionposition, connection_countconnection_count, is_connectableconnections.length() connection_count.guess_type(), ) rx.memo def custom_node() - rx.el.Div: return rx.el.div( custom_handle(typetarget, positionleft, connection_count1), rx.el.div(← Only one edge allowed), class_nameborder border-1 p-2 rounded-sm, border_colorrx.color_mode_cond(black, ), colorblack, bgwhite, )实现原理rx.memo装饰器与 React.memo 语义对应按声明式参数type、position、connection_count构建组件参数类型用rx.Var[...]标注是自定义节点的标准写法可对比 nodes.md 中颜色选择器节点示例rxe.flow.api.get_node_connections()返回当前节点上的连接数组支持按节点、Handle 类型、Handle ID 过滤见 hooks.md。这里无参调用取该节点全部连接is_connectable动态判定connections.length() connection_count.guess_type()在客户端求值——连接数已满时返回假Handle 自动禁用。connection_count作为rx.Var[int]传入通过.guess_type()得到比较用的数值调用custom_handle(typetarget, positionleft, connection_count1)即把上限设为 1rxe.flow.handle的connection_count属性将上限值同步给底层 Handle 实现与is_connectable配合完成达到上限即不可再连的限制逻辑。渲染画布def connection_limit(): return rx.box( rxe.flow( rxe.flow.background(), nodesConnectionLimitState.nodes, edgesConnectionLimitState.edges, default_nodesConnectionLimitState.nodes, default_edgesConnectionLimitState.edges, on_nodes_changelambda changes: ConnectionLimitState.set_nodes( rxe.flow.util.apply_node_changes(ConnectionLimitState.nodes, changes) ), on_edges_changelambda changes: ConnectionLimitState.set_edges( rxe.flow.util.apply_edge_changes(ConnectionLimitState.edges, changes) ), on_connectlambda connection: ConnectionLimitState.set_edges( rxe.flow.util.add_edge(connection, ConnectionLimitState.edges) ), node_types{ custom: rx.vars.function.ArgsFunctionOperation.create( (), custom_node() ) }, color_modelight, fit_viewTrue, ), height100vh, width100vw, )接线要点node_types注册自定义节点将custom类型映射到custom_node()渲染函数。此处通过rx.vars.function.ArgsFunctionOperation.create((), custom_node())将无参数组件封装为可序列化的函数操作对象供前端按节点类型调用on_connect成功连线时用add_edge追加边新连线一旦到达custom节点其目标 Handle 的连接数变为 1is_connectable随即为假后续拖线到该 Handle 将被拒绝color_modelight固定浅色主题fit_viewTrue自动适配视口。两示例背后的通用模式综合两个示例可以看出 Reflex Flow 交互式流程图的固定骨架状态层Node/Edge字典列表 自增 id 计数器 纯 setter 事件set_nodes/set_edges变更应用on_nodes_change/on_edges_change内用rxe.flow.util.apply_node_changes/apply_edge_changes在客户端求值on_connect内用rxe.flow.util.add_edge生成新边——这些工具函数只能在组件代码中使用严禁在rx.event内部调用interactivity.md 明确强调进阶交互需要画布级操作如按屏幕坐标建节点时用rxe.flow.api.screen_to_flow_position换算坐标用rxe.flow.api.get_node_connections读取连接状态这些 hooks 依赖外层rxe.flow.providerhooks.md自定义渲染rx.memorx.Var参数构建节点/Handle/边组件通过node_types/edge_types注册。若需进一步扩展可将示例一与示例二组合在handle_connect_end中同时检查目标节点的剩余容量实现容量不足则拒绝、容量有余则落点建节点的完整交互边与路径的更多自定义get_bezier_path、base_edge、edge_label_renderer可参考 edges.md 中的按钮边示例。【免费下载链接】reflex️ Web apps in pure Python 项目地址: https://gitcode.com/GitHub_Trending/re/reflex创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考