同期と非同期のパターン

これらの例では、CWSANDBOX_API_KEY に設定した CoreWeave API アクセストークンを使用します。Forge APIキーによる認証については、はじめにを参照してください。CKS の要件については、配置モードを参照してください。このガイドでは、CoreWeave サンドボックス Python SDK (cwsandbox) における同期パターンと非同期パターンの選択方法を説明し、一般的な各操作の同期形式と非同期形式を示します。 SDK の内部実装は Async のみです。sync/Async の柔軟性は結果の取得方法によるものです。sync の場合は .result() を、Async の場合は await を使用します。

クイック決定ガイド

使用例	パターン	理由
ほとんどの操作	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)

オペレーション

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

Sync
Async

from cwsandbox import Sandbox

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

from cwsandbox import Sandbox

# 構築して await で RUNNING に到達し、その後 async コンテキストマネージャーでクリーンアップ
sb = Sandbox()
async with sb as sb:
    result = await sb.exec(["echo", "hello"])
# 終了時に自動的に停止

Sandbox.run() は内部で start().result() を呼び出すため、イベントループをブロックします。非同期コードでは Sandbox(...) で構築し、async with または await sandbox を使用して RUNNING ステータスに到達してください。ブロックは発生しません。

start()

Sync
Async

sb = Sandbox()
sb.start().result()  # バックエンドが受け入れるまでブロックします
print(sb.sandbox_id)

sb = Sandbox()
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 ステータスに到達するまでブロックします。これは Sync 版のみです。ブロッキングが意図されているためです。async コードでは、await sandbox で同じことを実現できます。

# 同期: RUNNING になるまでブロック
sb = Sandbox.run().wait()
result = sb.exec(["echo", "ready"]).result()

# Async: サンドボックスを直接待機します
sb = Sandbox()
await sb  # RUNNING になるまで待機します
result = await sb.exec(["echo", "ready"])

wait_until_complete()

Sync
Async

sb = Sandbox.run("python", "-c", "print('done')")
sb.wait_until_complete().result()
print(f"Exit code: {sb.returncode}")

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

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

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

Sync
Async

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

Sync
Async

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

# すでに削除されている場合は無視します
Sandbox.delete("sandbox-abc123", missing_ok=True).result()

await Sandbox.delete("sandbox-abc123")

# すでに削除されている場合は無視します
await Sandbox.delete("sandbox-abc123", missing_ok=True)

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) を返します。サンドボックスは RUNNING ステータスに到達したときに解決され、完了時には解決されません。

import cwsandbox

# すべてのサンドボックスがRUNNINGに到達するまで待機します
sandboxes = [Sandbox.run() 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

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

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

    # 並列マップ
    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 cwsandbox import Sandbox, Session, SandboxDefaults

# Option 1: session.sandbox() - sandboxes auto-start on first 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() + collect start() refs
sandboxes = [Sandbox() 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 なしで動作します:

# セル 1 - サンドボックスを作成
from cwsandbox import Sandbox
sandbox = Sandbox.run()
sandbox.wait()  # RUNNING になるまで待機

# セル 2 - コマンドを実行
result = sandbox.exec(["python", "-c", "print(1+1)"]).result()
print(result.stdout)

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

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

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

エラー処理

同期と非同期のどちらの処理経路でも、同じ例外が送出されます。.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 ノートブック

エラー処理