Skip to main content
Direct volume paths are absolute volume-internal paths rooted at /. Relative paths are rejected, .. cannot escape the volume root, and there is no sandbox cwd or user context for direct volume file operations.

Read and write

from nullspace import Volume

volume = Volume.from_name("team-data", create_if_missing=True)
volume.files.make_dir("/datasets")
volume.files.write("/datasets/report.txt", "ready\n")
print(volume.files.read("/datasets/report.txt"))
print(volume.files.read("/datasets/report.txt", format="bytes"))

List and inspect

for item in volume.files.list("/datasets"):
    print(item.path, item.size)

print(volume.files.exists("/datasets/report.txt"))
print(volume.files.info("/datasets/report.txt").size)

Search and replace

matches = volume.files.find_files("/datasets", "ready")
paths = volume.files.search_files("/datasets", "*.txt")
result = volume.files.replace_in_files(paths, "ready", "done", dry_run=True)
print(result.dry_run, result.replacements, result.files_modified)
VolumeReplaceResult also includes the matched file list, so dry runs can show the planned changes before writing.

Move and remove

volume.files.rename("/datasets/report.txt", "/datasets/report-old.txt")
volume.files.set_permissions("/datasets/report-old.txt", "0644")
volume.files.remove("/datasets/report-old.txt")

Watch

def on_event(event):
    print(event.type, event.path)

with volume.files.watch_dir("/", on_event=on_event, recursive=True):
    volume.files.write("/datasets/changed.txt", "changed\n")
Direct volume watches are best-effort notifications from one selected data-plane host. In multi-host deployments, mutations performed through another host may require a resync; clients should treat resync_required as a prompt to list or stat the watched tree again. volume.files.replace_in_files() returns a VolumeReplaceResult. Sandbox filesystem replacement returns a plain integer count. Direct watches are opened against a concrete volume ID under the hood. If you look up a volume by name, the SDK resolves it first and then watches that ID.

Batch writes

from nullspace import BatchWriteError

try:
    volume.files.write_files([
        ("/datasets/a.txt", "a\n"),
        ("/datasets/b.txt", "b\n"),
    ])
except BatchWriteError as exc:
    print(exc.successes, exc.failures)

Async

The async client mirrors the sync surface above. Volume content search and replace (shown earlier) are SDK-only — the CLI does not expose them.
from nullspace import AsyncVolume

volume = await AsyncVolume.from_name("team-data", create_if_missing=True)
await volume.files.write("/hello.txt", "persistent\n")
print(await volume.files.read("/hello.txt"))