(
self,
config: Config,
enable_load_plugins: bool = True,
enable_monitor: bool = True,
enable_workflow: bool = True,
)
| 17 | |
| 18 | class PluginActor: |
| 19 | def __init__( |
| 20 | self, |
| 21 | config: Config, |
| 22 | enable_load_plugins: bool = True, |
| 23 | enable_monitor: bool = True, |
| 24 | enable_workflow: bool = True, |
| 25 | ): |
| 26 | if enable_load_plugins: |
| 27 | load_plugins() |
| 28 | self.config = config |
| 29 | if enable_monitor: |
| 30 | self.monitor = MONITOR.get("my_monitor")( |
| 31 | project=self.config.project, |
| 32 | group=self.config.group, |
| 33 | name=self.config.name, |
| 34 | role=self.config.explorer.name, |
| 35 | config=config, |
| 36 | ) |
| 37 | else: |
| 38 | self.monitor = None |
| 39 | if enable_workflow: |
| 40 | workflow = WORKFLOWS.get("my_workflow") |
| 41 | assert workflow is not None, "Workflow 'my_workflow' not found in registry" |
| 42 | |
| 43 | def run(self, workflow_cls=Type[Workflow]): |
| 44 | if self.monitor: |
nothing calls this directly
no test coverage detected