Cycle de vie d’une sandbox

from sandbox_auth import AUTH
from cwsandbox import Sandbox

# Rend la main dès que le backend accepte la requête de démarrage
sandbox = Sandbox.run(
    "echo", "hello",
    resources={"cpu": "1", "memory": "1Gi"},
    auth=AUTH,
)

# sandbox_id est défini immédiatement
print(sandbox.sandbox_id)  # "sandbox-abc123"

# Mais le statut peut encore être PENDING, et pas nécessairement déjà RUNNING

from sandbox_auth import AUTH
from cwsandbox import Sandbox

Sandbox.run(
    "echo",
    "hello",
    "world",
    resources={"cpu": "1", "memory": "1Gi"},
    auth=AUTH,
)

from sandbox_auth import AUTH
import cwsandbox
from cwsandbox import SandboxDefaults

with cwsandbox.Session(SandboxDefaults(
    resources={"cpu": "1", "memory": "1Gi"},
), auth=AUTH) as session:
    sandbox = session.sandbox()

    # Aucun appel réseau pour l’instant, sandbox_id vaut None
    print(sandbox.sandbox_id)  # None

    # La première opération déclenche automatiquement le démarrage
    result = sandbox.exec(["echo", "hello"]).result()

from sandbox_auth import AUTH
# Ce sandbox se termine presque immédiatement, car echo s’arrête aussitôt
sandbox = Sandbox.run("echo", "hi", resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
sandbox.wait()  # Peut déjà être à l’état COMPLETED

# Pour les flux de travail basés sur exec, utilisez une commande principale de longue durée
sandbox = Sandbox.run(
    "sleep", "infinity",
    resources={"cpu": "1", "memory": "1Gi"},
    auth=AUTH,
)
# Ou omettez la commande pour utiliser la commande de longue durée par défaut
sandbox = Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)

sandbox = session.sandbox()

# Démarrage explicite pour contrôler les erreurs
sandbox.start().result()  # Bloquant jusqu'à ce que le backend accepte
print(sandbox.sandbox_id)  # Désormais défini

from sandbox_auth import AUTH
from cwsandbox import Sandbox, SandboxStatus

with Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH) as sandbox:
    # start() a été appelé, mais la sandbox peut encore être à l'état PENDING/CREATING
    sandbox.wait()
    if sandbox.status == SandboxStatus.RUNNING:
        result = sandbox.exec(["echo", "ready"]).result()

from sandbox_auth import AUTH
from cwsandbox import Sandbox, SandboxStatus

sandbox = Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
sandbox.wait()  # Attendre le statut de démarrage, puis vérifier le statut

# Vérifier le statut avant de tenter une opération
if sandbox.status == SandboxStatus.RUNNING:
    result = sandbox.exec(["echo", "hello"]).result()

from sandbox_auth import AUTH
sandbox = Sandbox.run(
    "python", "train.py",
    resources={"cpu": "1", "memory": "1Gi"},
    auth=AUTH,
)
sandbox.wait_until_complete(timeout=3600.0).result()
print(sandbox.returncode)  # 0 si l’entraînement a réussi

from cwsandbox import SandboxTerminatedError

# Par défaut : lève une exception après l’appel de stop() par ce client
try:
    sandbox.wait_until_complete().result()
except SandboxTerminatedError:
    print("This client stopped the sandbox")

# Désactiver l’exception pour une gestion sans erreur
sandbox.wait_until_complete(raise_on_termination=False).result()
print(f"Exit code: {sandbox.returncode}")

|<--------------------- max_lifetime_seconds --------------------->|
start() --> [startup wait] --> RUNNING --> exec() --> [operation timeout]

# Sandbox de la Session, pas encore démarrée
sandbox = session.sandbox()

# exec() gère tout : démarrage, attente de l’état RUNNING, puis exécution de la commande
result = sandbox.exec(["echo", "hello"]).result()

sandbox.stop().result()  # Bloquant jusqu’à l’état terminal

# Capturer le montage d’instantané configuré
sandbox.stop(snapshot_on_stop=True).result()

# Nettoyage idempotent : l’appel est sans risque même si la sandbox n’existe plus
sandbox.stop(missing_ok=True).result()

# Laisser plus de temps aux processus pour s’arrêter
sandbox.stop(graceful_shutdown_seconds=30.0).result()

sandbox.stop().result()

# Ces appels lèvent SandboxNotRunningError
sandbox.exec(["echo", "hello"])  # Lève l'exception
sandbox.read_file("/path")       # Lève l'exception

fresh_status = sandbox.get_status()
if fresh_status == SandboxStatus.RUNNING:
    sandbox.stop().result()

from sandbox_auth import AUTH
with Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH) as sandbox:
    result = sandbox.exec(["echo", "hello"]).result()
# stop() est appelé ici, même si une exception a été levée dans le bloc

from sandbox_auth import AUTH
# Arrêter une sandbox que vous avez créée
sandbox.stop().result()

# Supprimer une sandbox orpheline à partir de son ID
Sandbox.delete("sandbox-abc123", missing_ok=True, auth=AUTH).result()

from sandbox_auth import AUTH
from cwsandbox import Sandbox

with Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH) as sandbox:
    result = sandbox.exec(["echo", "hello"]).result()
    print(result.stdout)

from sandbox_auth import AUTH
from cwsandbox import Sandbox, SandboxFailedError, SandboxStatus

sandbox = Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
try:
    sandbox.wait()
except SandboxFailedError:
    print("Failed to start. Check container image and resources")
    raise

if sandbox.status == SandboxStatus.RUNNING:
    result = sandbox.exec(["echo", "ready"]).result()
sandbox.stop().result()

from sandbox_auth import AUTH
from cwsandbox import Sandbox

sandbox = Sandbox.run(
    "python",
    "train.py",
    max_lifetime_seconds=7200,
    resources={"cpu": "1", "memory": "1Gi"},
    auth=AUTH,
)
sandbox.wait_until_complete(timeout=7200.0).result()

print(f"Training exited with code {sandbox.returncode}")

from sandbox_auth import AUTH
from cwsandbox import Sandbox, SandboxStatus

sandbox = Sandbox.from_id("sandbox-abc123", auth=AUTH).result()

# from_id() récupère le statut actuel, mais ne démarre ni ne vérifie la sandbox
if sandbox.status == SandboxStatus.RUNNING:
    result = sandbox.exec(["echo", "reconnected"]).result()
else:
    print(f"Sandbox is {sandbox.status}, not RUNNING")

from sandbox_auth import AUTH
import cwsandbox
from cwsandbox import SandboxDefaults

with cwsandbox.Session(SandboxDefaults(
    tags=("batch-job",),
    resources={"cpu": "1", "memory": "1Gi"},
), auth=AUTH) as session:
    sandboxes = [session.sandbox() for _ in range(5)]

    processes = [
        sb.exec(["python", "-c", f"print({i} ** 2)"])
        for i, sb in enumerate(sandboxes)
    ]

    done, pending = cwsandbox.wait(processes)
    for p in done:
        print(p.result().stdout)