Volumes

from declaw import Sandbox, Volumes, VolumeAttachment

# From a bytes object
with open("dataset.tar.gz", "rb") as f:
    vol = Volumes.create(name="training-set-v1", data=f)

# From a filesystem path — directory or file is tarred in-memory
vol = Volumes.create(name="training-set-v1", data="/path/to/dir")

# From raw bytes already in memory
vol = Volumes.create(name="training-set-v1", data=raw_bytes)

print(vol.volume_id, vol.size_bytes)

for vol in Volumes.list():
    print(vol.volume_id, vol.name, vol.size_bytes)

vol = Volumes.get("vol-abc123")

Volumes.delete("vol-abc123")

from declaw import Sandbox, VolumeAttachment

vol = Volumes.create(name="dataset", data="/path/to/dir")

sbx = Sandbox.create(
    template="python",
    timeout=600,
    volumes=[VolumeAttachment(volume_id=vol.volume_id, mount_path="/data")],
)

# The files are already visible by the time the first command runs
print(sbx.commands.run("ls -la /data").stdout)

vol = Volumes.empty(name="scratch")              # empty
vol = Volumes.ingest(name="seed", data="/dir")   # or pre-populated from a tar.gz
print(vol.backend)   # "juicefs" / "local" (not "tarball")

files = Volumes.files(vol.volume_id)
files.write("/config/app.json", b'{"k": "v"}')   # parent dirs auto-created
files.mkdir("/data")
print(files.read("/config/app.json"))            # b'{"k": "v"}'
for e in files.list("/"):
    print(e.path, e.is_dir, e.size)
files.rename("/config/app.json", "/config/app.prod.json")
files.remove("/data", recursive=True)

files = Volumes.files(vol.volume_id)
files.write("/greeting.txt", b"hello from the files API")

sbx = Sandbox.create(
    template="base",
    volumes=[VolumeAttachment(volume_id=vol.volume_id, mount_path="/data", mode="mount")],
)
# The sandbox reads the files-API write over a live NFS mount...
print(sbx.commands.run("cat /data/greeting.txt").stdout)
# ...and its writes are visible back through the files API immediately:
sbx.commands.run("echo 'from the sandbox' > /data/out.txt")
print(files.read("/out.txt"))   # b'from the sandbox\n'

sbx = Sandbox.create(
    template="base",
    volumes=[VolumeAttachment(
        volume_id=vol.volume_id,
        mount_path="/data",
        mode="mount",
        subpath="datasets/train",   # mounts <volume>/datasets/train at /data
    )],
)

# Any absolute in-sandbox path -> new volume
vol = Volumes.snapshot(sbx.sandbox_id, path="/workspace/out", name="run-42")

# An already-attached volume's mount path -> new volume
vol = Volumes.commit(sbx.sandbox_id, volume_id=src.volume_id, name="checkpoint")

locks = Volumes.locks(vol.volume_id)

lease = locks.acquire("/data/model.bin", ttl_seconds=60)   # ConflictException (409) if already held
token = lease["token"]

locks.renew("/data/model.bin", token, ttl_seconds=60)
print(locks.status("/data/model.bin"))                     # {"held": True, "expires_in_ms": ...}
locks.release("/data/model.bin", token)                    # True if released

from declaw import AsyncSandbox, AsyncVolumes, VolumeAttachment

vol = await AsyncVolumes.create(name="dataset", data=open("dataset.tar.gz", "rb"))
sbx = await AsyncSandbox.create(
    template="python",
    volumes=[VolumeAttachment(volume_id=vol.volume_id, mount_path="/data")],
)

@dataclass
class Volume:
    volume_id: str        # "vol-..."
    owner_id: str
    name: str             # human-readable, supplied at create
    blob_key: str         # object-store path, for reference
    size_bytes: int
    content_type: str     # "application/gzip"
    created_at: str
    metadata: dict[str, str]

    def attach(self, mount_path: str) -> VolumeAttachment:
        ...