Apheris CLI Python API Reference

apheris

Convenience namespace containing some common Apheris functions.

login(username=None, password=None, login_mode='sso')

Authenticate a user, either through their Apheris account or using their company account. Programmatic login can be achieved by supplying the apheris username/passwords directly. Alternatively the user can jump right through to using their company login by setting the login mode to "sso"

logout(verbose=True)

Logs the user out.

Parameters:

Name	Type	Description	Default
verbose	bool	If True, print information on logout process.	True

list_datasets(n=None, to_table=True)

List the n most recently updated remote datasets.

Parameters:

Name	Type	Description	Default
n	Optional[int]	number of remote datasets to list. If None, list all. Default: None	None
to_table	bool	If True, a prettytable.PrettyTable is returned. If False, a list of Dictionaries will be returned.	True

Returns:

Type	Description
Union[PrettyTable, List[dict]]	If to_table is True, a prettytable.PrettyTable is returned. The datasets are sorted by their updated_at time, starting from the most recent one. most recently updated remote datasets is returned. If n is provided, will return the n most recent rows.
Union[PrettyTable, List[dict]]	If to_table is False, a list of Dictionaries is returned. The datasets are sorted by their updated_at time, starting from the most recent one. most recently updated remote datasets is returned. If n is provided, will return the n most recent rows.

list_models(to_table=True)

Convenience function to query the model list and parse the response in one call.

Parameters:

Name	Type	Description	Default
to_table	bool	If True, a prettytable.PrettyTable is returned. If False, a list of Dictionaries will be returned.	True

Returns:

Type	Description
Union[PrettyTable, List[Dict[str, str]]]	If to_table is True, a prettytable.PrettyTable is returned.
Union[PrettyTable, List[Dict[str, str]]]	If to_table is False, a list of Dictionaries is returned.

list_compute_specs(limit=10, to_table=True, verbose=False)

Convenience function to list the limit most recent Compute Specs and optionally output as a table.

Parameters:

Name	Type	Description	Default
limit	Optional[int]	The number of most recent Compute Specs to list. If None, all compute specs are listed. Defaults to 10.	10
to_table	bool	Whether to output the results as a pretty table. Defaults to True.	True
verbose	bool	Whether to include all details in the table. Defaults to False.	False

Returns:

Type	Description
Union[PrettyTable, List[Dict[str, str]]]	If to_table is True, a prettytable.PrettyTable is returned.
Union[PrettyTable, List[Dict[str, str]]]	If to_table is False, a list of Dictionaries is returned. In future versions this will return a list of ComputeSpec objects.

list_jobs(compute_spec_id=None, use_v2=False)

List all jobs of a certain Compute Spec, or for the current user if no Compute Spec ID is provided.

Parameters:

Name	Type	Description	Default
compute_spec_id	Optional[UUID]	The ID of the Compute Spec. If None, show all jobs for the current user. Note: ignored when use_v2=True (the v2 endpoint does not support filtering by compute spec).	None
use_v2	bool	If True, use the v2 endpoint (GET /v2/jobs). All pages are fetched automatically.	False

Returns: list: List of all jobs on the specified Compute Spec (or for the provided user if no Compute Spec ID is provided).

aphcli.api.compute

Functions related to the handling of Compute Specs (creation, activation, list, etc.).

ComputeSpec dataclass

dataset_ids: List of dataset IDs client_n_cpu: number of vCPUs of Compute Clients client_n_gpu: number of GPUs of Compute Clients client_memory: memory of Compute Clients [MByte] server_n_cpu: number of vCPUs of Compute Aggregators server_n_gpu: number of GPUs of Compute Aggregators server_memory: memory of Compute Aggregators [MByte] model: model to use num_clients_per_gateway: number of compute clients to spawn per gateway (optional)

set_ignore_limits(ignore)

Limits have been set on

match(cs1, cs2)

Compare two ComputeSpec objects for equality, ignoring internal fields.

Parameters:

Name	Type	Description	Default
cs1	ComputeSpec	First ComputeSpec object to compare	required
cs2	ComputeSpec	Second ComputeSpec object to compare	required

Returns:

Name	Type	Description
bool	bool	True if all non-internal fields match between the two objects, False otherwise

deactivate_all()

Deactivate all Compute Specs that are in an active or transitional state.

This includes Compute Specs with status: - running - creating - failed - waiting_for_resources

create(compute_spec, verbose=False, cache=True)

Create a Compute Spec.

Parameters:

Name	Type	Description	Default
compute_spec	ComputeSpec	Compute Spec to be created.	required
verbose	bool	If True, print more detailed information.	False
cache	bool	If True, cache the Compute Spec ID after creation.	True

Returns:

Name	Type	Description
UUID	UUID	ID of the created Compute Spec.

create_from_args(dataset_ids=None, client_n_cpu=None, client_n_gpu=None, client_memory=None, server_n_cpu=None, server_n_gpu=None, server_memory=None, model=None, model_id=None, model_version=None)

Create a Compute Spec from raw arguments.

To describe the model, the user can supply either a dictionary with id and version fields, or provide them as individual arguments.

Parameters:

Name	Type	Description	Default
dataset_ids	Optional[List[str]]	List of dataset IDs.	None
client_n_cpu	Optional[float]	Number of vCPUs for the client.	None
client_n_gpu	Optional[int]	Number of GPUs for the client.	None
client_memory	Optional[int]	Amount of memory in MB for the client.	None
server_n_cpu	Optional[float]	Number of vCPUs for the server.	None
server_n_gpu	Optional[int]	Number of GPUs for the server.	None
server_memory	Optional[int]	Amount of memory in MB for the server.	None
model	Optional[Dict[str, str]]	Dictionary with id and version fields.	None
model_id	Optional[str]	ID of the model.	None
model_version	Optional[str]	Version of the model.	None

Returns:

Name	Type	Description
UUID	UUID	ID of the created Compute Spec.

get(compute_spec_id=None, cache=True)

Get a Compute Spec ID.

Parameters:

Name	Type	Description	Default
compute_spec_id	Optional[UUID]	Compute Spec ID that shall be fetched. If None, use the most recently used Compute Spec ID.	None
cache	bool	If True, cache the Compute Spec ID after getting the Compute Spec.	True

Returns:

Name	Type	Description
ComputeSpec	ComputeSpec	definition of a Compute Spec

activate(compute_spec_id=None, verbose=False, cache=True)

Activate a Compute Spec ID.

Parameters:

Name	Type	Description	Default
compute_spec_id	Optional[UUID]	Compute Spec ID that shall be activated. If None, use the most recently used Compute Spec ID.	None
verbose	bool	If True, print more detailed information.	False
cache	bool	If True, cache the Compute Spec ID after activation.	True

deactivate(compute_spec_id=None, verbose=False, cache=True)

Deactivate a Compute Spec ID.

Parameters:

Name	Type	Description	Default
compute_spec_id	Optional[UUID]	Compute Spec ID that shall be de-activated. If None, use the most recently used Compute Spec ID.	None
verbose	bool	If True, print more detailed information.	False
cache	bool	If True, cache the Compute Spec ID after deactivation.	True

get_activation_status(compute_spec_id=None)

Get the activation status of a Compute Spec.

Parameters:

Name	Type	Description	Default
compute_spec_id	Optional[UUID]	ID of the Compute Spec that shall be queried. If None, use the most recently used Compute Spec ID.	None

get_status(compute_spec_id=None)

Get the detailed status of a Compute Spec.

Parameters:

Name	Type	Description	Default
compute_spec_id	Optional[UUID]	ID of the Compute Spec that shall be queried. If None, use the most recently used Compute Spec ID.	None

Returns:

Name	Type	Description
dict	Dict[str, str]	A dict with keys status and message. The ´statusshows the activation status, e.g.creatingorrunning. Themessage` contains detail information. For example, if the Compute Spec isn't activating, the message might show that there is not enough hardware resource of a certain type.

list_compute_specs(limit=10, active_only=False)

List most recent Compute Specs.

Parameters:

Name	Type	Description	Default
limit	Optional[int]	The maximum number of Compute Specs to return. If None, return all.	10
active_only	bool	If True, only return Compute Specs that are active.	False

Returns:

Type	Description
List[dict]	List[dict]: A list of dictionaries containing the Compute Spec information.

get_compute_spec_count(active_only=False)

Get the number of compute specs.

Parameters:

Name	Type	Description	Default
active_only	bool	If True, only return the number of active Compute Specs.	False

Returns:

Name	Type	Description
int	int	Number of Compute Specs.

get_compute_specs_details(compute_specs)

Parse the Compute Spec response from the API end point into the data required for the tabular format.

Parameters:

Name	Type	Description	Default
compute_specs	Union[Dict[str, str], List[Dict[str, str]]]	A list of dictionaries containing Compute Spec information. Can also be a single dictionary containing one Compute Spec.	required

Returns:

Type	Description
List[Dict[str, str]]	List[dict]: A list of dictionaries containing the Compute Spec information.

wait_until_running(compute_spec_id, check_interval=10, timeout=COMPUTE_SPEC_ACTIVATION_TIMEOUT)

Wait until a Compute Spec's activation status is running.

Parameters:

Name	Type	Description	Default
compute_spec_id	Optional[UUID]	ID of the Compute Spec that we want to observe. If None, use the most recently used Compute Spec id.	required
check_interval	float	Interval to check the status in seconds	10
timeout	Optional[float]	Timeout in seconds. If the target status is not reached within this time a TimeoutError is raised. If set to None, the function will not time out.	COMPUTE_SPEC_ACTIVATION_TIMEOUT

aphcli.api.job

Functions related to the handling of jobs (run, abort, list, status, etc.).

list_jobs(compute_spec_id=None, use_v2=False)

List all jobs of a certain Compute Spec, or for the current user if no Compute Spec ID is provided.

Parameters:

Name	Type	Description	Default
compute_spec_id	Optional[UUID]	The ID of the Compute Spec. If None, show all jobs for the current user. Note: ignored when use_v2=True (the v2 endpoint does not support filtering by compute spec).	None
use_v2	bool	If True, use the v2 endpoint (GET /v2/jobs). All pages are fetched automatically.	False

Returns: list: List of all jobs on the specified Compute Spec (or for the provided user if no Compute Spec ID is provided).

get(job_id=None, verbose=False, use_v2=False)

Get details on a job

Parameters:

Name	Type	Description	Default
job_id	Union[str, UUID, None]	The ID of the job whose details shall be fetched	None
verbose	bool	If True, provide more detailed information.	False
use_v2	bool	If True, use the v2 endpoint (GET /v2/jobs/{id}).	False

Returns: dict: A dictionary with information on the job

get_steps_v2(job_id=None)

Fetch execution steps for a V2 job.

Parameters:

Name	Type	Description	Default
job_id	Optional[str]	The ID of the job. If None, the cached job ID is used.	None

Returns: List of JobV2Step objects in order of creation.

status(job_id=None, verbose=False)

Get the status of a job

Parameters:

Name	Type	Description	Default
job_id	Optional[UUID]	The ID of the job whose status shall be fetched	None
verbose	bool	If True, provide more detailed information.	False

Returns: str: A string with details on the job's status

run(datasets, payload, model, version, resources=Resources(), num_clients_per_gateway=1, timeout=COMPUTE_SPEC_ACTIVATION_TIMEOUT, new_cs=False, queue_id=None, no_auto_cleanup=None)

Runs a job on given datasets, resources can be optionally specified.

Parameters:

Name	Type	Description	Default
datasets	List[str]	List of dataset IDs.	required
payload	dict	Dictionary containing the job arguments.	required
model	str	Model ID.	required
version	str	Model version.	required
resources	Resources	Resources required for the job.	Resources()
num_clients_per_gateway	Optional[int]	Number of compute clients to spawn per gateway.	1
timeout	Optional[float]	Timeout in seconds for waiting for compute spec to be running. Defaults to 600 seconds (10 minutes). If None, the timeout is disabled entirely. Ignored when queue_id or no_auto_cleanup are set (v2 manages the lifecycle server-side).	COMPUTE_SPEC_ACTIVATION_TIMEOUT
new_cs	bool	If True, always create a new compute spec regardless of existing matches. Ignored when queue_id or no_auto_cleanup are set.	False
queue_id	Optional[str]	Optional queue ID for sequential execution. Jobs sharing the same queue ID run one at a time. Setting this uses the v2 endpoint which handles the full lifecycle server-side.	None
no_auto_cleanup	Optional[bool]	If set (True or False), uses the v2 endpoint. When True, skips compute spec deactivation after job completion.	None

Returns:

Name	Type	Description
Job	Job	The submitted job.

submit(job_args, compute_spec_id=None, verbose=False)

Submit a job

Parameters:

Name	Type	Description	Default
job_args	dict	Arguments for the job that you want to submit.	required
compute_spec_id	Optional[UUID]	The ID of the Compute Spec. If None, use the most recently used Compute Spec id.	None
verbose	bool	If True, provide more detailed information.	False

Returns: UUID: This UUID is your reference to the submitted job

abort(job_id=None, use_v2=False)

Abort a job

Parameters:

Name	Type	Description	Default
job_id	Optional[Union[UUID, str]]	The ID of the job that shall be aborted	None
use_v2	bool	If True, use the v2 endpoint (POST /v2/jobs/{id}/abort).	False

abort_jobs(jobs, use_v2=False)

Abort a pre-matched list of jobs.

Separated from abort_all so the CLI can abort exactly the jobs shown in the preview without re-listing, avoiding a race between what the user confirmed and what is actually aborted.

Parameters:

Name	Type	Description	Default
jobs	List	Pre-matched list of Job or JobV2 objects to abort.	required
use_v2	bool	If True, use the V2 abort endpoint; otherwise V1.	False

Returns:

Type	Description
Tuple[List[str], List[str]]	Tuple of (aborted_ids, failed_ids).

abort_all(only_statuses=None, only_created_by_me=False, only_queue_id=None, use_v2=False)

Abort all jobs matching the given criteria.

Parameters:

Name	Type	Description	Default
only_statuses	Optional[List[str]]	If provided, only abort jobs with these statuses. V2: must be a subset of ABORTABLE_STATUSES {"CREATED", "PROVISIONING", "RUNNING"}. V1: any status that does not start with "finished." is accepted. Defaults to all abortable statuses for the selected API version.	None
only_created_by_me	bool	If True, only abort jobs created by the currently logged-in user. V2 only — V1 jobs do not carry creator information.	False
only_queue_id	Optional[str]	If provided, only abort jobs belonging to this queue ID. V2 only — V1 jobs are not queue-aware.	None
use_v2	bool	If True, use the V2 endpoint and filters; otherwise use V1. Defaults to False to match the convention of all other job API helpers.	False

Returns:

Type	Description
Tuple[List[str], List[str]]	Tuple of (aborted_ids, failed_ids).

Raises:

Type	Description
ValueError	if only_statuses contains terminal/non-abortable statuses, or if V2-only filters are passed with use_v2=False.

download_results(download_path, job_id=None, use_v2=False)

Download the results of a job.

Parameters:

Name	Type	Description	Default
download_path	Union[str, Path]	File path to download the results to	required
job_id	Optional[Union[UUID, str]]	The ID of the job	None
use_v2	bool	If True, use the V2 endpoint (GET /v2/jobs/{id}/download).	False

logs(job_id=None)

Get the logs of a job

Parameters:

Name	Type	Description	Default
job_id	Optional[UUID]	The ID of the job whose logs shall be fetched	None

Returns:

Name	Type	Description
str	str	A string that contains the logs

logs_from_orchestrator(job_id=None)

Get the logs of a job from the orchestrator service

Parameters:

Name	Type	Description	Default
job_id	Union[str, UUID, None]	The ID of the job whose logs shall be fetched	None

Returns:

Type	Description
List[LogEntry]	List[LogEntry]: A list of log entries with client_id, container, created_at, and line

wait_until_job_finished(job_id=None, check_interval=None, timeout=100, warn_time=None)

Wait until a job is finished

Parameters:

Name	Type	Description	Default
job_id	Union[UUID, str, None]	The ID of the job	None
check_interval	Optional[float]	Interval between http-requests that query the status in seconds.	None
timeout	Optional[float]	Maximum time to wait in seconds before a TimeoutError is raised.	100
warn_time	Optional[float]	After this time in seconds, a warning is fired.	None

aphcli.api.models

Functions related to the Apheris Model Registry.

get_models()

Returns a dictionary with detailed information on models available in the model registry. This returns the raw response from the Apheris Registry API and should be processed using models_from_response.

Returns:

Type	Description
dict	A dictionary containing detailed information on models in the Registry.

models_from_response(response)

Parses the raw data from the models response and extracts the list of models and available versions.

Parameters:

Name	Type	Description	Default
response	dict	The raw response from the API.	required

Returns:

Type	Description
List[Dict[str, str]]	A list of dictionaries, each containing the model ID and version.

list_models(to_table=True)

Convenience function to query the model list and parse the response in one call.

Parameters:

Name	Type	Description	Default
to_table	bool	If True, a prettytable.PrettyTable is returned. If False, a list of Dictionaries will be returned.	True

Returns:

Type	Description
Union[PrettyTable, List[Dict[str, str]]]	If to_table is True, a prettytable.PrettyTable is returned.
Union[PrettyTable, List[Dict[str, str]]]	If to_table is False, a list of Dictionaries is returned.

indexed_model_list(models)

Convert the list of models to a tabular format for interaction

Parameters:

Name	Type	Description	Default
models	List[Dict[str, str]]	A list of dictionaries containing model information.	required

Returns:

Type	Description
List[Tuple[int, str, str]]	A list of tuples containing the index, model ID, and model version.

indexed_model_list_to_table(rows)

Convert the list of models to a prettytable for interaction.

Assumes the input is a list of tuples containing the index, model ID, and model, such as that provided by indexed_model_list

Parameters:

Name	Type	Description	Default
rows	List[Tuple[int, str, str]]	A list of tuples containing the index, model ID, and model version.	required

Returns:

Type	Description
PrettyTable	A prettytable containing the model information.

get_model(model_id)

Returns a dictionary with detailed information on a specific model in the model registry.

Parameters:

Name	Type	Description	Default
model_id	str	The ID of the model to get.	required

Returns:

Type	Description
ModelDetails	A ModelDetails object containing detailed information on the model.

add_model_version(model_id, version, digest, commit_hash, engine_version)

Adds a new version to an existing model in the Apheris Registry.

Parameters:

Name	Type	Description	Default
model_id	str	The ID of the model to add a version to.	required
version	str	The version tag for the new model version.	required
digest	str	The digest of the model.	required
commit_hash	str	The commit hash associated with this model version.	required
engine_version	str	Engine version, for example nvflare:2.6.0. For information on supported engines and versions, please contact your Apheris representative.	required

Returns:

Type	Description
Dict[str, Any]	A dictionary containing the response from the API.

Raises:

Type	Description
ComputeSpecException	If there's an error in adding the model version.

list_model_versions(model_id)

Convenience function to list all the versions of a model.

Parameters:

Name	Type	Description	Default
model_id	str	The ID of the model whose versions to list.	required

Returns:

Type	Description
List[ModelVersion]	A list ofModelVersion objects

add_model(model_id, model_name, source_repository_url, model_description='', logo_url='')

Adds a new model to the Apheris Registry.

Parameters:

Name	Type	Description	Default
model_id	str	The ID of the new model.	required
model_name	str	The Name of the new model.	required
source_repository_url	str	The URL of the source repository.	required
model_description	str	The description of the model.	''
logo_url	str	The URL of the logo or base64 encoded image.	''

Returns: A ModelDetails object containing detailed information on the model.

Raises:

Type	Description
ModelException	If there's an error in adding the new model.

get_robot()

Returns the user's robot account to use for pushing model images to the model's OCI repository.

Returns:

Type	Description
RobotDetails	A RobotDetails object containing detailed information on the robot.