Implement collection references in tool request API.

Based on work in #20004.

Author: jmchilton

lib/galaxy/managers/jobs.py

@@ -91,7 +91,9 @@
     StructuredApp,
 )
 from galaxy.tool_util.parameters import (
+    DataRequestCollectionUri,
     DataRequestInternalHda,
+    DataRequestInternalHdca,
     DataRequestUri,
     dereference,
     RequestInternalDereferencedToolState,
@@ -1978,21 +1980,44 @@ def dereference(
     ) -> Tuple[RequestInternalDereferencedToolState, list[DereferencedDatasetPair]]:
         new_hdas: list[DereferencedDatasetPair] = []
 
+        def dereference_collection_callback(data_request: DataRequestCollectionUri) -> DataRequestInternalHdca:
+            # a deferred dataset corresponding to request
+            history = tool_request.history
+            if not history:
+                raise InconsistentDatabase("Tool request has no history associated")
+
+            hdca = dereference_input(trans, data_request, history)
+            assert isinstance(hdca, model.HistoryDatasetCollectionAssociation)
+
+            # we need the HDCA to have an ID - so we force a commit here - for
+            # consistency it would be great if this happened in the dereference_input
+            # since the HDA is committed in the other branch.
+            history.add_pending_items()
+            trans.sa_session.commit()
+
+            def find_new_hdas(collection: model.DatasetCollection, request_elements) -> None:
+                for dce, dce_request in zip(collection.elements, request_elements):
+                    if dce.is_collection:
+                        find_new_hdas(dce.child_collection, dce_request.elements)
+                    else:
+                        new_hdas.append(DereferencedDatasetPair(dce.hda, dce_request))
+
+            find_new_hdas(hdca.collection, data_request.elements)
+            return DataRequestInternalHdca(id=hdca.id, src="hdca")
+
         def dereference_callback(data_request: DataRequestUri) -> DataRequestInternalHda:
             # a deferred dataset corresponding to request
             history = tool_request.history
             if not history:
                 raise InconsistentDatabase("Tool request has no history associated")
 
             hda = dereference_input(trans, data_request, history)
-            if not isinstance(hda, model.HistoryDatasetAssociation):
-                raise RequestParameterInvalidException("Input dataset is not a history dataset association")
-
+            assert isinstance(hda, model.HistoryDatasetAssociation)
             new_hdas.append(DereferencedDatasetPair(hda, data_request))
             return DataRequestInternalHda(id=hda.id, src="hda")
 
         tool_state = RequestInternalToolState(tool_request.request)
-        return dereference(tool_state, tool, dereference_callback), new_hdas
+        return dereference(tool_state, tool, dereference_callback, dereference_collection_callback), new_hdas
 
     def queue_jobs(self, tool: Tool, request: QueueJobs) -> None:
         tool_request: ToolRequest = self._tool_request(request.tool_request_id)

lib/galaxy/tool_util/parameters/__init__.py

@@ -16,8 +16,10 @@
     DataCollectionRequest,
     DataParameterModel,
     DataRequest,
+    DataRequestCollectionUri,
     DataRequestHda,
     DataRequestInternalHda,
+    DataRequestInternalHdca,
     DataRequestUri,
     FloatParameterModel,
     GalaxyParameterT,
@@ -105,7 +107,9 @@
     "ToolParameterBundle",
     "ToolParameterBundleModel",
     "DataRequest",
+    "DataRequestCollectionUri",
     "DataRequestInternalHda",
+    "DataRequestInternalHdca",
     "DataRequestHda",
     "DataRequestUri",
     "DataCollectionRequest",

lib/galaxy/tool_util/parameters/convert.py

@@ -20,8 +20,10 @@
     DataCollectionRequest,
     DataColumnParameterModel,
     DataParameterModel,
+    DataRequestCollectionUri,
     DataRequestHda,
     DataRequestInternalHda,
+    DataRequestInternalHdca,
     DataRequestUri,
     DiscriminatorType,
     DrillDownParameterModel,
@@ -63,6 +65,7 @@
 DecodeFunctionT = Callable[[str], int]
 EncodeFunctionT = Callable[[int], str]
 DereferenceCallable = Callable[[DataRequestUri], DataRequestInternalHda]
+DereferenceCollectionCallable = Callable[[DataRequestCollectionUri], DataRequestInternalHdca]
 # interfaces for adapting test data dictionaries to tool request dictionaries
 # e.g. {class: File, path: foo.bed} => {src: hda, id: ab1235cdfea3}
 AdaptDatasets = Callable[[JsonTestDatasetDefDict], DataRequestHda]
@@ -213,15 +216,25 @@ def _strictify_parameters(tool_state: Dict[str, Any], input_models: ToolParamete
 
 
 def dereference(
-    internal_state: RequestInternalToolState, input_models: ToolParameterBundle, dereference: DereferenceCallable
+    internal_state: RequestInternalToolState,
+    input_models: ToolParameterBundle,
+    dereference: DereferenceCallable,
+    dereference_collection: DereferenceCollectionCallable,
 ) -> RequestInternalDereferencedToolState:
 
     def dereference_dict(src_dict: dict):
         src = src_dict.get("src")
+        clazz = src_dict.get("class")
         if src == "url":
             data_request_uri: DataRequestUri = DataRequestUri.model_validate(src_dict)
             data_request_hda: DataRequestInternalHda = dereference(data_request_uri)
             return data_request_hda.model_dump()
+        elif clazz == "Collection":
+            data_request_collection_from_uri: DataRequestCollectionUri = DataRequestCollectionUri.model_validate(
+                src_dict
+            )
+            data_request_hdca: DataRequestInternalHdca = dereference_collection(data_request_collection_from_uri)
+            return data_request_hdca.model_dump()
         else:
             return src_dict
 
@@ -234,6 +247,11 @@ def dereference_callback(parameter: ToolParameterT, value: Any):
             else:
                 assert isinstance(value, dict), str(value)
                 return dereference_dict(value)
+        elif isinstance(parameter, DataCollectionParameterModel):
+            if value is None:
+                return VISITOR_NO_REPLACEMENT
+            assert isinstance(value, dict), str(value)
+            return dereference_dict(value)
         else:
             return VISITOR_NO_REPLACEMENT
 

lib/galaxy_test/api/test_tool_execution.py

@@ -124,9 +124,10 @@ def test_execution_with_deferred_src_urls(self):
             assert input_dataset_details["state"] == "deferred", input_dataset_details
 
     @skip_without_tool("gx_data_collection_list")
-    def test_execution_with_deferred_collection(self):
+    def test_execution_with_deferred_list(self):
         with self.dataset_populator.test_history() as history_id:
-            input_b64_1 = base64.b64encode(b"1 2 3").decode("utf-8")
+            input_b64_1 = base64.b64encode(b"Hello World!\n").decode("utf-8")
+            input_b64_2 = base64.b64encode(b"It is me - a collection!\n").decode("utf-8")
             response = self._run(
                 "gx_data_collection_list",
                 history_id,
@@ -141,7 +142,14 @@ def test_execution_with_deferred_collection(self):
                                 "url": f"base64://{input_b64_1}",
                                 "ext": "txt",
                                 "deferred": True,
-                            }
+                            },
+                            {
+                                "class": "File",
+                                "identifier": "mycoolelement2",
+                                "url": f"base64://{input_b64_2}",
+                                "ext": "txt",
+                                "deferred": False,
+                            },
                         ],
                     },
                 },
@@ -164,11 +172,67 @@ def test_execution_with_deferred_collection(self):
             job_output = job_outputs[0]
             assert job_output["name"] == "output"
             content = self.dataset_populator.get_history_dataset_content(history_id, dataset=job_output["dataset"])
-            assert content == "Hello World!"
+            assert content == "Hello World!\nIt is me - a collection!\n"
 
-            # verify input was left deferred and infer must have been materialized just for the job
-            input_dataset_details = self.dataset_populator.get_history_dataset_details(history_id, hid=1)
-            assert input_dataset_details["state"] == "deferred", input_dataset_details
+    @skip_without_tool("gx_data_collection")
+    def test_execution_with_deferred_nested_list(self):
+        with self.dataset_populator.test_history() as history_id:
+            input_b64_1 = base64.b64encode(b"Hello World!\n").decode("utf-8")
+            input_b64_2 = base64.b64encode(b"It is me - a collection!\n").decode("utf-8")
+            response = self._run(
+                "gx_data_collection",
+                history_id,
+                {
+                    "parameter": {
+                        "class": "Collection",
+                        "collection_type": "list:list",
+                        "elements": [
+                            {
+                                # Why is this needed? Planemo doesn't require this class right?
+                                # wait is it needed?
+                                "class": "Collection",
+                                "identifier": "outer_element",
+                                "collection_type": "list",
+                                "elements": [
+                                    {
+                                        "class": "File",
+                                        "identifier": "mycoolelement",
+                                        "url": f"base64://{input_b64_1}",
+                                        "ext": "txt",
+                                        "deferred": True,
+                                    },
+                                    {
+                                        "class": "File",
+                                        "identifier": "mycoolelement2",
+                                        "url": f"base64://{input_b64_2}",
+                                        "ext": "txt",
+                                        "deferred": False,
+                                    },
+                                ],
+                            }
+                        ],
+                    },
+                },
+            )
+            assert_status_code_is_ok(response)
+            response_json = response.json()
+            tool_request_id = response_json.get("tool_request_id")
+            task_result = response_json["task_result"]
+            self.dataset_populator.wait_on_task_object(task_result)
+            state = self.dataset_populator.wait_on_tool_request(tool_request_id)
+            assert state, str(self.dataset_populator.get_tool_request(tool_request_id))
+            jobs = self.galaxy_interactor.jobs_for_tool_request(tool_request_id)
+            self.dataset_populator.wait_for_jobs(jobs, assert_ok=True)
+            if len(jobs) != 1:
+                raise Exception(f"Found incorrect number of jobs for tool request - was expecting a single job {jobs}")
+            assert len(jobs) == 1, jobs
+            job_id = jobs[0]["id"]
+            job_outputs = self.galaxy_interactor.job_outputs(job_id)
+            assert len(job_outputs) == 1
+            job_output = job_outputs[0]
+            assert job_output["name"] == "output"
+            content = self.dataset_populator.get_history_dataset_content(history_id, dataset=job_output["dataset"])
+            assert content == "Hello World!\nIt is me - a collection!\n"
 
     def _assert_request_validates(self, tool_id: str, history_id: str, inputs: Dict[str, Any]):
         response = self._run(tool_id, history_id, inputs)

test/functional/tools/parameters/gx_data_collection.xml

@@ -1,6 +1,6 @@
 <tool id="gx_data_collection" name="gx_data_collection" version="1.0.0">
     <command><![CDATA[
-cat '$parameter' >> '$output'
+cat #for $f in $parameter# #if $f.is_collection# #for $inner in $f# '${inner}' #end for# #else# '$f' # #end if# #end for# >> '$output'
     ]]></command>
     <inputs>
         <param name="parameter" type="data_collection" ext="data" />

test/functional/tools/parameters/gx_data_collection_list.xml

@@ -1,6 +1,6 @@
 <tool id="gx_data_collection_list" name="gx_data_collection_list" version="1.0.0">
     <command><![CDATA[
-cat '$parameter' >> '$output'
+cat #for $q in $parameter# '$q' #end for# > '$output'
     ]]></command>
     <inputs>
         <param name="parameter" type="data_collection" collection_type="list" ext="data" />