Instantiate all nodes from configuration.
(self)
| 32 | self._build_topology_and_metadata() |
| 33 | |
| 34 | def _instantiate_nodes(self) -> None: |
| 35 | """Instantiate all nodes from configuration.""" |
| 36 | self.graph.nodes.clear() |
| 37 | for node_def in self.graph.config.get_node_definitions(): |
| 38 | node_id = node_def.id |
| 39 | if node_id in self.graph.nodes: |
| 40 | print(f"Duplicated node id detected: {node_id}") |
| 41 | continue |
| 42 | node_instance = copy.deepcopy(node_def) |
| 43 | node_instance.predecessors = [] |
| 44 | node_instance.successors = [] |
| 45 | node_instance._outgoing_edges = [] |
| 46 | node_instance.vars = dict(self.graph.vars) |
| 47 | self.graph.nodes[node_id] = node_instance |
| 48 | |
| 49 | if node_instance.node_type == "subgraph": |
| 50 | self._build_subgraph(node_id) |
| 51 | |
| 52 | def _build_subgraph(self, node_id: str) -> None: |
| 53 | """Build a subgraph for the given node ID.""" |
no test coverage detected