Exécution de commandes

Ce guide explique comment exécuter des commandes dans des sandboxes à l’aide de la méthode exec(). Utilisez-la pour exécuter des commandes shell, des scripts ou des interpréteurs interactifs dans une sandbox lorsque vous souhaitez capturer leur sortie, la diffuser en temps réel, envoyer des données sur stdin ou contrôler la gestion des codes de sortie non nuls. Ce guide s’adresse aux développeurs qui utilisent déjà le client Python cwsandbox.

Choisir l’authentification

Ces exemples s’exécutent sur une capacité sans serveur, sauf si une section indique CKS. Choisissez l’identifiant d’authentification du compte propriétaire de vos Sandbox. Consultez Choisir et obtenir des identifiants d’authentification pour obtenir une clé ou un jeton. Exécutez les commandes d’installation dans un environnement virtuel Python actif. Enregistrez le code Python de l’onglet correspondant à votre identifiant d’authentification sous le nom sandbox_auth.py, dans le même répertoire que les scripts Python que vous exécutez depuis cette page. Ce fichier enregistre la méthode d’authentification choisie, et non votre identifiant d’authentification. Les exemples importent AUTH depuis ce fichier et le transmettent explicitement à chaque nouvelle Sandbox ou session.

Clé API Forge
Jeton d’accès à l’API CoreWeave

Utilisez votre clé API Forge via la stratégie d’authentification W&B du SDK :

uv pip install 'cwsandbox[wandb]>=1.14.2'
export WANDB_API_KEY="[WANDB-API-KEY]"

sandbox_auth.py

from cwsandbox import AuthStrategy

AUTH = AuthStrategy.WANDB

Utilisez le jeton de votre organisation CoreWeave :

uv pip install 'cwsandbox>=1.14.2'
export CWSANDBOX_API_KEY="[API-ACCESS-TOKEN]"

sandbox_auth.py

from cwsandbox import AuthStrategy

AUTH = AuthStrategy.COREWEAVE_API_KEY

Pour un placement sur CKS, choisissez le jeton CoreWeave, activez un exécuteur et définissez placement_mode="cks" lors de la création de la Sandbox ou dans SandboxDefaults. Créer une session avec auth=AUTH authentifie également les Sandbox qui lui appartiennent. Utilisez le même identifiant d’authentification pour vous reconnecter aux Sandbox, les lister ou les nettoyer.

Exécuter une commande de base

Commencez par une commande unique qui s’exécute jusqu’à son achèvement et renvoie la sortie capturée. La méthode exec() renvoie un handle Process :

from sandbox_auth import AUTH
from cwsandbox import Sandbox

with Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH) as sandbox:
    # Exécuter une commande et obtenir le résultat
    result = sandbox.exec(["echo", "Hello, World!"]).result()

    print(result.stdout)      # "Hello, World!\n"
    print(result.returncode)  # 0

Obtenir les résultats

exec() rend la main immédiatement, ce qui vous permet de choisir comment attendre la fin de la commande. Appelez .result() sur le handle Process pour bloquer jusqu’à l’obtention de la sortie :

# Renvoie immédiatement un objet Process
process = sandbox.exec(["python", "-c", "print('hello')"])

# Attendre le résultat (appel bloquant)
result = process.result()
print(result.stdout)      # "hello\n"
print(result.stderr)      # ""
print(result.returncode)  # 0

# Variante en une seule ligne
result = sandbox.exec(["ls", "-la"]).result()

Sortie en flux

Un appel bloquant à .result() attend que la commande se termine avant d’afficher la moindre sortie. Pour suivre la progression en temps réel, itérez sur process.stdout avant d’appeler .result() :

# Renvoie immédiatement un objet Process
process = sandbox.exec(["python", "long_script.py"])

# Lire stdout en flux, ligne par ligne
for line in process.stdout:
    print(f"[stdout] {line}", end="")

# Obtenir le résultat final
result = process.result()
print(f"Exit code: {result.returncode}")

Utilisez la diffusion en continu lorsque vous devez :

Surveiller des processus de longue durée.
Traiter la sortie au fur et à mesure qu’elle arrive.
Mettre en place des indicateurs de progression.

Transmettre une entrée en flux vers stdin

Certaines commandes ont besoin de recevoir une entrée sur stdin pendant leur exécution, comme les pipelines, les interpréteurs interactifs ou les outils qui lisent jusqu’à EOF. Pour envoyer une entrée à des commandes en cours d’exécution, activez stdin avec stdin=True :

from sandbox_auth import AUTH
with Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH) as sandbox:
    process = sandbox.exec(["cat"], stdin=True)

    process.stdin.write(b"hello world\n").result()
    process.stdin.close().result()

    result = process.result()
    print(result.stdout)  # "hello world\n"

Méthodes de StreamWriter

Lorsque stdin=True, process.stdin est un StreamWriter doté de trois méthodes :

write(data: bytes) : écrit des octets bruts. Renvoie OperationRef[None].
writeline(text: str) : écrit du texte suivi d’un saut de ligne (encodé en UTF-8). Renvoie OperationRef[None].
close() : signale la fin de fichier (EOF). Le système finalise d’abord les écritures en attente. Renvoie OperationRef[None].

Lorsque stdin=False (valeur par défaut), process.stdin vaut None.

Effectuer plusieurs écritures

Envoyez les données de manière incrémentielle avant de fermer le flux :

process = sandbox.exec(["cat"], stdin=True)

process.stdin.writeline("line 1").result()
process.stdin.writeline("line 2").result()
process.stdin.writeline("line 3").result()
process.stdin.close().result()

result = process.result()
print(result.stdout)  # "line 1\nline 2\nline 3\n"

Exécuter du code Python en mode interactif via stdin

Transmettez du code Python à un interpréteur interactif :

process = sandbox.exec(["python3"], stdin=True)

process.stdin.writeline("x = 40 + 2").result()
process.stdin.writeline("print(f'answer: {x}')").result()
process.stdin.close().result()

result = process.result()
print(result.stdout)  # "answer: 42\n"

Combiner la diffusion en continu sur stdin et stdout

Diffusez la sortie en continu tout en envoyant des entrées :

process = sandbox.exec(["cat"], stdin=True)

# Envoyer l’entrée
process.stdin.writeline("hello").result()
process.stdin.writeline("world").result()
process.stdin.close().result()

# Lire la sortie en flux au fur et à mesure
for line in process.stdout:
    print(f"[out] {line}", end="")

result = process.result()

Gérer les commandes qui dépendent d’EOF

Certaines commandes (comme sort) lisent l’intégralité de l’entrée avant de produire une sortie. Fermez stdin pour signaler la fin de fichier (EOF) :

process = sandbox.exec(["sort"], stdin=True)

process.stdin.writeline("banana").result()
process.stdin.writeline("apple").result()
process.stdin.writeline("cherry").result()
process.stdin.close().result()  # sort attend EOF avant de produire sa sortie

result = process.result()
print(result.stdout)  # "apple\nbanana\ncherry\n"

Utiliser stdin dans un contexte asynchrone

Dans un contexte asynchrone, utilisez Sandbox(...) afin que le gestionnaire de contexte asynchrone démarre la sandbox sans bloquer la boucle d’événements. Appliquez await à chaque OperationRef :

from sandbox_auth import AUTH
async with Sandbox(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH) as sandbox:
    process = sandbox.exec(["cat"], stdin=True)

    await process.stdin.write(b"async hello\n")
    await process.stdin.close()

    result = await process
    print(result.stdout)  # "async hello\n"

Quand utiliser stdin=True ou stdin=False

Scénario	stdin	Raison
Exécuter une commande avec des arguments	`False`	L’entrée provient des arguments, et non de stdin
Transmettre des données à une commande via un pipe	`True`	La commande lit depuis stdin
Interpréteur interactif	`True`	L’interpréteur lit les commandes depuis stdin
Processus qui lit jusqu’à EOF	`True`	Nécessite `close()` pour signaler EOF
Commande fire-and-forget	`False`	Aucune entrée requise

Définir le répertoire de travail

Par défaut, les commandes s’exécutent dans le répertoire de travail par défaut de la sandbox. Pour en utiliser un autre, remplacez-le avec cwd :

result = sandbox.exec(
    ["ls", "-la"],
    cwd="/app/data",
).result()

Le chemin doit être absolu.

Définir un délai d’expiration

Pour une commande plus longue exécutée en mode attaché, transmettez une valeur timeout_seconds explicite à exec(), par exemple 1800 pour un amorçage de 30 minutes. Consommez la sortie du processus à l’aide des méthodes décrites dans Sortie en flux. Par défaut, le délai d’expiration d’une requête exec est de 300 secondes. La durée de vie de la sandbox doit également couvrir toute la durée d’exécution de la commande. Un délai d’expiration exec plus long ne permet pas à la commande de survivre à une déconnexion du client. Pour arrêter une commande susceptible de se bloquer ou de s’exécuter plus longtemps que prévu, définissez un délai d’expiration avec timeout_seconds :

from cwsandbox import SandboxTimeoutError

try:
    result = sandbox.exec(
        ["sleep", "60"],
        timeout_seconds=5.0,
    ).result()
except SandboxTimeoutError:
    print("Command timed out")

Gérer les erreurs avec check

Le paramètre check détermine le comportement à adopter lorsque le code de sortie est non nul :

Comportement par défaut avec check=False

Renvoie le résultat, quel que soit le code de sortie :

result = sandbox.exec(["false"]).result()
print(result.returncode)  # 1 (aucune exception levée)

Lever une exception en cas d’échec avec check=True

Lève une exception SandboxExecutionError si le code de sortie est non nul :

from cwsandbox import SandboxExecutionError

try:
    result = sandbox.exec(
        ["python", "-c", "raise ValueError('oops')"],
        check=True,
    ).result()
except SandboxExecutionError as e:
    print(f"Command failed: {e.exec_result.returncode}")
    print(f"stderr: {e.exec_result.stderr}")

Exécuter du code Python

Les bacs à sable servent couramment à exécuter du code Python, qu’il s’agisse de courtes commandes d’une seule ligne ou de scripts plus longs transmis directement en ligne :

# Commande en une ligne
result = sandbox.exec(
    ["python", "-c", "import sys; print(sys.version)"],
).result()

# Script à partir d’une chaîne de caractères
code = '''
import json
data = {"result": 42}
print(json.dumps(data))
'''
result = sandbox.exec(["python", "-c", code]).result()
output = json.loads(result.stdout)

Exécution séquentielle ou exécution parallèle

Choisissez l’exécution séquentielle lorsque des commandes dépendent de commandes précédentes, et l’exécution parallèle lorsque les commandes sont indépendantes et que vous souhaitez les exécuter simultanément sur plusieurs bacs à sable.

Exécuter les commandes séquentiellement lorsque l’ordre est important

# Les dépendances nécessitent une exécution séquentielle
sandbox.exec(["pip", "install", "requests"]).result()
sandbox.exec(["python", "script_using_requests.py"]).result()

Exécuter des commandes indépendantes en parallèle

from sandbox_auth import AUTH
# Démarrer plusieurs bacs à sable
sandboxes = [
    Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
    for _ in range(3)
]

# Lancer des commandes dans chacun d’eux
processes = [
    sb.exec(["python", "-c", f"print({i})"])
    for i, sb in enumerate(sandboxes)
]

# Récupérer tous les résultats
results = [p.result() for p in processes]
for r in results:
    print(r.stdout)

Attendre la fin de N processus sur M

Utilisez cwsandbox.wait() pour attendre qu’un sous-ensemble de processus se termine :

import cwsandbox

processes = [sb.exec(["python", "task.py"]) for sb in sandboxes]

# Attendre la fin des 2 premiers processus
done, pending = cwsandbox.wait(processes, num_returns=2)

# Traiter immédiatement les processus terminés
for p in done:
    print(p.result().stdout)

# Attendre les processus restants
for p in pending:
    print(p.result().stdout)

Contrôler les processus

Lorsqu’une commande continue de s’exécuter une fois que l’appel exec() a rendu la main, vous pouvez vérifier son état sans bloquer, ou attendre qu’elle se termine au moment qui vous convient. Le handle Process fournit des méthodes de surveillance et de contrôle :

process = sandbox.exec(["python", "server.py"])

# Vérifier si le processus est toujours en cours d'exécution (non-blocking)
if process.poll() is None:
    print("Still running")

# Attendre l'achèvement du processus
exit_code = process.wait()
print(f"Exited with code: {exit_code}")

Sandbox

Utiliser les Sandbox

Gérer les bacs à sable

Référence

Choisir l’authentification

Exécuter une commande de base

Obtenir les résultats

Sortie en flux

Transmettre une entrée en flux vers stdin

Méthodes de StreamWriter

Effectuer plusieurs écritures

Exécuter du code Python en mode interactif via stdin

Combiner la diffusion en continu sur stdin et stdout

Gérer les commandes qui dépendent d’EOF

Utiliser stdin dans un contexte asynchrone

Quand utiliser stdin=True ou stdin=False

Définir le répertoire de travail

Définir un délai d’expiration

Gérer les erreurs avec check

Comportement par défaut avec check=False

Lever une exception en cas d’échec avec check=True

Exécuter du code Python

Exécution séquentielle ou exécution parallèle

Exécuter les commandes séquentiellement lorsque l’ordre est important

Exécuter des commandes indépendantes en parallèle

Attendre la fin de N processus sur M

Contrôler les processus

​Choisir l’authentification

​Exécuter une commande de base

​Obtenir les résultats

​Sortie en flux

​Transmettre une entrée en flux vers stdin

​Méthodes de StreamWriter

​Effectuer plusieurs écritures

​Exécuter du code Python en mode interactif via stdin

​Combiner la diffusion en continu sur stdin et stdout

​Gérer les commandes qui dépendent d’EOF

​Utiliser stdin dans un contexte asynchrone

​Quand utiliser stdin=True ou stdin=False

​Définir le répertoire de travail

​Définir un délai d’expiration

​Gérer les erreurs avec check

​Comportement par défaut avec check=False

​Lever une exception en cas d’échec avec check=True

​Exécuter du code Python

​Exécution séquentielle ou exécution parallèle

​Exécuter les commandes séquentiellement lorsque l’ordre est important

​Exécuter des commandes indépendantes en parallèle

​Attendre la fin de N processus sur M

​Contrôler les processus

Choisir l’authentification

Exécuter une commande de base

Obtenir les résultats

Sortie en flux

Transmettre une entrée en flux vers stdin

Méthodes de StreamWriter

Effectuer plusieurs écritures

Exécuter du code Python en mode interactif via stdin

Combiner la diffusion en continu sur stdin et stdout

Gérer les commandes qui dépendent d’EOF

Utiliser stdin dans un contexte asynchrone

Quand utiliser stdin=True ou stdin=False

Définir le répertoire de travail

Définir un délai d’expiration

Gérer les erreurs avec check

Comportement par défaut avec check=False

Lever une exception en cas d’échec avec check=True

Exécuter du code Python

Exécution séquentielle ou exécution parallèle

Exécuter les commandes séquentiellement lorsque l’ordre est important

Exécuter des commandes indépendantes en parallèle

Attendre la fin de N processus sur M

Contrôler les processus