Apothic Client Storage, Secrets, and Sandboxes#

apothic-client now exposes three distinct resource families that often get used together:

secrets for credentials and tokens
mounted storage for caches, datasets, artifacts, and portable buckets
stateful resources through @app.cls(...) and app.sandbox(...)

The runtime treats them differently, so it helps to model them deliberately instead of treating all storage as one interchangeable mount.

Secrets#

Secrets are account-scoped runtime records. Use them when your deployed code needs tokens, API keys, or bucket credentials.

`Secret` signatures#

Secret.from_name(name: str) -> Secret

Secret.create(
    name: str,
    value: str,
    *,
    client: ControlPlaneClient | None = None,
    base_url: str | None = None,
) -> Secret

Secret.set(
    name: str,
    value: str,
    *,
    client: ControlPlaneClient | None = None,
    base_url: str | None = None,
) -> Secret

Secret.list(*, client: ControlPlaneClient | None = None, base_url: str | None = None) -> list[Secret]
secret.delete(*, client: ControlPlaneClient | None = None, base_url: str | None = None) -> SecretResponse

Create or update them through the SDK:

import os

from apothic import Secret

Secret.create("OPENAI_API_KEY", os.environ["OPENAI_API_KEY"])
Secret.set("HF_TOKEN", os.environ["HF_TOKEN"])
print([secret.name for secret in Secret.list()])

Attach them to a function or service by name:

from apothic import App, Secret

app = App("secret-demo")


@app.function(secrets=[Secret.from_name("HF_TOKEN")])
def whoami() -> str:
    return "secret mounted"

The current public client surface is SDK-first here. There is not yet a dedicated apothic secret ... CLI group.

Portable named volumes#

Use Volume.from_name(...) or Volume.cloud(...) when you want a portable named mount that is not tied to one execution host.

`Volume` signatures#

Volume.from_name(
    name: str,
    *,
    create_if_missing: bool = False,
    backend: str = "cloud_bucket",
    size_gb: int | None = None,
    mount_path: str | None = None,
    mount_path_default: str | None = None,
    geolocation: str | list[str] | None = None,
    client: ControlPlaneClient | None = None,
    base_url: str | None = None,
) -> Volume
Volume.cloud(name: str, *, mount_path: str | None = None) -> Volume
Volume.vast_local(name: str, *, size_gb: int, mount_path: str | None = None) -> Volume

Volume.create(
    name: str,
    *,
    backend: str = "cloud_bucket",
    size_gb: int | None = None,
    mount_path_default: str | None = None,
    geolocation: str | list[str] | None = None,
    client: ControlPlaneClient | None = None,
    base_url: str | None = None,
) -> Volume

Volume.list(*, client: ControlPlaneClient | None = None, base_url: str | None = None) -> list[Volume]
Volume.ephemeral(
    *,
    prefix: str = "apothic-ephemeral",
    mount_path: str | None = None,
    client: ControlPlaneClient | None = None,
    base_url: str | None = None,
) -> _EphemeralVolumeContext

volume.mount(mount_path: str) -> Volume
volume.mounted_at(mount_path: str) -> Volume
volume.delete(*, client: ControlPlaneClient | None = None, base_url: str | None = None) -> VolumeResponse

from pathlib import Path

from apothic import App, Volume

app = App("portable-volume-demo")
shared = Volume.from_name(
    "model-cache",
    create_if_missing=True,
    mount_path_default="/cache",
).mounted_at("/cache")


@app.function(volumes=[shared])
def write_marker() -> str:
    target = Path("/cache") / "marker.txt"
    target.write_text("hello", encoding="utf-8")
    return target.read_text(encoding="utf-8")

This is the right default when you want:

durable portable state
simple reuse across different worker placements
storage that should survive worker teardown without host affinity

Use create_if_missing=True when you want the concise "declare or reuse" path directly from application code.

Fast host-local volumes with `vast_local`#

Use vast_local when the hot path matters more than portability.

from apothic import App, Image, Volume

app = App("fast-cache-demo")

fast_cache = Volume.create(
    "fast-cache",
    backend="vast_local",
    size_gb=50,
    mount_path_default="/cache",
    geolocation=["US", "CA"],
)


@app.function(
    gpu=["RTX_4070", "RTX_4070S", "RTX_4070_TI", "RTX_3080", "RTX_3060", "RTX_3060_TI"],
    gpu_count=1,
    min_vram_gb=10,
    geolocation=["US", "CA"],
    disk_gb=160,
    image=Image.debian_slim().pip_install("numpy"),
    volumes=[fast_cache.mounted_at("/cache")],
)
def warm_cache() -> str:
    return "/cache"

Important semantics:

Volume.create(..., backend="vast_local") creates a reservation first
the runtime binds it on first real workload placement
after binding, it is host-affine by design
if the bound host is unavailable later, the runtime does not silently pretend it is portable

That makes vast_local a good fit for:

model weights caches
compiled kernels
fast working sets that should stay close to one GPU host

Current limitation: one vast_local volume per resource is supported at a time.

Temporary volumes#

For scratch storage that should clean itself up, use Volume.ephemeral(...):

from apothic import Volume

with Volume.ephemeral(prefix="training-scratch", mount_path="/scratch") as scratch:
    print(scratch.name, scratch.mount_path)

This is useful for one-shot workflows, tests, or batch jobs that need a temporary named mount and then want it removed automatically.

Mount an external S3-compatible bucket#

When you already have a bucket, mount it explicitly with CloudBucketMount.

`CloudBucketMount` signatures#

CloudBucketMount(
    bucket_name: str,
    access_key_secret: Secret | str,
    secret_key_secret: Secret | str,
    endpoint_url: str,
    region: str = "auto",
    prefix: str | None = None,
    read_only: bool = False,
    force_path_style: bool = False,
    mount_path: str | None = None,
)

bucket.mount(mount_path: str) -> CloudBucketMount
bucket.mounted_at(mount_path: str) -> CloudBucketMount

from apothic import App, CloudBucketMount

app = App("bucket-demo")

bucket = CloudBucketMount(
    bucket_name="team-datasets",
    access_key_secret="APOTHIC_BUCKET_ACCESS_KEY_SECRET",
    secret_key_secret="APOTHIC_BUCKET_SECRET_KEY_SECRET",
    endpoint_url="https://fly.storage.tigris.dev",
    region="auto",
    prefix="examples/datasets",
).mounted_at("/bucket")


@app.function(volumes=[bucket])
def list_bucket() -> str:
    return "/bucket"

Use this path when:

you need an existing bucket, not a runtime-managed named volume
you want explicit control over bucket credentials and prefixing
you are mounting Tigris or another S3-compatible store directly

Stateful classes with `@app.cls(...)`#

@app.cls(...) gives you a stateful resource shape instead of one stateless function call per request.

`@app.cls(...)` and `Cls` signatures#

@app.cls(...) accepts the same capacity, image, secret, and volume options as @app.function(...), plus:

@app.cls(
    *,
    name: str | None = None,
    ...,
    tags: Iterable[str] | None = None,
    metadata: dict[str, Any] | None = None,
)

Once deployed, the class handle surface is:

Cls.from_name(
    app_name: str,
    class_name: str,
    *,
    client: ControlPlaneClient | None = None,
    base_url: str | None = None,
) -> Cls

cls_handle.bind(*args, **kwargs) -> ClassInstance
cls_handle.remote(*args, **kwargs) -> ClassInstance
cls_handle.partial(*args, **kwargs) -> BoundCls
cls_handle.new(*args, **kwargs) -> ClassInstance
cls_handle.named(instance_id: str, *args, **kwargs) -> ClassInstance
cls_handle.local(*args, **kwargs) -> LocalClassInstance
cls_handle.attach(handle: ClassInstanceHandle | dict[str, Any] | str) -> ClassInstance
cls_handle.close_handle(handle: ClassInstanceHandle | dict[str, Any] | str) -> Any
cls_handle.close_named(instance_id: str, *args, **kwargs) -> Any
Cls.from_handle(handle, *, client: ControlPlaneClient | None = None, base_url: str | None = None) -> ClassInstance

from apothic import App, enter, exit, method

app = App("stateful-demo")


@app.cls(cpu=2, memory_mb=2048, timeout_s=120)
class Model:
    def __init__(self, prefix: str = "hi: ") -> None:
        self.prefix = prefix

    @enter()
    def load(self) -> None:
        return None

    @method()
    def infer(self, prompt: str) -> str:
        return self.prefix + prompt

    @exit()
    def shutdown(self) -> None:
        return None

Once deployed:

from apothic import App

remote_app = App.lookup("stateful-demo")
instance = remote_app.Model.named("session-1", "hi: ")
print(instance.infer.remote("world"))
instance.close()

Useful patterns include:

named instances for sticky per-session state
.new(...) for unique ephemeral instances
.attach(handle) when you need to resume from a serialized handle
.partial(...) when you want a reusable partially bound constructor

Sandboxes#

app.sandbox(...) is built on the same stateful instance machinery, but exposes a filesystem and command-execution surface instead of arbitrary methods.

`Sandbox` signatures#

app.sandbox(...) accepts the same capacity, image, secret, and volume options as @app.function(...):

app.sandbox(
    *,
    name: str = "default",
    ...,
    image: Image | None = None,
    secrets: list[Secret] | None = None,
    volumes: dict[str, StorageMount] | list[StorageMount] | None = None,
) -> Sandbox

Sandbox.from_name(
    app_name: str,
    sandbox_name: str,
    *,
    client: ControlPlaneClient | None = None,
    base_url: str | None = None,
) -> Sandbox

sandbox.create(*, cwd: str = "/", env: dict[str, Any] | None = None) -> SandboxSession
sandbox.named(sandbox_id: str, *, cwd: str = "/", env: dict[str, Any] | None = None) -> SandboxSession
sandbox.local(*, cwd: str = "/", env: dict[str, Any] | None = None) -> SandboxSession
sandbox.attach(handle: SandboxHandle | dict[str, Any] | str) -> SandboxSession
sandbox.inspect() -> SandboxInfo
sandbox.close_handle(handle: SandboxHandle | dict[str, Any] | str) -> Any
sandbox.close_named(sandbox_id: str, *, cwd: str = "/", env: dict[str, Any] | None = None) -> Any

from apothic import App, Image, Volume

app = App("sandbox-demo")

sandbox = app.sandbox(
    name="shell",
    image=Image.debian_slim(python_version="3.11"),
    volumes={"/workspace": Volume.from_name("sandbox-workspace")},
)

Use it like this:

app.deploy()

session = sandbox.named("demo-session", cwd="/workspace", env={"PYTHONUNBUFFERED": "1"})
result = session.exec(["python", "-c", "print('hello from sandbox')"])
print(result.stdout)
session.write_text("/workspace/notes.txt", "hello\n", make_parents=True)
print(session.read_text("/workspace/notes.txt"))
session.close()

The current sandbox API includes:

exec(...)
exec_stream(...)
read_text(...) and write_text(...)
read_bytes(...) and write_bytes(...)
mkdir(...), listdir(...), exists(...), pwd(...), and chdir(...)
environment inspection and mutation through get_env(), update_env(), and unset_env()

`SandboxSession` signatures#

session.inspect() -> SandboxSessionInfo
session.pwd() -> str
session.get_env() -> dict[str, str]
session.chdir(path: str) -> str
session.update_env(values: dict[str, Any] | None = None, /, **kwargs: Any) -> dict[str, str]
session.set_env(**kwargs: Any) -> dict[str, str]
session.unset_env(*names: str) -> dict[str, str]
session.exists(path: str = ".") -> bool
session.listdir(path: str = ".") -> list[str]
session.mkdir(path: str, *, parents: bool = True, exist_ok: bool = True) -> str
session.read_text(path: str, *, encoding: str = "utf-8") -> str
session.write_text(
    path: str,
    content: str,
    *,
    encoding: str = "utf-8",
    append: bool = False,
    make_parents: bool = False,
) -> str
session.read_bytes(path: str) -> bytes
session.write_bytes(
    path: str,
    content: bytes | bytearray | memoryview,
    *,
    append: bool = False,
    make_parents: bool = False,
) -> str
session.upload_file(local_path: str | Path, remote_path: str, *, make_parents: bool = False) -> str
session.download_file(remote_path: str, local_path: str | Path, *, make_parents: bool = False) -> Path
session.exec(
    command: str | list[str] | tuple[str, ...],
    *,
    shell: bool = False,
    cwd: str | None = None,
    env: dict[str, Any] | None = None,
    timeout_s: float | None = None,
) -> SandboxExecResult
session.exec_stream(
    command: str | list[str] | tuple[str, ...],
    *,
    shell: bool = False,
    cwd: str | None = None,
    env: dict[str, Any] | None = None,
    timeout_s: float | None = None,
) -> Iterator[SandboxStreamEvent]
session.close() -> Any

How to choose#

Use this rule of thumb:

Volume.from_name(...) or Volume.cloud(...) for portable runtime-managed storage
Volume.from_name(..., create_if_missing=True) when you want the concise reuse-or-create path
Volume.create(..., backend="vast_local") for the fastest persistent path on one host
CloudBucketMount(...) when you already have a bucket and credentials
Secret.from_name(...) for sensitive configuration
@app.cls(...) when you need persistent in-memory object state
app.sandbox(...) when you need filesystem and shell-style operations