Modèles synchrones ou asynchrones

Ce guide vous aide à choisir entre les modèles synchrones et asynchrones dans le SDK Python de sandbox CoreWeave (cwsandbox) et présente les variantes synchrone et asynchrone de chaque opération courante. En interne, le SDK repose sur une implémentation asynchrone unique. Le choix entre synchrone et asynchrone tient uniquement à la manière dont vous consommez les résultats : utilisez .result() en mode synchrone et await en mode asynchrone.

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. Pour obtenir une clé ou un jeton, consultez Choisir et obtenir des identifiants d’authentification. 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 le mode d’authentification choisi, 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 avec 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.

Guide de décision rapide

Cas d’utilisation	Approche	Raison
La plupart des opérations	Synchrone	Code plus simple, sans code passe-partout asyncio
Exécution parallèle	Synchrone	Les opérations sont conçues pour être non-blocking
Notebooks Jupyter	Synchrone	Pas besoin de `nest_asyncio`
Base de code asynchrone	Asynchrone	S’intègre au code asynchrone existant

Règle générale : privilégiez l’approche synchrone (.result()) pour plus de simplicité. Ne passez à await que si vous travaillez déjà dans une base de code asynchrone.

Concept fondamental : OperationRef

La plupart des méthodes du SDK renvoient OperationRef[T], un encapsuleur qui prend en charge à la fois l’appel à .result() et await. Les deux approches lèvent les mêmes exceptions.

ref = sandbox.read_file("/output/data.txt")

# Synchrone : bloquer jusqu’à l’obtention de la valeur
data = ref.result()

# Asynchrone : utiliser await à la place
data = await ref

Sandbox.exec() renvoie Process, qui étend OperationRef[ProcessResult] en y ajoutant la diffusion en continu et stdin.

N’utilisez jamais .result() dans un contexte asynchrone. Appeler .result() bloque le thread. Dans un contexte asynchrone, cela bloque la boucle d’événements et peut provoquer un interblocage de votre application. Utilisez plutôt await.

Comportement de démarrage automatique

session.sandbox() renvoie une sandbox non démarrée. La sandbox démarre automatiquement lors de la première opération qui la nécessite : Déclenchent le démarrage automatique : exec(), read_file(), write_file(), wait(), wait_until_complete() Ne déclenchent pas le démarrage automatique : get_status() (lève SandboxNotRunningError), stop() (no-op si la sandbox n’a jamais démarré)

Opérations

Créer des Sandbox

Synchrone
Asynchrone

from sandbox_auth import AUTH
from cwsandbox import Sandbox

# Gestionnaire de contexte (recommandé)
with Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH) as sb:
    result = sb.exec(["echo", "hello"]).result()
# Arrêté automatiquement à la sortie

from sandbox_auth import AUTH
from cwsandbox import Sandbox

# Instancier, puis utiliser un gestionnaire de contexte asynchrone pour le démarrage et le nettoyage
sb = Sandbox(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
async with sb as sb:
    result = await sb.exec(["echo", "hello"])
# Arrêté automatiquement à la sortie

Sandbox.run() appelle start().result() en interne, ce qui bloque la boucle d’événements. Dans du code asynchrone, instanciez l’objet avec Sandbox(...) et utilisez async with pour démarrer la sandbox. exec() attend le statut RUNNING avant de s’exécuter. Pour attendre le démarrage séparément, utilisez await sandbox et vérifiez son statut avant toute opération.

start()

Synchrone
Asynchrone

from sandbox_auth import AUTH
sb = Sandbox(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
sb.start().result()  # Bloque jusqu'à ce que le backend accepte la requête
print(sb.sandbox_id)

from sandbox_auth import AUTH
sb = Sandbox(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
await sb.start()  # Attend que le backend accepte la requête
print(sb.sandbox_id)

exec()

Synchrone
Asynchrone

result = sb.exec(["echo", "hello"]).result()
print(result.stdout)  # "hello\n"

# Lever une exception si le code de sortie est non nul
result = sb.exec(["python", "-c", "exit(1)"], check=True).result()

result = await sb.exec(["echo", "hello"])
print(result.stdout)  # "hello\n"

# Lever une exception si le code de sortie est non nul
result = await sb.exec(["python", "-c", "exit(1)"], check=True)

read_file()

Synchrone
Asynchrone

data = sb.read_file("/output/result.txt").result()
print(data.decode())

data = await sb.read_file("/output/result.txt")
print(data.decode())

write_file()

Synchrone
Asynchrone

sb.write_file("/input/data.txt", b"content").result()

await sb.write_file("/input/data.txt", b"content")

stop()

Synchrone
Asynchrone

sb.stop().result()

# Ignorer si déjà supprimé
sb.stop(missing_ok=True).result()

await sb.stop()

# Ignorer si déjà supprimé
await sb.stop(missing_ok=True)

wait()

wait() attend le statut de démarrage. Par exemple, la méthode peut rendre la main lorsque la sandbox est RUNNING ou TERMINATING, ou une fois qu’elle s’est terminée avec succès. Elle est sync-only. Dans du code asynchrone, await sandbox effectue la même attente de démarrage sans bloquer la boucle d’événements. Un retour normal ne garantit pas l’état RUNNING. Vérifiez status avant toute opération qui nécessite une sandbox en cours d’exécution. Pour la gestion des états terminaux et des erreurs, consultez Attendre une sandbox.

from sandbox_auth import AUTH
from cwsandbox import Sandbox, SandboxStatus

# Synchrone : attendre le statut de démarrage, puis vérifier le statut
sb = Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH).wait()
if sb.status == SandboxStatus.RUNNING:
    result = sb.exec(["echo", "ready"]).result()

from sandbox_auth import AUTH
from cwsandbox import Sandbox, SandboxStatus

# Asynchrone : attendre avec await la même phase de démarrage, puis vérifier le statut
sb = Sandbox(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
await sb
if sb.status == SandboxStatus.RUNNING:
    result = await sb.exec(["echo", "ready"])

wait_until_complete()

Synchrone
Asynchrone

from sandbox_auth import AUTH
sb = Sandbox.run(
    "python",
    "-c",
    "print('done')",
    resources={"cpu": "1", "memory": "1Gi"},
    auth=AUTH,
)
sb.wait_until_complete().result()
print(f"Exit code: {sb.returncode}")

# Gérer les bacs à sable arrêtés de l'extérieur sans lever d'exception
sb.wait_until_complete(raise_on_termination=False).result()

from sandbox_auth import AUTH
sb = Sandbox(
    command="python",
    args=["-c", "print('done')"],
    resources={"cpu": "1", "memory": "1Gi"},
    auth=AUTH,
)
await sb.wait_until_complete()
print(f"Exit code: {sb.returncode}")

# Gérer les bacs à sable arrêtés de l'extérieur sans lever d'exception
await sb.wait_until_complete(raise_on_termination=False)

get_status()

get_status() est sync-only. Cette méthode récupère le statut actuel auprès de l’API.

status = sb.get_status()
print(f"Sandbox is {status}")  # par ex. SandboxStatus.RUNNING

Sandbox.list()

Synchrone
Asynchrone

from sandbox_auth import AUTH
sandboxes = Sandbox.list(tags=["my-job"], auth=AUTH).result()
for sb in sandboxes:
    print(f"{sb.sandbox_id}: {sb.status}")

from sandbox_auth import AUTH
sandboxes = await Sandbox.list(tags=["my-job"], auth=AUTH)
for sb in sandboxes:
    print(f"{sb.sandbox_id}: {sb.status}")

Sandbox.from_id()

Synchrone
Asynchrone

from sandbox_auth import AUTH
sb = Sandbox.from_id("sandbox-abc123", auth=AUTH).result()
result = sb.exec(["echo", "reconnected"]).result()

from sandbox_auth import AUTH
sb = await Sandbox.from_id("sandbox-abc123", auth=AUTH)
result = await sb.exec(["echo", "reconnected"])

Sandbox.delete()

Synchrone
Asynchrone

from sandbox_auth import AUTH
Sandbox.delete("sandbox-abc123", auth=AUTH).result()

# Ignorer si déjà supprimé
Sandbox.delete("sandbox-abc123", missing_ok=True, auth=AUTH).result()

from sandbox_auth import AUTH
await Sandbox.delete("sandbox-abc123", auth=AUTH)

# Ignorer si déjà supprimé
await Sandbox.delete("sandbox-abc123", missing_ok=True, auth=AUTH)

session.list()

Synchrone
Asynchrone

sandboxes = session.list().result()
for sb in sandboxes:
    print(f"{sb.sandbox_id}: {sb.status}")

sandboxes = await session.list()
for sb in sandboxes:
    print(f"{sb.sandbox_id}: {sb.status}")

session.from_id()

Synchrone
Asynchrone

sb = session.from_id("sandbox-abc123").result()
result = sb.exec(["echo", "adopted"]).result()

sb = await session.from_id("sandbox-abc123")
result = await sb.exec(["echo", "adopted"])

Diffuser la sortie stdout en continu

Synchrone
Asynchrone

process = sb.exec(["python", "-c", "import time; [print(i) or time.sleep(0.1) for i in range(5)]"])
for line in process.stdout:
    print(line, end="")
result = process.result()

process = sb.exec(["python", "-c", "import time; [print(i) or time.sleep(0.1) for i in range(5)]"])
async for line in process.stdout:
    print(line, end="")
result = await process

Transmettre stdin en flux

Activez stdin avec stdin=True. Utilisez write() pour les octets bruts, writeline() pour les lignes de texte et close() pour signaler la fin de fichier (EOF).

Synchrone
Asynchrone

process = sb.exec(["cat"], stdin=True)
process.stdin.write(b"hello ").result()
process.stdin.writeline("world").result()
process.stdin.close().result()  # Signale la fin de fichier (EOF) au processus
result = process.result()
print(result.stdout)  # "hello world\n"

process = sb.exec(["cat"], stdin=True)
await process.stdin.write(b"hello ")
await process.stdin.writeline("world")
await process.stdin.close()  # Signale la fin de fichier (EOF) au processus
result = await process
print(result.stdout)  # "hello world\n"

cwsandbox.results()

cwsandbox.results() est un utilitaire de traitement par lot sync-only. Il appelle .result() sur un ou plusieurs objets OperationRef.

import cwsandbox

# Réf. unique
data = cwsandbox.results(sandbox.read_file("/path"))

# Réf. multiples
all_data = cwsandbox.results([sb.read_file(f) for f in files])

cwsandbox.wait()

cwsandbox.wait() est sync-only. Cette fonction attend une séquence d’objets Sandbox, OperationRef ou Process et renvoie (done, pending). Pour les Sandbox, elle applique la même attente de démarrage que Sandbox.wait(), plutôt que d’attendre la fin de la commande principale. Une sandbox présente dans done n’est pas nécessairement à l’état RUNNING : elle peut par exemple être déjà terminée ou en cours d’arrêt. Vérifiez le status de chaque sandbox avant toute opération qui exige qu’elle soit en cours d’exécution.

from sandbox_auth import AUTH
import cwsandbox

# Attendre la fin de la phase de démarrage de toutes les sandboxes
sandboxes = [
    Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
    for _ in range(5)
]
done, pending = cwsandbox.wait(sandboxes)

# Attendre la fin des 2 premières opérations
refs = [sb.read_file(f) for f in files]
done, pending = cwsandbox.wait(refs, num_returns=2)

# Attendre avec un délai d’expiration
done, pending = cwsandbox.wait(procs, timeout=30.0)

@session.function()

Synchrone
Asynchrone

from sandbox_auth import AUTH
with Session(defaults, auth=AUTH) as session:
    @session.function()
    def compute(x: int, y: int) -> int:
        return x + y

    result = compute.remote(2, 3).result()
    print(result)  # 5

    # Map en parallèle
    refs = compute.map([(1, 2), (3, 4), (5, 6)])
    results = [r.result() for r in refs]  # [3, 7, 11]

from sandbox_auth import AUTH
async with Session(defaults, auth=AUTH) as session:
    @session.function()
    def compute(x: int, y: int) -> int:
        return x + y

    result = await compute.remote(2, 3)
    print(result)  # 5

    # Map en parallèle
    refs = compute.map([(1, 2), (3, 4), (5, 6)])
    results = [await r for r in refs]  # [3, 7, 11]

Exécution parallèle

L’API synchrone prend en charge l’exécution parallèle, car les opérations sont non-blocking dès la conception. Les méthodes telles que exec(), read_file() et write_file() rendent la main immédiatement. Le blocage n’intervient que lorsque vous appelez .result(). Sandbox.run() reste bloquante jusqu’à ce que le backend accepte la requête de démarrage (elle appelle start().result() en interne). Pour un démarrage parallèle, utilisez session.sandbox() (démarrage automatique à la première opération) ou collectez les réf. renvoyées par start() :

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

# Option 1 : session.sandbox() - les sandboxes démarrent automatiquement au premier appel à exec()
with Session(
    SandboxDefaults(resources={"cpu": "1", "memory": "1Gi"}),
    auth=AUTH,
) as session:
    sandboxes = [session.sandbox() for _ in range(3)]
    processes = [sb.exec(["echo", f"sb-{i}"]) for i, sb in enumerate(sandboxes)]
    results = [p.result() for p in processes]

# Option 2 : Sandbox() + collecte des réf. renvoyées par start()
sandboxes = [
    Sandbox(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
    for _ in range(3)
]
start_refs = [sb.start() for sb in sandboxes]
for ref in start_refs:
    ref.result()  # Tous les démarrages se déroulent en parallèle

processes = [sb.exec(["echo", f"sb-{i}"]) for i, sb in enumerate(sandboxes)]
results = [p.result() for p in processes]
for sb in sandboxes:
    sb.stop().result()

Notebooks Jupyter

L’API synchrone fonctionne dans Jupyter sans nest_asyncio, car le SDK exécute sa propre boucle d’événements en arrière-plan dans un thread démon :

from sandbox_auth import AUTH
# Cellule 1 - Créer la sandbox
from cwsandbox import Sandbox, SandboxStatus
sandbox = Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
sandbox.wait()  # Attendre le statut de démarrage

# Cellule 2 - Vérifier le statut avant d’exécuter des commandes
if sandbox.status == SandboxStatus.RUNNING:
    result = sandbox.exec(["python", "-c", "print(1+1)"]).result()
    print(result.stdout)

# Cellule 3 - Nettoyage
sandbox.stop().result()

Pour le code asynchrone dans Jupyter, vous pouvez utiliser await directement, car Jupyter dispose d’une boucle d’événements intégrée :

from sandbox_auth import AUTH
# Fonctionne dans Jupyter sans asyncio.run()
sandboxes = await Sandbox.list(tags=["notebook"], auth=AUTH)

Gestion des erreurs

Les modes synchrone et asynchrone lèvent les mêmes exceptions. .result() relance toute exception issue de l’opération sous-jacente, tout comme await. Consultez Que signifient les messages d’erreur courants du SDK Sandbox ? pour connaître la hiérarchie complète des exceptions et les stratégies de récupération.

Sandbox

Utiliser les Sandbox

Gérer les bacs à sable

Référence

Modèles synchrones ou asynchrones

Choisir l’authentification

Guide de décision rapide

Concept fondamental : OperationRef

Comportement de démarrage automatique

Opérations

Créer des Sandbox

start()

exec()

read_file()

write_file()

stop()

wait()

wait_until_complete()

get_status()

Sandbox.list()

Sandbox.from_id()

Sandbox.delete()

session.list()

session.from_id()

Diffuser la sortie stdout en continu

Transmettre stdin en flux

cwsandbox.results()

cwsandbox.wait()

@session.function()

Exécution parallèle

Notebooks Jupyter

Gestion des erreurs

​Choisir l’authentification

​Guide de décision rapide

​Concept fondamental : OperationRef

​Comportement de démarrage automatique

​Opérations

​Créer des Sandbox

​start()

​exec()

​read_file()

​write_file()

​stop()

​wait()

​wait_until_complete()

​get_status()

​Sandbox.list()

​Sandbox.from_id()

​Sandbox.delete()

​session.list()

​session.from_id()

​Diffuser la sortie stdout en continu

​Transmettre stdin en flux

​cwsandbox.results()

​cwsandbox.wait()

​@session.function()

​Exécution parallèle

​Notebooks Jupyter

​Gestion des erreurs

Choisir l’authentification

Guide de décision rapide

Concept fondamental : OperationRef

Comportement de démarrage automatique

Opérations

Créer des Sandbox

start()

exec()

read_file()

write_file()

stop()

wait()

wait_until_complete()

get_status()

Sandbox.list()

Sandbox.from_id()

Sandbox.delete()

session.list()

session.from_id()

Diffuser la sortie stdout en continu

Transmettre stdin en flux

cwsandbox.results()

cwsandbox.wait()

@session.function()

Exécution parallèle

Notebooks Jupyter

Gestion des erreurs