Merge pull request #20004 from mvdbeek/dataset_collection_support_in_landing_request

Implement dataset collection support in workflow landing requests

lib/galaxy/celery/tasks.py

@@ -112,10 +112,11 @@ def purge_datasets(
 def materialize(
     hda_manager: HDAManager,
     request: MaterializeDatasetInstanceTaskRequest,
+    sa_session: galaxy_scoped_session,
     task_user_id: Optional[int] = None,
 ):
     """Materialize datasets using HDAManager."""
-    hda_manager.materialize(request)
+    hda_manager.materialize(request, sa_session())
 
 
 @galaxy_task(action="set metadata for job")

lib/galaxy/exceptions/utils.py

@@ -1,4 +1,44 @@
-from galaxy.exceptions import error_codes
+from typing import (
+    TYPE_CHECKING,
+    Union,
+)
+
+from galaxy.exceptions import (
+    error_codes,
+    MessageException,
+    RequestParameterInvalidException,
+    RequestParameterMissingException,
+)
+from galaxy.util.json import safe_dumps
+
+if TYPE_CHECKING:
+    from fastapi.exceptions import RequestValidationError
+    from pydantic import ValidationError
+
+
+def validation_error_to_message_exception(e: Union["ValidationError", "RequestValidationError"]) -> MessageException:
+    invalid_found = False
+    missing_found = False
+    messages = []
+    clean_validation_errors = []
+    for error in e.errors():
+        messages.append(f"{error['msg']} in {error['loc']}")
+        if error["type"] == "message_exception" and "ctx" in error:
+            return error["ctx"]["exception"]
+        if error["type"] == "missing" or error["type"] == "type_error.none.not_allowed":
+            missing_found = True
+        elif error["type"].startswith("type_error"):
+            invalid_found = True
+        # ctx contains data that can't be serialized, like exception instances
+        error.pop("ctx", None)
+        try:
+            clean_validation_errors.append(safe_dumps(error))
+        except TypeError:
+            pass
+    if missing_found and not invalid_found:
+        return RequestParameterMissingException("\n".join(messages), validation_errors=clean_validation_errors)
+    else:
+        return RequestParameterInvalidException("\n".join(messages), validation_errors=clean_validation_errors)
 
 
 def api_error_to_dict(**kwds):

lib/galaxy/managers/hdas.py

@@ -14,6 +14,7 @@ from typing import (
     List,
     Optional,
     Set,
+    Union,
 )
 
 from sqlalchemy import (
@@ -28,6 +29,7 @@ from sqlalchemy import (
     select,
     true,
 )
+from sqlalchemy.orm import Session
 from sqlalchemy.orm.session import object_session
 
 from galaxy import (
@@ -47,12 +49,16 @@ from galaxy.managers import (
 from galaxy.managers.context import ProvidesHistoryContext
 from galaxy.model import (
     HistoryDatasetAssociation,
+    HistoryDatasetCollectionAssociation,
     Job,
     JobStateHistory,
     JobToOutputDatasetAssociation,
 )
 from galaxy.model.deferred import materializer_factory
-from galaxy.model.dereference import dereference_to_model
+from galaxy.model.dereference import (
+    derefence_collection_to_model,
+    dereference_to_model,
+)
 from galaxy.schema.schema import DatasetSourceType
 from galaxy.schema.storage_cleaner import (
     CleanableItemsSummary,
@@ -70,7 +76,11 @@ from galaxy.structured_app import (
     MinimalManagerApp,
     StructuredApp,
 )
-from galaxy.tool_util.parameters import DataRequestUri
+from galaxy.tool_util_models.parameters import (
+    DataRequestCollectionUri,
+    DataRequestUri,
+    FileRequestUri,
+)
 from galaxy.util.compression_utils import get_fileobj
 
 log = logging.getLogger(__name__)
@@ -160,13 +170,15 @@ class HDAManager(
             session.commit()
         return hda
 
-    def materialize(self, request: MaterializeDatasetInstanceTaskRequest, in_place: bool = False) -> bool:
+    def materialize(
+        self, request: MaterializeDatasetInstanceTaskRequest, session: Session, in_place: bool = False
+    ) -> bool:
         request_user: RequestUser = request.user
         materializer = materializer_factory(
             True,  # attached...
             object_store=self.app.object_store,
             file_sources=self.app.file_sources,
-            sa_session=self.app.model.session(),
+            sa_session=session,
         )
         user = self.user_manager.by_id(request_user.user_id)
         if request.source == DatasetSourceType.hda:
@@ -179,7 +191,6 @@ class HDAManager(
             history.add_dataset(new_hda, set_hid=True)
         else:
             new_hda.set_total_size()
-        session = self.session()
         session.commit()
         return new_hda.is_ok
 
@@ -332,11 +343,17 @@ class HDAManager(
 
 
 def dereference_input(
-    trans: ProvidesHistoryContext, data_request: DataRequestUri, history: Optional[model.History] = None
-) -> HistoryDatasetAssociation:
-    target_history = history or trans.history
-    hda = dereference_to_model(trans.sa_session, trans.user, target_history, data_request)
-    permissions = trans.app.security_agent.history_get_default_permissions(target_history)
+    trans: ProvidesHistoryContext,
+    data_request: Union[DataRequestUri, FileRequestUri, DataRequestCollectionUri],
+    history: model.History,
+) -> Union[HistoryDatasetAssociation, HistoryDatasetCollectionAssociation]:
+    permissions = trans.app.security_agent.history_get_default_permissions(history)
+    if isinstance(data_request, DataRequestCollectionUri):
+        hdca = derefence_collection_to_model(trans.sa_session, trans.user, history, data_request)
+        for hda in hdca.dataset_instances:
+            trans.app.security_agent.set_all_dataset_permissions(hda.dataset, permissions, new=True, flush=False)
+        return hdca
+    hda = dereference_to_model(trans.sa_session, trans.user, history, data_request)
     trans.app.security_agent.set_all_dataset_permissions(hda.dataset, permissions, new=True, flush=False)
     trans.sa_session.commit()
     return hda

lib/galaxy/model/__init__.py

@@ -8835,7 +8835,9 @@ class InputWithRequest:
 @dataclass
 class InputToMaterialize:
     hda: "HistoryDatasetAssociation"
-    input_dataset: "WorkflowRequestToInputDatasetAssociation"
+    input_dataset: Union[
+        "WorkflowRequestToInputDatasetAssociation", "WorkflowRequestToInputDatasetCollectionAssociation"
+    ]
 
 
 class WorkflowInvocation(Base, UsesCreateAndUpdateTime, Dictifiable, Serializable):
@@ -8855,8 +8857,10 @@ class WorkflowInvocation(Base, UsesCreateAndUpdateTime, Dictifiable, Serializabl
     input_parameters = relationship("WorkflowRequestInputParameter", back_populates="workflow_invocation")
     step_states = relationship("WorkflowRequestStepState", back_populates="workflow_invocation")
     input_step_parameters = relationship("WorkflowRequestInputStepParameter", back_populates="workflow_invocation")
-    input_datasets = relationship("WorkflowRequestToInputDatasetAssociation", back_populates="workflow_invocation")
-    input_dataset_collections = relationship(
+    input_datasets: Mapped[List["WorkflowRequestToInputDatasetAssociation"]] = relationship(
+        "WorkflowRequestToInputDatasetAssociation", back_populates="workflow_invocation"
+    )
+    input_dataset_collections: Mapped[List["WorkflowRequestToInputDatasetCollectionAssociation"]] = relationship(
         "WorkflowRequestToInputDatasetCollectionAssociation",
         back_populates="workflow_invocation",
     )
@@ -9153,13 +9157,22 @@ class WorkflowInvocation(Base, UsesCreateAndUpdateTime, Dictifiable, Serializabl
             request = input_dataset_assoc.request
             if request:
                 deferred = request.get("deferred", False)
-                if not deferred:
+                if not deferred and input_dataset_assoc.dataset:
                     hdas_to_materialize.append(
                         InputToMaterialize(
                             input_dataset_assoc.dataset,
                             input_dataset_assoc,
                         )
                     )
+        for input_dataset_collection_association in self.input_dataset_collections:
+            request = input_dataset_collection_association.request
+            if request:
+                deferred = request.get("deferred", False)
+                if not deferred and input_dataset_collection_association.dataset_collection:
+                    hdas_to_materialize.extend(
+                        InputToMaterialize(hda, input_dataset_collection_association)
+                        for hda in input_dataset_collection_association.dataset_collection.dataset_instances
+                    )
         return hdas_to_materialize
 
     def _serialize(self, id_encoder, serialization_options):
@@ -9350,7 +9363,7 @@ class WorkflowInvocation(Base, UsesCreateAndUpdateTime, Dictifiable, Serializabl
             self.input_step_parameters.append(request_to_content)
 
     def recover_inputs(self) -> Tuple[Dict[str, Any], str]:
-        inputs = {}
+        inputs: Dict[str, Any] = {}
         inputs_by = "name"
 
         have_referenced_steps_by_order_index = False

lib/galaxy/model/deferred.py

@@ -8,10 +8,7 @@ from typing import (
     Union,
 )
 
-from sqlalchemy.orm import (
-    object_session,
-    Session,
-)
+from sqlalchemy.orm import Session
 from sqlalchemy.orm.exc import DetachedInstanceError
 
 from galaxy.datatypes.sniff import (
@@ -30,6 +27,7 @@ from galaxy.model import (
     HistoryDatasetAssociation,
     HistoryDatasetCollectionAssociation,
     LibraryDatasetDatasetAssociation,
+    required_object_session,
 )
 from galaxy.objectstore import (
     ObjectStore,
@@ -131,8 +129,7 @@ class DatasetInstanceMaterializer:
                 # we need a flush...
                 sa_session = self._sa_session
                 if sa_session is None:
-                    sa_session = object_session(dataset_instance)
-                assert sa_session
+                    sa_session = required_object_session(dataset_instance)
                 sa_session.add(materialized_dataset)
                 sa_session.commit()
             object_store_populator.set_dataset_object_store_id(materialized_dataset)
@@ -179,8 +176,7 @@ class DatasetInstanceMaterializer:
         if attached:
             sa_session = self._sa_session
             if sa_session is None:
-                sa_session = object_session(dataset_instance)
-            assert sa_session
+                sa_session = required_object_session(dataset_instance)
             sa_session.add(materialized_dataset_instance)
         if not in_place:
             materialized_dataset_instance.copy_from(

lib/galaxy/model/dereference.py

@@ -1,17 +1,42 @@
 import os.path
-from typing import List
+from typing import (
+    List,
+    Union,
+)
 
 from galaxy.model import (
+    DatasetCollection,
+    DatasetCollectionElement,
     DatasetSource,
     DatasetSourceHash,
+    History,
     HistoryDatasetAssociation,
+    HistoryDatasetCollectionAssociation,
     TransformAction,
+    User,
+)
+from galaxy.model.scoped_session import galaxy_scoped_session
+from galaxy.tool_util_models.parameters import (
+    CollectionElementCollectionRequestUri,
+    CollectionElementDataRequestUri,
+    DataRequestCollectionUri,
+    DataRequestUri,
+    FileRequestUri,
 )
-from galaxy.tool_util.parameters import DataRequestUri
 
 
-def dereference_to_model(sa_session, user, history, data_request_uri: DataRequestUri) -> HistoryDatasetAssociation:
-    name = data_request_uri.name or os.path.basename(data_request_uri.url)
+def dereference_to_model(
+    sa_session: galaxy_scoped_session,
+    user: User,
+    history: History,
+    data_request_uri: Union[DataRequestUri, FileRequestUri, CollectionElementDataRequestUri],
+    add_to_history=True,
+    visible=True,
+) -> HistoryDatasetAssociation:
+    if isinstance(data_request_uri, CollectionElementDataRequestUri):
+        name = data_request_uri.identifier
+    else:
+        name = data_request_uri.name or os.path.basename(data_request_uri.url)
     dbkey = data_request_uri.dbkey or "?"
     hda = HistoryDatasetAssociation(
         name=name,
@@ -20,6 +45,8 @@ def dereference_to_model(sa_session, user, history, data_request_uri: DataReques
         history=history,
         create_dataset=True,
         sa_session=sa_session,
+        visible=visible,
+        flush=False,
     )
     hda.state = hda.states.DEFERRED
     dataset_source = DatasetSource()
@@ -43,5 +70,81 @@ def dereference_to_model(sa_session, user, history, data_request_uri: DataReques
 
     sa_session.add(hda)
     sa_session.add(dataset_source)
-    history.add_dataset(hda, genome_build=dbkey, quota=False)
+    if add_to_history:
+        history.add_dataset(hda, genome_build=dbkey, quota=False)
     return hda
+
+
+def derefence_collection_element(
+    sa_session: galaxy_scoped_session,
+    user: User,
+    history: History,
+    element: CollectionElementCollectionRequestUri,
+    parent_dataset_collection: DatasetCollection,
+    element_index: int,
+):
+    child_dataset_collection = DatasetCollection(collection_type=element.collection_type)
+    DatasetCollectionElement(
+        collection=parent_dataset_collection,
+        element=child_dataset_collection,
+        element_identifier=element.identifier,
+        element_index=element_index,
+    )
+    sa_session.add(child_dataset_collection)
+    for index, child_element in enumerate(element.elements):
+        if child_element.class_ == "File":
+            dereference_collection_dataset_element(
+                sa_session, user, history, child_element, child_dataset_collection, element_index=index
+            )
+        elif child_element.class_ == "Collection":
+            derefence_collection_element(
+                sa_session, user, history, child_element, child_dataset_collection, element_index=element_index
+            )
+    child_dataset_collection.populated_state = "ok"
+    child_dataset_collection.element_count = len(element.elements)
+
+
+def dereference_collection_dataset_element(
+    sa_session: galaxy_scoped_session,
+    user: User,
+    history: History,
+    element: CollectionElementDataRequestUri,
+    parent_dataset_collection: DatasetCollection,
+    element_index: int,
+):
+    hda = dereference_to_model(sa_session, user, history, element, add_to_history=False, visible=False)
+    history.stage_addition(hda)
+    dce = DatasetCollectionElement(
+        collection=parent_dataset_collection,
+        element=hda,
+        element_identifier=element.identifier,
+        element_index=element_index,
+    )
+    parent_dataset_collection.elements.append(dce)
+
+
+def derefence_collection_to_model(
+    sa_session: galaxy_scoped_session,
+    user: User,
+    history: History,
+    data_request_uri: DataRequestCollectionUri,
+    collection_name: str = "Collection",
+) -> HistoryDatasetCollectionAssociation:
+    name = data_request_uri.name or collection_name
+    hdca = HistoryDatasetCollectionAssociation(
+        name=name,
+        history=history,
+    )
+    sa_session.add(hdca)
+    dc = DatasetCollection(collection_type=data_request_uri.collection_type)
+    sa_session.add(dc)
+    hdca.collection = dc
+    for i, element in enumerate(data_request_uri.elements):
+        if element.class_ == "File":
+            dereference_collection_dataset_element(sa_session, user, history, element, dc, element_index=i)
+        elif element.class_ == "Collection":
+            derefence_collection_element(sa_session, user, history, element, dc, i)
+    dc.populated_state = "ok"
+    dc.element_count = len(data_request_uri.elements)
+    history.stage_addition(hdca)
+    return hdca

lib/galaxy/model/store/__init__.py

@@ -2260,11 +2260,13 @@ class DirectoryModelExportStore(ModelExportStore):
     ) -> None:
         self.included_invocations.append(workflow_invocation)
         for input_dataset in workflow_invocation.input_datasets:
-            self.add_dataset(input_dataset.dataset)
+            if input_dataset.dataset:
+                self.add_dataset(input_dataset.dataset)
         for output_dataset in workflow_invocation.output_datasets:
             self.add_dataset(output_dataset.dataset)
         for input_dataset_collection in workflow_invocation.input_dataset_collections:
-            self.export_collection(input_dataset_collection.dataset_collection)
+            if input_dataset_collection.dataset_collection:
+                self.export_collection(input_dataset_collection.dataset_collection)
         for output_dataset_collection in workflow_invocation.output_dataset_collections:
             self.export_collection(output_dataset_collection.dataset_collection)
         for workflow_invocation_step in workflow_invocation.steps:

lib/galaxy/model/store/ro_crate_utils.py

@@ -125,7 +125,7 @@ class WorkflowRunCrateProfileBuilder:
 
     def _add_files(self, crate: ROCrate):
         for wfda in self.invocation.input_datasets:
-            if not self.file_entities.get(wfda.dataset.dataset.id):
+            if wfda.dataset and not self.file_entities.get(wfda.dataset.dataset.id):
                 dataset_formal_param = self._add_dataset_formal_parameter(wfda.dataset, crate)
                 crate.mainEntity.append_to("input", dataset_formal_param)
                 properties = {
@@ -135,7 +135,7 @@ class WorkflowRunCrateProfileBuilder:
                 self.create_action.append_to("object", file_entity)
 
         for wfda in self.invocation.output_datasets:
-            if not self.file_entities.get(wfda.dataset.dataset.id):
+            if wfda.dataset and not self.file_entities.get(wfda.dataset.dataset.id):
                 dataset_formal_param = self._add_dataset_formal_parameter(wfda.dataset, crate)
                 crate.mainEntity.append_to("output", dataset_formal_param)
                 properties = {
@@ -190,16 +190,18 @@ class WorkflowRunCrateProfileBuilder:
 
     def _add_collections(self, crate: ROCrate):
         for wfdca in self.invocation.input_dataset_collections:
-            collection_formal_param = self._add_collection_formal_parameter(wfdca.dataset_collection, crate)
-            collection_entity = self._add_collection(wfdca.dataset_collection, crate, collection_formal_param)
-            crate.mainEntity.append_to("input", collection_formal_param)
-            self.create_action.append_to("object", collection_entity)
+            if wfdca.dataset_collection:
+                collection_formal_param = self._add_collection_formal_parameter(wfdca.dataset_collection, crate)
+                collection_entity = self._add_collection(wfdca.dataset_collection, crate, collection_formal_param)
+                crate.mainEntity.append_to("input", collection_formal_param)
+                self.create_action.append_to("object", collection_entity)
 
         for wfdca in self.invocation.output_dataset_collections:
-            collection_formal_param = self._add_collection_formal_parameter(wfdca.dataset_collection, crate)
-            collection_entity = self._add_collection(wfdca.dataset_collection, crate, collection_formal_param)
-            crate.mainEntity.append_to("output", collection_formal_param)
-            self.create_action.append_to("result", collection_entity)
+            if wfdca.dataset_collection:
+                collection_formal_param = self._add_collection_formal_parameter(wfdca.dataset_collection, crate)
+                collection_entity = self._add_collection(wfdca.dataset_collection, crate, collection_formal_param)
+                crate.mainEntity.append_to("output", collection_formal_param)
+                self.create_action.append_to("result", collection_entity)
 
     def _add_workflows(self, crate: ROCrate):
         workflows_directory = self.model_store.workflows_directory

lib/galaxy/selenium/navigates_galaxy.py

@@ -1653,7 +1653,8 @@ class NavigatesGalaxy(HasDriver):
         workflow_run = self.components.workflow_run
         for label, value in inputs.items():
             input_div_element = workflow_run.input_data_div(label=label).wait_for_visible()
-            self.select_set_value(input_div_element, "{}: ".format(value["hid"]))
+            hid = value.pop("hid")
+            self.select_set_value(input_div_element, f"{hid}: ")
 
     def workflow_run_submit(self):
         self.components.workflow_run.run_workflow_disabled.wait_for_absent()

lib/galaxy/tool_util_models/parameters.py

@@ -27,13 +27,16 @@ from pydantic import (
     Field,
     field_validator,
     HttpUrl,
+    model_validator,
     RootModel,
     StrictBool,
     StrictFloat,
     StrictInt,
     StrictStr,
     Tag,
+    TypeAdapter,
 )
+from pydantic.json_schema import SkipJsonSchema
 from typing_extensions import (
     Annotated,
     Literal,
@@ -338,17 +341,30 @@ CollectionStrT = Literal["hdca"]
 TestCaseDataSrcT = Literal["File"]
 
 
-class DataRequestHda(StrictModel):
+class LegacyRequestModelAttributes(StrictModel):
+    # Here for bioblend's sake, should be stripped
+    map_over_type: SkipJsonSchema[Optional[str]] = Field(None, exclude=True)
+    hid: SkipJsonSchema[Optional[int]] = Field(None, exclude=True)
+    workflow_step_id: SkipJsonSchema[Optional[str]] = Field(None, exclude=True)
+    label: SkipJsonSchema[Optional[str]] = Field(None, exclude=True)
+
+
+class DataRequestHda(LegacyRequestModelAttributes):
     src: Literal["hda"] = "hda"
     id: StrictStr
 
 
-class DataRequestLdda(StrictModel):
+class DataRequestLdda(LegacyRequestModelAttributes):
     src: Literal["ldda"] = "ldda"
     id: StrictStr
 
 
-class DataRequestHdca(StrictModel):
+class DataRequestLd(LegacyRequestModelAttributes):
+    src: Literal["ld"] = Field(deprecated=True)
+    id: StrictStr
+
+
+class DataRequestHdca(LegacyRequestModelAttributes):
     src: Literal["hdca"] = "hdca"
     id: StrictStr
 
@@ -358,10 +374,8 @@ class DatasetHash(StrictModel):
     hash_value: StrictStr
 
 
-class DataRequestUri(StrictModel):
-    # calling it url instead of uri to match data fetch schema...
-    src: Literal["url"] = "url"
-    url: StrictStr
+class BaseDataRequest(StrictModel):
+    url: StrictStr = Field(..., alias="location")
     name: Optional[StrictStr] = None
     ext: StrictStr
     dbkey: StrictStr = "?"
@@ -371,13 +385,94 @@ class DataRequestUri(StrictModel):
     hashes: Optional[List[DatasetHash]] = None
     space_to_tab: bool = False
     to_posix_lines: bool = False
+
     # to implement:
     # tags
+    model_config = ConfigDict(extra="forbid", populate_by_name=True)
+
+    @model_validator(mode="before")
+    @classmethod
+    def allow_filetype_and_extension(cls, data: Any):
+        if isinstance(data, dict):
+            extension = data.get("filetype")
+            if extension:
+                data = data.copy()
+                data.pop("filetype")
+                data["ext"] = extension
+            extension = data.get("extension")
+            if extension:
+                data = data.copy()
+                data.pop("extension")
+                data["ext"] = extension
+        return data
 
 
-DataRequest: Type = cast(
-    Type, Annotated[union_type([DataRequestHda, DataRequestLdda, DataRequestUri]), Field(discriminator="src")]
-)
+class DataRequestUri(BaseDataRequest):
+    # calling it url instead of uri to match data fetch schema...
+    src: Literal["url"] = "url"
+
+
+class FileRequestUri(BaseDataRequest):
+    class_: Literal["File"] = Field(..., alias="class")
+    src: None = Field(None, exclude=True)
+
+
+class CollectionElementDataRequestUri(FileRequestUri):
+    class_: Literal["File"] = Field(..., alias="class")
+    identifier: StrictStr
+
+
+class CollectionElementCollectionRequestUri(StrictModel):
+    class_: Literal["Collection"] = Field(..., alias="class")
+    identifier: StrictStr
+    collection_type: StrictStr
+    elements: List[
+        Annotated[
+            Union["CollectionElementCollectionRequestUri", CollectionElementDataRequestUri],
+            Field(discriminator="class_"),
+        ]
+    ]
+
+    @model_validator(mode="before")
+    @classmethod
+    def allow_collection_type_by_type(cls, data: Any):
+        if isinstance(data, dict):
+            collection_type = data.get("type")
+            if collection_type:
+                data = data.copy()
+                data.pop("type")
+                data["collection_type"] = collection_type
+        return data
+
+
+class DataRequestCollectionUri(StrictModel):
+    class_: Literal["Collection"] = Field(..., alias="class")
+    collection_type: str
+    elements: List[
+        Annotated[
+            Union[CollectionElementCollectionRequestUri, CollectionElementDataRequestUri], Field(discriminator="class_")
+        ]
+    ]
+    deferred: StrictBool = False
+    name: Optional[StrictStr] = None
+    src: None = Field(None, exclude=True)
+
+
+_DataRequest = Annotated[
+    Union[DataRequestHda, DataRequestLdda, DataRequestLd, DataRequestUri], Field(discriminator="src")
+]
+DataRequest: Type = cast(Type, _DataRequest)
+
+DataOrCollectionRequest = Union[_DataRequest, FileRequestUri, DataRequestCollectionUri, DataRequestHdca]
+
+DataRequestHda.model_rebuild()
+DataRequestLd.model_rebuild()
+DataRequestLdda.model_rebuild()
+DataRequestUri.model_rebuild()
+DataRequestHdca.model_rebuild()
+DataRequestCollectionUri.model_rebuild()
+
+DataOrCollectionRequestAdapter: TypeAdapter[DataOrCollectionRequest] = TypeAdapter(DataOrCollectionRequest)
 
 
 class BatchDataInstance(StrictModel):
@@ -395,17 +490,17 @@ MultiDataRequest: Type = union_type([MultiDataInstance, list_type(MultiDataInsta
 
 
 class DataRequestInternalHda(StrictModel):
-    src: Literal["hda"] = "hda"
+    src: Literal["hda"]
     id: StrictInt
 
 
 class DataRequestInternalLdda(StrictModel):
-    src: Literal["ldda"] = "ldda"
+    src: Literal["ldda"]
     id: StrictInt
 
 
 class DataRequestInternalHdca(StrictModel):
-    src: Literal["hdca"] = "hdca"
+    src: Literal["hdca"]
     id: StrictInt
 
 
@@ -413,7 +508,8 @@ DataRequestInternal: Type = cast(
     Type, Annotated[Union[DataRequestInternalHda, DataRequestInternalLdda, DataRequestUri], Field(discriminator="src")]
 )
 DataRequestInternalDereferenced: Type = cast(
-    Type, Annotated[Union[DataRequestInternalHda, DataRequestInternalLdda], Field(discriminator="src")]
+    Type,
+    Annotated[Union[DataRequestInternalHda, DataRequestInternalLdda], Field(discriminator="src")],
 )
 DataJobInternal = DataRequestInternalDereferenced
 

lib/galaxy/web/framework/decorators.py

@@ -3,10 +3,6 @@ from functools import wraps
 from inspect import getfullargspec
 from json import loads
 from traceback import format_exc
-from typing import (
-    TYPE_CHECKING,
-    Union,
-)
 
 import paste.httpexceptions
 from pydantic import (
@@ -17,10 +13,11 @@ from pydantic import (
 from galaxy.exceptions import (
     error_codes,
     MessageException,
-    RequestParameterInvalidException,
-    RequestParameterMissingException,
 )
-from galaxy.exceptions.utils import api_error_to_dict
+from galaxy.exceptions.utils import (
+    api_error_to_dict,
+    validation_error_to_message_exception,
+)
 from galaxy.util import (
     parse_non_hex_float,
     unicodify,
@@ -28,9 +25,6 @@ from galaxy.util import (
 from galaxy.util.json import safe_dumps
 from galaxy.web.framework import url_for
 
-if TYPE_CHECKING:
-    from fastapi.exceptions import RequestValidationError
-
 log = logging.getLogger(__name__)
 
 JSON_CONTENT_TYPE = "application/json; charset=UTF-8"
@@ -390,31 +384,6 @@ def format_return_as_json(rval, jsonp_callback=None, pretty=False):
     return json
 
 
-def validation_error_to_message_exception(e: Union[ValidationError, "RequestValidationError"]) -> MessageException:
-    invalid_found = False
-    missing_found = False
-    messages = []
-    clean_validation_errors = []
-    for error in e.errors():
-        messages.append(f"{error['msg']} in {error['loc']}")
-        if error["type"] == "message_exception" and "ctx" in error:
-            return error["ctx"]["exception"]
-        if error["type"] == "missing" or error["type"] == "type_error.none.not_allowed":
-            missing_found = True
-        elif error["type"].startswith("type_error"):
-            invalid_found = True
-        # ctx contains data that can't be serialized, like exception instances
-        error.pop("ctx", None)
-        try:
-            clean_validation_errors.append(safe_dumps(error))
-        except TypeError:
-            pass
-    if missing_found and not invalid_found:
-        return RequestParameterMissingException("\n".join(messages), validation_errors=clean_validation_errors)
-    else:
-        return RequestParameterInvalidException("\n".join(messages), validation_errors=clean_validation_errors)
-
-
 def __api_error_dict(trans, **kwds):
     error_dict = api_error_to_dict(debug=trans.debug, **kwds)
     exception = kwds.get("exception", None)

lib/galaxy/webapps/base/api.py

@@ -33,10 +33,12 @@ from starlette_context.plugins import (
 )
 
 from galaxy.exceptions import MessageException
-from galaxy.exceptions.utils import api_error_to_dict
+from galaxy.exceptions.utils import (
+    api_error_to_dict,
+    validation_error_to_message_exception,
+)
 from galaxy.util.path import StrPath
 from galaxy.web.framework.base import walk_controller_modules
-from galaxy.web.framework.decorators import validation_error_to_message_exception
 
 if TYPE_CHECKING:
     from starlette.background import BackgroundTask

lib/galaxy/webapps/galaxy/api/history_contents.py

@@ -27,6 +27,7 @@ from starlette.responses import (
 from typing_extensions import Annotated
 
 from galaxy import util
+from galaxy.exceptions.utils import validation_error_to_message_exception
 from galaxy.managers.context import ProvidesHistoryContext
 from galaxy.schema import (
     FilterQueryParams,
@@ -59,7 +60,6 @@ from galaxy.schema.schema import (
     UpdateHistoryContentsPayload,
     WriteStoreToPayload,
 )
-from galaxy.web.framework.decorators import validation_error_to_message_exception
 from galaxy.webapps.galaxy.api import (
     depends,
     DependsOnTrans,

lib/galaxy/workflow/run_request.py

@@ -7,9 +7,13 @@ from typing import (
     List,
     Optional,
     TYPE_CHECKING,
+    Union,
 )
 
+from pydantic import ValidationError
+
 from galaxy import exceptions
+from galaxy.exceptions.utils import validation_error_to_message_exception
 from galaxy.managers.hdas import dereference_input
 from galaxy.model import (
     EffectiveOutput,
@@ -24,7 +28,7 @@ from galaxy.model import (
     WorkflowRequestStepState,
 )
 from galaxy.model.base import ensure_object_added_to_session
-from galaxy.tool_util.parameters import DataRequestUri
+from galaxy.tool_util_models.parameters import DataOrCollectionRequestAdapter
 from galaxy.tools.parameters.basic import ParameterValueError
 from galaxy.tools.parameters.meta import expand_workflow_inputs
 from galaxy.tools.parameters.workflow_utils import NO_REPLACEMENT
@@ -383,35 +387,29 @@ def build_workflow_run_configs(
                         f"{step.label or step.order_index + 1}: {e.message_suffix}"
                     )
                 continue
-            if "src" not in input_dict:
-                raise exceptions.RequestParameterInvalidException(
-                    f"Not input source type defined for input '{input_dict}'."
-                )
-            input_source = input_dict["src"]
-            if "id" not in input_dict and input_source != "url":
-                raise exceptions.RequestParameterInvalidException(f"No input id defined for input '{input_dict}'.")
-            elif input_source == "url" and not input_dict.get("url"):
-                raise exceptions.RequestParameterInvalidException(
-                    f"Supplied 'url' is empty or absent for input '{input_dict}'."
-                )
-            if "content" in input_dict:
-                raise exceptions.RequestParameterInvalidException(
-                    f"Input cannot specify explicit 'content' attribute {input_dict}'."
-                )
-            input_id = input_dict.get("id")
             try:
                 added_to_history = False
-                if input_source == "ldda":
-                    assert input_id
-                    ldda = trans.sa_session.get(LibraryDatasetDatasetAssociation, trans.security.decode_id(input_id))
+                try:
+                    data_request = DataOrCollectionRequestAdapter.validate_python(input_dict)
+                except ValidationError as e:
+                    raise validation_error_to_message_exception(e)
+                if data_request.src == "ldda":
+                    ldda = trans.sa_session.get(
+                        LibraryDatasetDatasetAssociation, trans.security.decode_id(data_request.id)
+                    )
                     assert ldda
                     assert trans.user_is_admin or trans.app.security_agent.can_access_dataset(
                         trans.get_current_user_roles(), ldda.dataset
                     )
                     content = ldda.to_history_dataset_association(history, add_to_history=add_to_history)
-                elif input_source == "ld":
-                    assert input_id
-                    library_dataset = trans.sa_session.get(LibraryDataset, trans.security.decode_id(input_id))
+                elif data_request.src == "hda":
+                    # Get dataset handle, add to dict and history if necessary
+                    content = trans.sa_session.get(HistoryDatasetAssociation, trans.security.decode_id(data_request.id))
+                    assert trans.user_is_admin or trans.app.security_agent.can_access_dataset(
+                        trans.get_current_user_roles(), content.dataset
+                    )
+                elif data_request.src == "ld":
+                    library_dataset = trans.sa_session.get(LibraryDataset, trans.security.decode_id(data_request.id))
                     assert library_dataset
                     ldda = library_dataset.library_dataset_dataset_association
                     assert ldda
@@ -419,28 +417,22 @@ def build_workflow_run_configs(
                         trans.get_current_user_roles(), ldda.dataset
                     )
                     content = ldda.to_history_dataset_association(history, add_to_history=add_to_history)
-                elif input_source == "hda":
-                    assert input_id
-                    # Get dataset handle, add to dict and history if necessary
-                    content = trans.sa_session.get(HistoryDatasetAssociation, trans.security.decode_id(input_id))
-                    assert trans.user_is_admin or trans.app.security_agent.can_access_dataset(
-                        trans.get_current_user_roles(), content.dataset
+                elif data_request.src == "hdca":
+                    content = app.dataset_collection_manager.get_dataset_collection_instance(
+                        trans, "history", data_request.id
                     )
-                elif input_source == "hdca":
-                    content = app.dataset_collection_manager.get_dataset_collection_instance(trans, "history", input_id)
-                elif input_source == "url":
-                    data_request = DataRequestUri.model_validate(input_dict)
-                    hda: HistoryDatasetAssociation = dereference_input(trans, data_request, history)
+                elif data_request.src == "url" or data_request.class_ == "File" or data_request.class_ == "Collection":
+                    request_input = dereference_input(trans, data_request, history)
                     added_to_history = True
                     content = InputWithRequest(
-                        input=hda,
+                        input=request_input,
                         request=data_request.model_dump(mode="json"),
                     )
                     if not data_request.deferred:
                         requires_materialization = True
                 else:
                     raise exceptions.RequestParameterInvalidException(
-                        f"Unknown workflow input source '{input_source}' specified."
+                        f"Unknown workflow input source for '{key}' specified."
                     )
                 if not added_to_history and add_to_history and content.history != history:
                     if isinstance(content, HistoryDatasetCollectionAssociation):
@@ -450,7 +442,9 @@ def build_workflow_run_configs(
                     history.stage_addition(content)
                 input_dict["content"] = content
             except AssertionError:
-                raise exceptions.ItemAccessibilityException(f"Invalid workflow input '{input_id}' specified")
+                raise exceptions.ItemAccessibilityException(
+                    f"Invalid workflow input '{input_dict.get('id')}' specified"
+                )
         for key in set(normalized_inputs.keys()):
             value = normalized_inputs[key]
             if isinstance(value, dict) and "content" in value:
@@ -635,7 +629,9 @@ def workflow_request_to_run_config(
     param_types = WorkflowRequestInputParameter.types
     history = workflow_invocation.history
     replacement_dict = {}
-    inputs = {}
+    inputs: Dict[
+        int, Union[HistoryDatasetAssociation, HistoryDatasetCollectionAssociation, str, int, float, bool, None]
+    ] = {}
     param_map = {}
     resource_params = {}
     copy_inputs_to_history = None
@@ -666,10 +662,12 @@ def workflow_request_to_run_config(
             resource_params[parameter.name] = parameter.value
         elif parameter_type == param_types.STEP_PARAMETERS:
             param_map[int(parameter.name)] = json.loads(parameter.value)
-    for input_association in workflow_invocation.input_datasets:
-        inputs[input_association.workflow_step_id] = input_association.dataset
-    for input_association in workflow_invocation.input_dataset_collections:
-        inputs[input_association.workflow_step_id] = input_association.dataset_collection
+    for dataset_input_association in workflow_invocation.input_datasets:
+        assert dataset_input_association.workflow_step_id
+        inputs[dataset_input_association.workflow_step_id] = dataset_input_association.dataset
+    for collection_input_association in workflow_invocation.input_dataset_collections:
+        assert collection_input_association.workflow_step_id
+        inputs[collection_input_association.workflow_step_id] = collection_input_association.dataset_collection
     for input_association in workflow_invocation.input_step_parameters:
         parameter_value = input_association.parameter_value
         inputs[input_association.workflow_step_id] = parameter_value

lib/galaxy/workflow/scheduling_manager.py

@@ -2,6 +2,8 @@ import os
 from functools import partial
 from typing import Optional
 
+from sqlalchemy.orm import Session
+
 import galaxy.workflow.schedulers
 from galaxy import model
 from galaxy.exceptions import HandlerAssignmentError
@@ -16,6 +18,7 @@ from galaxy.schema.tasks import (
     MaterializeDatasetInstanceTaskRequest,
     RequestUser,
 )
+from galaxy.structured_app import MinimalManagerApp
 from galaxy.util import plugin_config
 from galaxy.util.custom_logging import get_logger
 from galaxy.util.monitors import Monitors
@@ -285,7 +288,8 @@ class WorkflowSchedulingManager(ConfiguresHandlers):
 
 
 class WorkflowRequestMonitor(Monitors):
-    def __init__(self, app, workflow_scheduling_manager):
+
+    def __init__(self, app: MinimalManagerApp, workflow_scheduling_manager):
         self.app = app
         self.workflow_scheduling_manager = workflow_scheduling_manager
         self._init_monitor_thread(
@@ -335,7 +339,7 @@ class WorkflowRequestMonitor(Monitors):
             if not self.monitor_running:
                 return
 
-    def __attempt_materialize(self, workflow_invocation, session) -> bool:
+    def __attempt_materialize(self, workflow_invocation: model.WorkflowInvocation, session: Session) -> bool:
         try:
             inputs_to_materialize = workflow_invocation.inputs_requiring_materialization()
             for input_to_materialize in inputs_to_materialize:
@@ -347,9 +351,10 @@ class WorkflowRequestMonitor(Monitors):
                     source="hda",
                     content=hda.id,
                 )
-                materialized_okay = self.app.hda_manager.materialize(task_request, in_place=True)
+                materialized_okay = self.app.hda_manager.materialize(task_request, session, in_place=True)
                 if not materialized_okay:
                     workflow_invocation.fail()
+                    assert input_to_materialize.input_dataset.workflow_step
                     workflow_invocation.add_message(
                         InvocationFailureDatasetFailed(
                             workflow_step_id=input_to_materialize.input_dataset.workflow_step.id,

lib/galaxy_test/api/test_workflows.py

@@ -1749,6 +1749,165 @@ steps:
             message = invocation_details["messages"][0]
             assert message["reason"] == "dataset_failed"
 
+    @skip_without_tool("multi_data_optional")
+    def test_run_workflow_with_url_collection(self):
+        with self.dataset_populator.test_history() as history_id:
+            invocation = self._run_multi_data_workflow(history_id)
+            assert invocation["state"] == "scheduled", invocation
+            invocation_jobs = self.workflow_populator.get_invocation_jobs(invocation["id"])
+            assert len(invocation_jobs) == 1
+            job = invocation_jobs[0]
+            assert job["state"] == "ok"
+            job = self.dataset_populator.get_job_details(job_id=job["id"], full=True).json()
+            assert (
+                self.dataset_populator.get_history_dataset_content(
+                    history_id=history_id, content_id=job["outputs"]["out1"]["id"]
+                ).strip()
+                == "1 2 3"
+            )
+
+    @skip_without_tool("multi_data_optional")
+    def test_run_workflow_with_url_invalid_hash_collection(self):
+        with self.dataset_populator.test_history() as history_id:
+            invocation = self._run_multi_data_workflow(history_id, invalid_hash=True)
+            assert invocation["state"] == "failed", invocation
+            assert invocation["state"] == "failed"
+            assert len(invocation["messages"]) == 1
+            message = invocation["messages"][0]
+            assert message["reason"] == "dataset_failed"
+
+    def _run_multi_data_workflow(self, history_id, invalid_hash=False):
+        workflow_id = self._upload_yaml_workflow(
+            """
+class: GalaxyWorkflow
+inputs:
+  input:
+    type: collection
+    collection_type: "list"
+steps:
+  multi_data_optional:
+    tool_id: multi_data_optional
+    in:
+      input1: input
+    """
+        )
+        input_b64_1 = base64.b64encode(b"1 2 3").decode("utf-8")
+        deferred = False
+        hashes_1 = [
+            {"hash_function": "MD5", "hash_value": "abcdef" if invalid_hash else "5ba48b6e5a7c4d4930fda256f411e55b"}
+        ]
+        inputs = {
+            "input": {
+                "class": "Collection",
+                "collection_type": "list",
+                "elements": [
+                    {
+                        "class": "File",
+                        "identifier": "my cool element",
+                        "url": f"base64://{input_b64_1}",
+                        "ext": "txt",
+                        "deferred": deferred,
+                        "hashes": hashes_1,
+                    }
+                ],
+            },
+        }
+        workflow_request = dict(
+            history=f"hist_id={history_id}",
+        )
+        workflow_request["inputs"] = json.dumps(inputs)
+        workflow_request["inputs_by"] = "name"
+        invocation_id = self.workflow_populator.invoke_workflow_and_wait(
+            workflow_id, request=workflow_request, assert_ok=not invalid_hash
+        ).json()["id"]
+        return self._invocation_details(workflow_id, invocation_id)
+
+    @skip_without_tool("collection_paired_default")
+    def test_run_workflow_with_url_paired_collection(self):
+        with self.dataset_populator.test_history() as history_id:
+            workflow_id = self._upload_yaml_workflow(
+                """
+class: GalaxyWorkflow
+inputs:
+  input:
+    type: collection
+    collection_type: "list:paired"
+steps:
+  collection_paired_default:
+    tool_id: collection_paired_default
+    in:
+      f1: input
+    """
+            )
+            forward_b64_1 = base64.b64encode(b"1 2 3").decode("utf-8")
+            reverse_b64_2 = base64.b64encode(b"4 5 6").decode("utf-8")
+            deferred = False
+            hashes_1 = [{"hash_function": "MD5", "hash_value": "5ba48b6e5a7c4d4930fda256f411e55b"}]
+            hashes_2 = [{"hash_function": "MD5", "hash_value": "ad0f811416f7ed2deb9122007d649fb0"}]
+            inputs = {
+                "input": {
+                    "class": "Collection",
+                    "collection_type": "list:paired",
+                    "elements": [
+                        {
+                            "class": "Collection",
+                            "identifier": "my cool pair",
+                            "collection_type": "paired",
+                            "elements": [
+                                {
+                                    "class": "File",
+                                    "identifier": "forward",
+                                    "url": f"base64://{forward_b64_1}",
+                                    "ext": "txt",
+                                    "deferred": deferred,
+                                    "hashes": hashes_1,
+                                },
+                                {
+                                    "class": "File",
+                                    "identifier": "reverse",
+                                    "url": f"base64://{reverse_b64_2}",
+                                    "ext": "txt",
+                                    "deferred": deferred,
+                                    "hashes": hashes_2,
+                                },
+                            ],
+                        }
+                    ],
+                },
+            }
+            workflow_request = dict(
+                history=f"hist_id={history_id}",
+            )
+            workflow_request["inputs"] = json.dumps(inputs)
+            workflow_request["inputs_by"] = "name"
+            invocation_id = self.workflow_populator.invoke_workflow_and_wait(
+                workflow_id, request=workflow_request
+            ).json()["id"]
+            invocation = self._invocation_details(workflow_id, invocation_id)
+            assert invocation["state"] == "scheduled", invocation
+            assert invocation["inputs"]["0"]["src"] == "hdca"
+            input_hdca = self.dataset_populator.get_history_collection_details(
+                history_id=history_id, content_id=invocation["inputs"]["0"]["id"]
+            )
+            assert input_hdca["collection_type"] == "list:paired"
+            invocation_jobs = self.workflow_populator.get_invocation_jobs(invocation_id)
+            assert len(invocation_jobs) == 1
+            job = invocation_jobs[0]
+            assert job["state"] == "ok"
+            job = self.dataset_populator.get_job_details(job_id=job["id"], full=True).json()
+            hdca = self.dataset_populator.get_history_collection_details(
+                history_id=history_id, content_id=job["output_collections"]["out1"]["id"]
+            )
+            assert hdca["collection_type"] == "list"
+            assert hdca["elements"][0]["element_identifier"] == "my cool pair"
+            concatenated_output = hdca["elements"][0]["object"]
+            assert (
+                self.dataset_populator.get_history_dataset_content(
+                    history_id=history_id, content_id=concatenated_output["id"]
+                ).strip()
+                == "1 2 34 5 6"
+            )
+
     def __run_cat_workflow(self, inputs_by, history_id: Optional[str] = None):
         workflow = self.workflow_populator.load_workflow(name="test_for_run")
         workflow["steps"]["0"]["uuid"] = str(uuid4())
@@ -7086,16 +7245,16 @@ steps:
                 "coolinput": {
                     "batch": True,
                     "values": [
-                        {"id": hda1.get("id"), "hid": hda1.get("hid"), "src": "hda"},
-                        {"id": hda2.get("id"), "hid": hda2.get("hid"), "src": "hda"},
-                        {"id": hda3.get("id"), "hid": hda2.get("hid"), "src": "hda"},
-                        {"id": hda4.get("id"), "hid": hda2.get("hid"), "src": "hda"},
+                        {"id": hda1.get("id"), "src": "hda"},
+                        {"id": hda2.get("id"), "src": "hda"},
+                        {"id": hda3.get("id"), "src": "hda"},
+                        {"id": hda4.get("id"), "src": "hda"},
                     ],
                 }
             }
             parameters = {
                 "1": {
-                    "input": {"batch": False, "values": [{"id": hda1.get("id"), "hid": hda1.get("hid"), "src": "hda"}]},
+                    "input": {"batch": False, "values": [{"id": hda1.get("id"), "src": "hda"}]},
                     "exp": "2",
                 }
             }

lib/galaxy_test/base/populators.py

@@ -2133,6 +2133,7 @@ class BaseWorkflowPopulator(BasePopulator):
         assert_ok: bool = True,
     ) -> Response:
         invoke_return = self.invoke_workflow(workflow_id, history_id=history_id, inputs=inputs, request=request)
+        assert invoke_return.status_code < 300, invoke_return.json()
         invoke_return.raise_for_status()
         invocation_id = invoke_return.json()["id"]
 
@@ -3426,7 +3427,6 @@ def load_data_dict(
                 ).json()
             hdca_output = fetch_response["outputs"][0]
             hdca = dataset_populator.ds_entry(hdca_output)
-            hdca["hid"] = hdca_output["hid"]
             label_map[key] = hdca
             inputs[key] = hdca
             has_uploads = True

lib/galaxy_test/selenium/framework.py

@@ -635,6 +635,22 @@ class RunsWorkflows(GalaxyTestSeleniumContext):
             inputs, _, _ = load_data_dict(
                 history_id, test_data, self.dataset_populator, self.dataset_collection_populator
             )
+            for input_value in inputs.values():
+                if (
+                    isinstance(input_value, dict)
+                    and (src := input_value.get("src"))
+                    and (content_id := input_value.get("id"))
+                ):
+                    if src == "hda":
+                        content_item = self.dataset_populator.get_history_dataset_details(
+                            history_id, content_id=content_id
+                        )
+                        input_value["hid"] = content_item["hid"]
+                    elif src == "hdca":
+                        content_item = self.dataset_populator.get_history_collection_details(
+                            history_id=history_id, content_id=content_id
+                        )
+                        input_value["hid"] = content_item["hid"]
             self.dataset_populator.wait_for_history(history_id)
         else:
             inputs = {}

test/unit/data/model/test_model_store.py

@@ -18,11 +18,11 @@ from typing import (
 import pytest
 from rocrate.rocrate import ROCrate
 from sqlalchemy import select
-from sqlalchemy.orm.scoping import scoped_session
 
 from galaxy import model
 from galaxy.model import store
 from galaxy.model.metadata import MetadataTempFile
+from galaxy.model.scoped_session import galaxy_scoped_session as scoped_session
 from galaxy.model.store import SessionlessContext
 from galaxy.model.unittest_utils import GalaxyDataTestApp
 from galaxy.model.unittest_utils.store_fixtures import (

test/unit/tool_util/test_parameter_convert.py

@@ -232,7 +232,7 @@ def test_fill_defaults():
 
 
 def _fake_dereference(input: DataRequestUri) -> DataRequestInternalHda:
-    return DataRequestInternalHda(id=EXAMPLE_ID_1)
+    return DataRequestInternalHda(id=EXAMPLE_ID_1, src="hda")
 
 
 def _fake_decode(input: str) -> int: