Command execution

이 가이드에서는 exec() 방법을 사용하여 샌드박스에서 명령어를 실행하는 방법을 설명합니다. 셸 명령어, 스크립트 또는 대화형 인터프리터를 샌드박스 내부에서 실행하고 출력을 캡처하거나 실시간으로 스트림하거나 stdin에 입력을 보내거나 0이 아닌 종료 코드를 처리하는 방법을 제어해야 할 때 사용하세요. 이 가이드는 cwsandbox Python 클라이언트를 이미 사용하는 개발자를 위한 것입니다.

인증 선택

별도로 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로 세션을 생성하면 해당 세션이 소유한 샌드박스도 함께 인증됩니다. 샌드박스에 다시 연결하거나 목록을 조회하거나 정리할 때도 같은 사용자 인증 정보를 사용하세요.

기본 명령어 실행

완료될 때까지 실행되고 캡처한 출력을 반환하는 단일 명령어로 시작하세요. exec() 방법은 Process 핸들을 반환합니다:

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

# stdout을 한 줄씩 스트림으로 처리
for line in process.stdout:
    print(f"[stdout] {line}", end="")

# 최종 결과 조회
result = process.result()
print(f"Exit code: {result.returncode}")

다음과 같은 경우 스트리밍을 사용하세요:

장시간 실행되는 프로세스를 모니터링하세요.
출력이 도착하는 대로 처리하세요.
진행 표시기를 구현하세요.

stdin으로 입력 스트림

일부 명령어는 실행 중에 stdin에 입력을 작성해야 합니다. 예를 들어 파이프라인, 대화형 인터프리터 또는 EOF까지 읽는 도구가 있습니다. stdin=True로 stdin을 활성화하여 실행 중인 명령어에 입력을 보냅니다:

from sandbox_auth import AUTH
with Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH) 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은 세 가지 메서드를 가진 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)는 출력을 생성하기 전에 모든 입력을 조회합니다. EOF를 알리려면 stdin을 닫으세요:

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"

비동기 컨텍스트에서 stdin 사용하기

비동기 컨텍스트에서는 Sandbox(...)를 사용하세요. 그러면 비동기 컨텍스트 관리자가 이벤트 루프를 차단하지 않고 샌드박스를 시작합니다. 각 OperationRef는 await하세요.

from sandbox_auth import AUTH
async with Sandbox(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH) 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=False 대신 stdin=True를 사용해야 하는 경우

시나리오	stdin	이유
인수가 있는 명령어 실행	`False`	입력이 stdin이 아닌 인수에서 제공됨
명령어에 데이터 파이프로 전달	`True`	명령어가 stdin에서 읽음
대화형 인터프리터	`True`	인터프리터가 stdin에서 명령어를 읽음
EOF까지 읽는 프로세스	`True`	EOF를 알리려면 `close()`가 필요함
완료를 기다리지 않는 명령어	`False`	입력이 필요하지 않음

작업 디렉터리 설정

기본적으로 명령어는 샌드박스의 기본 작업 디렉터리에서 실행됩니다. cwd로 작업 디렉터리를 재정의하세요:

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

경로는 절대 경로여야 합니다.

timeout 설정

실행 시간이 긴 연결된 명령어에는 exec()에 timeout_seconds 값을 명시적으로 전달하세요. 예를 들어 30분짜리 부트스트랩에는 1800을 지정합니다. 프로세스 출력은 스트림 출력에 설명된 방법으로 처리하세요. exec 요청의 기본 timeout은 300초입니다. 샌드박스의 수명도 명령어 실행 시간보다 길어야 합니다. exec timeout을 늘려도 클라이언트 연결이 끊어지면 명령어가 계속 실행되지는 않습니다. 예상보다 오래 실행되거나 멈출 수 있는 명령어를 중지하려면 timeout_seconds로 timeout을 설정하세요:

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로 실패 시 예외 발생

non-zero exit 시 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 코드를 실행하는 데 주로 사용됩니다:

# 한 줄 코드
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()

독립적인 명령어를 병렬로 실행하세요

from sandbox_auth import AUTH
# 여러 샌드박스 시작
sandboxes = [
    Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
    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"])

# 실행 중인지 확인 (non-blocking)
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에 의존하는 명령어 처리

비동기 컨텍스트에서 stdin 사용하기

stdin=False 대신 stdin=True를 사용해야 하는 경우

작업 디렉터리 설정

timeout 설정

check로 오류 처리

check=False일 때의 기본 동작

check=True로 실패 시 예외 발생

Python 코드 실행

순차 실행과 병렬 실행 비교

순서가 중요할 때 명령어를 순차적으로 실행하세요

독립적인 명령어를 병렬로 실행하세요

M개 프로세스 중 N개가 완료될 때까지 대기

프로세스 제어

​인증 선택

​기본 명령어 실행

​결과 조회

​스트림 출력

​stdin으로 입력 스트림

​StreamWriter 메서드

​여러 쓰기 전송

​stdin을 통해 대화형 Python 실행하기

​stdin과 stdout 스트리밍 결합

​EOF에 의존하는 명령어 처리

​비동기 컨텍스트에서 stdin 사용하기

​stdin=False 대신 stdin=True를 사용해야 하는 경우

​작업 디렉터리 설정

​timeout 설정

​check로 오류 처리

​check=False일 때의 기본 동작

​check=True로 실패 시 예외 발생

​Python 코드 실행

​순차 실행과 병렬 실행 비교

​순서가 중요할 때 명령어를 순차적으로 실행하세요

​독립적인 명령어를 병렬로 실행하세요

​M개 프로세스 중 N개가 완료될 때까지 대기

​프로세스 제어

인증 선택

기본 명령어 실행

결과 조회

스트림 출력

stdin으로 입력 스트림

StreamWriter 메서드

여러 쓰기 전송

stdin을 통해 대화형 Python 실행하기

stdin과 stdout 스트리밍 결합

EOF에 의존하는 명령어 처리

비동기 컨텍스트에서 stdin 사용하기

stdin=False 대신 stdin=True를 사용해야 하는 경우

작업 디렉터리 설정

timeout 설정

check로 오류 처리

check=False일 때의 기본 동작

check=True로 실패 시 예외 발생

Python 코드 실행

순차 실행과 병렬 실행 비교

순서가 중요할 때 명령어를 순차적으로 실행하세요

독립적인 명령어를 병렬로 실행하세요

M개 프로세스 중 N개가 완료될 때까지 대기

프로세스 제어