파일 오퍼레이션

import json

from sandbox_auth import AUTH
from cwsandbox import Sandbox

with Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH) as sandbox:
    sandbox.write_file("/tmp/data.txt", b"Hello, World!").result()

    config = {"key": "value", "count": 42}
    sandbox.write_file("/tmp/config.json", json.dumps(config).encode()).result()

    content = sandbox.read_file("/tmp/data.txt").result()
    print(content.decode())  # "Hello, World!"

    config_bytes = sandbox.read_file("/tmp/config.json").result()
    restored_config = json.loads(config_bytes.decode())
    print(restored_config)

from cwsandbox import results

# 모든 업로드를 동시에 시작합니다
write_refs = [
    sandbox.write_file("/app/config.json", config_bytes),
    sandbox.write_file("/app/data.csv", data_bytes),
    sandbox.write_file("/app/model.pkl", model_bytes),
]

# 모든 업로드가 완료될 때까지 기다립니다
results(write_refs)

# 모든 다운로드를 동시에 시작합니다
read_refs = [
    sandbox.read_file("/app/output.json"),
    sandbox.read_file("/app/metrics.json"),
    sandbox.read_file("/app/logs.txt"),
]

# 모든 결과 조회
output, metrics, logs = results(read_refs)

from sandbox_auth import AUTH
from cwsandbox import Sandbox, results

with Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH) as sandbox:
    # 1. 병렬 업로드
    results([
        sandbox.write_file("/app/config.json", config_bytes),
        sandbox.write_file("/app/input.csv", input_bytes),
    ])

    # 2. 순차 처리
    sandbox.exec(["pip", "install", "-r", "requirements.txt"]).result()
    sandbox.exec(["python", "/app/process.py"]).result()

    # 3. 병렬 다운로드
    output, metrics = results([
        sandbox.read_file("/app/output.json"),
        sandbox.read_file("/app/metrics.json"),
    ])

from cwsandbox import SandboxFileError

try:
    content = sandbox.read_file("/nonexistent/file.txt").result()
except SandboxFileError as e:
    print(f"File error: {e.filepath}")

try:
    sandbox.write_file("/readonly/path.txt", b"data").result()
except SandboxFileError as e:
    print(f"Cannot write to: {e.filepath}")

# 이미지
with open("image.png", "rb") as f:
    sandbox.write_file("/app/image.png", f.read()).result()

# Pickle 파일 (신뢰할 수 있는 소스에서만 데이터를 언피클하세요)
import pickle
model_bytes = pickle.dumps(my_model)
sandbox.write_file("/app/model.pkl", model_bytes).result()

# 다운로드 및 언피클
model_bytes = sandbox.read_file("/app/trained_model.pkl").result()
trained_model = pickle.loads(model_bytes)

# 텍스트 쓰기
text = "Hello, Unicode! "
sandbox.write_file("/app/text.txt", text.encode("utf-8")).result()

# 텍스트 조회
content = sandbox.read_file("/app/text.txt").result()
text = content.decode("utf-8")

from cwsandbox import AuthStrategy, ObjectStorageAccess, ObjectStoragePermission, Sandbox

with Sandbox.run(
    auth=AuthStrategy.COREWEAVE_API_KEY,
    container_image="python:3.11",
    object_storage_access=ObjectStorageAccess(
        buckets=["[BUCKET-NAME]"],
        permission=ObjectStoragePermission.READ_WRITE,
    ),
    resources={"cpu": "1", "memory": "1Gi"},
) as sandbox:
    result = sandbox.exec(["printenv", "AWS_ENDPOINT_URL_S3"]).result()
    print(result.stdout)