From 168d65d19077064ea20031bfef3ab680fa91ac77 Mon Sep 17 00:00:00 2001 From: John Chilton Date: Tue, 10 Mar 2026 18:40:02 -0400 Subject: [PATCH] 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 --- lib/galaxy/webapps/galaxy/api/workflows.py | 2 +- lib/galaxy/workflow/modules.py | 193 ++++++++++ lib/galaxy_test/api/test_workflows.py | 398 +++++++++++++++++++++ 3 files changed, 592 insertions(+), 1 deletion(-) diff --git a/lib/galaxy/webapps/galaxy/api/workflows.py b/lib/galaxy/webapps/galaxy/api/workflows.py index fde7a414533..e9883c4476e 100644 --- a/lib/galaxy/webapps/galaxy/api/workflows.py +++ b/lib/galaxy/webapps/galaxy/api/workflows.py @@ -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"] diff --git a/lib/galaxy/workflow/modules.py b/lib/galaxy/workflow/modules.py index 997b3957da3..dcbc983ebdb 100644 --- a/lib/galaxy/workflow/modules.py +++ b/lib/galaxy/workflow/modules.py @@ -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, ) diff --git a/lib/galaxy_test/api/test_workflows.py b/lib/galaxy_test/api/test_workflows.py index 00fd842e866..a5ffc01237c 100644 --- a/lib/galaxy_test/api/test_workflows.py +++ b/lib/galaxy_test/api/test_workflows.py @@ -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(