graflag.api

GraFlag Python API for GUI Integration.

Thin wrapper around GraFlag core that provides error-safe access and returns structured dataclass objects suitable for GUI/web consumption.

class graflag.api.GraFlagAPI(config_file: str | None = None, log_level: int = 20)

Bases: object

High-level Python API for GraFlag operations.

All methods return structured data (dataclasses) and catch exceptions to avoid crashing the GUI. Use the core GraFlag class directly for CLI-style usage where exceptions should propagate.

Usage:

api = GraFlagAPI(config_file=”.env”)

cluster = api.get_cluster_info() methods = api.list_methods() experiments = api.list_experiments()

get_cluster_info() → ClusterInfo

Get cluster status information.

setup_cluster() → Dict[str, Any]

Setup GraFlag cluster.

list_methods() → List[MethodInfo]

List available methods.

get_method_details(method_name: str) → MethodInfo | None

Get details for a specific method.

list_datasets() → List[DatasetInfo]

List available datasets.

count_experiments() → int

Total experiments on the share, for reporting truncation.

list_experiments(limit: int = 50, offset: int = 0) → List[ExperimentInfo]

List recent experiments.

get_experiment_details(experiment_name: str) → ExperimentInfo | None

Get details for a specific experiment.

One probe of that experiment; this used to list 500 and search them, which also missed any experiment older than the 500 newest.

run(method: str, dataset: str, tag: str = 'latest', build: bool = False, gpu: bool = True, method_params: Dict[str, Any] | None = None, on_progress: Callable[[RunProgress], None] | None = None, exp_name: str | None = None, keep_service: bool = False) → str

Run an experiment. Returns experiment name.

Waits for the run without streaming its output (follow=False): the dashboard calls this from a background thread, where the stream only reached the server’s console, and following it kept a Docker SDK call in a loop for the whole run.

get_experiment_results(experiment_name: str) → ExperimentResults | None

Get experiment results.

get_evaluation_results(experiment_name: str) → EvaluationResults | None

Get evaluation results.

evaluate_experiment(experiment_name: str) → bool

Run evaluation on an experiment (quietly, see run()).

list_running_services() → List[Dict[str, str]]

List running Docker services.

Deliberately not error-safe, unlike its neighbours. Returning [] on failure made a dropped SSH tunnel look exactly like a cluster with nothing running: the Services panel emptied, the count read 0, and nothing anywhere said the cluster had not been reached. All three callers in the GUI handle the exception – the route answers 500 so the panel can say so, and the updater declines to broadcast a list it does not have.

stop_experiment(experiment_name: str) → bool

Stop a running experiment.

delete_experiment(experiment_name: str) → bool

Stop and delete an experiment.

cleanup_services(experiment: str | None = None, dry_run: bool = False) → List[ServiceCleanupResult]

Remove Swarm services for finished experiments.

clear(apply: bool = False, share: bool = True, images: bool = True, collect: bool = False) → ClearReport

Report – or remove – storage and images nothing owns any more.

A failure comes back as an empty report carrying the message, not as an exception: the GUI polls this beside the experiment list and a raise here would take the page down over housekeeping.

get_experiment_logs(experiment_name: str, tail: int = 100) → List[str]

Get recent logs for an experiment.

download_file(remote_path: str, local_path: str) → bool

Download a file from remote shared directory.

download_directory(remote_path: str, local_path: str) → bool

Download a directory from remote shared directory.