Create executors for every registered node type. Args: context: Shared execution context subgraphs: Mapping of subgraph nodes (used by Subgraph executors) Returns: Mapping from node type to executor instance
(context: ExecutionContext, subgraphs: dict = None)
| 14 | |
| 15 | @staticmethod |
| 16 | def create_executors(context: ExecutionContext, subgraphs: dict = None) -> Dict[str, NodeExecutor]: |
| 17 | """Create executors for every registered node type. |
| 18 | |
| 19 | Args: |
| 20 | context: Shared execution context |
| 21 | subgraphs: Mapping of subgraph nodes (used by Subgraph executors) |
| 22 | |
| 23 | Returns: |
| 24 | Mapping from node type to executor instance |
| 25 | """ |
| 26 | subgraphs = subgraphs or {} |
| 27 | |
| 28 | executors: Dict[str, NodeExecutor] = {} |
| 29 | for name, registration in iter_node_registrations().items(): |
| 30 | executors[name] = registration.build_executor(context, subgraphs=subgraphs) |
| 31 | return executors |
| 32 | |
| 33 | @staticmethod |
| 34 | def create_executor( |
no test coverage detected