ファイル操作

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)