Spaces:
Sleeping
Sleeping
File size: 1,378 Bytes
399b80c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 | from typing import Any, Dict, Optional
from abc import ABC, abstractmethod
from ..types import SandboxOptions, BackendType
class BaseSandbox(ABC):
def __init__(self, options: SandboxOptions):
self.options = options
self._active = False
@abstractmethod
async def start(self) -> bool:
"""Set self._active to True"""
pass
@abstractmethod
async def stop(self) -> None:
"""Set self._active to False"""
pass
@abstractmethod
async def execute_safe(self, command: str, **kwargs) -> Any:
pass
@abstractmethod
def get_connector(self) -> Any:
pass
@property
def is_active(self) -> bool:
return self._active
class SandboxManager:
def __init__(self):
self._sandboxes: Dict[BackendType, BaseSandbox] = {}
def register_sandbox(self, backend_type: BackendType, sandbox: BaseSandbox) -> None:
self._sandboxes[backend_type] = sandbox
def get_sandbox(self, backend_type: BackendType) -> Optional[BaseSandbox]:
return self._sandboxes.get(backend_type)
async def start_all(self) -> None:
for sandbox in self._sandboxes.values():
await sandbox.start()
async def stop_all(self) -> None:
for sandbox in self._sandboxes.values():
await sandbox.stop() |