节点列表
| 105 | |
| 106 | |
| 107 | class Workflow: |
| 108 | """ |
| 109 | 节点列表 |
| 110 | """ |
| 111 | nodes: List[Node] |
| 112 | """ |
| 113 | 线列表 |
| 114 | """ |
| 115 | edges: List[Edge] |
| 116 | """ |
| 117 | 节点id:node |
| 118 | """ |
| 119 | node_map: Dict[str, Node] |
| 120 | """ |
| 121 | 节点id:当前节点id上面的所有节点 |
| 122 | """ |
| 123 | up_node_map: Dict[str, List[EdgeNode]] |
| 124 | """ |
| 125 | 节点id:当前节点id下面的所有节点 |
| 126 | """ |
| 127 | next_node_map: Dict[str, List[EdgeNode]] |
| 128 | |
| 129 | workflow_mode: WorkflowMode |
| 130 | |
| 131 | def __init__(self, nodes: List[Node], edges: List[Edge], |
| 132 | workflow_mode: WorkflowMode = WorkflowMode.APPLICATION.value): |
| 133 | self.nodes = nodes |
| 134 | self.edges = edges |
| 135 | self.node_map = {node.id: node for node in nodes} |
| 136 | |
| 137 | self.up_node_map = {key: [EdgeNode(edge, self.node_map.get(edge.sourceNodeId)) for |
| 138 | edge in edges] for |
| 139 | key, edges in |
| 140 | group_by(edges, key=lambda edge: edge.targetNodeId).items()} |
| 141 | |
| 142 | self.next_node_map = {key: [EdgeNode(edge, self.node_map.get(edge.targetNodeId)) for edge in edges] for |
| 143 | key, edges in |
| 144 | group_by(edges, key=lambda edge: edge.sourceNodeId).items()} |
| 145 | self.workflow_mode = workflow_mode |
| 146 | |
| 147 | def get_node(self, node_id): |
| 148 | """ |
| 149 | 根据node_id 获取节点信息 |
| 150 | @param node_id: node_id |
| 151 | @return: 节点信息 |
| 152 | """ |
| 153 | return self.node_map.get(node_id) |
| 154 | |
| 155 | def get_up_edge_nodes(self, node_id) -> List[EdgeNode]: |
| 156 | """ |
| 157 | 根据节点id 获取当前连接前置节点和连线 |
| 158 | @param node_id: 节点id |
| 159 | @return: 节点连线列表 |
| 160 | """ |
| 161 | return self.up_node_map.get(node_id) |
| 162 | |
| 163 | def get_next_edge_nodes(self, node_id) -> List[EdgeNode]: |
| 164 | """ |