コマンドの実行

これらの例では、CWSANDBOX_API_KEY に設定した CoreWeave API アクセストークンを使用します。Forge の APIキーによる認証については、はじめにを参照してください。CKS の要件については、配置モードを参照してください。このガイドでは、exec() メソッドを使用してサンドボックスでコマンドを実行する方法を説明します。シェルコマンド、スクリプト、または対話型インタープリターをサンドボックス内で実行し、その出力を取得したり、リアルタイムでストリーミングしたり、stdin に送信したり、非ゼロの終了コードの処理方法を制御したりする場合に使用します。このガイドは、すでに cwsandbox Python クライアントを使用している開発者向けです。

基本的なコマンドを実行する

まず、完了まで実行され、取得した出力を返す単一のコマンドから始めます。exec() メソッドは Process ハンドルを返します：

from cwsandbox import Sandbox

with Sandbox.run() as sandbox:
    # コマンドを実行して結果を取得します
    result = sandbox.exec(["echo", "Hello, World!"]).result()

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

結果を取得

exec() はすぐに返るので、コマンドの待機方法を決定できます。出力が得られるまで待機するには、Process ハンドルの .result() を呼び出します：

# Process をすぐに返します
process = sandbox.exec(["python", "-c", "print('hello')"])

# 結果を待機します
result = process.result()
print(result.stdout)      # "hello\n"
print(result.stderr)      # ""
print(result.returncode)  # 0

# ワンライナーのパターン
result = sandbox.exec(["ls", "-la"]).result()

出力のストリーミング

.result() でブロックすると、コマンドが終了するまで出力は表示されません。進捗をリアルタイムで確認するには、.result() を呼び出す前に process.stdout を反復処理します：

# Process を即座に返します
process = sandbox.exec(["python", "long_script.py"])

# 標準出力を1行ずつストリーミングします
for line in process.stdout:
    print(f"[stdout] {line}", end="")

# 最終結果を取得します
result = process.result()
print(f"Exit code: {result.returncode}")

次のことが必要な場合は、ストリーミングを使用します。

長時間実行されるプロセスを監視する。
出力を受信するたびに処理する。
進行状況を示すインジケーターを実装する。

stdin への入力のストリーミング

一部のコマンドは、パイプラインや対話型インタープリター、EOF まで読み込むツールなど、実行中に stdin への入力書き込みを必要とします。stdin=True を指定して stdin を有効にすることで、実行中のコマンドに入力を送信できます：

with Sandbox.run() 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"

StreamWriter のメソッド

stdin=True の場合、process.stdin は次の 3 つのメソッドを持つ StreamWriter です。

write(data: bytes): 生のバイト列を書き込みます。OperationRef[None] を返します。
writeline(text: str): 末尾に改行を付けてテキストを書き込みます (UTF-8 にエンコードされます)。OperationRef[None] を返します。
close(): EOF を通知します。システムは保留中の書き込みを先に完了します。OperationRef[None] を返します。

stdin=False (デフォルト) の場合、process.stdin は None です。

複数回の書き込みを送信する

閉じる前にデータを段階的に送信します：

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"

stdin 経由で対話型 Python を実行する

対話型インタープリターに Python コードを入力します：

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"

stdin と stdout のストリーミングを組み合わせる

入力を送信しながら出力をストリーミングします：

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

# 入力を送信
process.stdin.writeline("hello").result()
process.stdin.writeline("world").result()
process.stdin.close().result()

# 到着した出力をストリーミング
for line in process.stdout:
    print(f"[out] {line}", end="")

result = process.result()

EOF に依存するコマンドの処理

一部のコマンド (sort など) は、すべての入力を読み取ってから出力を生成します。stdin を閉じて 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 が出力を生成するには EOF が必要です

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

async コンテキストで stdin を使用する

async コンテキストでは、各 OperationRef を直接 await します:

async with Sandbox.run() 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"

stdin=True と stdin=False の使い分け

シナリオ	stdin	理由
引数付きでコマンドを実行する	`False`	入力は stdin ではなく引数から来る
パイプでデータをコマンドに渡す	`True`	コマンドは stdin から読み取る
対話型インタープリター	`True`	インタープリターは stdin からコマンドを読み取る
EOF まで読み取るプロセス	`True`	EOF を通知するために `close()` が必要
ファイア・アンド・フォーゲットのコマンド	`False`	入力は不要

作業ディレクトリを設定する

デフォルトでは、コマンドはサンドボックスのデフォルトの作業ディレクトリから実行されます。これを cwd で上書きします：

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

パスは絶対パスでなければなりません。

タイムアウトを設定する

コマンドがフリーズしたり予想以上に長く実行される可能性がある場合に停止するには、timeout_seconds でタイムアウトを設定します：

from cwsandbox import SandboxTimeoutError

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

check でエラーを処理する

check パラメーターは、終了コードが 0 以外の場合のエラー処理の動作を制御します。

check=False のデフォルトの動作

終了コードに関係なく結果を返します：

result = sandbox.exec(["false"]).result()
print(result.returncode)  # 1 (例外なし)

失敗時に check=True で例外を送出

非ゼロ終了時に SandboxExecutionError 送出される例外：

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

Python コードを実行する

サンドボックスは、短いワンライナーや、インラインで渡す長いスクリプトとして Python コードを実行するためによく使用されます。

# 1 行のコード
result = sandbox.exec(
    ["python", "-c", "import sys; print(sys.version)"],
).result()

# 文字列からスクリプトを実行
code = '''
import json
data = {"result": 42}
print(json.dumps(data))
'''
result = sandbox.exec(["python", "-c", code]).result()
output = json.loads(result.stdout)

逐次実行と並列実行の比較

後続のコマンドが先行するコマンドに依存する場合は逐次実行を選びます。コマンドが互いに独立しており、複数のサンドボックスで同時に実行したい場合は並列実行を選びます。

順序が重要な場合のコマンドの実行

# 依存関係のため逐次実行が必要です
sandbox.exec(["pip", "install", "requests"]).result()
sandbox.exec(["python", "script_using_requests.py"]).result()

独立したコマンドを並列に実行する

# 複数のサンドボックスを起動します
sandboxes = [Sandbox.run() for _ in range(3)]

# 各サンドボックスでコマンドを実行します
processes = [
    sb.exec(["python", "-c", f"print({i})"])
    for i, sb in enumerate(sandboxes)
]

# すべての結果を取得します
results = [p.result() for p in processes]
for r in results:
    print(r.stdout)

M 個のプロセスのうち N 個の完了を待機する

cwsandbox.wait() を使用して、一部のプロセスの完了を待機します：

import cwsandbox

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

# 最初の2つが完了するのを待つ
done, pending = cwsandbox.wait(processes, num_returns=2)

# 完了したものをすぐに処理する
for p in done:
    print(p.result().stdout)

# 残りを待つ
for p in pending:
    print(p.result().stdout)

プロセスの制御

exec() の呼び出しが戻った後もコマンドが実行され続ける場合、処理をブロックせずに状態を確認したり、必要に応じて完了を待ったりできます。Process ハンドルには、モニタリングと制御のためのメソッドが用意されています。

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

# 実行中かどうかを確認します（ノンブロッキング）
if process.poll() is None:
    print("Still running")

# 完了を待ちます
exit_code = process.wait()
print(f"Exited with code: {exit_code}")

サンドボックス

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

サンドボックスの管理

リファレンス

基本的なコマンドを実行する

結果を取得

出力のストリーミング

stdin への入力のストリーミング

StreamWriter のメソッド

複数回の書き込みを送信する

stdin 経由で対話型 Python を実行する

stdin と stdout のストリーミングを組み合わせる

EOF に依存するコマンドの処理

async コンテキストで stdin を使用する

stdin=True と stdin=False の使い分け

作業ディレクトリを設定する

タイムアウトを設定する

check でエラーを処理する

check=False のデフォルトの動作

失敗時に check=True で例外を送出

Python コードを実行する

逐次実行と並列実行の比較

順序が重要な場合のコマンドの実行

独立したコマンドを並列に実行する

M 個のプロセスのうち N 個の完了を待機する

プロセスの制御

​基本的なコマンドを実行する

​結果を取得

​出力のストリーミング

​stdin への入力のストリーミング

​StreamWriter のメソッド

​複数回の書き込みを送信する

​stdin 経由で対話型 Python を実行する

​stdin と stdout のストリーミングを組み合わせる

​EOF に依存するコマンドの処理

​async コンテキストで stdin を使用する

​stdin=True と stdin=False の使い分け

​作業ディレクトリを設定する

​タイムアウトを設定する

​check でエラーを処理する

​check=False のデフォルトの動作

​失敗時に check=True で例外を送出

​Python コードを実行する

​逐次実行と並列実行の比較

​順序が重要な場合のコマンドの実行

​独立したコマンドを並列に実行する

​M 個のプロセスのうち N 個の完了を待機する

​プロセスの制御

基本的なコマンドを実行する

結果を取得

出力のストリーミング

stdin への入力のストリーミング

StreamWriter のメソッド

複数回の書き込みを送信する

stdin 経由で対話型 Python を実行する

stdin と stdout のストリーミングを組み合わせる

EOF に依存するコマンドの処理

async コンテキストで stdin を使用する

stdin=True と stdin=False の使い分け

作業ディレクトリを設定する

タイムアウトを設定する

check でエラーを処理する

check=False のデフォルトの動作

失敗時に check=True で例外を送出

Python コードを実行する

逐次実行と並列実行の比較

順序が重要な場合のコマンドの実行

独立したコマンドを並列に実行する

M 個のプロセスのうち N 個の完了を待機する

プロセスの制御