pyaqueduct.API

Aqueduct API interface to interact with experiments.

Parameters:

Name	Type	Description	Default
url	str	URL of the Aqueduct server including the prefix.	required
timeout	float	Timeout of operations in seconds.	0.5

create_experiment(title, description)

Create an experiment with specific title and description.

Parameters:

Name	Type	Description	Default
title	str	Title of the experiment.	required
description	str	Description of the experiment.	required

Returns:

Type	Description
Experiment	Experiment object to interact with its data.

find_experiments(search=None, limit=10, offset=0, tags=None, start_datetime=None, end_datetime=None)

Find the experiments that have the search criteria provided in arguments.

Parameters:

Name	Type	Description	Default
search	Optional[str]	The string to search for in the title field of experiments.	None
limit	PositiveInt	The maximum number of experiments to fetch in a single request.	10
offset	NonNegativeInt	The number of experiments to skip from the beginning of the search results.	0
tags	Optional[List[str]]	List of tags to filter the experiments by.	None
start_datetime	Optional[datetime]	Start datetime to filter the experiments after this date and time.	None
end_datetime	Optional[datetime]	End datetime to filter the experiments before this date and time.	None

Returns:

Type	Description
List[Experiment]	List of experiment objects to operate on their data.

get_experiment_by_eid(eid)

Get the experiment by the specified identifier to operate on.

Parameters:

Name	Type	Description	Default
eid	str	EID of the specified experiment.	required

Returns:

Type	Description
Experiment	Experiment object to interact with the experiment data.

get_experiment_by_uuid(uuid)

Get the experiment by the specified identifier to operate on.

Parameters:

Name	Type	Description	Default
uuid	UUID	UUID of the specified experiment.	required

Returns:

Type	Description
Experiment	Experiment object to interact with the experiment data.

get_extensions()

Gets the current fresh extension list from the server. Extension list may change without server restart.

Returns:

Type	Description
List[Extension]	List of extension objects.

get_task(task_id)

Get task by passing task_id.

Returns:

Type	Description
Task	Task object

get_tasks(limit=10, offset=0, extension_name=None, experiment_uuid=None, action_name=None, username=None, start_date=None, end_date=None)

Get task by passing task_id.

Returns:

Type	Description
List[Task]	Task object

remove_experiment_by_eid(eid)

Remove experiment from the database. Experiment's files will be also removed.

Parameters:

Name	Type	Description	Default
eid	str	EID of the specified experiment.	required

Bases: BaseModel

Experiment model.

created_at: datetime instance-attribute

Creation datetime of the experiment.

description: str property writable

Get description of experiment.

eid: str instance-attribute

EID of the experiment. User-readable identifier, it is unique within one Aqueduct installation

files: List[Tuple[str, datetime]] property

Get file names of expriment.

tags: List[str] property

Gets tags of experiment.

title: str property writable

Get title of experiment.

updated_at: datetime property

Get last updated datetime of the experiment.

uuid: UUID instance-attribute

UUID of the experiment. This is an internal experiment identifier in the database

add_tags(tags)

Add new tags to experiment.

Parameters:

Name	Type	Description	Default
tags	List[str]	List of tags to be added to the experiment.	required

download_file(file_name, destination_dir)

Download the specified file of experiment.

remove_files(files)

Remove files from experiment.

Parameters:

Name	Type	Description	Default
files	List[str]	List of file names to be removed from the experiment.	required

remove_tag(tag)

Remove tag from experiment.

upload_file(file)

Upload the specified file to experiment.

pyaqueduct.extensions

The module contains classes representing interface with extensions. Extension list may be retrieved from the server using api method API.get_extensions(). Each extension may have one or more actions. Each action has a list of expected parameters.

Extension

Bases: BaseModel

Class represents an extension as a collection of actions.

Source code in pyaqueduct/extensions.py

class Extension(BaseModel):
    """Class represents an extension as a collection of actions."""

    name: str
    """Extension name. Unique name within a server"""

    description: Optional[str]
    """Description of extension scope and overview of its actions."""

    authors: str
    """Authors of the extension."""

    actions: List[ExtensionAction]

    def __init__(
        self,
        name: str,
        description: Optional[str],
        authors: str,
        actions: List[ExtensionActionData],
        client: AqueductClient,
    ):
        super().__init__(name=name, description=description, authors=authors, actions=[])
        for action in actions:
            self.actions.append(ExtensionAction(self, action, client))

authors: str instance-attribute

Authors of the extension.

description: Optional[str] instance-attribute

Description of extension scope and overview of its actions.

name: str instance-attribute

Extension name. Unique name within a server

ExtensionAction

Bases: BaseModel

Extension action representation. Contains an execution method which trigger extension action execution on the side of Aqueduct server.

Source code in pyaqueduct/extensions.py

class ExtensionAction(BaseModel):
    """Extension action representation. Contains an execution method
    which trigger extension action execution on the side of Aqueduct server."""

    parameters: List[ExtensionParameterData]
    """List of parameters which extension action expects to accept."""

    data: ExtensionActionData
    extension: Extension = None
    """Extension to which this action belongs."""

    _client: AqueductClient = None

    def __init__(
        self,
        extension: Extension,
        action_data: ExtensionAction,
        client: AqueductClient,
    ):
        super().__init__(extension=extension, data=action_data, parameters=action_data.parameters)
        self._client = client

    @property
    def name(self) -> str:
        """Extension action name. Unique inside an extension."""
        return self.data.name

    @property
    def description(self) -> str:
        """Detailed description of the extension action."""
        return self.data.description

    @property
    def experiment_variable_name(self) -> str:
        """Name of the variable which is used to define a default experiment.
        This experiment will be used to save logs and validate variables
        of `file` type."""
        return self.data.experimentVariableName

    def execute(self, parameters: Dict[str, Any]) -> ExtensionExecutionResultData:
        """Execute an extension action on a server.

        Args:
            parameters: dictionary of parameters to pass to an extension.

        Returns:
            result of extension execution on server. `returnCode==0` corresponds to success.
        """
        return self._client.execute_extension_action(
            extension=self.extension.name,
            action=self.data.name,
            params=parameters,
        )

description: str property

Detailed description of the extension action.

experiment_variable_name: str property

Name of the variable which is used to define a default experiment. This experiment will be used to save logs and validate variables of file type.

extension: Extension = None class-attribute instance-attribute

Extension to which this action belongs.

name: str property

Extension action name. Unique inside an extension.

parameters: List[ExtensionParameterData] instance-attribute

List of parameters which extension action expects to accept.

execute(parameters)

Execute an extension action on a server.

Parameters:

Name	Type	Description	Default
parameters	Dict[str, Any]	dictionary of parameters to pass to an extension.	required

Returns:

Type	Description
ExtensionExecutionResultData	result of extension execution on server. returnCode==0 corresponds to success.

Source code in pyaqueduct/extensions.py

def execute(self, parameters: Dict[str, Any]) -> ExtensionExecutionResultData:
    """Execute an extension action on a server.

    Args:
        parameters: dictionary of parameters to pass to an extension.

    Returns:
        result of extension execution on server. `returnCode==0` corresponds to success.
    """
    return self._client.execute_extension_action(
        extension=self.extension.name,
        action=self.data.name,
        params=parameters,
    )