Experiment Spec Types

Python classes for the corresponding Beaker Experiment Spec API.

class BeakerExperimentSpec(version: ~beaker.types.BeakerSpecVersion = BeakerSpecVersion.v2, description: str | None = None, tasks: list[~beaker.types.BeakerTaskSpec] = <factory>, budget: str | None = None, retry: ~beaker.types.BeakerRetrySpec | None = None)

Defines an experiment workload.

Tip

See the spec format for an explanation of the fields.

version: BeakerSpecVersion = 'v2'

description: str | None = None

tasks: list[BeakerTaskSpec]

budget: str | None = None

retry: BeakerRetrySpec | None = None

classmethod from_file(path: PathLike | str) → BeakerExperimentSpec

Load a BeakerExperimentSpec from a YAML file.

classmethod from_json(data: dict[str, Any]) → BeakerExperimentSpec

Load a BeakerExperimentSpec from a JSON-compatible dictionary.

classmethod new(*, budget: str | None = None, task_name: str = 'main', description: str | None = None, cluster: str | list[str] | None = None, beaker_image: str | None = None, docker_image: str | None = None, result_path: str = '/unused', priority: BeakerJobPriority | str | None = None, **kwargs) → BeakerExperimentSpec

A convenience method for quickly creating a new BeakerExperimentSpec.

to_file(path: PathLike | str) → None

Write the experiment spec to a YAML file.

with_task(task: BeakerTaskSpec) → BeakerExperimentSpec

Returns a new spec with an additional task.

with_description(description: str) → BeakerExperimentSpec

Returns a new spec with the given description.

with_retries(allowed_task_retries: int) → BeakerExperimentSpec

Returns a new spec with the given retries.

class BeakerTaskSpec(image: BeakerImageSource, context: BeakerTaskContext, result: BeakerResultSpec | None, constraints: BeakerConstraints | None = None, name: str | None = None, command: list[str] | None = None, arguments: list[str] | None = None, env_vars: list[BeakerEnvVar] | None = None, datasets: list[BeakerDataMount] | None = None, resources: BeakerTaskResources | None = None, host_networking: bool = False, replicas: int | None = None, leader_selection: bool | None = False, propagate_failure: bool | None = None, propagate_preemption: bool | None = None, synchronized_start_timeout: int | None = None, timeout: int | None = None)

See the Task spec docs for an explanation of the fields.

image: BeakerImageSource

context: BeakerTaskContext

result: BeakerResultSpec | None

constraints: BeakerConstraints | None = None

name: str | None = None

command: list[str] | None = None

arguments: list[str] | None = None

env_vars: list[BeakerEnvVar] | None = None

datasets: list[BeakerDataMount] | None = None

resources: BeakerTaskResources | None = None

host_networking: bool = False

replicas: int | None = None

leader_selection: bool | None = False

propagate_failure: bool | None = None

propagate_preemption: bool | None = None

synchronized_start_timeout: int | None = None

timeout: int | None = None

classmethod from_json(data: dict[str, Any]) → BeakerTaskSpec

classmethod new(name: str, cluster: str | list[str] | None = None, beaker_image: str | None = None, docker_image: str | None = None, result_path: str = '/unused', priority: BeakerJobPriority | str | None = None, preemptible: bool | None = None, **kwargs) → BeakerTaskSpec

with_image(**kwargs) → BeakerTaskSpec

with_result(**kwargs) → BeakerTaskSpec

with_context(**kwargs) → BeakerTaskSpec

with_name(name: str) → BeakerTaskSpec

with_command(command: list[str | int | float]) → BeakerTaskSpec

with_arguments(arguments: list[str | int | float]) → BeakerTaskSpec

with_resources(**kwargs) → BeakerTaskSpec

with_dataset(mount_path: str, **kwargs) → BeakerTaskSpec

with_env_var(name: str, value: str | None = None, secret: str | None = None) → BeakerTaskSpec

with_constraint(**kwargs: list[str]) → BeakerTaskSpec

class BeakerImageSource(beaker: str | None = None, docker: str | None = None)

See the ImageSource spec docs for an explanation of the fields.

beaker: str | None = None

docker: str | None = None

class BeakerEnvVar(name: str, value: str | None = None, secret: str | None = None)

See the EnvVar spec docs for an explanation of the fields.

name: str

value: str | None = None

secret: str | None = None

class BeakerDataSource(beaker: str | None = None, host_path: str | None = None, weka: str | None = None, result: str | None = None, secret: str | None = None)

See the DataSource spec docs for an explanation of the fields.

beaker: str | None = None

host_path: str | None = None

weka: str | None = None

result: str | None = None

secret: str | None = None

class BeakerDataMount(source: BeakerDataSource, mount_path: str, sub_path: str | None = None)

See the DataMount spec docs for an explanation of the fields.

source: BeakerDataSource

mount_path: str

sub_path: str | None = None

classmethod new(mount_path: str, sub_path: str | None = None, beaker: str | None = None, host_path: str | None = None, weka: str | None = None, result: str | None = None, secret: str | None = None) → BeakerDataMount

class BeakerResultSpec(path: str)

See the ResultSpec docs for an explanation of the fields.

path: str

class BeakerTaskResources(cpu_count: float | None = None, gpu_count: int | None = None, memory: str | None = None, shared_memory: str | None = None)

See the TaskResources spec docs for an explanation of the fields.

cpu_count: float | None = None

gpu_count: int | None = None

memory: str | None = None

shared_memory: str | None = None

class BeakerTaskContext(cluster: str | None = None, priority: BeakerJobPriority | None = None, preemptible: bool | None = None)

See the Context spec docs for an explanation of the fields.

cluster: str | None = None

priority: BeakerJobPriority | None = None

preemptible: bool | None = None

class BeakerRetrySpec(allowed_task_retries: int | None = None)

See the Retry spec docs for an explanation of the fields.

allowed_task_retries: int | None = None

class BeakerConstraints(cluster: list[str] | None = None, hostname: list[str] | None = None)

See the Constraints spec docs for an explanation of the fields.

cluster: list[str] | None = None

hostname: list[str] | None = None