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"

Source code in .env/lib/python3.10/site-packages/apheris_auth/core/auth.py

def login(
    username: str = None,
    password: str = None,
    login_mode: Union[None, Literal["sso"]] = "sso",
) -> None:
    """
    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"
    """
    try:
        if username and password:
            if settings.LOGIN_TYPE == LoginType.LEGACY:
                __login(username, password)
                return
            else:
                _login_for_audiences(
                    partial(
                        resource_owner_grant.login, username=username, password=password
                    )
                )
                return

        if login_mode == "sso":
            if not is_logged_in():
                sso_login()
            else:
                cprint("You are already logged in", "green")
            return

        answer = input(
            """Please type 1 or 2 to specify how you would like to login:
      1. With your Apheris account
      2. With your company account
      (1|2): """
        )

        if answer == "2":
            if not is_logged_in():
                cprint(
                    "Please note this is a two step process and you "
                    "will be asked to interact with this application "
                    "during each of the steps",
                    color="grey",
                    on_color="on_white",
                    attrs=["bold"],
                )
                sso_login()
            else:
                cprint("You are already logged in", "green")
        elif answer == "1":
            if not username:
                username = input("E-mail: ")
            if not password:
                password = getpass()
            if settings.LOGIN_TYPE == LoginType.LEGACY:
                __login(username, password)
            else:
                _login_for_audiences(
                    partial(
                        resource_owner_grant.login, username=username, password=password
                    )
                )
        else:
            cprint("Invalid answer, please provide either 1 or 2", "red")
    except KeyboardInterrupt:
        cprint("\nLogin aborted by user...", "red")

logout()

Logs the user out.

Source code in .env/lib/python3.10/site-packages/apheris_auth/core/auth.py

def logout():
    """
    Logs the user out.
    """
    try:
        sso_logout()
    except exceptions.NotSSOSession:
        logger.debug("Will logout form an Apheris session.")
        # revoke apheris access token
        _logout()
    except exceptions.AlreadyLoggedOut:
        logger.debug("It appears that you are already logged out!")
        cprint("Already logged out", "green")
        return

    cprint("Successfully logged out", "green")

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.

Source code in .env/lib/python3.10/site-packages/aphcli/api/datasets.py

def list_datasets(
    n: Optional[int] = None,
    to_table: bool = True,
) -> Union[PrettyTable, List[dict]]:
    """
    List the `n` most recently updated remote datasets.

    Args:
        n: number of remote datasets to list. If None, list all. Default: None
        to_table: If True, a prettytable.PrettyTable is returned. If False, a list of
            Dictionaries will be returned.

    Returns:
        - 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.
        - 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.
    """
    client = get_client()
    datasets = client.get_datasets()

    def date_sort(x, key):
        return datetime.strptime(x[key], "%Y-%m-%dT%H:%M:%S.%fZ")

    if to_table:
        rows = [
            [
                x["slug"],
                x["organization"],
                x["owner"]["full_name"],
                x["updated_at"],
            ]
            for x in datasets
        ]
        rows.sort(key=lambda x: date_sort(x, 3))

        # Now remove the updated_at column as it's only used for sorting
        if n is not None:
            rows = rows[-n:]

        rows = [[i] + r[:3] for i, r in enumerate(rows)]
        return _create_dataset_table(rows)
    else:
        datasets = sorted(datasets, key=lambda x: date_sort(x, "updated_at"))
        if n is not None:
            datasets = datasets[-n:]
        return datasets

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.

Source code in .env/lib/python3.10/site-packages/aphcli/api/models.py

def list_models(to_table: bool = True) -> Union[PrettyTable, List[Dict[str, str]]]:
    """
    Convenience function to query the model list and parse the response in one call.

    Args:
        to_table: If True, a prettytable.PrettyTable is returned. If False, a list of
            Dictionaries will be returned.

    Returns:
        - If `to_table` is True, a `prettytable.PrettyTable` is returned.
        - If `to_table` is False, a list of Dictionaries is returned.
    """
    validate_is_logged_in()
    response = get_models()
    models = models_from_response(response)

    if to_table:
        rows = indexed_model_list(models)
        table = indexed_model_list_to_table(rows)
        return table
    else:
        return models

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.

Source code in .env/lib/python3.10/site-packages/apheris/__init__.py

def list_compute_specs(
    limit: Optional[int] = 10, to_table: bool = True, verbose: bool = False
) -> Union[prettytable.PrettyTable, List[Dict[str, str]]]:
    """
    Convenience function to list the `limit` most recent compute specs and optionally
    output as a table.
    """
    if limit is not None and (not isinstance(limit, int) or limit < 1):
        raise ValueError("The limit must either be None or an integer >= 1")

    compute_specs = compute_api.list_compute_specs()
    if limit is not None:
        compute_specs = compute_specs[:limit]

    if to_table:
        return generate_table(compute_specs, verbose)
    else:
        return compute_specs

list_jobs(compute_spec_id=None)

List all jobs of a certain compute spec

Parameters:

Name	Type	Description	Default
compute_spec_id	Optional[UUID]	The ID of the compute spec. If None, use the most recently used compute spec id.	None

Returns: list: List of all jobs on the specified compute spec

Source code in .env/lib/python3.10/site-packages/aphcli/api/job.py

def list_jobs(compute_spec_id: Optional[UUID] = None) -> List[Job]:
    """
    List all jobs of a certain compute spec

    Arguments:
        compute_spec_id: The ID of the compute spec. If `None`, use the most recently
            used compute spec id.
    Returns:
        list: List of all jobs on the specified compute spec
    """
    url = f"{settings.API_ORCHESTRATOR_BASE_URL}/jobs"
    if compute_spec_id:
        url += f"?computespecID={compute_spec_id}"
        cache_compute_spec_id(compute_spec_id)
    response = exception_handled_request(get_oauth_session(), url, "get")

    payload = response.json()
    try:
        return [Job(**item) for item in payload]
    except ValidationError as exc:
        raise JobsException(
            f"Could not interpret the server response. Errors: {exc.errors()}"
        )
    except TypeError as exc:
        raise JobsException(f"Could not interpret the server response. Error: {exc}")

aphcli.api.compute

Functions related to the handling of compute specs (creation, activation, list, etc.).

activate(compute_spec_id=None, verbose=False)

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

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 specification 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.

deactivate(compute_spec_id=None, verbose=False)

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

get(compute_spec_id=None)

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

Returns:

Name	Type	Description
ComputeSpec	ComputeSpec	definition of a compute spec

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.

set_ignore_limits(ignore)

Limits have been set on

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

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.	600

aphcli.api.job

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

abort(job_id=None, compute_spec_id=None)

Abort a job

Parameters:

Name	Type	Description	Default
job_id	Optional[UUID]	The ID of the job that shall be aborted	None
compute_spec_id	Optional[UUID]	The ID of the compute spec. If None, use the most recently used compute spec id.	None

download_results(download_path, job_id=None, compute_spec_id=None)

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[UUID]	The ID of the job	None
compute_spec_id	Optional[UUID]	The ID of the compute spec. If None, use the most recently used compute spec id.	None

get(job_id=None, compute_spec_id=None, verbose=True)

Get details on a job

Parameters:

Name	Type	Description	Default
job_id	Optional[UUID]	The ID of the job whose details shall be fetched	None
compute_spec_id	Optional[UUID]	deprecated, stays to prevent breaking changes	None
verbose		If True, provide more detailed information.	True

Returns: dict: A dictionary with information on the job

list_jobs(compute_spec_id=None)

List all jobs of a certain compute spec

Parameters:

Name	Type	Description	Default
compute_spec_id	Optional[UUID]	The ID of the compute spec. If None, use the most recently used compute spec id.	None

Returns: list: List of all jobs on the specified compute spec

logs(job_id=None, compute_spec_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
compute_spec_id	Optional[UUID]	The ID of the compute spec. If None, use the most recently used compute spec id.	None

Returns:

Name	Type	Description
str	str	A string that contains the logs

status(job_id=None, compute_spec_id=None, verbose=True)

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
compute_spec_id	Optional[UUID]	This argument is deprecated and	None
verbose		If True, provide more detailed information.	True

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

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

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

Wait until a job is finished

Parameters:

Name	Type	Description	Default
job_id	Optional[UUID]	The ID of the job	None
compute_spec_id	Optional[UUID]	The ID of the compute spec. If None, use the most recently used compute spec id.	None
check_interval	float	Interval between http-requests that query the status in seconds.	3
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.

indexed_model_list(models)

Convert the list of models to a tabular format for interaction

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.

models_from_response(response)

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