API Reference

class nova.galaxy.BasicTool[source]

Base tool class.

Provides methods required for tool runner to handle the tool execution.

abstractmethod get_results(tool: Tool) → bytes[source]: Get tool results as bytes.

abstractmethod prepare_tool() → Tuple[Tool, Parameters | None][source]: Prepare tool to run.

set_store(store: Datastore) → None[source]: Set datastore for the tool.

validate_for_run() → None[source]: Validate tool inputs.

class nova.galaxy.Connection(galaxy_url: str, galaxy_key: str)[source]

Class to manage a connection to the NDIP platform.

galaxy_url(str)

Type:: URL of the Galaxy instance.

galaxy_api_key(str)

Type:: API key for the Galaxy instance.

connect() → ConnectionHelper[source]

Connects to the Galaxy instance using the provided URL and API key.

Raises a ValueError if the URL or API key is not provided.

Raises:: ValueError – If the Galaxy URL or API key is not provided.:

class nova.galaxy.Dataset(path: str = '', name: str | None = None, remote_file: bool = False, force_upload: bool = True)[source]

Singular file that can be uploaded and used in a Galaxy tool.

If needing to change the path of the Dataset, it is recommended to create a new Dataset instead.

Parameters:

path (str) – The path to the file that this dataset is representing. Can be left blank if manually providing content.
name (Optional[str]) – The name of this dataset. Defaults to the filename from the path if provided.
remote_file (bool) – Whether this file is a remote file that upstream has access to. Defaults to False (local file).
force_upload (bool) – Whether to explicitly upload this dataset every time despite another dataset with the same name existing upstream. If False, Nova Galaxy will attempt to link this dataset with an upstream copy. Defaults to True.

download(local_path: str) → AbstractData[source]: Downloads this dataset to the local path given.

get_content() → Any[source]

Get the content of this dataset.

If the content is not already present in memory, this method will download and/or load the file content into memory. If not careful, this can cause performance issues with large datasets. For larger files, consider using the download() method and writing the file to a local path.

set_content(content: Any, file_type: str = '') → None[source]

Directly set the content of this dataset.

Use this method if instead of having a dataset load from a file, you want to directly pass in content. Note, the content must be able to be serialized as a string in order to facilitate the uploading process.

upload(store: Datastore, name: str | None = None) → None[source]

Uploads this dataset to the data store given.

This method will automatically set the id, and store class variables for future use.

Parameters:

store (Datastore) – The data store to upload this dataset to.
name (Optional[str]) – The name that will be used for the dataset upstream. Defaults to the local name.

class nova.galaxy.DatasetCollection(path: str, name: str | None = None)[source]

A group of files that can be uploaded as a collection and collectively be used in a Galaxy tool.

download(local_path: str) → AbstractData[source]: Downloads this dataset collection to the local path given.

get_content() → Any[source]: Get a list of the content of this Collection along with info on each element.

upload(store: Datastore) → None[source]: Will need to handle this differently than single datasets.

class nova.galaxy.Datastore(name: str, nova_connection: ConnectionHelper, history_id: str)[source]

Groups tool outputs together.

The constructor is not intended for external use. Use nova.galaxy.Connection.create_data_store() instead.

mark_for_cleanup() → None[source]: Clean up and delete all content related to this Data Store after the associated connection is closed.

persist() → None[source]

Persist this store even after the nova connection is closed.

Should be used carefully as tools will continue to run after even if this object is garbage collected. Use recover_tools() to with the same data store name to retrieve all running tools again.

recover_tools(filter_running: bool = True) → List[Tool][source]

Recovers all running tools in this data_store.

Mainly used to recover all the running tools inside of this data store or any past persisted data stores that used the same name. Can also be used to simply get a list of all running tools in a store as well.

Parameters:: filter_running (bool) – If this should only recover tools that are running (true).
Return type:: List of tools from this data store.

class nova.galaxy.Outputs[source]: Contains the output datasets and collections for a Tool.

class nova.galaxy.Parameters[source]: Specialized map wrapper used as an input to a Galaxy tool.

class nova.galaxy.Tool(id: str)[source]

Represents a tool from Galaxy that can be run.

It’s recommended to create a new Tool object every time you want to run a tool to prevent results from being overridden.

assign_id(new_id: str, data_store: Datastore) → None[source]

Assigns an id to this tool.

Assigns this tool a new id, so that it can track already existing tools. Useful for recovering old tools if you have kept track of the id.

Parameters:

new_id (str) – The new id to assign to this tool.
data_store (Datastore) – The datastore in which the tool should be tracked.

cancel() → None[source]: Cancels the tool execution and gets rid of any results collected.

get_full_status() → JobStatus[source]

Returns the full status and state of the tool. Includes any error or warning messages.

Raises:: Exception – If the job has not been started.
Returns:: Returns the full status of the tool including the WorkState and any error messages set by the tool.
Return type:: JobStatus

get_results() → Outputs | None[source]

Returns the results from running this tool.

Throws an Exception if the tool has not finished yet. Will be overridden if this tool is run again.

Returns:: An instance of Outputs that holds the datasets and collections from this tool execution if it is finished.
Return type:: Optional[Outputs]

get_status() → WorkState[source]

Returns the current status of the tool.

Returns:: Returns the status of the tool which will be a WorkState value.
Return type:: WorkState

get_stderr(position: int | None = None, length: int | None = None) → str | None[source]

Get the current STDERR for a tool.

Will be overridden everytime this tool is run.

Parameters:

position (int, optional) – The starting position from which to get the stderr. Useful if not wanting to fetch the entire stderr file.
length (int, optional) – How many characters of stdout to read.

Returns:

Returns the current STDERR of the tool if it is running or finished.

Return type:

Optional[str]

get_stdout(position: int | None = None, length: int | None = None) → str | None[source]

Get the current STDOUT for a tool.

Will be overridden everytime this tool is run.

Parameters:

position (int, optional) – The starting position from which to get the stdout. Useful if not wanting to fetch the entire stdout file.
length (int, optional) – How many characters of stdout to read.

Returns:

Returns the current STDOUT of the tool if it is running or finished.

Return type:

Optional[str]

get_uid() → str | None[source]

Get the unique ID for this tool.

Will only be available if Tool.run() has been successfully invoked.

Returns:: Returns the uid of this tool if it is running or finished.
Return type:: Optional[str]

get_url(max_tries: int = 5, check_url: bool = False) → str | None[source]

Get the URL for this tool.

If this is an interactive tool, then will return the endpoint to the tool. The first call may need to query Galaxy, but the url is cached for subsequent calls.

Parameters:

max_tries (int) – How many attempts to obtain the url.
check_url (bool) – Whether to check the URL for a 200 response before returning. If the request is unsuccessful, returns None.

run(data_store: Datastore, params: Parameters | None = None, wait: bool = True) → Outputs | None[source]

Run this tool.

By default, will be run in a blocking manner, unless wait is set to False. Will return the results as an instance of the Outputs class from nova.galaxy.outputs if run in a blocking way. Otherwise, will return None, and the user will be responsible for getting results by calling get_results.

Parameters:

data_store (Datastore) – The data store to run this tool in.
params (Parameters) – The input parameters for this tool.
wait (bool) – Whether to run this tool in a blocking manner (True) or not (False). Default is True.

Returns:

If run in a blocking manner, returns the Outputs once the tool is finished running. Otherwise, returns None.

Return type:

Optional[Outputs]

run_interactive(data_store: Datastore, params: Parameters | None = None, wait: bool = True, max_tries: int = 100, check_url: bool = True) → str | None[source]

Run tool interactively.

Interactive Tools typically are run exclusively with this method. Can poll for the interactive tool endpoint before returning, ensuring that the tool is reachable.

Parameters:

data_store (Datastore) – The data store to run this tool in.
params (Parameters) – The input parameters for this tool.
wait (bool) – Whether to wait for the interactive tool to start up before returning.
max_tries (int) – Timeout for how long to poll for the interactive tool endpoint.
check_url – Whether to check if the interactive tool endpoint is reachable before returning.

Returns:

Will return None, if not waiting for interactive tool to startup with wait parameter. Will return the URL to the interactive tool otherwise.

Return type:

Optional[str]

stop() → None[source]: Stop the tool, but keep any existing results.

wait_for_results() → None[source]: Wait for this Tool to finish running.

class nova.galaxy.ToolRunner(id: str, tool: BasicTool, store_factory: Callable[[], str], galaxy_url: str, galaxy_api_key: str)[source]

Class that manages running tools.

This class is responsible for managing the execution of tools in Galaxy.

Parameters:

id (str) – Unique identifier for the tool runner instance. Should be the same in all GUI components from nov a-trame libraray so that events could be passed between the Tool Runner and components
tool (BasicTool) – An instance of the tool to be executed.
store_factory (StoreFactoryFunction) – A factory function that returns a name of a store the tool should run with
galaxy_url (str) – The URL of the Galaxy server to interact with.
galaxy_api_key (str) – API key used for authentication with the Galaxy server.

class nova.galaxy.Workflow(id: str)[source]

Represents a Galaxy workflow that can be invoked (run).

It’s recommended to create a new Workflow object for each invocation to prevent state conflicts if run multiple times.

cancel() → bool[source]

Cancels the currently running workflow invocation.

Returns:: True if cancellation was successfully requested, False otherwise.
Return type:: bool

get_active_step() → Tool | None[source]

Gets the currently active (running) step in the workflow invocation.

This method iterates through all jobs associated with the workflow steps and returns the first one found to be in the ‘RUNNING’ state.

Returns:: The Job instance representing the currently running step. Returns None if no step is currently running, if the workflow hasn’t been run yet, or if step jobs cannot be retrieved.
Return type:: Optional[“Job”]

get_full_status() → InvocationStatus | None[source]

Returns the full status object of the last workflow invocation.

Returns:: The InvocationStatus object containing state and details. Returns None if run has not been called yet.
Return type:: Optional[InvocationStatus]

get_invocation_id() → str | None[source]

Gets the Galaxy invocation ID for the last run.

Returns:: The invocation ID if run() has been called, otherwise None.
Return type:: Optional[str]

get_results() → Outputs | None[source]

Returns the results from the last completed workflow invocation.

Should only be called after the workflow has finished successfully (i.e., get_status() returns FINISHED).

Returns:: An Outputs object containing the workflow results if finished, otherwise None.
Return type:: Optional[Outputs]
Raises:: Exception – If called before the invocation is finished, or if there was an error fetching or processing results.

get_status() → WorkState[source]

Returns the current status of the last workflow invocation.

Returns:: The current state (e.g., QUEUED, RUNNING, FINISHED, ERROR). Returns NOT_STARTED if run has not been called yet.
Return type:: WorkState

get_step_jobs(running_only: bool = True) → List[Tool][source]

Gets nova-galaxy Job instances for each step in the workflow.

Returns the individual jobs that make up the workflow steps, allowing access to step-level status, outputs, and console logs.

Parameters:: running_only (Optional[bool]) – A boolean that determines whether or not to return only jobs which are currently running.
Returns:: List of Job instances representing workflow steps. Returns empty list if workflow hasn’t been run yet.
Return type:: List[Job]

Examples

>>> workflow = Workflow("workflow_id")
>>> workflow.run(data_store, params, wait=False)
>>> jobs = workflow.get_step_jobs()
>>> for job in jobs:
...     print(f"Step {job.tool}: {job.status.state}")
...     if job.status.state == WorkState.RUNNING:
...         console = job.get_console_output(0, 1000)
...         print(console.get('stdout', ''))

get_step_name(step_number: int) → str[source]

Gets the name of the step in the workflow.

Returns the string of the name of the step associated with the number.

Returns:: Name of the step as declared in Galaxy. Empty if step doesn’t exist.
Return type:: str

run(data_store: Datastore, params: WorkflowParameters | None = None, wait: bool = True) → Outputs | None[source]

Invokes (runs) this workflow in the specified data store.

By default, runs in a blocking manner (waits for completion). Set wait=False for non-blocking execution.

Parameters:

data_store (Datastore) – The data store (history) where the workflow will be invoked.
params (Optional[Parameters]) – The input parameters and datasets for the workflow. The structure of these parameters needs to align with how the workflow expects inputs (e.g., keyed by step labels or IDs). See Invocation.submit for details.
wait (bool, optional) – If True (default), wait for the workflow invocation to complete before returning. If False, start the invocation and return None immediately.

Returns:

If wait is True and the invocation completes successfully, returns an Outputs object containing the workflow results. If wait is False or the invocation fails, returns None.

Return type:

Optional[Outputs]

Raises:

Exception – If the workflow is already running or if an error occurs during execution when wait is True.

stop() → bool[source]

Stops (cancels) the currently running workflow invocation.

Alias for cancel().

wait_for_results() → None[source]

Waits on the workflow to complete.

This method will wait for a running work to complete

class nova.galaxy.WorkflowParameters[source]

Handles workflow parameters using explicit bioblend-style approach.

add_step_param(step_id: str, param_path: str, value: Any) → None[source]

Add a step-level parameter.

Parameters:

step_id (str) – The workflow step ID (e.g., “2”, “4”)
param_path (str) – The parameter path within the step (e.g., “input”, “series_0|input_mode|export_folder”)
value (Any) – The parameter value

add_workflow_input(input_id: str, value: Any) → None[source]

Add a workflow-level input.

Parameters:

input_id (str) – The workflow input ID (e.g., “0”, “1”)
value (Any) – The input value (Dataset, DatasetCollection, or simple value)

get_bioblend_inputs() → Dict[str, Any][source]: Get the workflow inputs in bioblend format.

get_bioblend_params() → Dict[str, Dict[str, Any]][source]: Get the step parameters in bioblend format.