Workflows

The nova-galaxy library provides a Workflow class to interact with and run Galaxy workflows. This allows you to programmatically execute complex bioinformatic pipelines defined in Galaxy.

Key Concepts

  • Workflow ID: Each workflow in Galaxy has a unique ID. You’ll need this ID to instantiate a Workflow object. You can typically find this ID through the Galaxy UI or API (e.g., using galaxy_instance.workflows.get_workflows()).

  • Datastore: Workflows are run within a specific Galaxy history, which is represented by a Datastore object in nova-galaxy.

  • Parameters: Workflows often require input datasets and various parameters to control their execution. These are provided via a Parameters object.

  • Invocation: Each run of a workflow is called an “invocation”. The library manages the state and results of these invocations.

  • Outputs: Upon successful completion, a workflow produces output datasets, which can be accessed via an Outputs object.

Using the Workflow Class

The primary class for interacting with workflows is nova.galaxy.workflow.Workflow.

Initializing a Workflow

To start, you need the ID of the Galaxy workflow you want to run.

from nova.galaxy.workflow import Workflow

# Replace 'your_workflow_id' with the actual ID from Galaxy
workflow_id = "your_workflow_id"
my_workflow = Workflow(id=workflow_id)

Running a Workflow

To run the workflow, you use the run() method. This method requires a Datastore (representing the Galaxy history) and optionally a WorkflowParameters object for inputs and step-specific parameters.

from nova.galaxy.data_store import Datastore
from nova.galaxy.parameters import WorkflowParameters
from nova.galaxy.dataset import Dataset, DatasetCollection

# Assume 'galaxy_connection' is an established Connection object
# Assume 'history_id' is the ID of the target Galaxy history
data_store = Datastore(galaxy_connection, history_id=history_id)

# Prepare workflow parameters
workflow_params = WorkflowParameters()

# Example 1: Providing a dataset as a workflow-level input
# '0' is the input ID of the workflow (as defined in Galaxy)
# 'your_input_dataset_id' is the Galaxy ID of an existing dataset in the history.
input_dataset = Dataset(id="your_input_dataset_id")
workflow_params.add_workflow_input("0", input_dataset)

# Example 2: Providing a dataset collection as a workflow-level input
# '1' is another input ID of the workflow
input_collection = DatasetCollection(id="your_input_collection_id")
workflow_params.add_workflow_input("1", input_collection)

# Example 3: Setting a parameter for a specific step within the workflow
# '2' is the ID of the workflow step (as defined in Galaxy)
# 'some_tool_param' is the parameter path within that step
workflow_params.add_step_param("2", "some_tool_param", "some_value")

# Example 4: Setting a list of datasets for a parameter in a step
list_of_datasets = [Dataset(id="ds_id_1"), Dataset(id="ds_id_2")]
workflow_params.add_step_param("3", "multiple_inputs", list_of_datasets)

# Run the workflow and wait for completion (default behavior)
try:
    outputs = my_workflow.run(data_store=data_store, params=workflow_params, wait=True)
    if outputs:
        print("Workflow completed successfully!")
except Exception as e:
    print(f"Workflow execution failed: {e}")

Non-Blocking Execution

If you don’t want to wait for the workflow to complete, set wait=False.

my_workflow.run(data_store=data_store, params=params, wait=False)
print(f"Workflow started with invocation ID: {my_workflow.get_invocation_id()}")
# You'll need to check the status periodically

Checking Workflow Status

You can check the status of the last workflow invocation using get_status() or get_full_status().

from nova.common.job import WorkState

status = my_workflow.get_status()
print(f"Current workflow status: {status}")

if status == WorkState.RUNNING:
    print("Workflow is still running.")
elif status == WorkState.FINISHED:
    print("Workflow finished successfully.")
elif status == WorkState.ERROR:
    full_status = my_workflow.get_full_status()
    print(f"Workflow failed. Details: {full_status.details if full_status else 'N/A'}")

The get_status() method returns a WorkState enum member (e.g., WorkState.QUEUED, WorkState.RUNNING, WorkState.FINISHED, WorkState.ERROR).

The get_full_status() method returns an InvocationStatus object which contains both the state and a details string (useful for error messages).

Getting Workflow Results

Once a workflow has completed successfully (get_status() == WorkState.FINISHED), you can retrieve its outputs using get_results().

if my_workflow.get_status() == WorkState.FINISHED:
    outputs = my_workflow.get_results()
    if outputs:
        for output_name, dataset_or_collection in outputs.items():
            print(f"Output '{output_name}': ID {dataset_or_collection.id}")
    else:
        print("No outputs found, or an issue retrieving them.")

The get_results() method returns an Outputs object, which is a dictionary-like structure mapping output names (as defined in the workflow) to Dataset or DatasetCollection objects.

Cancelling a Workflow

If a workflow is running, you can attempt to cancel it using cancel() or its alias stop().

if my_workflow.get_status() == WorkState.RUNNING:
    was_cancelled = my_workflow.cancel()
    if was_cancelled:
        print("Workflow cancellation requested.")
    else:
        print("Failed to request workflow cancellation.")

Getting Invocation ID

Each workflow run (invocation) has a unique ID in Galaxy. You can retrieve this ID:

invocation_id = my_workflow.get_invocation_id()
if invocation_id:
    print(f"Galaxy Invocation ID: {invocation_id}")

Accessing Step-Level Tools

Workflows are composed of individual tool executions. You can access these as Tool objects using get_step_jobs(). This is useful for monitoring progress at a finer grain or retrieving logs from specific steps.

from nova.galaxy.tool import Tool

step_tools: List[Tool] = my_workflow.get_step_jobs()
for tool in step_tools:
    print(f"Step Tool ID: {tool.id}, Status: {tool.get_status()}")
    if tool.get_status() == WorkState.ERROR:
        full_tool_status = tool.get_full_status()
        print(f"  Tool Error Details: {full_tool_status.details if full_tool_status else 'N/A'}")

Important Notes

  • Workflow Definition: The structure of your WorkflowParameters object (workflow input IDs, step IDs, and parameter paths) must match how the workflow is defined in Galaxy. Use the Galaxy UI or API to inspect your workflow’s inputs and step details.

  • Dataset IDs: When providing Dataset or DatasetCollection objects as inputs, they must already exist in the Galaxy history and have their id attribute populated.

  • Error Handling: Always wrap run() calls (especially with wait=True) in try-except blocks to handle potential exceptions during workflow execution. Check get_full_status().details for more information on errors.

  • State Management: The Workflow object primarily manages the state of its last invocation. If you need to manage multiple concurrent runs of the same workflow definition, instantiate a new Workflow object for each run.

This guide provides an overview of using the Workflow class. For more detailed information on specific classes like Datastore, Parameters, Dataset, and Outputs, please refer to their respective documentation pages.