Apothic Client Classes and Sandboxes Reference#

Apothic supports two main stateful shapes:

object-style resources through @app.cls(...)
workspace-style resources through app.sandbox(...)

Class lifecycle decorators#

`@method(...)`#

@method(
    *,
    tags: Iterable[str] | None = None,
    metadata: dict[str, Any] | None = None,
)

`@enter()` and `@exit()`#

enter = @enter()
exit = @exit()

`@app.cls(...)`#

@app.cls(
    *,
    name: str | None = None,
    gpu: str | list[str] | None = None,
    gpu_count: int | None = None,
    min_vram_gb: int | None = None,
    geolocation: str | list[str] | None = None,
    offer_filters: OfferFiltersModel | dict[str, Any] | None = None,
    max_cost_per_hour_usd: float | None = None,
    max_cost_per_tflop_hour_usd: float | None = None,
    min_total_flops: float | None = None,
    min_gpu_ram_bandwidth_gbps: float | None = None,
    min_cpu_cores: float | None = None,
    min_cpu_ram_gb: float | None = None,
    min_cpu_ghz: float | None = None,
    min_disk_bandwidth_mb_s: float | None = None,
    min_internet_upload_mbps: float | None = None,
    min_internet_download_mbps: float | None = None,
    secure_cloud_only: bool | None = None,
    cpu: int = 1,
    memory_mb: int = 1024,
    disk_gb: float | None = None,
    timeout_s: int = 600,
    retries: int | Retries | None = None,
    max_retries: int = 0,
    concurrent_requests: int = 1,
    max_pending_tasks: int = 100,
    max_containers: int = 1,
    tasks_per_container: int = 1,
    image: Image | None = None,
    secrets: list[Secret] | None = None,
    volumes: dict[str, StorageMount] | list[StorageMount] | None = None,
    tags: Iterable[str] | None = None,
    metadata: dict[str, Any] | None = None,
)

`Cls`#

Lookup helpers#

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

Cls.from_handle(
    handle: ClassInstanceHandle | dict[str, Any] | str,
    *,
    client: ControlPlaneClient | None = None,
    base_url: str | None = None,
) -> ClassInstance

Properties#

cls_handle.name -> str
cls_handle.methods -> dict[str, Function[Any] | RemoteFunction]
cls_handle.method_names -> tuple[str, ...]
cls_handle.method_infos -> dict[str, MethodInfo]
cls_handle.close_function -> Function[Any] | RemoteFunction | None
cls_handle.signature -> inspect.Signature | None
cls_handle.parameters -> constructor parameter metadata
cls_handle.parameter_names -> tuple[str, ...]
cls_handle.tags -> tuple[str, ...]
cls_handle.metadata -> dict[str, Any]
cls_handle.enter_hook_names -> tuple[str, ...]
cls_handle.exit_hook_names -> tuple[str, ...]
cls_handle.remote_only -> bool
cls_handle.local_cls -> type[Any] | None
cls_handle.app_name -> str | None

Methods#

cls_handle.inspect() -> ClassInfo
cls_handle.has_method(name: str) -> bool
cls_handle.method_info(name: str) -> MethodInfo
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

`BoundCls`#

bound.cls -> Cls
bound.signature -> inspect.Signature | None
bound.init_args -> tuple[Any, ...]
bound.init_kwargs -> dict[str, Any]

bound.bind(*args, **kwargs) -> ClassInstance
bound.remote(*args, **kwargs) -> ClassInstance
bound.local(*args, **kwargs) -> LocalClassInstance
bound.new(*args, **kwargs) -> ClassInstance
bound.named(instance_id: str, *args, **kwargs) -> ClassInstance
bound.partial(*args, **kwargs) -> BoundCls
bound.close_handle(handle: ClassInstanceHandle | dict[str, Any] | str) -> Any
bound.close_named(instance_id: str, *args, **kwargs) -> Any

`Sandbox`#

Construction and lookup#

app.sandbox(
    *,
    name: str = "default",
    gpu: str | list[str] | None = None,
    gpu_count: int | None = None,
    min_vram_gb: int | None = None,
    geolocation: str | list[str] | None = None,
    offer_filters: OfferFiltersModel | dict[str, Any] | None = None,
    max_cost_per_hour_usd: float | None = None,
    max_cost_per_tflop_hour_usd: float | None = None,
    min_total_flops: float | None = None,
    min_gpu_ram_bandwidth_gbps: float | None = None,
    min_cpu_cores: float | None = None,
    min_cpu_ram_gb: float | None = None,
    min_cpu_ghz: float | None = None,
    min_disk_bandwidth_mb_s: float | None = None,
    min_internet_upload_mbps: float | None = None,
    min_internet_download_mbps: float | None = None,
    secure_cloud_only: bool | None = None,
    cpu: int = 1,
    memory_mb: int = 1024,
    disk_gb: float | None = None,
    timeout_s: int = 3600,
    retries: int | Retries | None = None,
    max_retries: int = 0,
    concurrent_requests: int = 1,
    max_pending_tasks: int = 100,
    max_containers: int = 1,
    tasks_per_container: int = 1000,
    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

Properties and methods#

sandbox.name -> str
sandbox.app_name -> str | None
sandbox.method_names -> tuple[str, ...]
sandbox.tags -> tuple[str, ...]
sandbox.metadata -> dict[str, Any]
sandbox.signature -> inspect.Signature | None

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

`SandboxSession`#

session.sandbox -> Sandbox
session.local -> bool
session.closed -> bool
session.id -> str
session.handle -> SandboxHandle
session.handle_json -> str
session.method_names -> tuple[str, ...]

session.has_method(name: str) -> bool
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