동기화 대 비동기 패턴

이 예시에서는 CWSANDBOX_API_KEY에 설정된 CoreWeave API 액세스 토큰을 사용합니다. Forge API 키 인증은 시작하기를 참조하세요. CKS 요구 사항은 배치 모드를 참조하세요. 이 가이드에서는 CoreWeave 샌드박스 Python SDK(cwsandbox)에서 동기화 및 비동기 패턴 중에서 선택하는 방법을 도와주며, 각 일반적인 오퍼레이션의 동기화 및 비동기 형식을 보여줍니다. SDK는 내부적으로 단일 비동기 구현을 가지고 있습니다. 동기화/비동기 유연성은 결과를 소비하는 방법에서 비롯됩니다: 동기화에는 .result()를 사용하고 비동기에는 await를 사용하세요.

빠른 결정 가이드

사용 사례	패턴	이유
대부분의 오퍼레이션	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 cwsandbox import Sandbox

# 컨텍스트 매니저 (권장)
with Sandbox.run() as sb:
    result = sb.exec(["echo", "hello"]).result()
# 종료 시 자동 중단

from cwsandbox import Sandbox

# RUNNING 상태로 전환하기 위해 생성 + await, 그 후 정리용 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()  # backend가 수락할 때까지 블록
print(sb.sandbox_id)

sb = Sandbox()
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 상태에 도달할 때까지 실행을 차단합니다. 실행 차단이 목적이므로 동기 방식으로만 사용할 수 있습니다. 비동기 코드에서는 await sandbox가 같은 역할을 합니다.

# 동기: RUNNING 상태가 될 때까지 차단
sb = Sandbox.run().wait()
result = sb.exec(["echo", "ready"]).result()

# 비동기: 샌드박스를 직접 await
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}")  # e.g. SandboxStatus.RUNNING

Sandbox.list()

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

동기
비동기

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()는 동기 방식으로만 사용할 수 있습니다. 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)

# timeout을 설정하여 대기
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]

병렬 실행

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

from cwsandbox import Sandbox, Session, SandboxDefaults

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

# 옵션 2: Sandbox() + start() Ref 수집
sandboxes = [Sandbox() 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가 데몬 스레드에서 자체 백그라운드 이벤트 루프를 실행하므로 sync API는 nest_asyncio 없이도 Jupyter에서 작동합니다.

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

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

오류 처리

동기화 경로와 비동기 경로는 모두 동일한 예외를 발생시킵니다. .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 노트북

오류 처리