mirror of
https://gitee.com/wanwujie/deer-flow
synced 2026-04-18 12:04:45 +08:00
54 lines
1.4 KiB
Python
54 lines
1.4 KiB
Python
from abc import ABC, abstractmethod
|
|
|
|
from src.config import get_app_config
|
|
from src.reflection import resolve_class
|
|
from src.sandbox.sandbox import Sandbox
|
|
|
|
|
|
class SandboxProvider(ABC):
|
|
"""Abstract base class for sandbox providers"""
|
|
|
|
@abstractmethod
|
|
def acquire(self, thread_id: str | None = None) -> str:
|
|
"""Acquire a sandbox environment and return its ID.
|
|
|
|
Returns:
|
|
The ID of the acquired sandbox environment.
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get(self, sandbox_id: str) -> Sandbox | None:
|
|
"""Get a sandbox environment by ID.
|
|
|
|
Args:
|
|
sandbox_id: The ID of the sandbox environment to retain.
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def release(self, sandbox_id: str) -> None:
|
|
"""Release a sandbox environment.
|
|
|
|
Args:
|
|
sandbox_id: The ID of the sandbox environment to destroy.
|
|
"""
|
|
pass
|
|
|
|
|
|
_default_sandbox_provider: SandboxProvider | None = None
|
|
|
|
|
|
def get_sandbox_provider(**kwargs) -> SandboxProvider:
|
|
"""Get the sandbox provider.
|
|
|
|
Returns:
|
|
A sandbox provider.
|
|
"""
|
|
global _default_sandbox_provider
|
|
if _default_sandbox_provider is None:
|
|
config = get_app_config()
|
|
cls = resolve_class(config.sandbox.use, SandboxProvider)
|
|
_default_sandbox_provider = cls(**kwargs)
|
|
return _default_sandbox_provider
|