샌드박스 설정

from sandbox_auth import AUTH
from cwsandbox import ResourceOptions, Sandbox, SandboxDefaults, Session

# SandboxDefaults를 통해
defaults = SandboxDefaults(
    container_image="python:3.11",
    max_lifetime_seconds=3600,
    resources=ResourceOptions(
        requests={"cpu": "2", "memory": "2Gi"},
        limits={"cpu": "2", "memory": "2Gi"},
    ),
)

# Sandbox.run() kwargs를 통해
sandbox = Sandbox.run(
    defaults=defaults,
    resources=ResourceOptions(
        requests={"cpu": "4", "memory": "4Gi"},
        limits={"cpu": "4", "memory": "4Gi"},
    ),
    auth=AUTH,
)

# @session.function() kwargs를 통해
with Session(defaults, auth=AUTH) as session:
    @session.function(resources={"cpu": "1", "memory": "1Gi"})
    def compute(x: int) -> int:
        return x * 2

from sandbox_auth import AUTH
from cwsandbox import ResourceOptions, Sandbox

# ResourceOptions 사용
sandbox = Sandbox.run(
    resources=ResourceOptions(
        requests={"cpu": "2", "memory": "2Gi"},
        limits={"cpu": "2", "memory": "2Gi"},
    ),
    auth=AUTH,
)

# dict 사용
sandbox = Sandbox.run(
    resources={
        "requests": {"cpu": "2", "memory": "2Gi"},
        "limits": {"cpu": "2", "memory": "2Gi"},
    },
    auth=AUTH,
)

from sandbox_auth import AUTH
from cwsandbox import ResourceOptions, Sandbox

sandbox = Sandbox.run(
    resources=ResourceOptions(
        requests={"cpu": "2", "memory": "4Gi"},
        limits={"cpu": "2", "memory": "4Gi"},
    ),
    auth=AUTH,
)

from sandbox_auth import AUTH
from cwsandbox import Sandbox

# Flat dict shorthand: requests와 limits가 동일한 값으로 설정됩니다
sandbox = Sandbox.run(
    resources={"cpu": "2", "memory": "4Gi"},
    auth=AUTH,
)

from cwsandbox import AuthStrategy, ResourceOptions, Sandbox

sandbox = Sandbox.run(
    auth=AuthStrategy.COREWEAVE_API_KEY,
    placement_mode="cks",
    runner_ids=["[RUNNER-ID]"],
    resources=ResourceOptions(
        requests={"cpu": "500m", "memory": "512Mi"},
        limits={"cpu": "4", "memory": "4Gi"},
    ),
)

from sandbox_auth import AUTH
from cwsandbox import ResourceOptions, Sandbox

sandbox = Sandbox.run(
    resources=ResourceOptions(
        requests={"cpu": "4", "memory": "16Gi"},
        limits={"cpu": "4", "memory": "16Gi"},
        gpu={"count": 1},
    ),
    auth=AUTH,
)

from cwsandbox import AuthStrategy, ResourceOptions, Sandbox

with Sandbox.run(
    auth=AuthStrategy.COREWEAVE_API_KEY,
    placement_mode="cks",
    runner_ids=["[RUNNER-ID]"],
    resources=ResourceOptions(
        requests={"cpu": "500m", "memory": "512Mi"},
        limits={"cpu": "2", "memory": "2Gi"},
        gpu={"count": 1},
    ),
) as sb:
    print(sb.resource_requests)  # {"cpu": "500m", "memory": "512Mi"}
    print(sb.resource_limits)    # {"cpu": "2", "memory": "2Gi"}
    print(sb.resource_gpu)       # {"count": 1}

# sandbox.yaml
container_image: "pytorch/pytorch:2.8.0-cuda12.8-cudnn9-runtime"
max_lifetime_seconds: 3600
tags:
  - training
  - experiment-42
resources:
  requests:
    cpu: "4"
    memory: "8Gi"
  limits:
    cpu: "4"
    memory: "8Gi"
  gpu:
    count: 1
network:
  egress:
    - dns_name: "pypi.org"
    - dns_name: "files.pythonhosted.org"
environment_variables:
  LOG_LEVEL: "info"

from sandbox_auth import AUTH
from omegaconf import OmegaConf
from cwsandbox import Sandbox, SandboxDefaults

cfg = OmegaConf.load("sandbox.yaml")
defaults = SandboxDefaults.from_dict(cfg)

with Sandbox.run(defaults=defaults, auth=AUTH) as sb:
    result = sb.exec(["python", "train.py"]).result()

from sandbox_auth import AUTH
from cwsandbox import Sandbox

sandbox = Sandbox.run(
    mounted_files=[
        {
            "mount_path": "/app/config.json",
            "file_content": b'{"debug": true}',
        },
        {
            "mount_path": "/app/script.py",
            "file_content": b"print('hello')",
        },
    ],
    resources={"cpu": "1", "memory": "1Gi"},
    auth=AUTH,
)

# 파일은 즉시 사용 가능합니다
result = sandbox.exec(["python", "/app/script.py"]).result()

import time
from urllib.error import URLError
from urllib.request import urlopen

from sandbox_auth import AUTH
from cwsandbox import (
    Endpoint,
    EndpointAuth,
    EndpointKind,
    Sandbox,
    Service,
    ServiceProtocol,
    ServiceVisibility,
)

sandbox = Sandbox.run(
    "python", "-m", "http.server", "8080", "--directory", "/tmp",
    container_image="python:3.11",
    placement_mode="serverless",
    services=[
        Service(
            port=8080,
            name="web",
            protocol=ServiceProtocol.TCP,
            visibility=ServiceVisibility.PUBLIC,
            endpoint=Endpoint(kind=EndpointKind.HTTPS, auth=EndpointAuth.OPEN),
        ),
    ],
    resources={"cpu": "1", "memory": "1Gi"},
    max_lifetime_seconds=300,
    auth=AUTH,
)

try:
    deadline = time.monotonic() + 120
    while time.monotonic() < deadline:
        sandbox.get_status()
        if sandbox.service_urls:
            url = sandbox.service_urls[0][2]
            try:
                with urlopen(url, timeout=10) as response:
                    body = response.read().decode("utf-8")
                    assert response.status == 200
                    assert "Directory listing for /" in body
                    print("HTTP 200: directory listing received")
                    break
            except (URLError, TimeoutError):
                pass  # 애플리케이션이 준비되기 전에 URL이 먼저 표시될 수 있습니다.
        time.sleep(2)
    else:
        raise TimeoutError("The HTTPS service did not become reachable")
finally:
    Sandbox.delete(sandbox.sandbox_id, auth=AUTH).result()

from sandbox_auth import AUTH
from cwsandbox import EgressRule, NetworkOptions, Sandbox

# NetworkOptions 사용 시
sandbox = Sandbox.run(
    network=NetworkOptions(
        egress=[
            EgressRule(dns_name="pypi.org"),
            EgressRule(dns_name="*.pypi.org"),
            EgressRule(dns_name="files.pythonhosted.org"),
        ],
    ),
    resources={"cpu": "1", "memory": "1Gi"},
    auth=AUTH,
)

# dict 사용 시
sandbox = Sandbox.run(
    network={
        "egress": [
            {"dns_name": "pypi.org"},
            {"dns_name": "*.pypi.org"},
            {"dns_name": "files.pythonhosted.org"},
        ],
    },
    resources={"cpu": "1", "memory": "1Gi"},
    auth=AUTH,
)

from sandbox_auth import AUTH
from cwsandbox import EgressRule, NetworkOptions, SandboxDefaults, Session

defaults = SandboxDefaults(
    network=NetworkOptions(
        egress=[
            EgressRule(dns_name="pypi.org"),
            EgressRule(dns_name="files.pythonhosted.org"),
        ],
    ),
    resources={"cpu": "1", "memory": "1Gi"},
)

with Session(defaults, auth=AUTH) as session:
    # 모든 샌드박스는 네트워크 설정을 상속합니다
    sb1 = session.sandbox()
    sb2 = session.sandbox()

    # 이 샌드박스에서는 대상 권한 부여를 차단합니다. DNS는 별도로 제어됩니다
    sb3 = session.sandbox(network=NetworkOptions(deny_egress=True))

from sandbox_auth import AUTH
from cwsandbox import Sandbox

sandbox = Sandbox.run(
    annotations={
        "team": "ml-infra",
        "experiment": "training-run-42",
    },
    resources={"cpu": "1", "memory": "1Gi"},
    auth=AUTH,
)

from sandbox_auth import AUTH
from cwsandbox import SandboxDefaults, Session

defaults = SandboxDefaults(
    annotations={
        "team": "ml-infra",
        "managed-by": "sandbox-sdk",
    },
    resources={"cpu": "1", "memory": "1Gi"},
)

with Session(defaults, auth=AUTH) as session:
    # 기본 어노테이션을 상속합니다
    sb1 = session.sandbox()

    # 병합: 키 충돌 시 명시적 어노테이션이 기본값을 재정의합니다
    sb2 = session.sandbox(
        annotations={"experiment": "run-42", "team": "ml-research"},
    )
    # sb2 receives: team=ml-research, managed-by=sandbox-sdk, experiment=run-42

from sandbox_auth import AUTH
from cwsandbox import Sandbox

with Sandbox.run(
    max_lifetime_seconds=7200,  # 2시간
    resources={"cpu": "1", "memory": "1Gi"},
    auth=AUTH,
) as sandbox:
    sandbox.exec(["python", "train.py"], timeout_seconds=7000).result()

from sandbox_auth import AUTH
from cwsandbox import Sandbox, SandboxDefaults

defaults = SandboxDefaults(
    container_image="python:3.11",
    max_lifetime_seconds=3600,  # 1시간
    resources={"cpu": "1", "memory": "1Gi"},
)

with Sandbox.run(defaults=defaults, auth=AUTH) as sandbox:
    sandbox.exec(["python", "-c", "print('working')"]).result()

result = sandbox.exec(
    ["python", "long_script.py"],
    timeout_seconds=300,  # 5분
).result()

from cwsandbox import SandboxDefaults

defaults = SandboxDefaults(
    request_timeout_seconds=600.0,
    resources={"cpu": "1", "memory": "1Gi"},
)

from sandbox_auth import AUTH
from cwsandbox import (
    EgressRule,
    NetworkOptions,
    ResourceOptions,
    Sandbox,
    SandboxDefaults,
    Service,
)

defaults = SandboxDefaults(
    container_image="python:3.11",
    max_lifetime_seconds=1800,  # 30분
    tags=("production", "ml-pipeline"),
    resources=ResourceOptions(
        requests={"cpu": "4", "memory": "8Gi"},
        limits={"cpu": "4", "memory": "8Gi"},
    ),
    network=NetworkOptions(
        egress=[
            EgressRule(dns_name="pypi.org"),
            EgressRule(dns_name="*.pypi.org"),
            EgressRule(dns_name="files.pythonhosted.org"),
        ],
    ),
)

with Sandbox.run(
    defaults=defaults,
    mounted_files=[
        {
            "mount_path": "/app/config.yaml",
            "file_content": b"model: gpt-4\nmax_tokens: 1000",
        },
    ],
    services=[
        Service(port=8000, name="api"),
    ],
    auth=AUTH,
) as sandbox:
    # 캐시를 사용하지 않고 의존성 설치
    sandbox.exec(
        ["pip", "install", "--no-cache-dir", "fastapi", "uvicorn"],
        check=True,
    ).result()

    # 설치된 의존성 확인
    result = sandbox.exec(
        ["python", "-c", "import fastapi, uvicorn; print('Dependencies imported')"],
        check=True,
        timeout_seconds=60,
    ).result()
    print(result.stdout)