mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-24 16:30:27 +08:00
Add PickValueModule workflow module for conditional branch selection
New workflow module type that selects among conditional branch outputs. Supports 4 modes: first_non_null, first_or_skip, the_only_non_null, all_non_null. Uses N named input terminals for explicit ordering. - PickValueModule class in modules.py with execute(), null detection, skipped HDA creation, and collection creation for all_non_null - Registered in module_types dict and build_module API bypass - 9 API tests covering all modes, error cases, and ordering verification Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.6
parent
2825bb09e4
commit
168d65d190
@@ -543,7 +543,7 @@ class WorkflowsAPIController(
|
||||
module_type = payload.get("type", "tool")
|
||||
inputs = payload.get("inputs", {})
|
||||
trans.workflow_building_mode = workflow_building_modes.ENABLED
|
||||
from_tool_form = True if module_type != "data_collection_input" else False
|
||||
from_tool_form = True if module_type not in ("data_collection_input", "pick_value") else False
|
||||
if not from_tool_form and "tool_state" not in payload and "inputs" in payload:
|
||||
# tool state not sent, use the manually constructed inputs
|
||||
payload["tool_state"] = payload["inputs"]
|
||||
|
||||
@@ -41,6 +41,7 @@ from galaxy.model import (
|
||||
WorkflowStepConnection,
|
||||
)
|
||||
from galaxy.model.base import ensure_object_added_to_session
|
||||
from galaxy.objectstore import ObjectStorePopulator
|
||||
from galaxy.model.dataset_collections import matching
|
||||
from galaxy.model.dataset_collections.adapters import PromoteCollectionElementToCollectionAdapter
|
||||
from galaxy.model.dataset_collections.query import HistoryQuery
|
||||
@@ -1947,6 +1948,197 @@ class PauseModule(WorkflowModule):
|
||||
return bool(action)
|
||||
|
||||
|
||||
class PickValueModule(WorkflowModule):
|
||||
"""Workflow module that selects among conditional branch outputs.
|
||||
|
||||
Accepts N inputs from conditional steps and produces a single output
|
||||
based on the configured selection mode. Supports first_non_null,
|
||||
first_or_skip, the_only_non_null, and all_non_null modes.
|
||||
"""
|
||||
|
||||
type = "pick_value"
|
||||
name = "Pick Value"
|
||||
|
||||
MODES = ("first_non_null", "first_or_skip", "the_only_non_null", "all_non_null")
|
||||
|
||||
def get_inputs(self):
|
||||
# State managed by frontend Vue component, not backend forms
|
||||
return {}
|
||||
|
||||
def validate_state(self, inputs: dict[str, Any]) -> None:
|
||||
mode = inputs.get("mode")
|
||||
if mode and mode not in self.MODES:
|
||||
raise ValueError(f"Invalid pick_value mode: {mode}")
|
||||
|
||||
def get_export_state(self):
|
||||
return self._get_state_dict()
|
||||
|
||||
def _get_state_dict(self):
|
||||
mode = self.state.inputs.get("mode", "first_non_null")
|
||||
num_inputs = self.state.inputs.get("num_inputs", 2)
|
||||
return {"mode": mode, "num_inputs": num_inputs}
|
||||
|
||||
def save_to_step(self, step, detached=False):
|
||||
step.type = self.type
|
||||
step.tool_inputs = self._get_state_dict()
|
||||
|
||||
@property
|
||||
def _num_inputs(self):
|
||||
"""Number of input terminals — at least 2, grows with connections."""
|
||||
num_from_state = self.state.inputs.get("num_inputs", 2)
|
||||
num_from_connections = 0
|
||||
if hasattr(self, "workflow_step") and self.workflow_step:
|
||||
num_from_connections = len(self.workflow_step.input_connections_by_name)
|
||||
return max(2, num_from_state, num_from_connections)
|
||||
|
||||
def get_all_inputs(self, data_only=False, connectable_only=False):
|
||||
inputs = []
|
||||
# N connected terminals + 1 empty terminal for grow-on-connect
|
||||
for i in range(self._num_inputs + 1):
|
||||
inputs.append(
|
||||
dict(
|
||||
name=f"input_{i}",
|
||||
label=f"Input {i}",
|
||||
multiple=False,
|
||||
extensions=["input"],
|
||||
input_type="dataset",
|
||||
optional=True,
|
||||
)
|
||||
)
|
||||
return inputs
|
||||
|
||||
def get_all_outputs(self, data_only=False):
|
||||
mode = self.state.inputs.get("mode", "first_non_null")
|
||||
if mode == "all_non_null":
|
||||
return [
|
||||
dict(
|
||||
name="output",
|
||||
label="Picked values",
|
||||
extensions=["input"],
|
||||
collection=True,
|
||||
collection_type="list",
|
||||
)
|
||||
]
|
||||
return [
|
||||
dict(
|
||||
name="output",
|
||||
label="Picked value",
|
||||
extensions=["input"],
|
||||
)
|
||||
]
|
||||
|
||||
def get_runtime_state(self):
|
||||
state = DefaultToolState()
|
||||
state.inputs = {}
|
||||
return state
|
||||
|
||||
@staticmethod
|
||||
def _is_null_or_skipped(value) -> bool:
|
||||
"""Check if a replacement value represents a skipped/null output."""
|
||||
if value is NO_REPLACEMENT:
|
||||
return True
|
||||
if isinstance(value, model.HistoryDatasetAssociation):
|
||||
if value.extension == "expression.json" and value.blurb == "skipped":
|
||||
return True
|
||||
return False
|
||||
|
||||
def execute(
|
||||
self, trans, progress: "WorkflowProgress", invocation_step, use_cached_job: bool = False
|
||||
) -> Optional[bool]:
|
||||
step = invocation_step.workflow_step
|
||||
mode = step.tool_inputs.get("mode", "first_non_null") if step.tool_inputs else "first_non_null"
|
||||
|
||||
# Gather replacements from each named input terminal, in order
|
||||
replacements = []
|
||||
for input_dict in self.get_all_inputs():
|
||||
replacement = progress.replacement_for_input(trans, step, input_dict)
|
||||
if replacement is not NO_REPLACEMENT:
|
||||
replacements.append(replacement)
|
||||
|
||||
# Separate non-null from null/skipped, preserving order
|
||||
non_null = [r for r in replacements if not self._is_null_or_skipped(r)]
|
||||
|
||||
if mode == "first_non_null":
|
||||
if not non_null:
|
||||
raise FailWorkflowEvaluation(
|
||||
why=InvocationFailureExpressionEvaluationFailed(
|
||||
reason=FailureReason.expression_evaluation_failed,
|
||||
workflow_step_id=step.id,
|
||||
)
|
||||
)
|
||||
output = non_null[0]
|
||||
|
||||
elif mode == "first_or_skip":
|
||||
if not non_null:
|
||||
output = self._create_skipped_output(trans, invocation_step)
|
||||
else:
|
||||
output = non_null[0]
|
||||
|
||||
elif mode == "the_only_non_null":
|
||||
if len(non_null) != 1:
|
||||
raise FailWorkflowEvaluation(
|
||||
why=InvocationFailureExpressionEvaluationFailed(
|
||||
reason=FailureReason.expression_evaluation_failed,
|
||||
workflow_step_id=step.id,
|
||||
)
|
||||
)
|
||||
output = non_null[0]
|
||||
|
||||
elif mode == "all_non_null":
|
||||
if not non_null:
|
||||
raise FailWorkflowEvaluation(
|
||||
why=InvocationFailureExpressionEvaluationFailed(
|
||||
reason=FailureReason.expression_evaluation_failed,
|
||||
workflow_step_id=step.id,
|
||||
)
|
||||
)
|
||||
output = self._create_collection_from_list(trans, invocation_step, non_null)
|
||||
|
||||
else:
|
||||
raise ValueError(f"Unknown pick_value mode: {mode}")
|
||||
|
||||
progress.set_step_outputs(invocation_step, {"output": output})
|
||||
return None
|
||||
|
||||
def _create_skipped_output(self, trans, invocation_step):
|
||||
"""Create a skipped HDA for first_or_skip when all inputs are null."""
|
||||
invocation = invocation_step.workflow_invocation
|
||||
history = invocation.history
|
||||
hda = model.HistoryDatasetAssociation(
|
||||
name="Pick Value - skipped",
|
||||
history=history,
|
||||
create_dataset=True,
|
||||
flush=False,
|
||||
)
|
||||
object_store_populator = ObjectStorePopulator(trans.app, trans.user)
|
||||
hda.set_skipped(object_store_populator, replace_dataset=False)
|
||||
trans.sa_session.add(hda)
|
||||
return hda
|
||||
|
||||
def _create_collection_from_list(self, trans, invocation_step, hdas):
|
||||
"""Create an HDCA from a list of non-null HDAs for all_non_null mode."""
|
||||
invocation = invocation_step.workflow_invocation
|
||||
history = invocation.history
|
||||
elements = []
|
||||
for i, hda in enumerate(hdas):
|
||||
elements.append(
|
||||
dict(
|
||||
name=str(i),
|
||||
src="hda",
|
||||
id=hda.id,
|
||||
)
|
||||
)
|
||||
collection_manager = trans.app.dataset_collection_manager
|
||||
hdca = collection_manager.create(
|
||||
trans,
|
||||
history,
|
||||
name="Pick Value - all non-null",
|
||||
collection_type="list",
|
||||
element_identifiers=elements,
|
||||
)
|
||||
return hdca
|
||||
|
||||
|
||||
class ToolModule(WorkflowModule):
|
||||
type = "tool"
|
||||
name = "Tool"
|
||||
@@ -2776,6 +2968,7 @@ module_types = dict(
|
||||
data_collection_input=InputDataCollectionModule,
|
||||
parameter_input=InputParameterModule,
|
||||
pause=PauseModule,
|
||||
pick_value=PickValueModule,
|
||||
tool=ToolModule,
|
||||
subworkflow=SubWorkflowModule,
|
||||
)
|
||||
|
||||
@@ -3154,6 +3154,404 @@ test_data:
|
||||
)
|
||||
self.dataset_populator.wait_for_history(history_id=history_id, assert_ok=True)
|
||||
|
||||
def test_pick_value_first_non_null(self):
|
||||
with self.dataset_populator.test_history() as history_id:
|
||||
summary = self._run_workflow(
|
||||
"""class: GalaxyWorkflow
|
||||
inputs:
|
||||
input_data:
|
||||
type: data
|
||||
steps:
|
||||
branch_a:
|
||||
tool_id: cat1
|
||||
in:
|
||||
input1: input_data
|
||||
when: $(true)
|
||||
branch_b:
|
||||
tool_id: cat1
|
||||
in:
|
||||
input1: input_data
|
||||
when: $(false)
|
||||
pick:
|
||||
type: pick_value
|
||||
state:
|
||||
mode: first_non_null
|
||||
in:
|
||||
input_0: branch_a/out_file1
|
||||
input_1: branch_b/out_file1
|
||||
outputs:
|
||||
picked:
|
||||
outputSource: pick/output
|
||||
""",
|
||||
test_data="""
|
||||
input_data:
|
||||
value: 1.bed
|
||||
type: File
|
||||
""",
|
||||
history_id=history_id,
|
||||
)
|
||||
invocation = self.workflow_populator.get_invocation(summary.invocation_id, step_details=True)
|
||||
output_details = self.dataset_populator.get_history_dataset_details(
|
||||
history_id, content_id=invocation["outputs"]["picked"]["id"]
|
||||
)
|
||||
assert output_details["state"] == "ok"
|
||||
|
||||
def test_pick_value_first_non_null_error_all_null(self):
|
||||
with self.dataset_populator.test_history() as history_id:
|
||||
summary = self._run_workflow(
|
||||
"""class: GalaxyWorkflow
|
||||
inputs:
|
||||
input_data:
|
||||
type: data
|
||||
steps:
|
||||
branch_a:
|
||||
tool_id: cat1
|
||||
in:
|
||||
input1: input_data
|
||||
when: $(false)
|
||||
branch_b:
|
||||
tool_id: cat1
|
||||
in:
|
||||
input1: input_data
|
||||
when: $(false)
|
||||
pick:
|
||||
type: pick_value
|
||||
state:
|
||||
mode: first_non_null
|
||||
in:
|
||||
input_0: branch_a/out_file1
|
||||
input_1: branch_b/out_file1
|
||||
outputs:
|
||||
picked:
|
||||
outputSource: pick/output
|
||||
""",
|
||||
test_data="""
|
||||
input_data:
|
||||
value: 1.bed
|
||||
type: File
|
||||
""",
|
||||
history_id=history_id,
|
||||
assert_ok=False,
|
||||
wait=True,
|
||||
)
|
||||
invocation = self.workflow_populator.get_invocation(summary.invocation_id, step_details=True)
|
||||
assert invocation["state"] == "failed"
|
||||
|
||||
def test_pick_value_first_or_skip(self):
|
||||
with self.dataset_populator.test_history() as history_id:
|
||||
summary = self._run_workflow(
|
||||
"""class: GalaxyWorkflow
|
||||
inputs:
|
||||
input_data:
|
||||
type: data
|
||||
steps:
|
||||
branch_a:
|
||||
tool_id: cat1
|
||||
in:
|
||||
input1: input_data
|
||||
when: $(true)
|
||||
branch_b:
|
||||
tool_id: cat1
|
||||
in:
|
||||
input1: input_data
|
||||
when: $(false)
|
||||
pick:
|
||||
type: pick_value
|
||||
state:
|
||||
mode: first_or_skip
|
||||
in:
|
||||
input_0: branch_a/out_file1
|
||||
input_1: branch_b/out_file1
|
||||
outputs:
|
||||
picked:
|
||||
outputSource: pick/output
|
||||
""",
|
||||
test_data="""
|
||||
input_data:
|
||||
value: 1.bed
|
||||
type: File
|
||||
""",
|
||||
history_id=history_id,
|
||||
)
|
||||
invocation = self.workflow_populator.get_invocation(summary.invocation_id, step_details=True)
|
||||
output_details = self.dataset_populator.get_history_dataset_details(
|
||||
history_id, content_id=invocation["outputs"]["picked"]["id"]
|
||||
)
|
||||
assert output_details["state"] == "ok"
|
||||
|
||||
def test_pick_value_first_or_skip_all_null(self):
|
||||
with self.dataset_populator.test_history() as history_id:
|
||||
summary = self._run_workflow(
|
||||
"""class: GalaxyWorkflow
|
||||
inputs:
|
||||
input_data:
|
||||
type: data
|
||||
steps:
|
||||
branch_a:
|
||||
tool_id: cat1
|
||||
in:
|
||||
input1: input_data
|
||||
when: $(false)
|
||||
branch_b:
|
||||
tool_id: cat1
|
||||
in:
|
||||
input1: input_data
|
||||
when: $(false)
|
||||
pick:
|
||||
type: pick_value
|
||||
state:
|
||||
mode: first_or_skip
|
||||
in:
|
||||
input_0: branch_a/out_file1
|
||||
input_1: branch_b/out_file1
|
||||
outputs:
|
||||
picked:
|
||||
outputSource: pick/output
|
||||
""",
|
||||
test_data="""
|
||||
input_data:
|
||||
value: 1.bed
|
||||
type: File
|
||||
""",
|
||||
history_id=history_id,
|
||||
)
|
||||
invocation = self.workflow_populator.get_invocation(summary.invocation_id, step_details=True)
|
||||
output_details = self.dataset_populator.get_history_dataset_details(
|
||||
history_id, content_id=invocation["outputs"]["picked"]["id"]
|
||||
)
|
||||
assert output_details["extension"] == "expression.json"
|
||||
assert output_details["misc_blurb"] == "skipped"
|
||||
|
||||
def test_pick_value_the_only_non_null(self):
|
||||
with self.dataset_populator.test_history() as history_id:
|
||||
summary = self._run_workflow(
|
||||
"""class: GalaxyWorkflow
|
||||
inputs:
|
||||
input_data:
|
||||
type: data
|
||||
steps:
|
||||
branch_a:
|
||||
tool_id: cat1
|
||||
in:
|
||||
input1: input_data
|
||||
when: $(true)
|
||||
branch_b:
|
||||
tool_id: cat1
|
||||
in:
|
||||
input1: input_data
|
||||
when: $(false)
|
||||
pick:
|
||||
type: pick_value
|
||||
state:
|
||||
mode: the_only_non_null
|
||||
in:
|
||||
input_0: branch_a/out_file1
|
||||
input_1: branch_b/out_file1
|
||||
outputs:
|
||||
picked:
|
||||
outputSource: pick/output
|
||||
""",
|
||||
test_data="""
|
||||
input_data:
|
||||
value: 1.bed
|
||||
type: File
|
||||
""",
|
||||
history_id=history_id,
|
||||
)
|
||||
invocation = self.workflow_populator.get_invocation(summary.invocation_id, step_details=True)
|
||||
output_details = self.dataset_populator.get_history_dataset_details(
|
||||
history_id, content_id=invocation["outputs"]["picked"]["id"]
|
||||
)
|
||||
assert output_details["state"] == "ok"
|
||||
|
||||
def test_pick_value_the_only_non_null_error_multiple(self):
|
||||
with self.dataset_populator.test_history() as history_id:
|
||||
summary = self._run_workflow(
|
||||
"""class: GalaxyWorkflow
|
||||
inputs:
|
||||
input_data:
|
||||
type: data
|
||||
steps:
|
||||
branch_a:
|
||||
tool_id: cat1
|
||||
in:
|
||||
input1: input_data
|
||||
when: $(true)
|
||||
branch_b:
|
||||
tool_id: cat1
|
||||
in:
|
||||
input1: input_data
|
||||
when: $(true)
|
||||
pick:
|
||||
type: pick_value
|
||||
state:
|
||||
mode: the_only_non_null
|
||||
in:
|
||||
input_0: branch_a/out_file1
|
||||
input_1: branch_b/out_file1
|
||||
outputs:
|
||||
picked:
|
||||
outputSource: pick/output
|
||||
""",
|
||||
test_data="""
|
||||
input_data:
|
||||
value: 1.bed
|
||||
type: File
|
||||
""",
|
||||
history_id=history_id,
|
||||
assert_ok=False,
|
||||
wait=True,
|
||||
)
|
||||
invocation = self.workflow_populator.get_invocation(summary.invocation_id, step_details=True)
|
||||
assert invocation["state"] == "failed"
|
||||
|
||||
def test_pick_value_all_non_null(self):
|
||||
with self.dataset_populator.test_history() as history_id:
|
||||
summary = self._run_workflow(
|
||||
"""class: GalaxyWorkflow
|
||||
inputs:
|
||||
input_data:
|
||||
type: data
|
||||
steps:
|
||||
branch_a:
|
||||
tool_id: cat1
|
||||
in:
|
||||
input1: input_data
|
||||
when: $(true)
|
||||
branch_b:
|
||||
tool_id: cat1
|
||||
in:
|
||||
input1: input_data
|
||||
when: $(true)
|
||||
branch_c:
|
||||
tool_id: cat1
|
||||
in:
|
||||
input1: input_data
|
||||
when: $(false)
|
||||
pick:
|
||||
type: pick_value
|
||||
state:
|
||||
mode: all_non_null
|
||||
in:
|
||||
input_0: branch_a/out_file1
|
||||
input_1: branch_b/out_file1
|
||||
input_2: branch_c/out_file1
|
||||
outputs:
|
||||
picked:
|
||||
outputSource: pick/output
|
||||
""",
|
||||
test_data="""
|
||||
input_data:
|
||||
value: 1.bed
|
||||
type: File
|
||||
""",
|
||||
history_id=history_id,
|
||||
)
|
||||
invocation = self.workflow_populator.get_invocation(summary.invocation_id, step_details=True)
|
||||
output_collection = self.dataset_populator.get_history_collection_details(
|
||||
history_id, content_id=invocation["output_collections"]["picked"]["id"]
|
||||
)
|
||||
assert output_collection["collection_type"] == "list"
|
||||
assert len(output_collection["elements"]) == 2
|
||||
|
||||
def test_pick_value_first_non_null_ordering(self):
|
||||
"""Verify first_non_null picks input_0 over input_1 when both are non-null."""
|
||||
with self.dataset_populator.test_history() as history_id:
|
||||
summary = self._run_workflow(
|
||||
"""class: GalaxyWorkflow
|
||||
inputs:
|
||||
input_a:
|
||||
type: data
|
||||
input_b:
|
||||
type: data
|
||||
steps:
|
||||
branch_a:
|
||||
tool_id: cat1
|
||||
in:
|
||||
input1: input_a
|
||||
when: $(true)
|
||||
branch_b:
|
||||
tool_id: cat1
|
||||
in:
|
||||
input1: input_b
|
||||
when: $(true)
|
||||
pick:
|
||||
type: pick_value
|
||||
state:
|
||||
mode: first_non_null
|
||||
in:
|
||||
input_0: branch_a/out_file1
|
||||
input_1: branch_b/out_file1
|
||||
outputs:
|
||||
picked:
|
||||
outputSource: pick/output
|
||||
""",
|
||||
test_data="""
|
||||
input_a:
|
||||
value: 1.bed
|
||||
type: File
|
||||
input_b:
|
||||
value: 2.bed
|
||||
type: File
|
||||
""",
|
||||
history_id=history_id,
|
||||
)
|
||||
invocation = self.workflow_populator.get_invocation(summary.invocation_id, step_details=True)
|
||||
picked_content = self.dataset_populator.get_history_dataset_content(
|
||||
history_id, content_id=invocation["outputs"]["picked"]["id"]
|
||||
)
|
||||
input_a_content = open(self.test_data_resolver.get_filename("1.bed")).read()
|
||||
assert picked_content == input_a_content
|
||||
|
||||
def test_pick_value_first_non_null_ordering_skipped_first(self):
|
||||
"""Verify first_non_null skips null input_0 and picks input_1."""
|
||||
with self.dataset_populator.test_history() as history_id:
|
||||
summary = self._run_workflow(
|
||||
"""class: GalaxyWorkflow
|
||||
inputs:
|
||||
input_a:
|
||||
type: data
|
||||
input_b:
|
||||
type: data
|
||||
steps:
|
||||
branch_a:
|
||||
tool_id: cat1
|
||||
in:
|
||||
input1: input_a
|
||||
when: $(false)
|
||||
branch_b:
|
||||
tool_id: cat1
|
||||
in:
|
||||
input1: input_b
|
||||
when: $(true)
|
||||
pick:
|
||||
type: pick_value
|
||||
state:
|
||||
mode: first_non_null
|
||||
in:
|
||||
input_0: branch_a/out_file1
|
||||
input_1: branch_b/out_file1
|
||||
outputs:
|
||||
picked:
|
||||
outputSource: pick/output
|
||||
""",
|
||||
test_data="""
|
||||
input_a:
|
||||
value: 1.bed
|
||||
type: File
|
||||
input_b:
|
||||
value: 2.bed
|
||||
type: File
|
||||
""",
|
||||
history_id=history_id,
|
||||
)
|
||||
invocation = self.workflow_populator.get_invocation(summary.invocation_id, step_details=True)
|
||||
picked_content = self.dataset_populator.get_history_dataset_content(
|
||||
history_id, content_id=invocation["outputs"]["picked"]["id"]
|
||||
)
|
||||
input_b_content = open(self.test_data_resolver.get_filename("2.bed")).read()
|
||||
assert picked_content == input_b_content
|
||||
|
||||
def test_run_workflow_simple_conditional_step(self):
|
||||
with self.dataset_populator.test_history() as history_id:
|
||||
summary = self._run_workflow(
|
||||
|
||||
Reference in New Issue
Block a user