from typing import List, Optional
from finetuner.client.base import _BaseClient
from finetuner.constants import (
API_VERSION,
CONFIG,
CPUS,
DELETE,
DESCRIPTION,
DEVICE,
EXPERIMENTS,
GET,
GPUS,
LOGS,
NAME,
POST,
RUNS,
STATUS,
)
[docs]class FinetunerV1Client(_BaseClient):
"""
The Finetuner v1 API client.
"""
""" Experiment API """
[docs] def create_experiment(self, name: str, description: Optional[str] = '') -> dict:
"""Create a new experiment.
:param name: The name of the experiment.
:param description: Optional description of the experiment.
:return: Created experiment.
"""
url = self._construct_url(self._base_url, API_VERSION, EXPERIMENTS)
return self._handle_request(
url=url, method=POST, json_data={NAME: name, DESCRIPTION: description}
)
[docs] def get_experiment(self, name: str) -> dict:
"""Get an experiment by its name.
:param name: The name of the experiment.
:return: Requested experiment.
"""
url = self._construct_url(self._base_url, API_VERSION, EXPERIMENTS, name)
return self._handle_request(url=url, method=GET)
[docs] def list_experiments(self) -> List[dict]:
"""List all available experiments.
:return: List of all experiments.
"""
url = self._construct_url(self._base_url, API_VERSION, EXPERIMENTS)
return self._handle_request(url=url, method=GET)
[docs] def delete_experiment(self, name: str) -> dict:
"""Delete an experiment given its name.
:param name: The name of the experiment.
:return: Experiment to be deleted.
"""
url = self._construct_url(self._base_url, API_VERSION, EXPERIMENTS, name)
return self._handle_request(url=url, method=DELETE)
[docs] def delete_experiments(self) -> List[dict]:
"""Delete all experiments.
:return: Experiments to be deleted.
"""
url = self._construct_url(self._base_url, API_VERSION, EXPERIMENTS)
return self._handle_request(url=url, method=DELETE)
""" Run API """
[docs] def get_run(self, experiment_name: str, run_name: str) -> dict:
"""Get a run by its name and experiment.
:param experiment_name: The name of the experiment.
:param run_name: The name of the run.
:return: Requested run.
"""
url = self._construct_url(
self._base_url, API_VERSION, EXPERIMENTS, experiment_name, RUNS, run_name
)
return self._handle_request(url=url, method=GET)
[docs] def list_runs(self, experiment_name: Optional[str] = None) -> List[dict]:
"""List all created runs inside a given experiment.
If no experiment is specified, list runs for all available experiments.
:param experiment_name: The name of the experiment.
:return: List of all runs.
"""
if not experiment_name:
target_experiments = [
experiment[NAME] for experiment in self.list_experiments()
]
else:
target_experiments = [experiment_name]
response = []
for experiment_name in target_experiments:
url = self._construct_url(
self._base_url, API_VERSION, EXPERIMENTS, experiment_name, RUNS
)
response.extend(self._handle_request(url=url, method=GET))
return response
[docs] def delete_run(self, experiment_name: str, run_name: str) -> dict:
"""Delete a run by its name and experiment.
:param experiment_name: The name of the experiment.
:param run_name: The name of the run.
:return: Deleted run.
"""
url = self._construct_url(
self._base_url, API_VERSION, EXPERIMENTS, experiment_name, RUNS, run_name
)
return self._handle_request(url=url, method=DELETE)
[docs] def delete_runs(self, experiment_name: str) -> List[dict]:
"""Delete all runs inside a given experiment.
:param experiment_name: The name of the experiment.
:return: List of all deleted runs.
"""
url = self._construct_url(
self._base_url, API_VERSION, EXPERIMENTS, experiment_name, RUNS
)
return self._handle_request(url=url, method=DELETE)
[docs] def get_run_status(self, experiment_name: str, run_name: str) -> dict:
"""Get a run status by its name and experiment.
:param experiment_name: The name of the experiment.
:param run_name: The name of the run.
:return: Run status.
"""
url = self._construct_url(
self._base_url,
API_VERSION,
EXPERIMENTS,
experiment_name,
RUNS,
run_name,
STATUS,
)
return self._handle_request(url=url, method=GET)
[docs] def get_run_logs(self, experiment_name: str, run_name: str) -> str:
"""Get a run logs by its name and experiment.
:param experiment_name: The name of the experiment.
:param run_name: The name of the run.
:return: Run logs.
"""
url = self._construct_url(
self._base_url,
API_VERSION,
EXPERIMENTS,
experiment_name,
RUNS,
run_name,
LOGS,
)
return self._handle_request(url=url, method=GET)
[docs] def create_run(
self,
experiment_name: str,
run_name: str,
run_config: dict,
device: str,
cpus: int,
gpus: int,
) -> dict:
"""Create a run inside a given experiment.
For optional parameters please visit our documentation (link).
:param experiment_name: The name of the experiment.
:param run_name: The name of the run.
:param run_config: The run configuration.
:param device: The device to use, either `cpu` or `gpu`.
:param cpus: The number of CPUs to use.
:param gpus: The number of GPUs to use.
:return: Created run.
"""
url = self._construct_url(
self._base_url, API_VERSION, EXPERIMENTS, experiment_name, RUNS
)
return self._handle_request(
url=url,
method=POST,
json_data={
NAME: run_name,
CONFIG: run_config,
DEVICE: device,
CPUS: cpus,
GPUS: gpus,
},
)
