|
|
import os |
|
|
from typing import Optional |
|
|
|
|
|
from injector import Module, provider |
|
|
|
|
|
from taskweaver.ces import code_execution_service_factory |
|
|
from taskweaver.ces.common import Manager |
|
|
from taskweaver.config.module_config import ModuleConfig |
|
|
|
|
|
|
|
|
class ExecutionServiceConfig(ModuleConfig): |
|
|
def _configure(self) -> None: |
|
|
self._set_name("execution_service") |
|
|
self.env_dir = self._get_path( |
|
|
"env_dir", |
|
|
os.path.join(self.src.app_base_path, "env"), |
|
|
) |
|
|
|
|
|
|
|
|
class ExecutionServiceModule(Module): |
|
|
def __init__(self) -> None: |
|
|
self.manager: Optional[Manager] = None |
|
|
|
|
|
@provider |
|
|
def provide_executor_manager(self, config: ExecutionServiceConfig) -> Manager: |
|
|
if self.manager is None: |
|
|
self.manager = code_execution_service_factory( |
|
|
config.env_dir, |
|
|
) |
|
|
return self.manager |
|
|
|
