同期と非同期のパターン

このガイドでは、CoreWeave サンドボックス Python SDK (cwsandbox) における同期パターンと非同期パターンの選択方法を説明し、一般的な各操作の同期形式と非同期形式を示します。 SDK の内部実装は Async のみです。sync/Async の柔軟性は結果の取得方法によるものです。sync の場合は .result() を、Async の場合は await を使用します。

認証を選択する

これらの例は、セクションで CKS が指定されている場合を除き、サーバーレスキャパシティ上で実行されます。サンドボックスを所有するアカウントの認証情報を選択してください。キーまたはトークンを取得するには、認証情報の選択と取得を参照してください。インストールコマンドは、有効化した Python 仮想環境内で実行してください。使用する認証情報のタブにある Python コードを、sandbox_auth.py という名前で、このページで実行する Python スクリプトと同じディレクトリに保存します。このファイルに保存されるのはどの認証方式を使うかという設定のみで、認証情報そのものは含まれません。各例ではこのファイルから AUTH をインポートし、新しいサンドボックスやセッションを作成するたびに明示的に渡します。

Forge API キー
CoreWeave API アクセストークン

SDK の W&B 認証方式を使用して、Forge の APIキーで認証します。

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

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

CKS に配置する場合は、CoreWeave トークンを選択し、ランナーを有効化したうえで、サンドボックスの作成時または SandboxDefaults で placement_mode="cks" を設定します。auth=AUTH を指定してセッションを作成すると、そのセッションが所有するサンドボックスも認証されます。サンドボックスへの再接続、一覧表示、クリーンアップを行う際も、同じ認証情報を使用してください。

クイック決定ガイド

使用例	パターン	理由
ほとんどの操作	Sync	コードがシンプルで、asyncio のボイラープレートが不要
並列実行	Sync	操作は設計上ノンブロッキング
Jupyter ノートブック	Sync	`nest_asyncio` は不要
Async コードベース	Async	既存の async コードと連携する

経験則: シンプルにするために sync パターン (.result()) を使用してください。すでに async コードベースにいる場合にのみ await に切り替えてください。

Core concept: OperationRef

ほとんどの SDK メソッドは OperationRef[T] を返します。これは .result() 可能かつ await 可能なラッパーです。どちらのパスも同じ例外を発生させます。

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

# Sync: 値をブロックして待機
data = ref.result()

# Async: 代わりに await
data = await ref

Sandbox.exec() は Process を返します。これは OperationRef[ProcessResult] を拡張し、ストリーミングと stdin をサポートします。

async コンテキストでは .result() を使用しないでください。.result() を呼び出すとスレッドがブロックされます。async コンテキストではイベントループがブロックされ、アプリケーションが deadlock する可能性があります。代わりに await を使用してください。

自動開始の動作

session.sandbox() は開始されていないサンドボックスを返します。サンドボックスは、必要となる最初の操作で自動的に開始されます: 自動開始をトリガーするもの: exec(), read_file(), write_file(), wait(), wait_until_complete() 自動開始しないもの: get_status() (送出される例外 SandboxNotRunningError), stop() (開始されていなければ no-op)

オペレーション

サンドボックスを作成する

同期
Async

from sandbox_auth import AUTH
from cwsandbox import Sandbox

# コンテキストマネージャー (推奨)
with Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH) as sb:
    result = sb.exec(["echo", "hello"]).result()
# 終了時に自動的に停止

from sandbox_auth import AUTH
from cwsandbox import Sandbox

# 構築した後、async コンテキストマネージャーで起動とクリーンアップを行う
sb = Sandbox(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
async with sb as sb:
    result = await sb.exec(["echo", "hello"])
# 終了時に自動的に停止

Sandbox.run() は内部で start().result() を呼び出すため、イベントループをブロックします。 Asyncコードでは Sandbox(...) で構築し、async with を使用してサンドボックスを起動してください。 exec() は RUNNING になるまで待機してから実行します。起動の完了を個別に待機するには、 await sandbox を使用し、操作の前にステータスを確認してください。

start()

同期
Async

from sandbox_auth import AUTH
sb = Sandbox(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
sb.start().result()  # バックエンドが受け入れるまでブロックします
print(sb.sandbox_id)

from sandbox_auth import AUTH
sb = Sandbox(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
await sb.start()  # バックエンドが受け入れるまで待機します
print(sb.sandbox_id)

exec()

Sync
Async

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

# 非ゼロの終了コードで例外を発生させる
result = sb.exec(["python", "-c", "exit(1)"], check=True).result()

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

# 非ゼロの終了コードで例外を発生させる
result = await sb.exec(["python", "-c", "exit(1)"], check=True)

read_file()

Sync
Async

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

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

write_file()

Sync
Async

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

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

stop()

Sync
Async

sb.stop().result()

# すでに削除されている場合は無視します
sb.stop(missing_ok=True).result()

await sb.stop()

# すでに削除されている場合は無視します
await sb.stop(missing_ok=True)

wait()

wait() は起動ステータスになるまで待機します。たとえば、サンドボックスが RUNNING または TERMINATING になったとき、あるいは正常に完了した後に制御を返します。このメソッドは同期版のみです。Asyncコードでは、 await sandbox を使用すると、イベントループをブロックせずに同じ起動待機を行えます。正常に制御が返っても、RUNNING であるとは限りません。実行中のサンドボックスを必要とする操作の前に、status を確認してください。終了状態の処理とエラー処理については、サンドボックスを待機するを参照してください。

from sandbox_auth import AUTH
from cwsandbox import Sandbox, SandboxStatus

# 同期: 起動時のステータスになるまで待機してから、ステータスを確認します
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

# Async: 同じ起動待機を await で待ってから、ステータスを確認します
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()

同期
Async

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}")

# 外部で終了したサンドボックスを例外を発生させずに処理する
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}")

# 外部で終了したサンドボックスを例外を発生させずに処理する
await sb.wait_until_complete(raise_on_termination=False)

get_status()

get_status() は同期版のみです。API から現在のステータスを取得します。

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

Sandbox.list()

同期
Async

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()

Sync
Async

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()

同期
Async

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

# すでに削除済みの場合は無視します
Sandbox.delete("sandbox-abc123", missing_ok=True, auth=AUTH).result()

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

# すでに削除済みの場合は無視します
await Sandbox.delete("sandbox-abc123", missing_ok=True, auth=AUTH)

session.list()

Sync
Async

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()

Sync
Async

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"])

Stream stdout

同期
Async

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

stdin をストリーミングする

stdin=True を指定して stdin を有効にします。生のバイトを書き込むには write()、テキスト行を書き込むには writeline() を使用し、 EOF を通知するには close() を使用します。

Sync
Async

process = sb.exec(["cat"], stdin=True)
process.stdin.write(b"hello ").result()
process.stdin.writeline("world").result()
process.stdin.close().result()  # プロセスに EOF をシグナルします
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()  # プロセスに EOF をシグナルします
result = await process
print(result.stdout)  # "hello world\n"

cwsandbox.results()

cwsandbox.results() は同期版のみのバッチヘルパーです。1つ以上の OperationRef オブジェクトに対して .result() を呼び出します。

import cwsandbox

# 単一の ref
data = cwsandbox.results(sandbox.read_file("/path"))

# 複数の ref
all_data = cwsandbox.results([sb.read_file(f) for f in files])

cwsandbox.wait()

cwsandbox.wait() は同期版のみです。Sandbox、OperationRef、または Process オブジェクトのシーケンスを待機し、(done, pending) を返します。サンドボックスに対しては、メインコマンドの終了を待つのではなく、Sandbox.wait() と同じ起動待機を行います。done に含まれるサンドボックスが RUNNING であるとは限りません。たとえば、すでに完了している場合や終了処理中の場合があります。実行中であることが必要な操作を行う前に、各サンドボックスの status を確認してください。

from sandbox_auth import AUTH
import cwsandbox

# すべてのサンドボックスの起動待機が完了するまで待機します
sandboxes = [
    Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
    for _ in range(5)
]
done, pending = cwsandbox.wait(sandboxes)

# 最初の2つの操作が完了するまで待機します
refs = [sb.read_file(f) for f in files]
done, pending = cwsandbox.wait(refs, num_returns=2)

# タイムアウトを指定して待機します
done, pending = cwsandbox.wait(procs, timeout=30.0)

@session.function()

Sync
Async

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

    # 並列マップ
    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

    # 並列マップ
    refs = compute.map([(1, 2), (3, 4), (5, 6)])
    results = [await r for r in refs]  # [3, 7, 11]

並列実行

同期 API は、操作が設計上ノンブロッキングであるため、並列実行をサポートします。exec()、read_file()、write_file() などのメソッドは即座に返ります。.result() を呼び出したときのみブロックします。 Sandbox.run() は、バックエンドが起動リクエストを受け入れるまでブロックします (内部で start().result() を呼び出します) 。並列起動には、session.sandbox() (最初の操作で自動起動) を使用するか、start() の ref を収集します：

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

# Option 1: session.sandbox() - sandboxes auto-start on first 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() + collect start() refs
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()  # All starts proceed in parallel

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()

Jupyter ノートブック

SDK がデーモンスレッド内で独自のバックグラウンドイベントループを実行するため、同期 API は Jupyter で nest_asyncio なしで動作します:

from sandbox_auth import AUTH
# セル 1 - サンドボックスを作成
from cwsandbox import Sandbox, SandboxStatus
sandbox = Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
sandbox.wait()  # 起動完了まで待機

# セル 2 - コマンドを実行する前にステータスを確認
if sandbox.status == SandboxStatus.RUNNING:
    result = sandbox.exec(["python", "-c", "print(1+1)"]).result()
    print(result.stdout)

# セル 3 - クリーンアップ
sandbox.stop().result()

Jupyter で Async を使用する場合、Jupyter にはイベントループが組み込まれているため、await をそのまま使用できます。

from sandbox_auth import AUTH
# Jupyter では asyncio.run() を使わずに動作します
sandboxes = await Sandbox.list(tags=["notebook"], auth=AUTH)

エラー処理

同期と非同期のどちらの処理経路でも、同じ例外が送出されます。.result() は基になる操作で発生した例外を再送出し、await も同様に動作します。例外階層全体と復旧パターンについては、Sandbox SDK の一般的なエラーメッセージの意味を参照してください。

サンドボックス

サンドボックスを使用する

サンドボックスの管理

リファレンス

同期と非同期のパターン

認証を選択する

クイック決定ガイド

Core concept: OperationRef

自動開始の動作

オペレーション

サンドボックスを作成する

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()

Stream stdout

stdin をストリーミングする

cwsandbox.results()

cwsandbox.wait()

@session.function()

並列実行

Jupyter ノートブック

エラー処理

​認証を選択する

​クイック決定ガイド

​Core concept: OperationRef

​自動開始の動作

​オペレーション

​サンドボックスを作成する

​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()

​Stream stdout

​stdin をストリーミングする

​cwsandbox.results()

​cwsandbox.wait()

​@session.function()

​並列実行

​Jupyter ノートブック

​エラー処理

認証を選択する

クイック決定ガイド

Core concept: OperationRef

自動開始の動作

オペレーション

サンドボックスを作成する

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()

Stream stdout

stdin をストリーミングする

cwsandbox.results()

cwsandbox.wait()

@session.function()

並列実行

Jupyter ノートブック

エラー処理