Configuration de la sandbox

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

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

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

# Via les kwargs de @session.function()
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

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

# Avec un 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

# Raccourci sous forme de dict plat : les demandes de ressources et les limites sont définies sur des valeurs identiques
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,
)

# Les fichiers sont immédiatement disponibles
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  # L’URL peut être disponible avant que l’application soit prête.
        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

# Avec 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,
)

# Avec un 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:
    # Toutes les sandboxes héritent de la configuration réseau
    sb1 = session.sandbox()
    sb2 = session.sandbox()

    # Désactiver les autorisations de destination pour cette sandbox. Le DNS est contrôlé séparément
    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:
    # Hérite des annotations par défaut
    sb1 = session.sandbox()

    # Fusion : en cas de conflit de clés, les annotations explicites remplacent les valeurs par défaut
    sb2 = session.sandbox(
        annotations={"experiment": "run-42", "team": "ml-research"},
    )
    # sb2 reçoit : 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 heures
    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 heure
    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 minutes
).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 minutes
    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:
    # Installer les dépendances sans utiliser le cache de téléchargement
    sandbox.exec(
        ["pip", "install", "--no-cache-dir", "fastapi", "uvicorn"],
        check=True,
    ).result()

    # Vérifier les dépendances installées
    result = sandbox.exec(
        ["python", "-c", "import fastapi, uvicorn; print('Dependencies imported')"],
        check=True,
        timeout_seconds=60,
    ).result()
    print(result.stdout)