| 239 | return json.dumps(graph_dict, indent=4, sort_keys=True) |
| 240 | |
| 241 | def dump( |
| 242 | self, name: str | None = None, description: str | None = None, endpoint_name: str | None = None |
| 243 | ) -> GraphDump: |
| 244 | if self.raw_graph_data != {"nodes": [], "edges": []}: |
| 245 | data_dict = self.raw_graph_data |
| 246 | else: |
| 247 | # we need to convert the vertices and edges to json |
| 248 | nodes = [node.to_data() for node in self.vertices] |
| 249 | edges = [edge.to_data() for edge in self.edges] |
| 250 | self.raw_graph_data = {"nodes": nodes, "edges": edges} |
| 251 | data_dict = self.raw_graph_data |
| 252 | graph_dict: GraphDump = { |
| 253 | "data": data_dict, |
| 254 | "is_component": len(data_dict.get("nodes", [])) == 1 and data_dict["edges"] == [], |
| 255 | } |
| 256 | if name: |
| 257 | graph_dict["name"] = name |
| 258 | elif name is None and self.flow_name: |
| 259 | graph_dict["name"] = self.flow_name |
| 260 | if description: |
| 261 | graph_dict["description"] = description |
| 262 | elif description is None and self.description: |
| 263 | graph_dict["description"] = self.description |
| 264 | graph_dict["endpoint_name"] = str(endpoint_name) |
| 265 | return graph_dict |
| 266 | |
| 267 | def add_nodes_and_edges(self, nodes: list[NodeData], edges: list[EdgeData]) -> None: |
| 268 | self._vertices = nodes |