동기화 대 비동기 패턴

이 가이드에서는 CoreWeave 샌드박스 Python SDK(cwsandbox)에서 동기화 및 비동기 패턴 중에서 선택하는 방법을 도와주며, 각 일반적인 오퍼레이션의 동기화 및 비동기 형식을 보여줍니다. SDK는 내부적으로 단일 비동기 구현을 가지고 있습니다. 동기화/비동기 유연성은 결과를 소비하는 방법에서 비롯됩니다: 동기화에는 .result()를 사용하고 비동기에는 await를 사용하세요.

인증 선택

이 예시는 섹션에서 CKS를 따로 지정하지 않는 한 서버리스 용량에서 실행됩니다. 샌드박스를 소유한 계정의 사용자 인증 정보를 선택하세요. 키 또는 토큰을 획득하는 방법은 사용자 인증 정보 선택 및 조회를 참조하세요. 설치 명령어는 활성화된 Python 가상 환경에서 실행하세요. 사용 중인 사용자 인증 정보 탭의 Python 코드를 sandbox_auth.py로 저장하고, 이 페이지에서 실행할 Python script와 같은 위치에 두세요. 이 파일에는 사용자 인증 정보 자체가 아니라 어떤 인증 방식을 사용할지만 저장됩니다. 예시에서는 이 파일에서 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	오퍼레이션이 기본적으로 non-blocking으로 설계됨
Jupyter 노트북	Sync	`nest_asyncio`가 필요 없음
Async 코드베이스	Async	기존 async 코드와 통합됨

기본 원칙: 단순하게 작성하려면 sync 패턴(.result())을 사용하세요. 이미 async 코드베이스에서 작업 중인 경우에만 await으로 전환하세요.

핵심 개념: OperationRef

대부분의 SDK 메서드는 .result() 호출과 await가 모두 가능한 래퍼인 OperationRef[T]를 반환합니다. 두 방식 모두 동일한 예외를 발생시킵니다.

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

# Sync: 값을 블로킹
data = ref.result()

# Async: 대신 await
data = await ref

Sandbox.exec()는 OperationRef[ProcessResult]를 확장한 Process를 스트림과 stdin을 지원하도록 반환합니다.

비동기 컨텍스트에서는 절대로 .result()를 사용하지 마세요. .result()를 호출하면 스레드가 차단됩니다. 비동기 컨텍스트에서는 이벤트 루프가 차단되어 애플리케이션에 교착 상태가 발생할 수 있습니다. 대신 await를 사용하세요.

자동 시작 동작

session.sandbox()는 시작되지 않은 샌드박스를 반환합니다. 샌드박스는 이를 필요로 하는 첫 번째 오퍼레이션에서 자동으로 시작됩니다: 자동 시작을 트리거합니다: exec(), read_file(), write_file(), wait(), wait_until_complete() 자동 시작하지 않습니다: get_status() (SandboxNotRunningError 예외 발생), stop() (시작된 적이 없으면 no-op)

오퍼레이션

샌드박스 생성

Sync
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()를 호출하며, 이는 이벤트 루프를 차단합니다. 비동기 코드에서는 Sandbox(...)로 생성한 후 async with를 사용하여 샌드박스를 시작하세요. exec()는 RUNNING 상태가 될 때까지 기다린 후 실행됩니다. 시작 완료를 따로 기다리려면 await sandbox를 사용하고, 오퍼레이션을 수행하기 전에 상태를 확인하세요.

start()

Sync
Async

from sandbox_auth import AUTH
sb = Sandbox(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
sb.start().result()  # backend가 수락할 때까지 블록
print(sb.sandbox_id)

from sandbox_auth import AUTH
sb = Sandbox(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
await sb.start()  # backend가 수락할 때까지 대기
print(sb.sandbox_id)

exec()

Sync
Async

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

# 0이 아닌 종료 코드에서 예외 발생
result = sb.exec(["python", "-c", "exit(1)"], check=True).result()

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

# 0이 아닌 종료 코드에서 예외 발생
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 상태가 되거나 성공적으로 완료되면 반환될 수 있습니다. 이 메서드는 동기화 전용입니다. 비동기 코드에서는 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

# 비동기: 동일한 시작 대기 작업을 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()

Sync
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}")  # e.g. SandboxStatus.RUNNING

Sandbox.list()

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

동기
비동기

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

동기
비동기

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

스트림 stdout

Sync
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()는 동기 방식으로만 사용할 수 있는 배치 헬퍼입니다. 하나 이상의 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()는 sync-only입니다. 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)

# timeout을 설정하여 대기
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는 오퍼레이션이 설계상 non-blocking이기 때문에 병렬 실행을 지원합니다. exec(), read_file(), write_file() 같은 메서드는 즉시 반환됩니다. .result()를 호출할 때만 블록됩니다. Sandbox.run()은 backend가 start request를 수락할 때까지 블록합니다(내부적으로 start().result()를 호출합니다). 병렬 시작을 위해 session.sandbox()(첫 오퍼레이션 시 자동 시작)를 사용하거나 start() Ref를 수집하세요:

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

# 옵션 1: session.sandbox() - 샌드박스가 첫 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]

# 옵션 2: Sandbox() + start() Ref 수집
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()  # 모든 시작이 병렬로 진행됩니다

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는 nest_asyncio 없이도 Jupyter에서 작동합니다.

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에는 이벤트 루프가 기본으로 내장되어 있으므로, Jupyter에서 비동기를 사용할 때는 await를 바로 쓸 수 있습니다.

from sandbox_auth import AUTH
# Jupyter에서는 asyncio.run() 없이도 동작합니다
sandboxes = await Sandbox.list(tags=["notebook"], auth=AUTH)

오류 처리

동기화 경로와 비동기 경로는 모두 동일한 예외를 발생시킵니다. .result()는 기본 오퍼레이션에서 발생한 모든 예외를 다시 발생시키며, await도 마찬가지입니다. 전체 예외 계층 구조와 복구 패턴은 일반적인 샌드박스 SDK 오류 메시지는 무엇을 의미하나요?를 참조하세요.

샌드박스

샌드박스 사용하기

샌드박스 관리

레퍼런스

동기화 대 비동기 패턴

인증 선택

빠른 결정 가이드

핵심 개념: 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()

스트림 stdout

stdin 스트림

cwsandbox.results()

cwsandbox.wait()

@session.function()

병렬 실행

Jupyter 노트북

오류 처리

​인증 선택

​빠른 결정 가이드

​핵심 개념: 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()

​스트림 stdout

​stdin 스트림

​cwsandbox.results()

​cwsandbox.wait()

​@session.function()

​병렬 실행

​Jupyter 노트북

​오류 처리

인증 선택

빠른 결정 가이드

핵심 개념: 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()

스트림 stdout

stdin 스트림

cwsandbox.results()

cwsandbox.wait()

@session.function()

병렬 실행

Jupyter 노트북

오류 처리