サンドボックスのライフサイクル

from sandbox_auth import AUTH
from cwsandbox import Sandbox

# バックエンドが起動リクエストを受け入れた後に戻ります
sandbox = Sandbox.run(
    "echo", "hello",
    resources={"cpu": "1", "memory": "1Gi"},
    auth=AUTH,
)

# sandbox_id は即座に設定されます
print(sandbox.sandbox_id)  # "sandbox-abc123"

# ただし、ステータスはまだ PENDING の場合があり、必ずしも RUNNING になっているとは限りません

from sandbox_auth import AUTH
from cwsandbox import Sandbox

Sandbox.run(
    "echo",
    "hello",
    "world",
    resources={"cpu": "1", "memory": "1Gi"},
    auth=AUTH,
)

from sandbox_auth import AUTH
import cwsandbox
from cwsandbox import SandboxDefaults

with cwsandbox.Session(SandboxDefaults(
    resources={"cpu": "1", "memory": "1Gi"},
), auth=AUTH) as session:
    sandbox = session.sandbox()

    # まだネットワーク呼び出しはありません。sandbox_id は None です
    print(sandbox.sandbox_id)  # None

    # 最初の操作で自動的に開始がトリガーされます
    result = sandbox.exec(["echo", "hello"]).result()

from sandbox_auth import AUTH
# echo がすぐに終了するため、このサンドボックスはほぼ即座に完了します
sandbox = Sandbox.run("echo", "hi", resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
sandbox.wait()  # すでに COMPLETED になっている場合があります

# exec ベースのワークフローでは、長時間実行されるメインコマンドを使用します
sandbox = Sandbox.run(
    "sleep", "infinity",
    resources={"cpu": "1", "memory": "1Gi"},
    auth=AUTH,
)
# または、コマンドを省略して、デフォルトの長時間実行コマンドを使用します
sandbox = Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)

sandbox = session.sandbox()

# エラーを制御するために明示的に起動します
sandbox.start().result()  # バックエンドが受け入れるまで待機します
print(sandbox.sandbox_id)  # この時点で設定済みです

from sandbox_auth import AUTH
from cwsandbox import Sandbox, SandboxStatus

with Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH) as sandbox:
    # start() は呼び出されていますが、サンドボックスはまだ PENDING/CREATING の可能性があります
    sandbox.wait()
    if sandbox.status == SandboxStatus.RUNNING:
        result = sandbox.exec(["echo", "ready"]).result()

from sandbox_auth import AUTH
from cwsandbox import Sandbox, SandboxStatus

sandbox = Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
sandbox.wait()  # 起動ステータスになるまで待機してから、ステータスを確認します

# 操作を試みる前にステータスを確認します
if sandbox.status == SandboxStatus.RUNNING:
    result = sandbox.exec(["echo", "hello"]).result()

from sandbox_auth import AUTH
sandbox = Sandbox.run(
    "python", "train.py",
    resources={"cpu": "1", "memory": "1Gi"},
    auth=AUTH,
)
sandbox.wait_until_complete(timeout=3600.0).result()
print(sandbox.returncode)  # トレーニングが成功した場合は 0

from cwsandbox import SandboxTerminatedError

# デフォルト: このクライアントが stop() を呼び出した後は例外を送出します
try:
    sandbox.wait_until_complete().result()
except SandboxTerminatedError:
    print("This client stopped the sandbox")

# 例外を抑制して適切に処理します
sandbox.wait_until_complete(raise_on_termination=False).result()
print(f"Exit code: {sandbox.returncode}")

|<--------------------- max_lifetime_seconds --------------------->|
start() --> [startup wait] --> RUNNING --> exec() --> [operation timeout]

# Session のサンドボックスはまだ起動していません
sandbox = session.sandbox()

# exec() がすべて処理します: 起動し、RUNNING になるまで待機してからコマンドを実行します
result = sandbox.exec(["echo", "hello"]).result()

sandbox.stop().result()  # 終端状態になるまで待機します

# 設定済みのスナップショットマウントを取得します
sandbox.stop(snapshot_on_stop=True).result()

# 冪等なクリーンアップ: サンドボックスがすでに存在しない場合でも安全に呼び出せます
sandbox.stop(missing_ok=True).result()

# プロセスが終了するまでの猶予時間を長くします
sandbox.stop(graceful_shutdown_seconds=30.0).result()

sandbox.stop().result()

# 以下は SandboxNotRunningError を送出します
sandbox.exec(["echo", "hello"])  # 送出される例外
sandbox.read_file("/path")       # 送出される例外

fresh_status = sandbox.get_status()
if fresh_status == SandboxStatus.RUNNING:
    sandbox.stop().result()

from sandbox_auth import AUTH
with Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH) as sandbox:
    result = sandbox.exec(["echo", "hello"]).result()
# stop() はここで呼び出されます。ブロック内で例外が発生した場合でも同様です。

from sandbox_auth import AUTH
# 作成したサンドボックスを停止します
sandbox.stop().result()

# 孤立したサンドボックスを ID で削除します
Sandbox.delete("sandbox-abc123", missing_ok=True, auth=AUTH).result()

from sandbox_auth import AUTH
from cwsandbox import Sandbox

with Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH) as sandbox:
    result = sandbox.exec(["echo", "hello"]).result()
    print(result.stdout)

from sandbox_auth import AUTH
from cwsandbox import Sandbox, SandboxFailedError, SandboxStatus

sandbox = Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
try:
    sandbox.wait()
except SandboxFailedError:
    print("Failed to start. Check container image and resources")
    raise

if sandbox.status == SandboxStatus.RUNNING:
    result = sandbox.exec(["echo", "ready"]).result()
sandbox.stop().result()

from sandbox_auth import AUTH
from cwsandbox import Sandbox

sandbox = Sandbox.run(
    "python",
    "train.py",
    max_lifetime_seconds=7200,
    resources={"cpu": "1", "memory": "1Gi"},
    auth=AUTH,
)
sandbox.wait_until_complete(timeout=7200.0).result()

print(f"Training exited with code {sandbox.returncode}")

from sandbox_auth import AUTH
from cwsandbox import Sandbox, SandboxStatus

sandbox = Sandbox.from_id("sandbox-abc123", auth=AUTH).result()

# from_id() は現在のステータスを取得しますが、サンドボックスの起動や検証は行いません
if sandbox.status == SandboxStatus.RUNNING:
    result = sandbox.exec(["echo", "reconnected"]).result()
else:
    print(f"Sandbox is {sandbox.status}, not RUNNING")

from sandbox_auth import AUTH
import cwsandbox
from cwsandbox import SandboxDefaults

with cwsandbox.Session(SandboxDefaults(
    tags=("batch-job",),
    resources={"cpu": "1", "memory": "1Gi"},
), auth=AUTH) as session:
    sandboxes = [session.sandbox() for _ in range(5)]

    processes = [
        sb.exec(["python", "-c", f"print({i} ** 2)"])
        for i, sb in enumerate(sandboxes)
    ]

    done, pending = cwsandbox.wait(processes)
    for p in done:
        print(p.result().stdout)