Apothic Client App and Decorators Reference#

App is the main authoring surface in apothic-client. It collects functions, services, classes, sandboxes, and local entrypoints into one deployable unit.

`App`#

Constructor#

App(name: str, *, client: ControlPlaneClient | None = None)

Class methods#

App.lookup(
    name: str,
    *,
    client: ControlPlaneClient | None = None,
    base_url: str | None = None,
) -> RemoteApp

Properties#

app.name -> str
app.client -> ControlPlaneClient
app.aio -> async app API
app.deployment_id -> str | None
app.functions -> dict[str, Function[Any]]
app.classes -> dict[str, Cls]
app.sandboxes -> dict[str, Sandbox]
app.sandbox_names -> tuple[str, ...]
app.namespaces -> dict[str, AppNamespace]
app.local_entrypoints -> dict[str, LocalEntrypoint[Any]]

Deployment and packaging#

app.manifest() -> AppManifest
app.deploy() -> str

Async mirror:

await app.aio.deploy() -> str

Local entrypoints#

@app.local_entrypoint(*, name: str | None = None)

app.get_local_entrypoint(name: str | None = None) -> LocalEntrypoint[Any]
app.run_local(
    *,
    entrypoint_name: str | None = None,
    args: list[Any] | None = None,
    kwargs: dict[str, Any] | None = None,
) -> Any

app.run_local(...) resolves awaitables, so async local entrypoints are supported by the current client and CLI path.

Resource lookup on a local `App`#

app.get_class(name: str) -> Cls
app.get_sandbox(name: str) -> Sandbox

Attribute access also works for declared resources and namespaces:

app.my_function
app.MyClass
app.shell
app.my_namespace

`RemoteApp`#

RemoteApp is the lookup surface returned by App.lookup(...).

Constructor#

RemoteApp(name: str, client: ControlPlaneClient)

Properties#

remote_app.name -> str
remote_app.functions -> dict[str, RemoteFunction]
remote_app.classes -> dict[str, Cls]
remote_app.sandboxes -> dict[str, Sandbox]
remote_app.sandbox_names -> tuple[str, ...]
remote_app.namespaces -> dict[str, AppNamespace]

Methods#

remote_app.get_function(name: str) -> RemoteFunction
remote_app.get_class(name: str) -> Cls
remote_app.get_sandbox(name: str) -> Sandbox

App composition#

Apothic supports composing larger apps from smaller locally defined pieces.

Include helpers#

app.include_function(
    function: Function[Any],
    *,
    name: str | None = None,
    namespace: str | None = None,
) -> Function[Any]

app.include_class(
    cls_handle: Cls,
    *,
    name: str | None = None,
    namespace: str | None = None,
    methods: Iterable[str] | dict[str, str] | None = None,
) -> Cls

app.include_local_entrypoint(
    entrypoint: LocalEntrypoint[Any],
    *,
    name: str | None = None,
    namespace: str | None = None,
) -> LocalEntrypoint[Any]

app.include(other: App, *, namespace: str | None = None) -> App

Shared decorator options#

The core remote decorators all share one large option family:

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

Service-shaped decorators add warm-capacity and startup controls:

keep_warm_s: int = 600
service_startup_timeout_s: float | None = None

Core decorators#

`@app.function(...)`#

@app.function(
    *,
    schedule: Cron | Period | None = None,
    liv: LivConfig | dict[str, Any] | tuple[Any, ...] | None = None,
    ...shared resource options...,
)

`@app.endpoint(...)`#

@app.endpoint(
    *,
    liv: LivConfig | dict[str, Any] | tuple[Any, ...] | None = None,
    ...shared resource options...,
    keep_warm_s: int = 600,
    service_startup_timeout_s: float | None = None,
)

`@app.asgi(...)`#

@app.asgi(
    *,
    ...shared resource options...,
    keep_warm_s: int = 600,
    service_startup_timeout_s: float | None = None,
)

`@app.fastapi_endpoint(...)`#

@app.fastapi_endpoint(
    *,
    ...shared resource options...,
    keep_warm_s: int = 600,
    service_startup_timeout_s: float | None = None,
)

`@app.wsgi_app(...)`#

@app.wsgi_app(
    *,
    ...shared resource options...,
    keep_warm_s: int = 600,
    service_startup_timeout_s: float | None = None,
)

`@app.web_server(...)`#

@app.web_server(
    port: int,
    *,
    startup_timeout: float = 5.0,
    ...shared resource options...,
    keep_warm_s: int = 600,
    service_startup_timeout_s: float | None = None,
)

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

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

Alias:

app.class_ = app.cls

`app.sandbox(...)`#

app.sandbox(
    *,
    name: str = "default",
    ...shared resource options except schedule / liv / keep_warm_s...,
) -> Sandbox