diff --git a/lib/galaxy/celery/tasks.py b/lib/galaxy/celery/tasks.py index 375d855848c..79ade5ccefb 100644 --- a/lib/galaxy/celery/tasks.py +++ b/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") diff --git a/lib/galaxy/exceptions/utils.py b/lib/galaxy/exceptions/utils.py index 81a9f8d4023..b5196f86bb7 100644 --- a/lib/galaxy/exceptions/utils.py +++ b/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): diff --git a/lib/galaxy/managers/hdas.py b/lib/galaxy/managers/hdas.py index abb31c1f0d9..fbd4f889277 100644 --- a/lib/galaxy/managers/hdas.py +++ b/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 diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py index 15e660dfae2..d3f6177faf9 100644 --- a/lib/galaxy/model/__init__.py +++ b/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 diff --git a/lib/galaxy/model/deferred.py b/lib/galaxy/model/deferred.py index 9b3defb868d..402641d9fee 100644 --- a/lib/galaxy/model/deferred.py +++ b/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( diff --git a/lib/galaxy/model/dereference.py b/lib/galaxy/model/dereference.py index b773c6b6626..4ea57095102 100644 --- a/lib/galaxy/model/dereference.py +++ b/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 diff --git a/lib/galaxy/model/store/__init__.py b/lib/galaxy/model/store/__init__.py index a8903afa6dc..3ff3a2d7209 100644 --- a/lib/galaxy/model/store/__init__.py +++ b/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: diff --git a/lib/galaxy/model/store/ro_crate_utils.py b/lib/galaxy/model/store/ro_crate_utils.py index 45884643a49..dd680b61e7e 100644 --- a/lib/galaxy/model/store/ro_crate_utils.py +++ b/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 diff --git a/lib/galaxy/selenium/navigates_galaxy.py b/lib/galaxy/selenium/navigates_galaxy.py index b9c87ffd971..6449355cb23 100644 --- a/lib/galaxy/selenium/navigates_galaxy.py +++ b/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() diff --git a/lib/galaxy/tool_util_models/parameters.py b/lib/galaxy/tool_util_models/parameters.py index 2cb9e70ee33..a8283d33259 100644 --- a/lib/galaxy/tool_util_models/parameters.py +++ b/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 diff --git a/lib/galaxy/web/framework/decorators.py b/lib/galaxy/web/framework/decorators.py index 6cef907ae17..13b8d7dd70f 100644 --- a/lib/galaxy/web/framework/decorators.py +++ b/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) diff --git a/lib/galaxy/webapps/base/api.py b/lib/galaxy/webapps/base/api.py index 2c8cc4b8b11..765bba28af7 100644 --- a/lib/galaxy/webapps/base/api.py +++ b/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 diff --git a/lib/galaxy/webapps/galaxy/api/history_contents.py b/lib/galaxy/webapps/galaxy/api/history_contents.py index 9e4c7709351..829d11f886c 100644 --- a/lib/galaxy/webapps/galaxy/api/history_contents.py +++ b/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, diff --git a/lib/galaxy/workflow/run_request.py b/lib/galaxy/workflow/run_request.py index 1177f5872d2..0a361700d2a 100644 --- a/lib/galaxy/workflow/run_request.py +++ b/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 diff --git a/lib/galaxy/workflow/scheduling_manager.py b/lib/galaxy/workflow/scheduling_manager.py index e30c90c74fb..9a617388e68 100644 --- a/lib/galaxy/workflow/scheduling_manager.py +++ b/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, diff --git a/lib/galaxy_test/api/test_workflows.py b/lib/galaxy_test/api/test_workflows.py index 69c4308f655..6920bb75c19 100644 --- a/lib/galaxy_test/api/test_workflows.py +++ b/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", } } diff --git a/lib/galaxy_test/base/populators.py b/lib/galaxy_test/base/populators.py index eb0d03953a7..d3e84be8bc0 100644 --- a/lib/galaxy_test/base/populators.py +++ b/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 diff --git a/lib/galaxy_test/selenium/framework.py b/lib/galaxy_test/selenium/framework.py index e5274319e2c..76302390969 100644 --- a/lib/galaxy_test/selenium/framework.py +++ b/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 = {} diff --git a/test/unit/data/model/test_model_store.py b/test/unit/data/model/test_model_store.py index b240eac77ad..e023b156537 100644 --- a/test/unit/data/model/test_model_store.py +++ b/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 ( diff --git a/test/unit/tool_util/test_parameter_convert.py b/test/unit/tool_util/test_parameter_convert.py index 7f5deff41e8..595bc578d61 100644 --- a/test/unit/tool_util/test_parameter_convert.py +++ b/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: