Sandbox lifecycle

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]

# 세션 샌드박스, 아직 시작되지 않음
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)