Modèles synchrones ou asynchrones

Ces exemples utilisent un jeton d’accès à l’API CoreWeave défini dans CWSANDBOX_API_KEY. Pour l’authentification par clé API Forge, voir Premiers pas. Pour les exigences relatives à CKS, voir Modes de placement. 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.

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 cwsandbox import Sandbox

# Gestionnaire de contexte (recommandé)
with Sandbox.run() as sb:
    result = sb.exec(["echo", "hello"]).result()
# Arrêté automatiquement à la sortie

from cwsandbox import Sandbox

# Instancier + await pour atteindre RUNNING, puis gestionnaire de contexte asynchrone pour le nettoyage
sb = Sandbox()
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 ou await sandbox pour atteindre le statut RUNNING sans bloquer l’exécution.

start()

Synchrone
Asynchrone

sb = Sandbox()
sb.start().result()  # Bloque jusqu'à ce que le backend accepte la requête
print(sb.sandbox_id)

sb = Sandbox()
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() bloque l’exécution jusqu’à ce que la sandbox atteigne le statut RUNNING. Cette méthode est sync-only, car le blocage est précisément le but recherché. Dans du code asynchrone, await sandbox produit le même effet.

# Synchrone : bloque jusqu’à l’état RUNNING
sb = Sandbox.run().wait()
result = sb.exec(["echo", "ready"]).result()

# Asynchrone : attendre directement la sandbox
sb = Sandbox()
await sb  # Attendre l’état RUNNING
result = await sb.exec(["echo", "ready"])

wait_until_complete()

Synchrone
Asynchrone

sb = Sandbox.run("python", "-c", "print('done')")
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()

sb = Sandbox(command="python", args=["-c", "print('done')"])
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

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

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

Sandbox.from_id()

Synchrone
Asynchrone

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

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

Sandbox.delete()

Synchrone
Asynchrone

Sandbox.delete("sandbox-abc123").result()

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

await Sandbox.delete("sandbox-abc123")

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

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). Une Sandbox est résolue dès qu’elle atteint le statut RUNNING, et non à son achèvement.

import cwsandbox

# Attendre que toutes les sandboxes atteignent l’état RUNNING
sandboxes = [Sandbox.run() 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

with Session(defaults) 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]

async with Session(defaults) 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 cwsandbox import Sandbox, Session, SandboxDefaults

# Option 1 : session.sandbox() - les sandboxes démarrent automatiquement au premier appel à exec()
with Session(SandboxDefaults()) 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() 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 :

# Cellule 1 - Créer la sandbox
from cwsandbox import Sandbox
sandbox = Sandbox.run()
sandbox.wait()  # Attendre l’état RUNNING

# Cellule 2 - Exécuter des commandes
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 :

# Fonctionne dans Jupyter sans asyncio.run()
sandboxes = await Sandbox.list(tags=["notebook"])

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

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

​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

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