interface

alchemiscale.interface.client — client for interacting with user-facing API

class alchemiscale.interface.client.AlchemiscaleClient(api_url: str, identifier: str, key: str, max_retries: int = 5, retry_base_seconds: float = 2.0, retry_max_seconds: float = 60.0, verify: bool = True)

Bases: AlchemiscaleBaseClient

Client for user interaction with API service.

action_tasks(tasks: List[ScopedKey], network: ScopedKey) → List[ScopedKey | None]

Action Tasks for execution via the given AlchemicalNetwork’s TaskHub.

A Task cannot be actioned:

  • to an AlchemicalNetwork in a different Scope.

  • if it extends another Task that is not complete.

  • if it has any status other than ‘waiting’, ‘running’, or ‘error’

Parameters:

  • tasks – Task ScopedKeys to action for execution.

  • network – The AlchemicalNetwork ScopedKey to action the Tasks for. The Tasks will be added to the network’s associated TaskHub.

Returns:

ScopedKeys for Tasks actioned, in the same order as given as tasks on input. If a Task couldn’t be actioned, then None will be returned in its place.

Return type:

List[Optional[ScopedKey]]

cancel_tasks(tasks: List[ScopedKey], network: ScopedKey) → List[ScopedKey]

Cancel Tasks for execution via the given AlchemicalNetwork’s TaskHub.

A Task cannot be canceled:

  • if it is not present in the AlchemicalNetwork’s TaskHub.

Parameters:

  • tasks – Task ScopedKeys to cancel for execution.

  • network – The AlchemicalNetwork ScopedKey to cancel the Tasks for. The Tasks will be removed from the network’s associated TaskHub.

Returns:

ScopedKeys for Tasks canceled, in the same order as given as tasks on input. If a Task couldn’t be canceled, then None will be returned in its place.

Return type:

List[Optional[ScopedKey]]

check_exists(scoped_key: Scope) → bool

Returns True if the given ScopedKey represents an object in the database.

create_network(network: gufe.AlchemicalNetwork, scope: Scope, compress: bool | int = True) → ScopedKey

Submit an AlchemicalNetwork to a specific Scope.

Parameters:

  • network – The AlchemicalNetwork to submit.

  • scope – The Scope in which to submit the AlchemicalNetwork. This must be a specific Scope; it must not contain wildcards.

  • compress –

    If True, compress the AlchemicalNetwork client-side before shipping to the API service. This can reduce submission time depending on the bandwidth of your connection to the API service. Set to False to submit without compressing. This is a performance optimization; it has no bearing on the result of this method call.

    Use an integer between 0 and 9 for finer control over the degree of compression; 0 means no compression, 9 means max compression. True is synonymous with level 5 compression.

Returns:

The ScopedKey of the AlchemicalNetwork.

Return type:

ScopedKey

create_tasks(transformation: ScopedKey, extends: ScopedKey | None = None, count: int = 1) → List[ScopedKey]

Create Tasks for the given Transformation.

Parameters:

  • transformation – The ScopedKey of the Transformation to create Tasks for.

  • extends – The ScopedKey for the Task to use as a starting point for the Tasks created.

  • count – The number of new Tasks to create.

Returns:

A list giving the ScopedKeys of the new Tasks created.

Return type:

List[ScopedKey]

get_chemicalsystem(chemicalsystem: ScopedKey | str, compress: bool = True, visualize: bool = True) → gufe.ChemicalSystem

Retrieve a ChemicalSystem given its ScopedKey.

Parameters:

  • chemicalsystem – The ScopedKey of the ChemicalSystem to retrieve.

  • compress – If True, compress the ChemicalSystem server-side before shipping it to the client. This can reduce retrieval time depending on the bandwidth of your connection to the API service. Set to False to retrieve without compressing. This is a performance optimization; it has no bearing on the result of this method call.

  • visualize – If True, show retrieval progress indicator.

Returns:

The retrieved ChemicalSystem.

Return type:

ChemicalSystem

get_chemicalsystem_networks(chemicalsystem: ScopedKey) → List[ScopedKey]

List ScopedKeys for the AlchemicalNetworks associated with the given ChemicalSystem.

get_chemicalsystem_transformations(chemicalsystem: ScopedKey) → List[ScopedKey]

List ScopedKeys for the Transformations associated with the given ChemicalSystem.

get_network(network: ScopedKey | str, compress: bool = True, visualize: bool = True) → gufe.AlchemicalNetwork

Retrieve an AlchemicalNetwork given its ScopedKey.

Parameters:

  • network – The ScopedKey of the AlchemicalNetwork to retrieve.

  • compress – If True, compress the AlchemicalNetwork server-side before shipping it to the client. This can reduce retrieval time depending on the bandwidth of your connection to the API service. Set to False to retrieve without compressing. This is a performance optimization; it has no bearing on the result of this method call.

  • visualize – If True, show retrieval progress indicator.

Returns:

The retrieved AlchemicalNetwork.

Return type:

AlchemicalNetwork

get_network_chemicalsystems(network: ScopedKey) → List[ScopedKey]

List ScopedKeys for the ChemicalSystems associated with the given AlchemicalNetwork.

get_network_failures(network: ScopedKey, compress: bool = True, visualize: bool = True) → Dict[str, List[gufe.protocols.ProtocolDAGResult]]

Get all failed ProtocolDAGResult`s for every `Transformation in the given AlchemicalNetwork.

A dict giving the ScopedKey of each Transformation in the network as keys, a list of the ProtocolDAGResults as values, is returned.

Parameters:

  • network – The ScopedKey of the AlchemicalNetwork to retrieve results for.

  • compress – If True, compress the ProtocolDAGResults server-side before shipping them to the client. This can reduce retrieval time depending on the bandwidth of your connection to the API service. Set to False to retrieve without compressing. This is a performance optimization; it has no bearing on the result of this method call.

  • visualize – If True, show retrieval progress indicators.

get_network_results(network: ScopedKey, return_protocoldagresults: bool = False, compress: bool = True, visualize: bool = True) → Dict[str, gufe.protocols.ProtocolResult | None | List[gufe.protocols.ProtocolDAGResult]]

Get a ProtocolResult for every Transformation in the given AlchemicalNetwork.

A dict giving the ScopedKey of each Transformation in the network as keys, ProtocolResult as values, is returned. If no ProtocolDAGResults exist for a given Transformation, None is given for its value.

If return_protocoldagresults is True, then a list of the ProtocolDAGResults themselves is given as values instead of ProtocolResults.

Parameters:

  • network – The ScopedKey of the AlchemicalNetwork to retrieve results for.

  • return_protocoldagresults – If True, return the raw ProtocolDAGResult`s instead of returning a processed `ProtocolResult. Only successful ProtocolDAGResults are returned.

  • compress – If True, compress the ProtocolDAGResults server-side before shipping them to the client. This can reduce retrieval time depending on the bandwidth of your connection to the API service. Set to False to retrieve without compressing. This is a performance optimization; it has no bearing on the result of this method call.

  • visualize – If True, show retrieval progress indicators.

get_network_status(network: ScopedKey, visualize: bool | None = True) → Dict[str, int]

Return status counts for all Tasks associated with the given AlchemicalNetwork.

Parameters:

  • network – ScopedKey for the Alchemicalnetwork to obtain status counts for.

  • visualize – If True, print a table of status counts.

Returns:

Dict giving statuses as keys, Task counts as values.

Return type:

status_counts

get_network_tasks(network: ScopedKey, status: str | None = None)

List ScopedKeys for all Tasks associated with the given AlchemicalNetwork.

get_network_transformations(network: ScopedKey) → List[ScopedKey]

List ScopedKeys for Transformations associated with the given AlchemicalNetwork.

get_scope_status(scope: Scope | None = None, visualize: bool | None = True) → Dict[str, int]

Return status counts for all Tasks within the given Scope.

Parameters:

  • scope – Scope to use for querying status. Non-specific Scopes are allowed, and will give back counts for all Tasks within that this user has Scope access to. Defaults to all Scopes.

  • visualize – If True, print a table of status counts.

Returns:

Dict giving statuses as keys, Task counts as values.

Return type:

status_counts

get_scoped_key(obj: gufe.tokenization.GufeTokenizable, scope: Scope) → ScopedKey

Given any gufe object and a fully-specified Scope, return corresponding ScopedKey.

This method does not check that this ScopedKey is represented in the database. It is only a convenience for properly constructing a ScopedKey from a gufe object and a Scope.

get_task_failures(task: ScopedKey, compress: bool = True, visualize: bool = True) → List[gufe.protocols.ProtocolDAGResult]

Get failed ProtocolDAGResult`s for the given `Task.

Parameters:

  • task – The ScopedKey of the Task to retrieve failures for.

  • compress – If True, compress the ProtocolDAGResults server-side before shipping them to the client. This can reduce retrieval time depending on the bandwidth of your connection to the API service. Set to False to retrieve without compressing. This is a performance optimization; it has no bearing on the result of this method call.

  • visualize – If True, show retrieval progress indicators.

get_task_networks(task: ScopedKey)

List ScopedKeys for all AlchemicalNetworks associated with the given Task.

get_task_results(task: ScopedKey, compress: bool = True, visualize: bool = True) → List[gufe.protocols.ProtocolDAGResult]

Get successful ProtocolDAGResult`s for the given `Task.

Parameters:

  • task – The ScopedKey of the Task to retrieve results for.

  • compress – If True, compress the ProtocolDAGResults server-side before shipping them to the client. This can reduce retrieval time depending on the bandwidth of your connection to the API service. Set to False to retrieve without compressing. This is a performance optimization; it has no bearing on the result of this method call.

  • visualize – If True, show retrieval progress indicators.

get_task_transformation(task: ScopedKey) → ScopedKey

Get the Transformation associated with the given Task.

get_tasks_status(tasks: List[ScopedKey], batch_size: int = 1000) → List[str]

Get the status of multiple Tasks.

Parameters:

  • tasks – The Tasks to get the status of.

  • batch_size – The number of Tasks to include in a single request; use to tune method call speed when requesting many statuses at once.

Returns:

The status of each Task in the same order as given in tasks. If a given Task doesn’t exist, None will be returned in its place.

Return type:

statuses

get_transformation(transformation: ScopedKey | str, compress: bool = True, visualize: bool = True) → gufe.Transformation

Retrieve a Transformation given its ScopedKey.

Parameters:

  • transformation – The ScopedKey of the Transformation to retrieve.

  • compress – If True, compress the Transformation server-side before shipping it to the client. This can reduce retrieval time depending on the bandwidth of your connection to the API service. Set to False to retrieve without compressing. This is a performance optimization; it has no bearing on the result of this method call.

  • visualize – If True, show retrieval progress indicator.

Returns:

The retrieved Transformation.

Return type:

Transformation

get_transformation_chemicalsystems(transformation: ScopedKey) → List[ScopedKey]

List ScopedKeys for the ChemicalSystems associated with the given Transformation.

get_transformation_failures(transformation: ScopedKey, compress: bool = True, visualize: bool = True) → List[gufe.protocols.ProtocolDAGResult]

Get failed ProtocolDAGResults for the given Transformation.

Parameters:

  • transformation – The ScopedKey of the Transformation to retrieve failures for.

  • compress – If True, compress the ProtocolDAGResults server-side before shipping them to the client. This can reduce retrieval time depending on the bandwidth of your connection to the API service. Set to False to retrieve without compressing. This is a performance optimization; it has no bearing on the result of this method call.

  • visualize – If True, show retrieval progress indicators.

get_transformation_networks(transformation: ScopedKey) → List[ScopedKey]

List ScopedKeys for AlchemicalNetworks associated with the given Transformation.

get_transformation_results(transformation: ScopedKey, return_protocoldagresults: bool = False, compress: bool = True, visualize: bool = True) → gufe.protocols.ProtocolResult | None | List[gufe.protocols.ProtocolDAGResult]

Get a ProtocolResult for the given Transformation.

A ProtocolResult object corresponds to the Protocol used for this Transformation. This is constructed from the available ProtocolDAGResults for this Transformation via Transformation.gather. If no ProtocolDAGResults exist for this Transformation, None is returned.

If return_protocoldagresults is True, then a list of the ProtocolDAGResults themselves is returned instead.

Parameters:

  • transformation – The ScopedKey of the Transformation to retrieve results for.

  • return_protocoldagresults – If True, return the raw ProtocolDAGResult`s instead of returning a processed `ProtocolResult. Only successful ProtocolDAGResults are returned.

  • compress – If True, compress the ProtocolDAGResults server-side before shipping them to the client. This can reduce retrieval time depending on the bandwidth of your connection to the API service. Set to False to retrieve without compressing. This is a performance optimization; it has no bearing on the result of this method call.

  • visualize – If True, show retrieval progress indicators.

get_transformation_status(transformation: ScopedKey, visualize: bool | None = True) → Dict[str, int]

Return status counts for all Tasks associated with the given Transformation.

Parameters:

  • transformation – ScopedKey for the Transformation to obtain status counts for.

  • visualize – If True, print a table of status counts.

Returns:

Dict giving statuses as keys, Task counts as values.

Return type:

status_counts

get_transformation_tasks(transformation: ScopedKey, extends: ScopedKey | None = None, return_as: str = 'list', status: str | None = None) → List[ScopedKey] | networkx.DiGraph

Return the Tasks associated with the given Transformation.

Parameters:

  • transformation – The ScopedKey of the Transformation to get Tasks for.

  • extends – If given, only return Tasks that extend from the given Task’s ScopedKey. This will also give any Tasks that extend from those Tasks, recursively. Using this keyword argument amounts to choosing the tree of Tasks that extend from the given Task.

  • return_as (['list', 'graph']) – If ‘list’, Tasks will be returned in no particular order. If ‘graph’, Tasks will be returned in a networkx.DiGraph, with a directed edge pointing from a given Task to the Task it extends.

query_chemicalsystems(name: str | None = None, scope: Scope | None = None) → List[ScopedKey]

Query for ChemicalSystems, optionally by name or Scope.

Calling this method with no query arguments will return ScopedKeys for all ChemicalSystems that are within the Scopes this user has access to.

query_networks(name: str | None = None, scope: Scope | None = None, return_gufe=False) → List[ScopedKey] | Dict[ScopedKey, gufe.AlchemicalNetwork]

Query for AlchemicalNetworks, optionally by name or Scope.

Calling this method with no query arguments will return ScopedKeys for all AlchemicalNetworks that are within the Scopes this user has access to.

query_tasks(scope: Scope | None = None, status: str | None = None) → List[ScopedKey]

Query for Tasks, optionally by status or Scope.

Calling this method with no query arguments will return ScopedKeys for all Tasks that are within the Scopes this user has access to.

query_transformations(name: str | None = None, scope: Scope | None = None) → List[ScopedKey]

Query for Transformations, optionally by name or Scope.

Calling this method with no query arguments will return ScopedKeys for all Transformations that are within the Scopes this user has access to.

set_strategy(network: ScopedKey, strategy: Strategy)

Set the Strategy for evaluating the given AlchemicalNetwork.

The Strategy will be applied to create and action tasks for the Transformations in the AlchemicalNetwork without user interaction.

set_tasks_status(tasks: List[ScopedKey], status: TaskStatusEnum | str, batch_size: int = 1000) → List[ScopedKey | None]

Set the status of one or multiple Tasks.

Task status can be set to ‘waiting’ if currently ‘error’. Status can be set to ‘invalid’ or ‘deleted’ from any other status.

Parameters:

  • tasks – The Tasks to set the status of.

  • status – The status to set the Tasks to. Can be one of ‘waiting’, ‘invalid’, or ‘deleted’.

Returns:

The ScopedKeys of the Tasks that were updated, in the same order as given in tasks. If a given Task doesn’t exist, None will be returned in its place.

Return type:

updated

exception alchemiscale.interface.client.AlchemiscaleClientError(*args, **kwargs)

Bases: AlchemiscaleBaseClientError