Initialize the cycle executor. Args: log_manager: Logger instance nodes: Mapping of node ids to nodes cycle_execution_order: Super-node execution order with cycles cycle_manager: Cycle manager coordinating iterations execut
(
self,
log_manager: LogManager,
nodes: Dict[str, Node],
cycle_execution_order: List[Dict[str, Any]],
cycle_manager: CycleManager,
execute_node_func: Callable[[Node], None],
)
| 21 | """ |
| 22 | |
| 23 | def __init__( |
| 24 | self, |
| 25 | log_manager: LogManager, |
| 26 | nodes: Dict[str, Node], |
| 27 | cycle_execution_order: List[Dict[str, Any]], |
| 28 | cycle_manager: CycleManager, |
| 29 | execute_node_func: Callable[[Node], None], |
| 30 | ): |
| 31 | """Initialize the cycle executor. |
| 32 | |
| 33 | Args: |
| 34 | log_manager: Logger instance |
| 35 | nodes: Mapping of node ids to nodes |
| 36 | cycle_execution_order: Super-node execution order with cycles |
| 37 | cycle_manager: Cycle manager coordinating iterations |
| 38 | execute_node_func: Callable that executes a single node |
| 39 | """ |
| 40 | self.log_manager = log_manager |
| 41 | self.nodes = nodes |
| 42 | self.cycle_execution_order = cycle_execution_order |
| 43 | self.cycle_manager = cycle_manager |
| 44 | self.execute_node_func = execute_node_func |
| 45 | self.parallel_executor = ParallelExecutor(log_manager, nodes) |
| 46 | |
| 47 | def execute(self) -> None: |
| 48 | """Run the workflow that contains cycles.""" |
nothing calls this directly
no test coverage detected