Cube Sandbox

简体中文

English

简体中文

English

Appearance

Python SDK 架构解析 ​

Python SDK 是 CubeSandbox 的 Python 客户端，与 Go SDK 共享同一套后端 API，但在实现细节上有不少差异。这篇文章拆解 Python SDK 的设计。

1. 对象模型 ​

2. Sandbox.create() 工厂方法 ​

python
# sandbox.py
@classmethod
def create(cls, template=None, *, timeout=None, env_vars=None, metadata=None, config=None, **kwargs):
    cfg = config or Config()
    tpl = template or cfg.template_id

    payload = {"templateID": tpl, "timeout": timeout or cfg.timeout}
    if env_vars:
        payload["envVars"] = env_vars
    if metadata:
        payload["metadata"] = metadata

    with httpx.Client(headers={"Content-Type": "application/json"}) as s:
        resp = s.post(f"{cfg.api_url}/sandboxes", json=payload)
    _check_response(resp)
    return cls(resp.json(), config=cfg)

错误处理 ​

python
def _check_response(resp):
    if resp.is_success:
        return
    code = resp.status_code
    if code in (401, 403):
        raise AuthenticationError(msg, code)
    if code == 404:
        raise TemplateNotFoundError(...) if "template" in msg else SandboxNotFoundError(...)
    raise ApiError(msg, code)

3. run_code 实现 ​

Python SDK 的 run_code 支持流式回调：

回调机制 ​

python
# sandbox.py
def run_code(self, code, *, language="python", timeout=None, on_stdout=None, on_stderr=None):
    # 构建请求
    payload = {"process": {"cmd": "...", "args": [code]}, "stdin": False}

    # 流式解析响应
    for event in stream_events(resp):
        if event.type == "data" and event.stdout:
            if on_stdout:
                on_stdout(OutputMessage(text=event.stdout))

    return Execution(text=stdout, logs=Logs(stderr=stderr_lines), error=error)

4. Commands.run 的巧妙实现 ​

Python SDK 的 Commands.run 与 Go SDK 不同，它用 Python 的 subprocess 模块包装 shell 命令：

python
# _commands.py
def run(self, cmd, *, timeout=None):
    code = (
        "import subprocess as _sp\n"
        f"_r = _sp.run({cmd!r}, shell=True, capture_output=True, text=True)\n"
        "import sys as _sys\n"
        "_sys.stdout.write(_r.stdout)\n"
        "_sys.stderr.write(_r.stderr)\n"
        "print(_r.returncode)\n"
    )
    execution = self._sandbox.run_code(code, timeout=timeout)

    # 最后一行是 exit code
    lines = execution.text.splitlines()
    exit_code = int(lines[-1])
    stdout = "\n".join(lines[:-1])
    return CommandResult(stdout=stdout, stderr=stderr, exit_code=exit_code)

这个设计的好处：

  • 复用 run_code 的流式通道
  • 自动获得 stdout/stderr 分离
  • 支持超时控制

5. 与 Go SDK 的设计差异 ​

特性Go SDKPython SDK
HTTP clientnet/http（双 client）httpx（单 client）
命令执行直接调用 envd API用 subprocess 包装后调用 run_code
流式回调手动解析 Connect envelope内置 on_stdout/on_stderr 回调
连接管理手动 Close()支持 context manager (with)
错误类型error 返回值异常体系（4 种异常类）
文件操作直接调用 /files API直接调用 /files API
配置Config struct + 环境变量Config class + 环境变量
异步无（标准库无 async HTTP）无（同步 httpx）

异常体系 ​

python
# _exceptions.py
CubeSandboxError          # 基类
├── ApiError              # HTTP 错误
├── AuthenticationError   # 401/403
├── SandboxNotFoundError  # 404 沙箱
└── TemplateNotFoundError # 404 模板

6. 配置 ​

python
# _config.py
class Config:
    api_url: str          # CubeAPI 地址
    api_key: str          # API Key
    template_id: str      # 默认模板
    sandbox_domain: str   # 沙箱域名
    timeout: int          # 超时秒数
环境变量说明
CUBESANDBOX_API_URLCubeAPI 地址
CUBESANDBOX_API_KEYAPI Key
CUBE_TEMPLATE_ID默认模板 ID
CUBESANDBOX_DOMAIN沙箱域名

7. 上下文管理器 ​

Python SDK 支持 with 语句自动管理生命周期：

python
with Sandbox.create(template="my-tpl") as sb:
    result = sb.run_code("print('hello')")
    print(result.text)
# 退出时自动 kill 沙箱

延伸阅读 ​