API

Gantry’s public API.

class GitRepoState(repo: str, repo_url: str, ref: str, branch: str | None = None)

Represents the state of a local git repository.

Tip

Use from_env() to instantiate this class.

repo: str

The repository name, e.g. "allenai/beaker-gantry".

repo_url: str

The repository URL for cloning, e.g. "https://github.com/allenai/beaker-gantry".

ref: str

The current ref.

branch: str | None = None

The current active branch, if any.

property is_dirty: bool

If the local repository state is dirty (uncommitted changes).

property is_public: bool

If the repository is public.

property ref_url: str

The URL to the current ref.

property branch_url: str | None

The URL to the current active branch.

is_in_tree(path: PathLike | str) → bool

Check if a file is in the tree.

classmethod from_env(ref: str | None = None, branch: str | None = None) → GitRepoState

Instantiate this class from the root of a git repository.

Raises:

• GitError – If this method isn’t called from the root of a valid git repository.

• UnpushedChangesError – If there are unpushed commits.

• RemoteBranchNotFoundError – If the local branch is not tracking a remote branch.

launch_experiment(args: Sequence[str], name: str | None = None, description: str | None = None, task_name: str = 'main', workspace: str | None = None, group_names: Sequence[str] | None = None, clusters: Sequence[str] | None = None, gpu_types: Sequence[str] | None = None, tags: Sequence[str] | None = None, hostnames: Sequence[str] | None = None, beaker_image: str | None = None, docker_image: str | None = None, cpus: float | None = None, gpus: int | None = None, memory: str | None = None, shared_memory: str | None = None, datasets: Sequence[str] | None = None, gh_token_secret: str = 'GITHUB_TOKEN', ref: str | None = None, branch: str | None = None, conda_file: PathLike | str | None = None, conda_env: str | None = None, python_manager: Literal['uv', 'conda'] | None = None, system_python: bool = False, uv_venv: str | None = None, uv_extras: Sequence[str] | None = None, uv_all_extras: bool | None = None, uv_torch_backend: str | None = None, env_vars: Sequence[str] | None = None, env_secrets: Sequence[str] | None = None, dataset_secrets: Sequence[str] | None = None, timeout: int | None = None, task_timeout: str | None = None, show_logs: bool | None = None, allow_dirty: bool = False, dry_run: bool = False, yes: bool | None = None, save_spec: PathLike | str | None = None, priority: str | None = None, install: str | None = None, no_python: bool = False, replicas: int | None = None, leader_selection: bool = False, host_networking: bool = False, propagate_failure: bool | None = None, propagate_preemption: bool | None = None, synchronized_start_timeout: str | None = None, mounts: Sequence[str] | None = None, weka: str | None = None, budget: str | None = None, preemptible: bool | None = None, retries: int | None = None, results: str = '/results', runtime_dir: str = '/gantry-runtime', exec_method: Literal['exec', 'bash'] = 'exec', skip_tcpxo_setup: bool = False, default_python_version: str = '3.10', pre_setup: str | None = None, post_setup: str | None = None)

Launch an experiment on Beaker. Same as the gantry run command.

follow_workload(beaker: Beaker, workload: Workload, *, task: Task | None = None, timeout: int = 0, tail: bool = False, show_logs: bool = True) → Job

Follow a workload until completion while streaming logs to stdout.

Parameters:

• task – A specific task in the workload to follow. Defaults to the first task.

• timeout – The number of seconds to wait for the workload to complete. Raises a timeout error if it doesn’t complete in time. Set to 0 (the default) to wait indefinitely.

• tail – Start tailing the logs if a job is already running. Otherwise shows all logs.

• show_logs – Set to False to avoid streaming the logs.

Returns:

The finalized BeakerJob from the task being followed.

Raises:

BeakerJobTimeoutError – If timeout is set to a positive number and the workload doesn’t complete in time.

update_workload_description(description: str, strategy: Literal['append', 'prepend', 'replace'] = 'replace', beaker_token: str | None = None)

Update the description of the Gantry workload that this process is running in.

Parameters:

• strategy – One of “append”, “prepend”, or “replace” to indicate how the new description should be combined with the original description. Defaults to “replace”.

• beaker_token – An optional Beaker API token to use. If not provided, the BEAKER_TOKEN environment variable will be used if set, or a Beaker config file.

write_metrics(metrics: Dict[str, Any])

Write result metrics for the Gantry workload that this process is running in.

Parameters:

metrics – A JSON-serializable dictionary of metrics to write.