From 2e4a50a38c75e103911142f490cc731c856e2052 Mon Sep 17 00:00:00 2001 From: John Chilton Date: Wed, 29 Oct 2025 16:33:22 -0400 Subject: [PATCH] Tool Request API... --- .github/workflows/framework_tools.yaml | 3 +- lib/galaxy/app.py | 8 +- lib/galaxy/celery/tasks.py | 19 +- lib/galaxy/managers/jobs.py | 140 +++++++++- lib/galaxy/model/__init__.py | 27 +- lib/galaxy/schema/jobs.py | 13 + lib/galaxy/schema/schema.py | 17 ++ lib/galaxy/schema/tasks.py | 13 + lib/galaxy/tool_util/parameters/__init__.py | 2 + lib/galaxy/tool_util/parameters/case.py | 8 +- lib/galaxy/tool_util/parameters/convert.py | 63 ++++- lib/galaxy/tool_util/parameters/factory.py | 5 +- lib/galaxy/tool_util/parser/interface.py | 2 + lib/galaxy/tool_util/verify/_types.py | 11 +- lib/galaxy/tool_util/verify/interactor.py | 207 ++++++++++++--- lib/galaxy/tool_util/verify/parse.py | 43 +++- lib/galaxy/tool_util_models/parameters.py | 21 +- lib/galaxy/tool_util_models/tool_source.py | 2 + lib/galaxy/tools/__init__.py | 143 ++++++++++- lib/galaxy/tools/_types.py | 6 + lib/galaxy/tools/execute.py | 128 ++++++++- lib/galaxy/tools/parameters/__init__.py | 203 +++++++++++++++ lib/galaxy/tools/parameters/meta.py | 112 +++++++- lib/galaxy/webapps/galaxy/api/histories.py | 12 + lib/galaxy/webapps/galaxy/api/jobs.py | 25 +- lib/galaxy/webapps/galaxy/api/tools.py | 78 +++++- lib/galaxy/webapps/galaxy/services/base.py | 39 ++- .../webapps/galaxy/services/histories.py | 9 + lib/galaxy/webapps/galaxy/services/jobs.py | 130 +++++++++- lib/galaxy/webapps/galaxy/services/tools.py | 13 +- lib/galaxy_test/api/conftest.py | 2 +- lib/galaxy_test/api/test_tool_execute.py | 242 ++++++++++++------ lib/galaxy_test/api/test_tool_execution.py | 134 ++++++++++ lib/galaxy_test/base/populators.py | 153 +++++++++-- test/functional/test_toolbox_pytest.py | 10 +- .../gx_conditional_select_dynamic.xml | 52 ++++ .../parameters/gx_repeat_select_dynamic.xml | 42 +++ .../parameters/gx_section_select_dynamic.xml | 42 +++ .../tool_util/parameter_specification.yml | 11 +- test/unit/tool_util/test_parameter_convert.py | 30 ++- .../tool_util/test_parameter_test_cases.py | 4 +- 41 files changed, 2008 insertions(+), 216 deletions(-) create mode 100644 lib/galaxy_test/api/test_tool_execution.py create mode 100644 test/functional/tools/parameters/gx_conditional_select_dynamic.xml create mode 100644 test/functional/tools/parameters/gx_repeat_select_dynamic.xml create mode 100644 test/functional/tools/parameters/gx_section_select_dynamic.xml diff --git a/.github/workflows/framework_tools.yaml b/.github/workflows/framework_tools.yaml index 3b38a477d90..2aa74ce2f7d 100644 --- a/.github/workflows/framework_tools.yaml +++ b/.github/workflows/framework_tools.yaml @@ -28,6 +28,7 @@ jobs: strategy: matrix: python-version: ['3.9'] + use-legacy-api: ['if_needed', 'always'] services: postgres: image: postgres:17 @@ -61,7 +62,7 @@ jobs: path: 'galaxy root/.venv' key: gxy-venv-${{ runner.os }}-${{ steps.full-python-version.outputs.version }}-${{ hashFiles('galaxy root/requirements.txt') }}-framework - name: Run tests - run: ./run_tests.sh --coverage --framework-tools + run: GALAXY_TEST_USE_LEGACY_TOOL_API="${{ matrix.use-legacy-api }}" ./run_tests.sh --coverage --framework-tools working-directory: 'galaxy root' - uses: codecov/codecov-action@v5 with: diff --git a/lib/galaxy/app.py b/lib/galaxy/app.py index 167b69be133..e1083e78f85 100644 --- a/lib/galaxy/app.py +++ b/lib/galaxy/app.py @@ -688,6 +688,10 @@ class GalaxyManagerApplication(MinimalManagerApp, MinimalGalaxyApplication): self._register_singleton(Registry, self.datatypes_registry) galaxy.model.set_datatypes_registry(self.datatypes_registry) self.configure_sentry_client() + # Load dbkey / genome build manager + self._configure_genome_builds(data_table_name="__dbkeys__", load_old_style=True) + # Tool Data Tables + self._configure_tool_data_tables(from_shed_config=False) self._configure_tool_shed_registry() self._register_singleton(tool_shed_registry.Registry, self.tool_shed_registry) @@ -766,10 +770,6 @@ class UniverseApplication(StructuredApp, GalaxyManagerApplication, InstallationT ) self.api_keys_manager = self._register_singleton(ApiKeyManager) - # Setup lazy variables - self._configure_tool_data_tables(from_shed_config=False) - self._configure_genome_builds() - # Genomes self.genomes = self._register_singleton(Genomes) # Data providers registry. diff --git a/lib/galaxy/celery/tasks.py b/lib/galaxy/celery/tasks.py index 974250cbf9e..51512beb5e4 100644 --- a/lib/galaxy/celery/tasks.py +++ b/lib/galaxy/celery/tasks.py @@ -31,6 +31,7 @@ from galaxy.managers.datasets import ( DatasetManager, ) from galaxy.managers.hdas import HDAManager +from galaxy.managers.jobs import JobSubmitter from galaxy.managers.lddas import LDDAManager from galaxy.managers.markdown_util import generate_branded_pdf from galaxy.managers.model_stores import ModelStoreManager @@ -56,6 +57,7 @@ from galaxy.schema.tasks import ( MaterializeDatasetInstanceTaskRequest, PrepareDatasetCollectionDownload, PurgeDatasetsTaskRequest, + QueueJobs, SetupHistoryExportJob, WriteHistoryContentTo, WriteHistoryTo, @@ -72,8 +74,10 @@ log = get_logger(__name__) @lru_cache -def cached_create_tool_from_representation(app: MinimalManagerApp, raw_tool_source: str): - return create_tool_from_representation(app=app, raw_tool_source=raw_tool_source, tool_source_class="XmlToolSource") +def cached_create_tool_from_representation(app: MinimalManagerApp, raw_tool_source: str, tool_dir: str = ""): + return create_tool_from_representation( + app=app, raw_tool_source=raw_tool_source, tool_dir=tool_dir, tool_source_class="XmlToolSource" + ) @galaxy_task(action="recalculate a user's disk usage") @@ -328,6 +332,17 @@ def fetch_data( return abort_when_job_stops(_fetch_data, session=sa_session, job_id=job_id, setup_return=setup_return) +@galaxy_task(action="queuing up submitted jobs") +def queue_jobs(request: QueueJobs, app: MinimalManagerApp, job_submitter: JobSubmitter): + tool = cached_create_tool_from_representation( + app, request.tool_source.raw_tool_source, tool_dir=request.tool_source.tool_dir + ) + job_submitter.queue_jobs( + tool, + request, + ) + + @galaxy_task(ignore_result=True, action="setting up export history job") def export_history( model_store_manager: ModelStoreManager, diff --git a/lib/galaxy/managers/jobs.py b/lib/galaxy/managers/jobs.py index 97e7787427d..1e3ad580c3d 100644 --- a/lib/galaxy/managers/jobs.py +++ b/lib/galaxy/managers/jobs.py @@ -1,6 +1,7 @@ import json import logging from collections.abc import Iterable +from dataclasses import dataclass from datetime import ( date, datetime, @@ -10,6 +11,7 @@ from typing import ( Any, cast, Optional, + Tuple, TYPE_CHECKING, Union, ) @@ -36,6 +38,7 @@ from typing_extensions import TypedDict from galaxy import model from galaxy.exceptions import ( ConfigDoesNotAllowException, + InconsistentDatabase, ItemAccessibilityException, ObjectNotFound, RequestParameterInvalidException, @@ -48,14 +51,20 @@ from galaxy.job_metrics import ( from galaxy.managers.collections import DatasetCollectionManager from galaxy.managers.context import ProvidesUserContext from galaxy.managers.datasets import DatasetManager -from galaxy.managers.hdas import HDAManager +from galaxy.managers.hdas import ( + dereference_input, + HDAManager, +) +from galaxy.managers.histories import HistoryManager from galaxy.managers.lddas import LDDAManager +from galaxy.managers.users import UserManager from galaxy.model import ( ImplicitCollectionJobs, ImplicitCollectionJobsJobAssociation, Job, JobMetricNumeric, JobParameter, + ToolRequest, User, Workflow, WorkflowInvocation, @@ -72,8 +81,23 @@ from galaxy.schema.schema import ( JobIndexQueryPayload, JobIndexSortByEnum, ) +from galaxy.schema.tasks import ( + MaterializeDatasetInstanceTaskRequest, + QueueJobs, +) from galaxy.security.idencoding import IdEncodingHelper -from galaxy.structured_app import StructuredApp +from galaxy.structured_app import ( + MinimalManagerApp, + StructuredApp, +) +from galaxy.tool_util.parameters import ( + DataRequestInternalHda, + DataRequestUri, + dereference, + RequestInternalDereferencedToolState, + RequestInternalToolState, +) +from galaxy.tools import Tool from galaxy.tools._types import ( ToolStateDumpedToJsonInternalT, ToolStateJobInstancePopulatedT, @@ -89,6 +113,7 @@ from galaxy.util.search import ( parse_filters_structured, RawTextTerm, ) +from galaxy.work.context import WorkRequestContext if TYPE_CHECKING: from sqlalchemy.sql.expression import Select @@ -155,6 +180,8 @@ class JobManager: workflow_id = payload.workflow_id invocation_id = payload.invocation_id implicit_collection_jobs_id = payload.implicit_collection_jobs_id + tool_request_id = payload.tool_request_id + search = payload.search order_by = payload.order_by @@ -171,6 +198,7 @@ class JobManager: def add_workflow_jobs(): wfi_step = select(WorkflowInvocationStep) + if workflow_id is not None: wfi_step = ( wfi_step.join(WorkflowInvocation).join(Workflow).where(Workflow.stored_workflow_id == workflow_id) @@ -185,6 +213,7 @@ class JobManager: ImplicitCollectionJobsJobAssociation.implicit_collection_jobs_id == wfi_step_sq.c.implicit_collection_jobs_id, ) + # Ensure the result is models, not tuples sq = stmt1.union(stmt2).subquery() # SQLite won't recognize Job.foo as a valid column for the ORDER BY clause due to the UNION clause, so we'll use the subquery `columns` collection (`sq.c`). @@ -262,6 +291,9 @@ class JobManager: if history_id is not None: stmt = stmt.where(Job.history_id == history_id) + if tool_request_id is not None: + stmt = stmt.filter(model.Job.tool_request_id == tool_request_id) + order_by_columns = Job if workflow_id or invocation_id: stmt, order_by_columns = add_workflow_jobs() @@ -1910,3 +1942,107 @@ def get_jobs_to_check_at_startup(session: galaxy_scoped_session, track_jobs_in_d def get_job(session: galaxy_scoped_session, *where_clauses): stmt = select(Job).where(*where_clauses).limit(1) return session.scalars(stmt).first() + + +@dataclass +class DereferencedDatasetPair: + hda: model.HistoryDatasetAssociation + request: DataRequestUri + + +class JobSubmitter: + def __init__( + self, + history_manager: HistoryManager, + user_manager: UserManager, + hda_manager: HDAManager, + app: MinimalManagerApp, + ): + self.history_manager = history_manager + self.user_manager = user_manager + self.hda_manager = hda_manager + self.app = app + + def materialize_request_for( + self, trans: WorkRequestContext, hda: model.HistoryDatasetAssociation + ) -> MaterializeDatasetInstanceTaskRequest: + return MaterializeDatasetInstanceTaskRequest( + user=trans.async_request_user, + history_id=trans.history.id, + source="hda", + content=hda.id, + ) + + def dereference( + self, trans: WorkRequestContext, tool: Tool, request: QueueJobs, tool_request: ToolRequest + ) -> Tuple[RequestInternalDereferencedToolState, list[DereferencedDatasetPair]]: + new_hdas: list[DereferencedDatasetPair] = [] + + def dereference_callback(data_request: DataRequestUri) -> DataRequestInternalHda: + # a deferred dataset corresponding to request + history = tool_request.history + if not history: + raise InconsistentDatabase("Tool request has no history associated") + + hda = dereference_input(trans, data_request, history) + if not isinstance(hda, model.HistoryDatasetAssociation): + raise RequestParameterInvalidException("Input dataset is not a history dataset association") + + new_hdas.append(DereferencedDatasetPair(hda, data_request)) + return DataRequestInternalHda(id=hda.id, src="hda") + + tool_state = RequestInternalToolState(tool_request.request) + return dereference(tool_state, tool, dereference_callback), new_hdas + + def queue_jobs(self, tool: Tool, request: QueueJobs) -> None: + tool_request: ToolRequest = self._tool_request(request.tool_request_id) + sa_session = self.app.model.context + try: + request_context = self._context(tool_request, request) + target_history = request_context.history + use_cached_jobs = request.use_cached_jobs + rerun_remap_job_id = request.rerun_remap_job_id + tool_state: RequestInternalDereferencedToolState + new_hdas: list[DereferencedDatasetPair] + tool_state, new_hdas = self.dereference(request_context, tool, request, tool_request) + to_materialize_list: list[DereferencedDatasetPair] = [p for p in new_hdas if not p.request.deferred] + for to_materialize in to_materialize_list: + materialize_request = self.materialize_request_for(request_context, to_materialize.hda) + # API dataset materialization is immutable and produces new datasets + # here we just created the datasets - lets just materialize them in place + # and avoid extra and confusing input copies + self.hda_manager.materialize(materialize_request, sa_session(), in_place=True) + tool.handle_input_async( + request_context, + tool_request, + tool_state, + history=target_history, + use_cached_job=use_cached_jobs, + rerun_remap_job_id=rerun_remap_job_id, + ) + tool_request.state = ToolRequest.states.SUBMITTED + sa_session.add(tool_request) + sa_session.commit() + except Exception as e: + log.exception("Problem validating tool state after request created") + tool_request.state = ToolRequest.states.FAILED + tool_request.state_message = str(e) + sa_session.add(tool_request) + sa_session.commit() + + def _context(self, tool_request: ToolRequest, request: QueueJobs) -> WorkRequestContext: + user = self.user_manager.by_id(request.user.user_id) + target_history = tool_request.history + trans = WorkRequestContext( + self.app, + user, + history=target_history, + ) + return trans + + def _tool_request(self, tool_request_id: int) -> ToolRequest: + sa_session = self.app.model.context + tool_request: ToolRequest = cast(ToolRequest, sa_session.query(ToolRequest).get(tool_request_id)) + if tool_request is None: + raise Exception(f"Problem fetching request with ID {tool_request_id}") + return tool_request diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py index b44a61685bb..8db5c6578f8 100644 --- a/lib/galaxy/model/__init__.py +++ b/lib/galaxy/model/__init__.py @@ -1408,6 +1408,28 @@ class ToolRequest(Base, Dictifiable, RepresentById): tool_source: Mapped["ToolSource"] = relationship() history: Mapped[Optional["History"]] = relationship(back_populates="tool_requests") + jobs: Mapped[list["Job"]] = relationship(back_populates="tool_request", order_by=lambda: asc(Job.id)) + implicit_collections: Mapped[list["ToolRequestImplicitCollectionAssociation"]] = relationship( + back_populates="tool_request" + ) + + +class ToolRequestImplicitCollectionAssociation(Base, Dictifiable, RepresentById): + __tablename__ = "tool_request_implicit_collection_association" + + id: Mapped[int] = mapped_column(primary_key=True) + tool_request_id: Mapped[int] = mapped_column(ForeignKey("tool_request.id", name="fk_trica_tri"), index=True) + dataset_collection_id: Mapped[int] = mapped_column( + ForeignKey("history_dataset_collection_association.id", name="fk_trica_dci"), index=True + ) + output_name: Mapped[str] = mapped_column(String(255)) + + tool_request: Mapped["ToolRequest"] = relationship(back_populates="implicit_collections") + dataset_collection: Mapped["HistoryDatasetCollectionAssociation"] = relationship( + back_populates="tool_request_association", uselist=False + ) + + dict_collection_visible_keys = ["id", "tool_request_id", "dataset_collection_id", "output_name"] class UserDynamicToolAssociation(Base, Dictifiable, RepresentById): @@ -1592,7 +1614,7 @@ class Job(Base, JobLike, UsesCreateAndUpdateTime, Dictifiable, Serializable): tool_request_id: Mapped[Optional[int]] = mapped_column(ForeignKey("tool_request.id"), index=True) dynamic_tool: Mapped[Optional["DynamicTool"]] = relationship() - tool_request: Mapped[Optional["ToolRequest"]] = relationship() + tool_request: Mapped[Optional["ToolRequest"]] = relationship(back_populates="jobs") user: Mapped[Optional["User"]] = relationship() galaxy_session: Mapped[Optional["GalaxySession"]] = relationship() history: Mapped[Optional["History"]] = relationship(back_populates="jobs") @@ -7546,6 +7568,9 @@ class HistoryDatasetCollectionAssociation( back_populates="dataset_collection", ) creating_job_associations: Mapped[list["JobToOutputDatasetCollectionAssociation"]] = relationship(viewonly=True) + tool_request_association: Mapped[Optional["ToolRequestImplicitCollectionAssociation"]] = relationship( + back_populates="dataset_collection" + ) dict_dbkeysandextensions_visible_keys = ["dbkeys", "extensions"] editable_keys = ("name", "deleted", "visible") diff --git a/lib/galaxy/schema/jobs.py b/lib/galaxy/schema/jobs.py index 404012e0206..46812b7c9a4 100644 --- a/lib/galaxy/schema/jobs.py +++ b/lib/galaxy/schema/jobs.py @@ -79,6 +79,19 @@ class JobOutputAssociation(JobAssociation): ) +class JobOutputCollectionAssociation(Model): + name: str = Field( + default=..., + title="name", + description="Name of the job parameter.", + ) + dataset_collection_instance: EncodedDataItemSourceId = Field( + default=..., + title="dataset_collection_instance", + description="Reference to the associated item.", + ) + + class ReportJobErrorPayload(Model): dataset_id: DecodedDatabaseIdField = Field( default=..., diff --git a/lib/galaxy/schema/schema.py b/lib/galaxy/schema/schema.py index 2d76c53043e..702dd4d08fe 100644 --- a/lib/galaxy/schema/schema.py +++ b/lib/galaxy/schema/schema.py @@ -1668,6 +1668,7 @@ class JobIndexQueryPayload(Model): workflow_id: Optional[DecodedDatabaseIdField] = None invocation_id: Optional[DecodedDatabaseIdField] = None implicit_collection_jobs_id: Optional[DecodedDatabaseIdField] = None + tool_request_id: Optional[DecodedDatabaseIdField] = None order_by: JobIndexSortByEnum = JobIndexSortByEnum.update_time search: Optional[str] = None limit: int = 500 @@ -3982,6 +3983,22 @@ class ToolRequestModel(Model): state_message: Optional[str] +class ToolRequestJobReference(Model): + src: Literal["job"] + id: EncodedDatabaseIdField + + +class ToolRequestImplicitCollectionReference(Model): + src: Literal["hdca"] + id: EncodedDatabaseIdField + output_name: str + + +class ToolRequestDetailedModel(ToolRequestModel): + jobs: list[ToolRequestJobReference] = Field(default=[]) + implicit_collections: list[ToolRequestImplicitCollectionReference] = Field(default=[]) + + class AsyncFile(Model): storage_request_id: UUID task: AsyncTaskResultSummary diff --git a/lib/galaxy/schema/tasks.py b/lib/galaxy/schema/tasks.py index b4475ff5d4e..85b27862267 100644 --- a/lib/galaxy/schema/tasks.py +++ b/lib/galaxy/schema/tasks.py @@ -151,3 +151,16 @@ class TaskResult(Model): title="Result", description="The result message of the task. Empty if the task is still running. If the task failed, this will contain the exception message.", ) + + +class ToolSource(Model): + raw_tool_source: str + tool_dir: str + + +class QueueJobs(Model): + tool_source: ToolSource + tool_request_id: int # links to request ("incoming") and history + user: RequestUser # TODO: test anonymous users through this submission path + use_cached_jobs: bool + rerun_remap_job_id: Optional[int] # link to a job to rerun & remap diff --git a/lib/galaxy/tool_util/parameters/__init__.py b/lib/galaxy/tool_util/parameters/__init__.py index c8c3875910d..4179dc7e0ec 100644 --- a/lib/galaxy/tool_util/parameters/__init__.py +++ b/lib/galaxy/tool_util/parameters/__init__.py @@ -1,3 +1,4 @@ +from galaxy.tool_util_models._types import is_optional from galaxy.tool_util_models.parameters import ( BooleanParameterModel, ColorParameterModel, @@ -141,6 +142,7 @@ __all__ = ( "validate_workflow_step", "validate_workflow_step_linked", "validate_explicit_conditional_test_value", + "is_optional", "ToolState", "TestCaseToolState", "ToolParameterT", diff --git a/lib/galaxy/tool_util/parameters/case.py b/lib/galaxy/tool_util/parameters/case.py index 09bb832e36b..776aae54828 100644 --- a/lib/galaxy/tool_util/parameters/case.py +++ b/lib/galaxy/tool_util/parameters/case.py @@ -149,7 +149,11 @@ def legacy_from_string(parameter: ToolParameterT, value: Optional[Any], warnings def test_case_state( - test_dict: ToolSourceTest, tool_parameter_bundle: List[ToolParameterT], profile: str, validate: bool = True + test_dict: ToolSourceTest, + tool_parameter_bundle: List[ToolParameterT], + profile: str, + validate: bool = True, + name: Optional[str] = None, ) -> TestCaseStateAndWarnings: warnings: List[str] = [] inputs: ToolSourceTestInputs = test_dict["inputs"] @@ -165,7 +169,7 @@ def test_case_state( tool_state = TestCaseToolState(state) if validate: - tool_state.validate(tool_parameter_bundle) + tool_state.validate(tool_parameter_bundle, name=name) for input_name in unhandled_inputs: raise Exception(f"Invalid parameter name found {input_name}") return TestCaseStateAndWarnings(tool_state, warnings, unhandled_inputs) diff --git a/lib/galaxy/tool_util/parameters/convert.py b/lib/galaxy/tool_util/parameters/convert.py index d9a075a5d3a..38c921194cb 100644 --- a/lib/galaxy/tool_util/parameters/convert.py +++ b/lib/galaxy/tool_util/parameters/convert.py @@ -23,6 +23,7 @@ from galaxy.tool_util_models.parameters import ( FloatParameterModel, HiddenParameterModel, IntegerParameterModel, + TextParameterModel, ToolParameterBundle, ToolParameterT, ) @@ -79,11 +80,11 @@ def cwl_runtime_model(input_models: ToolParameterBundle): def decode( - external_state: RequestToolState, input_models: ToolParameterBundle, decode_id: Callable[[str], int] + external_state: RequestToolState, input_models: ToolParameterBundle, decode_id: Callable[[str], int], name_base: Optional[str] = None ) -> RequestInternalToolState: """Prepare an internal representation of tool state (request_internal) for storing in the database.""" - external_state.validate(input_models) + external_state.validate(input_models, f"{name_base} (request model)") decode_callback = _decode_callback_for(decode_id) internal_state_dict = visit_input_values( input_models, @@ -92,7 +93,7 @@ def decode( ) internal_request_state = RequestInternalToolState(internal_state_dict) - internal_request_state.validate(input_models) + internal_request_state.validate(input_models, f"{name_base} (request internal model)") return internal_request_state @@ -163,8 +164,7 @@ def dereference( if parameter.parameter_type == "gx_data": if value is None: return VISITOR_NO_REPLACEMENT - if parameter.multiple: - assert isinstance(value, list), str(value) + if parameter.multiple and isinstance(value, list): return list(map(derefrence_dict, value)) else: assert isinstance(value, dict), str(value) @@ -328,6 +328,25 @@ def _fill_default_for(tool_state: Dict[str, Any], parameter: ToolParameterT) -> tool_state[parameter_name] = {} section_state = cast(Dict[str, Any], tool_state[parameter_name]) _fill_defaults(section_state, parameter) + elif parameter.parameter_type == "gx_data_collection": + collection_parameter = parameter + if parameter_name not in tool_state and collection_parameter.optional: + tool_state[parameter_name] = None + elif parameter.parameter_type in ["gx_text"]: + text_parameter = cast(TextParameterModel, parameter) + if parameter_name not in tool_state: + if not text_parameter.optional: + # restore legacy behavior of allowing empty string implicit default + # for these non-optional inputs. + tool_state[parameter_name] = "" + else: + tool_state[parameter_name] = None + else: + # legacy behavior of converting explicit None into implicit null. We should introduce + # a layer somewhere to deal with this behavior further up the stack and clean up these models. + if not text_parameter.optional and tool_state[parameter_name] is None: + tool_state[parameter_name] = "" + def _select_which_when( @@ -354,17 +373,25 @@ def _encode_callback_for(encode_id: EncodeFunctionT) -> Callback: else: return src_dict + def encode_element(element: dict): + if element.get("__class__") == "Batch": + encoded = element.copy() + values = encoded.pop("values") + encoded["values"] = list(map(encode_src_dict, values)) + return encoded + else: + return encode_src_dict(element) + def encode_callback(parameter: ToolParameterT, value: Any): if parameter.parameter_type == "gx_data": - if parameter.multiple: - assert isinstance(value, list), str(value) - return list(map(encode_src_dict, value)) + if parameter.multiple and isinstance(value, list): + return list(map(encode_element, value)) else: assert isinstance(value, dict), str(value) - return encode_src_dict(value) + return encode_element(value) elif parameter.parameter_type == "gx_data_collection": assert isinstance(value, dict), str(value) - return encode_src_dict(value) + return encode_element(value) else: return VISITOR_NO_REPLACEMENT @@ -381,16 +408,24 @@ def _decode_callback_for(decode_id: DecodeFunctionT) -> Callback: else: return src_dict + def decode_element(element: dict): + if element.get("__class__") == "Batch": + decoded = element.copy() + values = decoded.pop("values") + decoded["values"] = list(map(decode_src_dict, values)) + return decoded + else: + return decode_src_dict(element) + def decode_callback(parameter: ToolParameterT, value: Any): if parameter.parameter_type == "gx_data": if value is None: return VISITOR_NO_REPLACEMENT - if parameter.multiple: - assert isinstance(value, list), str(value) - return list(map(decode_src_dict, value)) + if parameter.multiple and isinstance(value, list): + return list(map(decode_element, value)) else: assert isinstance(value, dict), str(value) - return decode_src_dict(value) + return decode_element(value) elif parameter.parameter_type == "gx_data_collection": if value is None: return VISITOR_NO_REPLACEMENT diff --git a/lib/galaxy/tool_util/parameters/factory.py b/lib/galaxy/tool_util/parameters/factory.py index a13ee67ce04..54ea9bb1769 100644 --- a/lib/galaxy/tool_util/parameters/factory.py +++ b/lib/galaxy/tool_util/parameters/factory.py @@ -215,7 +215,10 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool select_validators: List[SelectCompatiableValidators] = [] for static_validator in static_validator_models: if static_validator.type == "no_options": - select_validators.append(static_validator) + # test case test_tool_execute::test_select_optional_null_by_default verifies + # these validators don't get applied effectively if the select is optional. + if not optional: + select_validators.append(static_validator) return SelectParameterModel( type="select", name=input_source.parse_name(), diff --git a/lib/galaxy/tool_util/parser/interface.py b/lib/galaxy/tool_util/parser/interface.py index eba5fd3f393..e2f3bc1a6e9 100644 --- a/lib/galaxy/tool_util/parser/interface.py +++ b/lib/galaxy/tool_util/parser/interface.py @@ -722,6 +722,8 @@ class TestCollectionDef: test_format_dict = JsonTestCollectionDefDict(**self._test_format_to_dict()) if self.name: test_format_dict["name"] = self.name + if self.fields is not None: + test_format_dict["fields"] = self.fields return test_format_dict def to_dict(self) -> XmlTestCollectionDefDict: diff --git a/lib/galaxy/tool_util/verify/_types.py b/lib/galaxy/tool_util/verify/_types.py index e5aa85f1ddb..c532dab9aa6 100644 --- a/lib/galaxy/tool_util/verify/_types.py +++ b/lib/galaxy/tool_util/verify/_types.py @@ -19,10 +19,15 @@ from galaxy.tool_util.parser.interface import ( ToolSourceTestOutputs, ) -# inputs that have been processed with parse.py and expanded out +# legacy inputs for working with POST /api/tools +# + inputs that have been processed with parse.py and expanded out ExpandedToolInputs = Dict[str, Any] -# ExpandedToolInputs where any model objects have been json-ified with to_dict() +# + ExpandedToolInputs where any model objects have been json-ified with to_dict() ExpandedToolInputsJsonified = Dict[str, Any] + +# modern inputs for working with POST /api/jobs* +RawTestToolRequest = Dict[str, Any] + ExtraFileInfoDictT = Dict[str, Any] RequiredFileTuple = Tuple[str, ExtraFileInfoDictT] RequiredFilesT = List[RequiredFileTuple] @@ -36,6 +41,8 @@ class ToolTestDescriptionDict(TypedDict): name: str test_index: int inputs: ExpandedToolInputsJsonified + request: NotRequired[Optional[Dict[str, Any]]] + request_schema: NotRequired[Optional[Dict[str, Any]]] outputs: ToolSourceTestOutputs output_collections: List[TestSourceTestOutputColllection] stdout: Optional[AssertionList] diff --git a/lib/galaxy/tool_util/verify/interactor.py b/lib/galaxy/tool_util/verify/interactor.py index 26c1c529e0f..3a019e37d75 100644 --- a/lib/galaxy/tool_util/verify/interactor.py +++ b/lib/galaxy/tool_util/verify/interactor.py @@ -36,6 +36,14 @@ from typing_extensions import ( from galaxy import util from galaxy.tool_util.client.staging import StagingInterface +from galaxy.tool_util.parameters import ( + DataCollectionRequest, + DataRequestHda, + encode_test, + input_models_from_json, + TestCaseToolState, + ToolParameterBundle, +) from galaxy.tool_util.parser.interface import ( AssertionList, TestCollectionDef, @@ -45,6 +53,10 @@ from galaxy.tool_util.parser.interface import ( XmlTestCollectionDefDict, ) from galaxy.tool_util.verify.test_data import TestDataResolver +from galaxy.tool_util_models.tool_source import ( + JsonTestCollectionDefDict, + JsonTestDatasetDefDict, +) from galaxy.util import requests from galaxy.util.bunch import Bunch from galaxy.util.compression_utils import CompressedFile @@ -56,6 +68,7 @@ from . import verify from ._types import ( ExpandedToolInputs, ExpandedToolInputsJsonified, + RawTestToolRequest, RequiredDataTablesT, RequiredFilesT, RequiredLocFileT, @@ -66,6 +79,9 @@ from .wait import wait_on log = getLogger(__name__) +UseLegacyApiT = Literal["always", "never", "if_needed"] +DEFAULT_USE_LEGACY_API: UseLegacyApiT = "always" + # Off by default because it can pound the database pretty heavily # and result in sqlite errors on larger tests or larger numbers of # tests. @@ -105,6 +121,8 @@ JobDataCallbackT = Callable[[JobDataT], None] class ValidToolTestDict(TypedDict): inputs: ExpandedToolInputs + request: NotRequired[Optional[RawTestToolRequest]] + request_schema: NotRequired[Optional[Dict[str, Any]]] outputs: ToolSourceTestOutputs output_collections: List[TestSourceTestOutputColllection] stdout: NotRequired[AssertionList] @@ -155,7 +173,7 @@ def stage_data_in_history( tool_version=None, test_data_resolver: Optional[TestDataResolver] = None, ): - assert tool_id + assert tool_id, "Tool id not set" staging_interface = InteractorStagingInterface(galaxy_interactor, maxseconds=maxseconds, upload_async=UPLOAD_ASYNC) job = {} @@ -268,6 +286,15 @@ class GalaxyInteractorApi: assert response.status_code == 200, f"Non 200 response from tool tests available API. [{response.content}]" return response.json() + def get_tool_inputs(self, tool_id: str, tool_version: Optional[str] = None) -> ToolParameterBundle: + url = f"tools/{tool_id}/inputs" + params = {"tool_version": tool_version} if tool_version else None + response = self._get(url, data=params) + assert response.status_code == 200, f"Non 200 response from tool inputs API. [{response.content}]" + raw_inputs_array = response.json() + tool_parameter_bundle = input_models_from_json(raw_inputs_array) + return tool_parameter_bundle + def get_tool_tests(self, tool_id: str, tool_version: Optional[str] = None) -> List[ToolTestDescriptionDict]: url = f"tools/{tool_id}/test_data" params = {"tool_version": tool_version} if tool_version else None @@ -398,9 +425,27 @@ class GalaxyInteractorApi: def wait_for_job(self, job_id: str, history_id: Optional[str] = None, maxseconds=DEFAULT_TOOL_TEST_WAIT) -> None: self.wait_for(lambda: self.__job_ready(job_id, history_id), maxseconds=maxseconds) + def wait_on_tool_request(self, tool_request_id: str): + def state(): + state_response = self._get(f"tool_requests/{tool_request_id}/state") + state_response.raise_for_status() + return state_response.json() + + def is_ready(): + is_complete = state() in ["submitted", "failed"] + return True if is_complete else None + + self.wait_for(is_ready, "waiting for tool request to submit") + return state() == "submitted" + + def get_tool_request(self, tool_request_id: str): + response_raw = self._get(f"tool_requests/{tool_request_id}") + response_raw.raise_for_status() + return response_raw.json() + def wait_for(self, func: Callable, what: str = "tool test run", **kwd) -> None: walltime_exceeded = int(kwd.get("maxseconds", DEFAULT_TOOL_TEST_WAIT)) - wait_on(func, what, walltime_exceeded) + return wait_on(func, what, walltime_exceeded) def get_job_stdio(self, job_id: str) -> Dict[str, Any]: return self.__get_job_stdio(job_id).json() @@ -632,39 +677,71 @@ class GalaxyInteractorApi: return location def run_tool( - self, testdef: "ToolTestDescription", history_id: str, resource_parameters: Optional[Dict[str, Any]] = None + self, + testdef: "ToolTestDescription", + history_id: str, + resource_parameters: Optional[Dict[str, Any]] = None, + use_legacy_api: UseLegacyApiT = DEFAULT_USE_LEGACY_API, ) -> RunToolResponse: # We need to handle the case where we've uploaded a valid compressed file since the upload # tool will have uncompressed it on the fly. resource_parameters = resource_parameters or {} - inputs_tree = testdef.inputs.copy() - for key, value in inputs_tree.items(): - values = [value] if not isinstance(value, list) else value - new_values = [] - for value in values: - if isinstance(value, TestCollectionDef): - hdca_id = self._create_collection(history_id, value) - new_values = [dict(src="hdca", id=hdca_id)] - elif value in self.uploads: - new_values.append(self.uploads[value]) - else: - new_values.append(value) - inputs_tree[key] = new_values + request = testdef.request + request_schema = testdef.request_schema + submit_with_legacy_api = use_legacy_api == "always" or (use_legacy_api == "if_needed" and request is None) + if submit_with_legacy_api: + inputs_tree = testdef.inputs.copy() + for key, value in inputs_tree.items(): + values = [value] if not isinstance(value, list) else value + new_values = [] + for value in values: + if isinstance(value, TestCollectionDef): + hdca_id = self._create_collection(history_id, value) + new_values = [dict(src="hdca", id=hdca_id)] + elif value in self.uploads: + new_values.append(self.uploads[value]) + else: + new_values.append(value) + inputs_tree[key] = new_values + + # HACK: Flatten single-value lists. Required when using expand_grouping + for key, value in inputs_tree.items(): + if isinstance(value, list) and len(value) == 1: + inputs_tree[key] = value[0] + else: + assert request is not None, "Request not set" + assert request_schema is not None, "Request schema not set" + parameters = request_schema["parameters"] + + def adapt_datasets(test_input: JsonTestDatasetDefDict) -> DataRequestHda: + # if path is not set it might be a composite file with a path, + # e.g. composite_shapefile + test_input_path = test_input.get("path", "") + return DataRequestHda(**self.uploads[test_input_path]) + + def adapt_collections(test_input: JsonTestCollectionDefDict) -> DataCollectionRequest: + test_collection_def = TestCollectionDef.from_dict(test_input) + hdca_id = self._create_collection(history_id, test_collection_def) + return DataCollectionRequest(src="hdca", id=hdca_id) + + test_case_state = TestCaseToolState(input_state=request) + inputs_tree = encode_test( + test_case_state, input_models_from_json(parameters), adapt_datasets, adapt_collections + ).input_state if resource_parameters: inputs_tree["__job_resource|__job_resource__select"] = "yes" for key, value in resource_parameters.items(): inputs_tree[f"__job_resource|{key}"] = value - # HACK: Flatten single-value lists. Required when using expand_grouping - for key, value in inputs_tree.items(): - if isinstance(value, list) and len(value) == 1: - inputs_tree[key] = value[0] - submit_response = None for _ in range(DEFAULT_TOOL_TEST_WAIT): submit_response = self.__submit_tool( - history_id, tool_id=testdef.tool_id, tool_input=inputs_tree, tool_version=testdef.tool_version + history_id, + tool_id=testdef.tool_id, + tool_input=inputs_tree, + tool_version=testdef.tool_version, + use_legacy_api=submit_with_legacy_api, ) if _are_tool_inputs_not_ready(submit_response): print("Tool inputs not ready yet") @@ -673,10 +750,36 @@ class GalaxyInteractorApi: else: break submit_response_object = ensure_tool_run_response_okay(submit_response, "execute tool", inputs_tree) - try: + if not submit_with_legacy_api: + tool_request_id = submit_response_object["tool_request_id"] + successful = self.wait_on_tool_request(tool_request_id) + if not successful: + request = self.get_tool_request(tool_request_id) or {} + raise RunToolException( + f"Tool request failure - state {request.get('state')}, message: {request.get('state_message')}", + inputs_tree, + ) + job_refs = self.jobs_for_tool_request(tool_request_id) + outputs = OutputsDict() + output_collections = {} + if len(job_refs) != 1: + raise Exception( + f"Found incorrect number of jobs for tool request - was expecting a single job {job_refs}" + ) + assert len(job_refs) == 1, job_refs + job_id = job_refs[0]["id"] + jobs = [self.__get_job(job_id).json()] + job_outputs = self.job_outputs(job_id) + for job_output in job_outputs: + if "dataset" in job_output: + outputs[job_output["name"]] = job_output["dataset"] + else: + output_collections[job_output["name"]] = job_output["dataset_collection_instance"] + else: outputs = self.__dictify_outputs(submit_response_object) output_collections = self.__dictify_output_collections(submit_response_object) jobs = submit_response_object["jobs"] + try: return RunToolResponse( inputs=inputs_tree, outputs=outputs, @@ -698,7 +801,9 @@ class GalaxyInteractorApi: ) if collection_def.fields: create_payload["fields"] = collection_def.fields - return self._post("dataset_collections", data=create_payload, json=True).json()["id"] + create_response = self._post("dataset_collections", data=create_payload, json=True) + create_response.raise_for_status() + return create_response.json()["id"] def _element_identifiers(self, collection_def): element_identifiers = [] @@ -834,6 +939,16 @@ class GalaxyInteractorApi: dataset_json = self._get(f"histories/{history_id}/contents/{id}").json() return dataset_json + def jobs_for_tool_request(self, tool_request_id: str) -> List[Dict[str, Any]]: + job_list_response = self._get(f"tool_requests/{tool_request_id}") + job_list_response.raise_for_status() + return job_list_response.json()["jobs"] + + def job_outputs(self, job_id: str) -> List[Dict[str, Any]]: + outputs = self._get(f"jobs/{job_id}/outputs") + outputs.raise_for_status() + return outputs.json() + def __contents(self, history_id: str) -> List[Dict[str, Any]]: history_contents_response = self._get(f"histories/{history_id}/contents") history_contents_response.raise_for_status() @@ -858,12 +973,25 @@ class GalaxyInteractorApi: extra_data: Optional[dict] = None, files: Optional[dict] = None, tool_version: Optional[str] = None, + use_legacy_api: bool = True, ): extra_data = extra_data or {} - data = dict( - history_id=history_id, tool_id=tool_id, inputs=dumps(tool_input), tool_version=tool_version, **extra_data - ) - return self._post("tools", files=files, data=data) + if use_legacy_api: + data = dict( + history_id=history_id, + tool_id=tool_id, + inputs=dumps(tool_input), + tool_version=tool_version, + **extra_data, + ) + return self._post("tools", files=files, data=data) + else: + assert files is None + data = dict( + history_id=history_id, tool_id=tool_id, inputs=tool_input, tool_version=tool_version, **extra_data + ) + submit_tool_request_response = self._post("jobs", data=data, json=True) + return submit_tool_request_response def ensure_user_with_email(self, email, password=None): admin_key = self.master_api_key @@ -1391,6 +1519,7 @@ def verify_tool( register_job_data: Optional[JobDataCallbackT] = None, test_index: int = 0, tool_version: Optional[str] = None, + use_legacy_api: UseLegacyApiT = DEFAULT_USE_LEGACY_API, quiet: bool = False, test_history: Optional[str] = None, no_history_cleanup: bool = False, @@ -1408,11 +1537,7 @@ def verify_tool( if client_test_config is None: client_test_config = NullClientTestConfig() tool_test_dicts = _tool_test_dicts or galaxy_interactor.get_tool_tests(tool_id, tool_version=tool_version) - tool_test_dict = tool_test_dicts[test_index] - if "test_index" not in tool_test_dict: - tool_test_dict["test_index"] = test_index - if "tool_id" not in tool_test_dict: - tool_test_dict["tool_id"] = tool_id + tool_test_dict: ToolTestDescriptionDict = tool_test_dicts[test_index] if tool_version is None and "tool_version" in tool_test_dict: tool_version = tool_test_dict.get("tool_version") @@ -1478,7 +1603,9 @@ def verify_tool( input_staging_exc_info = sys.exc_info() raise try: - tool_response = galaxy_interactor.run_tool(testdef, test_history, resource_parameters=resource_parameters) + tool_response = galaxy_interactor.run_tool( + testdef, test_history, resource_parameters=resource_parameters, use_legacy_api=use_legacy_api + ) data_list, jobs, tool_inputs = tool_response.outputs, tool_response.jobs, tool_response.inputs data_collection_list = tool_response.output_collections except RunToolException as e: @@ -1765,6 +1892,8 @@ def adapt_tool_source_dict(processed_dict: ToolTestDict) -> ToolTestDescriptionD expect_test_failure: bool = DEFAULT_EXPECT_TEST_FAILURE inputs: ExpandedToolInputsJsonified = {} maxseconds: Optional[int] = None + request: Optional[Dict[str, Any]] = None + request_schema: Optional[Dict[str, Any]] = None if not error_in_test_definition: processed_test_dict = cast(ValidToolTestDict, processed_dict) @@ -1790,6 +1919,8 @@ def adapt_tool_source_dict(processed_dict: ToolTestDict) -> ToolTestDescriptionD expect_failure = processed_test_dict.get("expect_failure", DEFAULT_EXPECT_FAILURE) expect_test_failure = processed_test_dict.get("expect_test_failure", DEFAULT_EXPECT_TEST_FAILURE) inputs = processed_test_dict.get("inputs", {}) + request = processed_test_dict.get("request", None) + request_schema = processed_test_dict.get("request_schema", None) else: invalid_test_dict = cast(InvalidToolTestDict, processed_dict) maxseconds = DEFAULT_TOOL_TEST_WAIT @@ -1817,6 +1948,8 @@ def adapt_tool_source_dict(processed_dict: ToolTestDict) -> ToolTestDescriptionD expect_failure=expect_failure, expect_test_failure=expect_test_failure, inputs=inputs, + request=request, + request_schema=request_schema, ) @@ -1875,6 +2008,8 @@ class ToolTestDescription: expect_test_failure: bool exception: Optional[str] inputs: ExpandedToolInputs + request: Optional[Dict[str, Any]] + request_schema: Optional[Dict[str, Any]] outputs: ToolSourceTestOutputs output_collections: List[TestCollectionOutputDef] maxseconds: Optional[int] @@ -1903,6 +2038,8 @@ class ToolTestDescription: self.expect_failure = json_dict.get("expect_failure", DEFAULT_EXPECT_FAILURE) self.expect_test_failure = json_dict.get("expect_test_failure", DEFAULT_EXPECT_TEST_FAILURE) self.inputs = expanded_inputs_from_json(json_dict.get("inputs", {})) + self.request = json_dict.get("request", None) + self.request_schema = json_dict.get("request_schema", None) self.tool_id = json_dict["tool_id"] self.tool_version = json_dict.get("tool_version") self.maxseconds = json_dict.get("maxseconds") @@ -1934,6 +2071,8 @@ class ToolTestDescription: "required_files": self.required_files, "required_data_tables": self.required_data_tables, "required_loc_files": self.required_loc_files, + "request": self.request, + "request_schema": self.request_schema, "error": self.error, "exception": self.exception, } diff --git a/lib/galaxy/tool_util/verify/parse.py b/lib/galaxy/tool_util/verify/parse.py index 343c8892c8c..c272cf6c44a 100644 --- a/lib/galaxy/tool_util/verify/parse.py +++ b/lib/galaxy/tool_util/verify/parse.py @@ -1,8 +1,10 @@ import logging import os import traceback +from dataclasses import dataclass from typing import ( Any, + Dict, Iterable, List, Optional, @@ -15,6 +17,8 @@ from packaging.version import Version from galaxy.tool_util.parameters import ( input_models_for_tool_source, test_case_state as case_state, + TestCaseToolState, + ToolParameterBundleModel, ) from galaxy.tool_util.parser.interface import ( InputSource, @@ -67,15 +71,18 @@ def parse_tool_test_descriptions( profile = tool_source.parse_profile() for i, raw_test_dict in enumerate(raw_tests_dict.get("tests", [])): validation_exception: Optional[Exception] = None - if validate_on_load: + request_and_schema: Optional[TestRequestAndSchema] = None + try: tool_parameter_bundle = input_models_for_tool_source(tool_source) - try: - case_state(raw_test_dict, tool_parameter_bundle.parameters, profile, validate=True) - except Exception as e: - # TOOD: restrict types of validation exceptions a bit probably? - validation_exception = e + validated_test_case = case_state(raw_test_dict, tool_parameter_bundle.parameters, profile, validate=True) + request_and_schema = TestRequestAndSchema( + validated_test_case.tool_state, + tool_parameter_bundle, + ) + except Exception as e: + validation_exception = e - if validation_exception: + if validation_exception and validate_on_load: tool_id, tool_version = _tool_id_and_version(tool_source, tool_guid) test = ToolTestDescription.from_tool_source_dict( InvalidToolTestDict( @@ -91,13 +98,23 @@ def parse_tool_test_descriptions( ) ) else: - test = _description_from_tool_source(tool_source, raw_test_dict, i, tool_guid) + test = _description_from_tool_source(tool_source, raw_test_dict, i, tool_guid, request_and_schema) tests.append(test) return tests +@dataclass +class TestRequestAndSchema: + request: TestCaseToolState + request_schema: ToolParameterBundleModel + + def _description_from_tool_source( - tool_source: ToolSource, raw_test_dict: ToolSourceTest, test_index: int, tool_guid: Optional[str] + tool_source: ToolSource, + raw_test_dict: ToolSourceTest, + test_index: int, + tool_guid: Optional[str], + request_and_schema: Optional[TestRequestAndSchema], ) -> ToolTestDescription: required_files: RequiredFilesT = [] required_data_tables: RequiredDataTablesT = [] @@ -110,6 +127,12 @@ def _description_from_tool_source( if maxseconds is not None: maxseconds = int(maxseconds) + request: Optional[Dict[str, Any]] = None + request_schema: Optional[Dict[str, Any]] = None + if request_and_schema: + request = request_and_schema.request.input_state + request_schema = request_and_schema.request_schema.dict() + tool_id, tool_version = _tool_id_and_version(tool_source, tool_guid) processed_test_dict: Union[ValidToolTestDict, InvalidToolTestDict] try: @@ -124,6 +147,8 @@ def _description_from_tool_source( processed_test_dict = ValidToolTestDict( { "inputs": processed_inputs, + "request": request, + "request_schema": request_schema, "outputs": raw_test_dict["outputs"], "output_collections": raw_test_dict["output_collections"], "num_outputs": num_outputs, diff --git a/lib/galaxy/tool_util_models/parameters.py b/lib/galaxy/tool_util_models/parameters.py index 15cc028c0ec..bbfa3f4d58c 100644 --- a/lib/galaxy/tool_util_models/parameters.py +++ b/lib/galaxy/tool_util_models/parameters.py @@ -122,6 +122,7 @@ def allow_batching(job_template: DynamicModelInformation, batch_type: Optional[T class BatchRequest(StrictModel): meta_class: Literal["Batch"] = Field(..., alias="__class__") values: List[batch_type] # type: ignore[valid-type] + linked: Optional[bool] = None # maybe True instead? request_type = union_type([job_py_type, BatchRequest]) @@ -273,8 +274,17 @@ class TextParameterModel(BaseGalaxyToolParameterModelDefinition): def py_type(self) -> Type: return optional_if_needed(StrictStr, self.optional) + @property + def py_type_request(self) -> Type: + # such a hack but explicit nulls are always allowed in the API even for non-optional + # parameters - it becomes "" in the internal state. + return optional(StrictStr) + def pydantic_template(self, state_representation: StateRepresentationT) -> DynamicModelInformation: - py_type = decorate_type_with_validators_if_needed(self.py_type, self.validators) + py_type = self.py_type + if state_representation in ["request", "request_internal", "request_internal_dereferenced"]: + py_type = self.py_type_request + py_type = decorate_type_with_validators_if_needed(py_type, self.validators) if state_representation == "workflow_step_linked": py_type = allow_connected_value(py_type) requires_value = self.request_requires_value @@ -303,8 +313,15 @@ class IntegerParameterModel(BaseGalaxyToolParameterModelDefinition): def py_type(self) -> Type: return optional_if_needed(StrictInt, self.optional) + #@property + #def py_type_request(self) -> Type: + # # ugh... we allow explicit nulls in the API even if the input is not optional + # return optional_if_needed(StrictInt, True) + def pydantic_template(self, state_representation: StateRepresentationT) -> DynamicModelInformation: py_type = self.py_type + # if state_representation == "request": + # py_type = self.py_type_request validators = self.validators[:] if self.min is not None or self.max is not None: validators.append(InRangeParameterValidatorModel(min=self.min, max=self.max, implicit=True)) @@ -1138,7 +1155,7 @@ class GenomeBuildParameterModel(BaseGalaxyToolParameterModelDefinition): py_type: Type = StrictStr if self.multiple: py_type = list_type(py_type) - return optional_if_needed(py_type, self.optional) + return optional_if_needed(py_type, self.optional or self.multiple) def pydantic_template(self, state_representation: StateRepresentationT) -> DynamicModelInformation: requires_value = self.request_requires_value diff --git a/lib/galaxy/tool_util_models/tool_source.py b/lib/galaxy/tool_util_models/tool_source.py index d8602d8f584..88f0452a062 100644 --- a/lib/galaxy/tool_util_models/tool_source.py +++ b/lib/galaxy/tool_util_models/tool_source.py @@ -216,6 +216,7 @@ BaseJsonTestCollectionDefCollectionElementDict = TypedDict( "class": Literal["Collection"], "collection_type": Optional[str], "elements": NotRequired[Optional[List[JsonTestCollectionDefElementDict]]], + "fields": NotRequired[Optional[List[FieldDict]]], }, ) @@ -236,5 +237,6 @@ JsonTestCollectionDefDict = TypedDict( "collection_type": Optional[str], "elements": NotRequired[Optional[List[JsonTestCollectionDefElementDict]]], "name": NotRequired[Optional[str]], + "fields": NotRequired[Optional[List[FieldDict]]], }, ) diff --git a/lib/galaxy/tools/__init__.py b/lib/galaxy/tools/__init__.py index 6c8ae533ad4..3a404c2047d 100644 --- a/lib/galaxy/tools/__init__.py +++ b/lib/galaxy/tools/__init__.py @@ -56,6 +56,7 @@ from galaxy.model import ( Job, JobToOutputDatasetAssociation, StoredWorkflow, + ToolRequest, ) from galaxy.model.dataset_collections.matching import MatchingCollections from galaxy.schema.credentials import CredentialsContext @@ -80,7 +81,10 @@ from galaxy.tool_util.ontologies.ontology_data import ( ) from galaxy.tool_util.output_checker import DETECTED_JOB_STATE from galaxy.tool_util.parameters import ( + fill_static_defaults, input_models_for_pages, + JobInternalToolState, + RequestInternalDereferencedToolState, ToolParameterBundle, ) from galaxy.tool_util.parser import ( @@ -138,12 +142,14 @@ from galaxy.tools.execution_helpers import ToolExecutionCache from galaxy.tools.imp_exp import JobImportHistoryArchiveWrapper from galaxy.tools.parameters import ( check_param, + fill_dynamic_defaults, params_from_strings, params_to_incoming, params_to_json, params_to_json_internal, params_to_strings, populate_state, + populate_state_async, visit_input_values, ) from galaxy.tools.parameters.basic import ( @@ -170,7 +176,10 @@ from galaxy.tools.parameters.grouping import ( UploadDataset, ) from galaxy.tools.parameters.input_translation import ToolInputTranslator -from galaxy.tools.parameters.meta import expand_meta_parameters +from galaxy.tools.parameters.meta import ( + expand_meta_parameters, + expand_meta_parameters_async, +) from galaxy.tools.parameters.populate_model import populate_model from galaxy.tools.parameters.workflow_utils import workflow_building_modes from galaxy.tools.parameters.wrapped_json import json_wrap @@ -215,6 +224,7 @@ from ._types import ( ToolRequestT, ToolStateDumpedToJsonInternalT, ToolStateDumpedToJsonT, + ToolStateJobInstanceExpansionT, ToolStateJobInstancePopulatedT, ToolStateJobInstanceT, ) @@ -225,7 +235,8 @@ from .execute import ( DEFAULT_RERUN_REMAP_JOB_ID, DEFAULT_SET_OUTPUT_HID, DEFAULT_USE_CACHED_JOB, - execute as execute_job, + execute as execute_sync, + execute_async, ExecutionSlice, JobCallbackT, MappingParameters, @@ -2060,6 +2071,64 @@ class Tool(UsesDictVisibleKeys, ToolParameterBundle): if self.check_values: visit_input_values(self.inputs, values, callback) + def expand_incoming_async( + self, + request_context: WorkRequestContext, + tool_request_internal_state: RequestInternalDereferencedToolState, + rerun_remap_job_id: Optional[int], + ) -> tuple[ + list[ToolStateJobInstancePopulatedT], + list[ToolStateJobInstancePopulatedT], + Optional[MatchingCollections], + list[JobInternalToolState], + ]: + """The tool request API+tasks version of expand_incoming. + + This is responsible for breaking the map over job requests into individual jobs for execution. + """ + if self.input_translator: + raise exceptions.RequestParameterInvalidException( + "Failure executing tool request with id '%s' (cannot validate inputs from this type of data source tool - please POST to /api/tools).", + self.id, + ) + + set_dataset_matcher_factory(request_context, self) + + expanded_incomings: list[ToolStateJobInstanceExpansionT] + job_tool_states: list[ToolStateJobInstanceT] + collection_info: Optional[MatchingCollections] + expanded_incomings, job_tool_states, collection_info = expand_meta_parameters_async( + request_context.app, self, tool_request_internal_state + ) + + self._ensure_expansion_is_valid(job_tool_states, rerun_remap_job_id) + + # Process incoming data + validation_timer = self.app.execution_timer_factory.get_timer( + "internals.galaxy.tools.validation", + "Validated and populated state for tool request", + ) + all_errors = [] + all_params: list[ToolStateJobInstancePopulatedT] = [] + internal_states: list[JobInternalToolState] = [] + for expanded_incoming, job_tool_state in zip(expanded_incomings, job_tool_states): + expanded_incoming = fill_static_defaults(expanded_incoming, self, self.profile) + job_tool_state = fill_static_defaults(job_tool_state, self, self.profile) + params, errors = self._populate_async(request_context, expanded_incoming) + # params have had dynamic defaults requiring like dataset contents expanded out + # so we can use that backfill job_tool_state + fill_dynamic_defaults(request_context, self.inputs, job_tool_state, params) + internal_tool_state = JobInternalToolState(job_tool_state) + internal_tool_state.validate(self, f"{self.id} (job internal model)") + + internal_states.append(internal_tool_state) + all_errors.append(errors) + all_params.append(params) + unset_dataset_matcher_factory(request_context) + + log.info(validation_timer) + return all_params, all_errors, collection_info, internal_states + def expand_incoming( self, request_context: WorkRequestContext, incoming: ToolRequestT, input_format: InputFormatT = "legacy" ) -> tuple[ @@ -2072,7 +2141,7 @@ class Tool(UsesDictVisibleKeys, ToolParameterBundle): set_dataset_matcher_factory(request_context, self) # Fixed set of input parameters may correspond to any number of jobs. # Expand these out to individual parameters for given jobs (tool executions). - expanded_incomings: list[ToolStateJobInstanceT] + expanded_incomings: list[ToolStateJobInstanceExpansionT] collection_info: Optional[MatchingCollections] expanded_incomings, collection_info = expand_meta_parameters( request_context, self, incoming, input_format=input_format @@ -2098,7 +2167,9 @@ class Tool(UsesDictVisibleKeys, ToolParameterBundle): return all_params, all_errors, rerun_remap_job_id, collection_info def _ensure_expansion_is_valid( - self, expanded_incomings: list[ToolStateJobInstanceT], rerun_remap_job_id: Optional[int] + self, + expanded_incomings: Union[list[JobInternalToolState], list[ToolStateJobInstanceT]], + rerun_remap_job_id: Optional[int], ) -> None: """If the request corresponds to multiple jobs but this doesn't work with request configuration - raise an error. @@ -2147,6 +2218,33 @@ class Tool(UsesDictVisibleKeys, ToolParameterBundle): self._handle_validate_input_hook(request_context, params, errors) return params, errors + def _populate_async( + self, request_context, expanded_incoming: ToolStateJobInstanceT + ) -> tuple[ToolStateJobInstancePopulatedT, ParameterValidationErrorsT]: + """Validate expanded parameters for a job to replace references with model objects. + + So convert a ToolStateJobInstanceT to a ToolStateJobInstancePopulatedT. + """ + params: ToolStateJobInstancePopulatedT = {} + errors: ParameterValidationErrorsT = {} + if self.input_translator: + self.input_translator.translate(expanded_incoming) + if not self.check_values: + # If `self.check_values` is false we don't do any checking or + # processing on input This is used to pass raw values + # through to/from external sites. + params = expanded_incoming + else: + populate_state_async( + request_context, + self.inputs, + expanded_incoming, + params, + errors, + ) + self._handle_validate_input_hook(request_context, params, errors) + return params, errors + def _handle_validate_input_hook( self, request_context, params: ToolStateJobInstancePopulatedT, errors: ParameterValidationErrorsT ) -> None: @@ -2183,6 +2281,39 @@ class Tool(UsesDictVisibleKeys, ToolParameterBundle): completed_jobs[i] = None return completed_jobs + def handle_input_async( + self, + request_context: WorkRequestContext, + tool_request: ToolRequest, + tool_state: RequestInternalDereferencedToolState, + history: Optional[model.History] = None, + use_cached_job: bool = DEFAULT_USE_CACHED_JOB, + preferred_object_store_id: Optional[str] = DEFAULT_PREFERRED_OBJECT_STORE_ID, + rerun_remap_job_id: Optional[int] = None, + input_format: str = "legacy", + ): + """The tool request API+tasks version of handle_input.""" + all_params, all_errors, collection_info, job_tool_states = self.expand_incoming_async( + request_context, tool_state, rerun_remap_job_id + ) + self.handle_incoming_errors(all_errors) + + mapping_params = MappingParameters(tool_request.request, all_params, tool_state, job_tool_states) + completed_jobs: dict[int, Optional[model.Job]] = self.completed_jobs( + request_context, use_cached_job, all_params + ) + execute_async( + request_context, + self, + mapping_params, + request_context.history, + tool_request, + completed_jobs, + rerun_remap_job_id=rerun_remap_job_id, + preferred_object_store_id=preferred_object_store_id, + collection_info=collection_info, + ) + def handle_input( self, trans, @@ -2207,11 +2338,11 @@ class Tool(UsesDictVisibleKeys, ToolParameterBundle): # If there were errors, we stay on the same page and display them self.handle_incoming_errors(all_errors) - mapping_params = MappingParameters(incoming, all_params) + mapping_params = MappingParameters(incoming, all_params, None, None) if use_cached_job: mapping_params.param_template["__use_cached_job__"] = use_cached_job completed_jobs: dict[int, Optional[Job]] = self.completed_jobs(trans, use_cached_job, all_params) - execution_tracker = execute_job( + execution_tracker = execute_sync( trans, self, mapping_params, diff --git a/lib/galaxy/tools/_types.py b/lib/galaxy/tools/_types.py index b2b87a28ce9..dd6057a11a1 100644 --- a/lib/galaxy/tools/_types.py +++ b/lib/galaxy/tools/_types.py @@ -8,6 +8,8 @@ provide strong traditional typing semantics. +================================+============+=================================+============+===========+ | ToolRequestT | request | src dicts of encoded ids | nope | | | ToolStateJobInstanceT | a job | src dicts of encoded ids | nope | | +| ToolStateJobInstanceExpansionT | a job | a mix I think, things that were | nope | | +| | | expanded are objects | nope | | | ToolStateJobInstancePopulatedT | a job | model objs loaded from db | check_param | | | ToolStateDumpedToJsonT | a job | src dicts of encoded ids | " | | | | | (normalized into values attr) | " | | @@ -34,6 +36,10 @@ ToolRequestT = dict[str, Any] # been "checked" (check_param has not been called). ToolStateJobInstanceT = dict[str, Any] +# After meta.expand_incoming stuff I think expanded parameters are in model object form but the other stuff is likely +# still encoded IDs? None of this is verified though. +ToolStateJobInstanceExpansionT = dict[str, Any] + # Input dictionary for an individual job where objects are their model objects and parameters have been # "checked" (check_param has been called). ToolStateJobInstancePopulatedT = dict[str, Any] diff --git a/lib/galaxy/tools/execute.py b/lib/galaxy/tools/execute.py index eab40a7d724..5b5cb396a48 100644 --- a/lib/galaxy/tools/execute.py +++ b/lib/galaxy/tools/execute.py @@ -22,12 +22,17 @@ from typing_extensions import TypeAlias from galaxy import model from galaxy.exceptions import ToolInputsNotOKException +from galaxy.model import ToolRequest from galaxy.model.dataset_collections.matching import MatchingCollections from galaxy.model.dataset_collections.structure import ( get_structure, tool_output_to_structure, ) from galaxy.schema.credentials import CredentialsContext +from galaxy.tool_util.parameters.state import ( + JobInternalToolState, + RequestInternalDereferencedToolState, +) from galaxy.tool_util.parser import ToolOutputCollectionPart from galaxy.tools.execution_helpers import ( filter_output, @@ -67,8 +72,58 @@ class PartialJobExecution(Exception): class MappingParameters(NamedTuple): + # the raw request - might correspond to multiple jobs param_template: ToolRequestT + # parameters corresponding to individual job param_combinations: list[ToolStateJobInstancePopulatedT] + # schema driven parameters + # model validated tool request - might correspond to multiple jobs + validated_param_template: Optional[RequestInternalDereferencedToolState] = None + # validated job parameters for individual jobs + validated_param_combinations: Optional[list[JobInternalToolState]] = None + + def ensure_validated(self): + assert self.validated_param_template is not None + assert self.validated_param_combinations is not None + + +def execute_async( + trans, + tool: "Tool", + mapping_params: MappingParameters, + history: model.History, + tool_request: ToolRequest, + completed_jobs: Optional[CompletedJobsT] = None, + rerun_remap_job_id: Optional[int] = None, + preferred_object_store_id: Optional[str] = None, + collection_info: Optional[MatchingCollections] = None, + workflow_invocation_uuid: Optional[str] = None, + invocation_step: Optional[model.WorkflowInvocationStep] = None, + max_num_jobs: Optional[int] = None, + job_callback: Optional[Callable] = None, + workflow_resource_parameters: Optional[dict[str, Any]] = None, + validate_outputs: bool = False, +) -> "ExecutionTracker": + """The tool request/async version of execute.""" + completed_jobs = completed_jobs or {} + mapping_params.ensure_validated() + return _execute( + trans, + tool, + mapping_params, + history, + tool_request, + rerun_remap_job_id, + preferred_object_store_id, + collection_info, + workflow_invocation_uuid, + invocation_step, + max_num_jobs, + job_callback, + completed_jobs, + workflow_resource_parameters, + validate_outputs, + ) def execute( @@ -87,12 +142,48 @@ def execute( completed_jobs: Optional[CompletedJobsT] = None, workflow_resource_parameters: Optional[WorkflowResourceParametersT] = None, validate_outputs: bool = False, -): +) -> "ExecutionTracker": """ Execute a tool and return object containing summary (output data, number of failures, etc...). """ completed_jobs = completed_jobs or {} + return _execute( + trans, + tool, + mapping_params, + history, + None, + rerun_remap_job_id, + preferred_object_store_id, + collection_info, + workflow_invocation_uuid, + invocation_step, + max_num_jobs, + job_callback, + completed_jobs, + workflow_resource_parameters, + validate_outputs, + ) + + +def _execute( + trans, + tool: "Tool", + mapping_params: MappingParameters, + history: model.History, + tool_request: Optional[ToolRequest], + rerun_remap_job_id: Optional[int], + preferred_object_store_id: Optional[str], + collection_info: Optional[MatchingCollections], + workflow_invocation_uuid: Optional[str], + invocation_step: Optional[model.WorkflowInvocationStep], + max_num_jobs: Optional[int], + job_callback: Optional[Callable], + completed_jobs: dict[int, Optional[model.Job]], + workflow_resource_parameters: Optional[dict[str, Any]], + validate_outputs: bool, +) -> "ExecutionTracker": if max_num_jobs is not None: assert invocation_step is not None if rerun_remap_job_id: @@ -117,10 +208,11 @@ def execute( "internals.galaxy.tools.execute.job_single", SINGLE_EXECUTION_SUCCESS_MESSAGE ) params = execution_slice.param_combination - if "__data_manager_mode" in mapping_params.param_template: - params["__data_manager_mode"] = mapping_params.param_template["__data_manager_mode"] - if "__use_cached_job__" in mapping_params.param_template: - params["__use_cached_job__"] = mapping_params.param_template["__use_cached_job__"] + request_state = mapping_params.param_template + if "__data_manager_mode" in request_state: + params["__data_manager_mode"] = request_state["__data_manager_mode"] + if "__use_cached_job__" in request_state: + params["__use_cached_job__"] = request_state["__use_cached_job__"] if workflow_invocation_uuid: params["__workflow_invocation_uuid__"] = workflow_invocation_uuid elif "__workflow_invocation_uuid__" in params: @@ -150,6 +242,8 @@ def execute( skip=skip, ) if job: + if tool_request: + job.tool_request = tool_request log.debug(job_timer.to_str(tool_id=tool.id, job_id=job.id)) execution_tracker.record_success(execution_slice, job, result) # associate dataset instances with the job that creates them @@ -177,7 +271,7 @@ def execute( except ToolInputsNotOKException as e: execution_tracker.record_error(e) - execution_tracker.ensure_implicit_collections_populated(history, mapping_params.param_template) + execution_tracker.ensure_implicit_collections_populated(history, mapping_params.param_template, tool_request) job_count = len(execution_tracker.param_combinations) jobs_executed = 0 @@ -190,7 +284,11 @@ def execute( has_remaining_jobs = True break else: - skip = execution_slice.param_combination.pop("__when_value__", None) is False + slice_params = execution_slice.param_combination + if isinstance(slice_params, JobInternalToolState): + slice_params = slice_params.input_state + + skip = slice_params.pop("__when_value__", None) is False execute_single_job(execution_slice, completed_jobs[i], skip=skip) history = execution_slice.history or history jobs_executed += 1 @@ -444,15 +542,15 @@ class ExecutionTracker: mapped_output_structure = mapping_structure.multiply(output_structure) return mapped_output_structure - def ensure_implicit_collections_populated(self, history, params): + def ensure_implicit_collections_populated(self, history, params, tool_request: Optional[ToolRequest]): if not self.collection_info: return history = history or self.tool.get_default_history_by_trans(self.trans) # params = param_combinations[0] if param_combinations else mapping_params.param_template - self.precreate_output_collections(history, params) + self.precreate_output_collections(history, params, tool_request) - def precreate_output_collections(self, history, params): + def precreate_output_collections(self, history, params, tool_request: Optional[ToolRequest]): # params is just one sample tool param execution with parallelized # collection replaced with a specific dataset. Need to replace this # with the collection and wrap everything up so can evaluate output @@ -497,6 +595,12 @@ class ExecutionTracker: ) collection_instance.implicit_collection_jobs = implicit_collection_jobs collection_instances[output_name] = collection_instance + if tool_request: + assoc = model.ToolRequestImplicitCollectionAssociation() + assoc.tool_request = tool_request + assoc.dataset_collection = collection_instance + assoc.output_name = output_name + tool_request.implicit_collections.append(assoc) trans.sa_session.add(collection_instance) # Needed to flush the association created just above with # job.add_output_dataset_collection. @@ -699,13 +803,13 @@ class WorkflowStepExecutionTracker(ExecutionTracker): yield ExecutionSlice(job_index, param_combination, dataset_collection_elements) - def ensure_implicit_collections_populated(self, history, params): + def ensure_implicit_collections_populated(self, history, params, tool_request: Optional[ToolRequest]): if not self.collection_info: return history = history or self.tool.get_default_history_by_trans(self.trans) if self.invocation_step.is_new: - self.precreate_output_collections(history, params) + self.precreate_output_collections(history, params, tool_request) for output_name, implicit_collection in self.implicit_collections.items(): self.invocation_step.add_output(output_name, implicit_collection) else: diff --git a/lib/galaxy/tools/parameters/__init__.py b/lib/galaxy/tools/parameters/__init__.py index 6e9532f2824..79abac27f78 100644 --- a/lib/galaxy/tools/parameters/__init__.py +++ b/lib/galaxy/tools/parameters/__init__.py @@ -14,6 +14,10 @@ from typing import ( from boltons.iterutils import remap from galaxy.exceptions import RequestParameterInvalidException +from galaxy.model import ( + HistoryDatasetAssociation, + HistoryDatasetCollectionAssociation, +) from galaxy.util import unicodify from galaxy.util.expressions import ExpressionContext from galaxy.util.json import safe_loads @@ -22,6 +26,7 @@ from .basic import ( DataToolParameter, ParameterValueError, SelectToolParameter, + TextToolParameter, ToolParameter, ) from .grouping import ( @@ -686,6 +691,204 @@ def _populate_state_legacy( state[input.name] = value +def populate_state_async( + request_context, + inputs: ToolInputsT, + incoming: ToolStateJobInstanceT, + state: ToolStateJobInstancePopulatedT, + errors: ParameterValidationErrorsT, + context=None, +): + context = ExpressionContext(state, context) + for input in inputs.values(): + initial_value = input.get_initial_value(request_context, context) + input_name = input.name + state[input_name] = initial_value + group_state = state[input_name] + if input.type == "repeat": + repeat_input = cast(Repeat, input) + if ( + len(incoming[repeat_input.name]) > repeat_input.max + or len(incoming[repeat_input.name]) < repeat_input.min + ): + errors[repeat_input.name] = "The number of repeat elements is outside the range specified by the tool." + else: + del group_state[:] + for rep in incoming[repeat_input.name]: + new_state: ToolStateJobInstancePopulatedT = {} + group_state.append(new_state) + repeat_errors: ParameterValidationErrorsT = {} + populate_state_async( + request_context, + repeat_input.inputs, + rep, + new_state, + repeat_errors, + context=context, + ) + if repeat_errors: + errors[repeat_input.name] = repeat_errors + + elif input.type == "conditional": + conditional_input = cast(Conditional, input) + test_param = cast(ToolParameter, conditional_input.test_param) + test_param_value = incoming.get(conditional_input.name, {}).get(test_param.name) + value, error = check_param(request_context, test_param, test_param_value, context) + if error: + errors[test_param.name] = error + else: + try: + current_case = conditional_input.get_current_case(value) + group_state = state[conditional_input.name] = {} + cast_errors: ParameterValidationErrorsT = {} + populate_state_async( + request_context, + conditional_input.cases[current_case].inputs, + cast(ToolStateJobInstanceT, incoming.get(conditional_input.name)), + group_state, + cast_errors, + context=context, + ) + if cast_errors: + errors[conditional_input.name] = cast_errors + group_state["__current_case__"] = current_case + except Exception: + errors[test_param.name] = "The selected case is unavailable/invalid." + group_state[test_param.name] = value + + elif input.type == "section": + section_input = cast(Section, input) + section_errors: ParameterValidationErrorsT = {} + populate_state_async( + request_context, + section_input.inputs, + cast(ToolStateJobInstanceT, incoming.get(section_input.name)), + group_state, + section_errors, + context=context, + ) + if section_errors: + errors[section_input.name] = section_errors + + elif input.type == "upload_dataset": + raise NotImplementedError + + else: + assert isinstance(input, ToolParameter) + param_value = _get_incoming_value(incoming, input.name, state.get(input.name)) + value, error = check_param(request_context, input, param_value, context, simple_errors=False) + if error: + errors[input.name] = error + state[input.name] = value + + def to_internal_single(value): + if isinstance(value, HistoryDatasetCollectionAssociation): + return {"src": "hdca", "id": value.id} + elif isinstance(value, HistoryDatasetAssociation): + return {"src": "hda", "id": value.id} + else: + # tests and such to confirm we need DCE, LDDA, etc... + return value + + def to_internal(value): + if isinstance(value, list): + return [to_internal_single(v) for v in value] + else: + return to_internal_single(value) + + if input_name not in incoming: + if input.type == "data_column": + if isinstance(value, str): + incoming[input_name] = int(value) + elif isinstance(value, list): + incoming[input_name] = [int(v) for v in value] + else: + incoming[input_name] = value + elif input.type == "text": + text_input = cast(TextToolParameter, input) + # see behavior of tools in test_tools.py::test_null_to_text_tools + # these parameters act as empty string in this context + if value is None and not text_input.optional: + incoming[input_name] = "" + else: + incoming[input_name] = value + else: + incoming[input_name] = to_internal(value) + + +def fill_dynamic_defaults( + request_context, + inputs: ToolInputsT, + job_tool_state: ToolStateJobInstanceT, + params: ToolStateJobInstancePopulatedT, + context=None, +): + """ + Expands incoming parameters with default values. + """ + context = ExpressionContext(job_tool_state, job_tool_state) + for input in inputs.values(): + if input.type == "repeat": + repeat_input = cast(Repeat, input) + repeat_name = repeat_input.name + for rep, rep_params in enumerate(job_tool_state[repeat_name]): + fill_dynamic_defaults( + request_context, + repeat_input.inputs, + rep_params, + params[repeat_name][rep], + context=context, + ) + + elif input.type == "conditional": + conditional_input = cast(Conditional, input) + test_param = cast(ToolParameter, conditional_input.test_param) + test_param_value = job_tool_state.get(conditional_input.name, {}).get(test_param.name) + try: + current_case = conditional_input.get_current_case(test_param_value) + fill_dynamic_defaults( + request_context, + conditional_input.cases[current_case].inputs, + cast(ToolStateJobInstanceT, job_tool_state.get(conditional_input.name)), + cast(ToolStateJobInstancePopulatedT, params.get(conditional_input.name)), + context=context, + ) + except Exception: + raise Exception("The selected case is unavailable/invalid.") + + elif input.type == "section": + section_input = cast(Section, input) + fill_dynamic_defaults( + request_context, + section_input.inputs, + cast(ToolStateJobInstanceT, job_tool_state.get(section_input.name)), + cast(ToolStateJobInstancePopulatedT, params.get(section_input.name)), + context=context, + ) + + elif input.type == "upload_dataset": + raise NotImplementedError + + else: + if input.name not in job_tool_state and input.name in params: + if input.type == "data_column": + if isinstance(params[input.name], str): + job_tool_state[input.name] = int(params[input.name]) + elif isinstance(params[input.name], list): + job_tool_state[input.name] = [int(v) for v in params[input.name]] + else: + job_tool_state[input.name] = params[input.name] + elif input.type == "data_collection": + data_collection = params[input.name] + if data_collection: + job_tool_state[input.name] = { + "src": "hdca", + "id": data_collection.id, + } + else: + job_tool_state[input.name] = params[input.name] + + def _get_incoming_value(incoming, key, default): """ Fetch value from incoming dict directly or check special nginx upload diff --git a/lib/galaxy/tools/parameters/meta.py b/lib/galaxy/tools/parameters/meta.py index 375d8f3460a..dd54eadcd03 100644 --- a/lib/galaxy/tools/parameters/meta.py +++ b/lib/galaxy/tools/parameters/meta.py @@ -15,13 +15,16 @@ from galaxy import ( from galaxy.model import ( DatasetCollectionElement, DatasetInstance, + HistoryDatasetAssociation, HistoryDatasetCollectionAssociation, + LibraryDatasetDatasetAssociation, ) from galaxy.model.dataset_collections import ( matching, subcollections, ) from galaxy.model.dataset_collections.adapters import PromoteCollectionElementToCollectionAdapter +from galaxy.tool_util.parameters import RequestInternalDereferencedToolState from galaxy.util.permutations import ( build_combos, input_classification, @@ -37,6 +40,7 @@ from .wrapped import process_key from .._types import ( InputFormatT, ToolRequestT, + ToolStateDumpedToJsonInternalT, ToolStateJobInstanceT, ) @@ -333,6 +337,80 @@ def split_inputs_nested(inputs, nested_dict, classifier): return (single_inputs_nested, matched_multi_inputs, multiplied_multi_inputs) +ExpandedAsyncT = tuple[ + list[ToolStateJobInstanceT], list[ToolStateDumpedToJsonInternalT], Optional[matching.MatchingCollections] +] + + +def expand_meta_parameters_async(app, tool, incoming: RequestInternalDereferencedToolState) -> ExpandedAsyncT: + collections_to_match = matching.CollectionsToMatch() + + def classifier_from_value(value, input_key): + if isinstance(value, dict) and "values" in value: + # Explicit meta wrapper for inputs... + is_batch = value.get("__class__", "Batch") == "Batch" + is_linked = value.get("linked", True) + if is_batch and is_linked: + classification = input_classification.MATCHED + elif is_batch: + classification = input_classification.MULTIPLIED + else: + classification = input_classification.SINGLE + if __collection_multirun_parameter(value): + collection_value = value["values"][0] + values = __expand_collection_parameter_async( + app, input_key, collection_value, collections_to_match, linked=is_linked + ) + else: + values = value["values"] + else: + classification = input_classification.SINGLE + values = value + return classification, values + + # is there a way to make Pydantic ensure reordering isn't needed - model and serialize out the parameters maybe? + reordered_incoming = reorder_parameters(tool, incoming.input_state, incoming.input_state, True) + incoming_template = reordered_incoming + + single_inputs, matched_multi_inputs, multiplied_multi_inputs = split_inputs_nested( + tool.inputs, incoming_template, classifier_from_value + ) + expanded_incomings = build_combos(single_inputs, matched_multi_inputs, multiplied_multi_inputs, nested=True) + # those all have sa model objects from expansion to be used within for additional logic (maybe?) + # but we want to record just src and IDS in the job state object - so undo that + expanded_job_states = build_combos( + to_decoded_json(single_inputs), + to_decoded_json(matched_multi_inputs), + to_decoded_json(multiplied_multi_inputs), + nested=True, + ) + if collections_to_match.has_collections(): + collection_info = app.dataset_collection_manager.match_collections(collections_to_match) + else: + collection_info = None + return expanded_incomings, expanded_job_states, collection_info + + +def to_decoded_json(has_objects): + if isinstance(has_objects, dict): + decoded_json = {} + for key, value in has_objects.items(): + decoded_json[key] = to_decoded_json(value) + return decoded_json + elif isinstance(has_objects, list): + return [to_decoded_json(o) for o in has_objects] + elif isinstance(has_objects, DatasetCollectionElement): + return {"src": "dce", "id": has_objects.id} + elif isinstance(has_objects, HistoryDatasetAssociation): + return {"src": "hda", "id": has_objects.id} + elif isinstance(has_objects, HistoryDatasetCollectionAssociation): + return {"src": "hdca", "id": has_objects.id} + elif isinstance(has_objects, LibraryDatasetDatasetAssociation): + return {"src": "ldda", "id": has_objects.id} + else: + return has_objects + + CollectionExpansionListT = Union[ list[Union[DatasetCollectionElement, PromoteCollectionElementToCollectionAdapter]], list[DatasetInstance] ] @@ -371,7 +449,9 @@ def __expand_collection_parameter( raise exceptions.ToolInputsNotReadyException("An input collection is not populated.") collections_to_match.add(input_key, item, subcollection_type=subcollection_type, linked=linked) if subcollection_type is not None: - subcollection_elements = subcollections._split_dataset_collection(collection, subcollection_type) + subcollection_elements: list[Union[DatasetCollectionElement, PromoteCollectionElementToCollectionAdapter]] = ( + subcollections._split_dataset_collection(collection, subcollection_type) + ) return subcollection_elements else: hdas: list[DatasetInstance] = [] @@ -382,8 +462,36 @@ def __expand_collection_parameter( return hdas +def __expand_collection_parameter_async( + app, input_key, incoming_val, collections_to_match: "matching.CollectionsToMatch", linked=False +) -> CollectionExpansionListT: + # If subcollection multirun of data_collection param - value will + # be "hdca_id|subcollection_type" else it will just be hdca_id + try: + src = incoming_val["src"] + if src != "hdca": + raise exceptions.ToolMetaParameterException(f"Invalid dataset collection source type {src}") + hdc_id = incoming_val["id"] + subcollection_type = incoming_val.get("map_over_type", None) + except TypeError: + hdc_id = incoming_val + subcollection_type = None + hdc = app.model.context.get(HistoryDatasetCollectionAssociation, hdc_id) + collections_to_match.add(input_key, hdc, subcollection_type=subcollection_type, linked=linked) + if subcollection_type is not None: + subcollection_elements = subcollections.split_dataset_collection_instance(hdc, subcollection_type) + return subcollection_elements + else: + hdas: list[DatasetInstance] = [] + for element in hdc.collection.dataset_elements: + hda = element.dataset_instance + hda.element_identifier = element.element_identifier + hdas.append(hda) + return hdas + + def __collection_multirun_parameter(value: dict[str, Any]) -> bool: - is_batch = value.get("batch", False) + is_batch = value.get("batch", False) or value.get("__class__", None) == "Batch" if not is_batch: return False diff --git a/lib/galaxy/webapps/galaxy/api/histories.py b/lib/galaxy/webapps/galaxy/api/histories.py index 00d4e91956b..4353b21ac11 100644 --- a/lib/galaxy/webapps/galaxy/api/histories.py +++ b/lib/galaxy/webapps/galaxy/api/histories.py @@ -60,6 +60,7 @@ from galaxy.schema.schema import ( ShareWithPayload, SharingStatus, StoreExportPayload, + ToolRequestModel, UpdateHistoryPayload, WriteStoreToPayload, ) @@ -373,6 +374,17 @@ class FastAPIHistories: ) -> list[Any]: return self.service.citations(trans, history_id) + @router.get( + "/api/histories/{history_id}/tool_requests", + summary="Return all the tool requests for the tools submitted to this history.", + ) + def tool_requests( + self, + history_id: HistoryIDPathParam, + trans: ProvidesHistoryContext = DependsOnTrans, + ) -> list[ToolRequestModel]: + return self.service.tool_requests(trans, history_id) + @router.post( "/api/histories", summary="Creates a new history.", diff --git a/lib/galaxy/webapps/galaxy/api/jobs.py b/lib/galaxy/webapps/galaxy/api/jobs.py index 5e442cb2190..5bd6e5c37a5 100644 --- a/lib/galaxy/webapps/galaxy/api/jobs.py +++ b/lib/galaxy/webapps/galaxy/api/jobs.py @@ -46,6 +46,7 @@ from galaxy.schema.jobs import ( JobInputAssociation, JobInputSummary, JobOutputAssociation, + JobOutputCollectionAssociation, ReportJobErrorPayload, SearchJobsPayload, ) @@ -70,11 +71,14 @@ from galaxy.webapps.galaxy.api import ( ) from galaxy.webapps.galaxy.api.common import query_parameter_as_list from galaxy.webapps.galaxy.services.jobs import ( + JobCreateResponse, JobIndexPayload, JobIndexViewEnum, + JobRequest, JobsService, ) from galaxy.work.context import proxy_work_context_for_history +from .tools import validate_not_protected log = logging.getLogger(__name__) @@ -158,6 +162,12 @@ ImplicitCollectionJobsIdQueryParam: Optional[DecodedDatabaseIdField] = Query( description="Limit listing of jobs to those that match the specified implicit collection job ID. If none, jobs from any implicit collection execution (or from no implicit collection execution) may be returned.", ) +ToolRequestIdQueryParam: Optional[DecodedDatabaseIdField] = Query( + default=None, + title="Tool Request ID", + description="Limit listing of jobs to those that were created from the supplied tool request ID. If none, jobs from any tool request (or from no workflows) may be returned.", +) + SortByQueryParam: JobIndexSortByEnum = Query( default=JobIndexSortByEnum.update_time, title="Sort By", @@ -261,6 +271,13 @@ class ShowFullJobResponse(EncodedJobDetails): class FastAPIJobs: service: JobsService = depends(JobsService) + @router.post("/api/jobs") + def create( + self, trans: ProvidesHistoryContext = DependsOnTrans, job_request: JobRequest = Body(...) + ) -> JobCreateResponse: + validate_not_protected(job_request.tool_id) + return self.service.create(trans, job_request) + @router.get("/api/jobs") def index( self, @@ -277,6 +294,7 @@ class FastAPIJobs: workflow_id: Optional[DecodedDatabaseIdField] = WorkflowIdQueryParam, invocation_id: Optional[DecodedDatabaseIdField] = InvocationIdQueryParam, implicit_collection_jobs_id: Optional[DecodedDatabaseIdField] = ImplicitCollectionJobsIdQueryParam, + tool_request_id: Optional[DecodedDatabaseIdField] = ToolRequestIdQueryParam, order_by: JobIndexSortByEnum = SortByQueryParam, search: Optional[str] = SearchQueryParam, limit: int = LimitQueryParam, @@ -295,6 +313,7 @@ class FastAPIJobs: workflow_id=workflow_id, invocation_id=invocation_id, implicit_collection_jobs_id=implicit_collection_jobs_id, + tool_request_id=tool_request_id, order_by=order_by, search=search, limit=limit, @@ -415,12 +434,14 @@ class FastAPIJobs: self, job_id: JobIdPathParam, trans: ProvidesUserContext = DependsOnTrans, - ) -> list[JobOutputAssociation]: + ) -> list[Union[JobOutputAssociation, JobOutputCollectionAssociation]]: job = self.service.get_job(trans=trans, job_id=job_id) associations = self.service.dictify_associations(trans, job.output_datasets, job.output_library_datasets) - output_associations = [] + output_associations: list[Union[JobOutputAssociation, JobOutputCollectionAssociation]] = [] for association in associations: output_associations.append(JobOutputAssociation(name=association.name, dataset=association.dataset)) + + output_associations.extend(self.service.dictify_output_collection_associations(trans, job)) return output_associations @router.get( diff --git a/lib/galaxy/webapps/galaxy/api/tools.py b/lib/galaxy/webapps/galaxy/api/tools.py index 3f7297c85dd..f0e99ef991d 100644 --- a/lib/galaxy/webapps/galaxy/api/tools.py +++ b/lib/galaxy/webapps/galaxy/api/tools.py @@ -14,6 +14,7 @@ from typing import ( from fastapi import ( Body, Depends, + Path, Query, Request, Response, @@ -38,6 +39,8 @@ from galaxy.managers.context import ( from galaxy.managers.hdas import HDAManager from galaxy.managers.histories import HistoryManager from galaxy.managers.landing import LandingRequestManager +from galaxy.managers.tools import ToolRunReference +from galaxy.model import ToolRequest from galaxy.model.dataset_collections.workbook_util import workbook_to_bytes from galaxy.schema.fetch_data import ( CreateDataLandingPayload, @@ -45,11 +48,14 @@ from galaxy.schema.fetch_data import ( FetchDataFormPayload, FetchDataPayload, ) +from galaxy.schema.fields import DecodedDatabaseIdField from galaxy.schema.schema import ( ClaimLandingPayload, CreateToolLandingRequestPayload, ToolLandingRequest, + ToolRequestDetailedModel, ) +from galaxy.tool_util.parameters import ToolParameterT from galaxy.tool_util.verify import ToolTestDescriptionDict from galaxy.tool_util_models import UserToolSource from galaxy.tools.evaluation import global_tool_errors @@ -76,6 +82,7 @@ from galaxy.web import ( from galaxy.webapps.base.controller import UsesVisualizationMixin from galaxy.webapps.base.webapp import GalaxyWebTransaction from galaxy.webapps.galaxy.api.common import serve_workbook +from galaxy.webapps.galaxy.services.base import tool_request_detailed_to_model from galaxy.webapps.galaxy.services.tools import ToolsService from . import ( APIContentTypeRoute, @@ -130,6 +137,13 @@ router = Router(tags=["tools"]) FetchDataForm = as_form(FetchDataFormPayload) +ToolIDPathParam: str = Path( + ..., + title="Tool ID", + description="The tool ID for the lineage stored in Galaxy's toolbox.", +) +ToolVersionQueryParam: Optional[str] = Query(default=None, title="Tool Version", description="") + async def get_files(request: Request, files: Optional[list[UploadFile]] = None): # FastAPI's UploadFile is a very light wrapper around starlette's UploadFile @@ -269,6 +283,57 @@ class FetchTools: tool_landing_request = self.service.data_landing_to_tool_landing(trans, data_landing_request) return self.landing_manager.create_tool_landing_request(tool_landing_request) + @router.get( + "/api/tool_requests/{id}", + summary="Get tool request state.", + ) + def get_tool_request( + self, + id: DecodedDatabaseIdField, + trans: ProvidesHistoryContext = DependsOnTrans, + ) -> ToolRequestDetailedModel: + tool_request = self._get_tool_request_or_raise_not_found(trans, id) + return tool_request_detailed_to_model(tool_request) + + @router.get( + "/api/tool_requests/{id}/state", + summary="Get tool request state.", + ) + def tool_request_state( + self, + id: DecodedDatabaseIdField, + trans: ProvidesHistoryContext = DependsOnTrans, + ) -> str: + tool_request = self._get_tool_request_or_raise_not_found(trans, id) + state = tool_request.state + if not state: + raise exceptions.InconsistentDatabase() + return state + + @router.get( + "/api/tools/{tool_id}/inputs", + summary="Get tool inputs.", + ) + def tool_inputs( + self, + tool_id: str = ToolIDPathParam, + tool_version: Optional[str] = ToolVersionQueryParam, + trans: ProvidesHistoryContext = DependsOnTrans, + ) -> list[ToolParameterT]: + tool_run_ref = ToolRunReference(tool_id=tool_id, tool_version=tool_version, tool_uuid=None) + return self.service.inputs(trans, tool_run_ref) + + def _get_tool_request_or_raise_not_found( + self, trans: ProvidesHistoryContext, id: DecodedDatabaseIdField + ) -> ToolRequest: + tool_request: Optional[ToolRequest] = cast( + Optional[ToolRequest], trans.app.model.context.query(ToolRequest).get(id) + ) + if tool_request is None: + raise exceptions.ObjectNotFound() + assert tool_request + return tool_request + @router.post("/api/tool_landings", public=True, allow_cors=True) def create_landing( self, @@ -299,7 +364,6 @@ class FetchTools: ) -> ToolLandingRequest: return self.landing_manager.get_tool_landing_request(trans, uuid) - class ToolsController(BaseGalaxyAPIController, UsesVisualizationMixin): """ RESTful controller for interactions with tools. @@ -787,10 +851,7 @@ class ToolsController(BaseGalaxyAPIController, UsesVisualizationMixin): """ tool_id = payload.get("tool_id") tool_uuid = payload.get("tool_uuid") - if tool_id in PROTECTED_TOOLS: - raise exceptions.RequestParameterInvalidException( - f"Cannot execute tool [{tool_id}] directly, must use alternative endpoint." - ) + validate_not_protected(tool_id) if tool_id is None and tool_uuid is None: raise exceptions.RequestParameterInvalidException("Must specify a valid tool_id to use this endpoint.") __tags = payload.get("__tags", []) @@ -799,6 +860,13 @@ class ToolsController(BaseGalaxyAPIController, UsesVisualizationMixin): return self.service._create(trans, payload, **kwd) +def validate_not_protected(tool_id: Optional[str]): + if tool_id in PROTECTED_TOOLS: + raise exceptions.RequestParameterInvalidException( + f"Cannot execute tool [{tool_id}] directly, must use alternative endpoint." + ) + + def _kwd_or_payload(kwd: dict[str, Any]) -> dict[str, Any]: if "payload" in kwd: kwd = cast(dict[str, Any], kwd.get("payload")) diff --git a/lib/galaxy/webapps/galaxy/services/base.py b/lib/galaxy/webapps/galaxy/services/base.py index 9e320540d5e..de0ab99da8b 100644 --- a/lib/galaxy/webapps/galaxy/services/base.py +++ b/lib/galaxy/webapps/galaxy/services/base.py @@ -22,13 +22,20 @@ from galaxy.managers.base import ( ) from galaxy.managers.context import ProvidesUserContext from galaxy.managers.model_stores import create_objects_from_store -from galaxy.model import User +from galaxy.model import ( + ToolRequest, + User, +) from galaxy.model.store import ( get_export_store_factory, ModelExportStore, ) from galaxy.schema.fields import EncodedDatabaseIdField -from galaxy.schema.schema import AsyncTaskResultSummary +from galaxy.schema.schema import ( + AsyncTaskResultSummary, + ToolRequestDetailedModel, + ToolRequestModel, +) from galaxy.security.idencoding import IdEncodingHelper from galaxy.short_term_storage import ( ShortTermStorageAllocator, @@ -192,3 +199,31 @@ def async_task_summary(async_result: AsyncResult) -> AsyncTaskResultSummary: name=name, queue=queue, ) + + +def tool_request_to_model(tool_request: ToolRequest) -> ToolRequestModel: + as_dict = { + "id": tool_request.id, + "request": tool_request.request, + "state": tool_request.state, + "state_message": tool_request.state_message, + } + return ToolRequestModel.model_validate(as_dict) + + +def tool_request_detailed_to_model(tool_request: ToolRequest) -> ToolRequestDetailedModel: + jobs = [{"src": "job", "id": job.id} for job in tool_request.jobs] + implicit_collections = [ + {"src": "hdca", "id": assoc.dataset_collection.id, "output_name": assoc.output_name} + for assoc in tool_request.implicit_collections + ] + as_dict = { + "id": tool_request.id, + "request": tool_request.request, + "state": tool_request.state, + "state_message": tool_request.state_message, + "jobs": jobs, + "implicit_collections": implicit_collections, + } + model = ToolRequestDetailedModel.model_validate(as_dict) + return model diff --git a/lib/galaxy/webapps/galaxy/services/histories.py b/lib/galaxy/webapps/galaxy/services/histories.py index 8dfdb16b93e..e55189c4dd8 100644 --- a/lib/galaxy/webapps/galaxy/services/histories.py +++ b/lib/galaxy/webapps/galaxy/services/histories.py @@ -68,6 +68,7 @@ from galaxy.schema.schema import ( ShareHistoryWithStatus, ShareWithPayload, StoreExportPayload, + ToolRequestModel, WriteStoreToPayload, ) from galaxy.schema.tasks import ( @@ -85,6 +86,7 @@ from galaxy.webapps.galaxy.services.base import ( model_store_storage_target, ServesExportStores, ServiceBase, + tool_request_to_model, ) from galaxy.webapps.galaxy.services.notifications import NotificationService from galaxy.webapps.galaxy.services.sharable import ShareableService @@ -536,6 +538,13 @@ class HistoriesService(ServiceBase, ConsumesModelStores, ServesExportStores): ] return rval + def tool_requests( + self, trans: ProvidesHistoryContext, history_id: DecodedDatabaseIdField + ) -> list[ToolRequestModel]: + history = self.manager.get_accessible(history_id, trans.user, current_history=trans.history) + tool_requests = history.tool_requests + return [tool_request_to_model(tr) for tr in tool_requests] + def citations(self, trans: ProvidesHistoryContext, history_id: DecodedDatabaseIdField): """ Return all the references for the tools used to produce the datasets in diff --git a/lib/galaxy/webapps/galaxy/services/jobs.py b/lib/galaxy/webapps/galaxy/services/jobs.py index 9e3960ed4fd..58f2baa2c7f 100644 --- a/lib/galaxy/webapps/galaxy/services/jobs.py +++ b/lib/galaxy/webapps/galaxy/services/jobs.py @@ -1,3 +1,4 @@ +import logging from enum import Enum from typing import ( Any, @@ -6,24 +7,60 @@ from typing import ( Union, ) +from pydantic import ( + BaseModel, + Field, +) + from galaxy import ( exceptions, model, ) +from galaxy.celery.tasks import queue_jobs from galaxy.managers import hdas from galaxy.managers.base import security_check -from galaxy.managers.context import ProvidesUserContext +from galaxy.managers.context import ( + ProvidesHistoryContext, + ProvidesUserContext, +) +from galaxy.managers.histories import HistoryManager from galaxy.managers.jobs import ( JobManager, JobSearch, view_show_job, ) -from galaxy.model import Job -from galaxy.schema.fields import DecodedDatabaseIdField -from galaxy.schema.jobs import JobAssociation -from galaxy.schema.schema import JobIndexQueryPayload +from galaxy.managers.tools import ToolRunReference +from galaxy.model import ( + Job, + ToolRequest, + ToolSource as ToolSourceModel, +) +from galaxy.schema.fields import ( + DecodedDatabaseIdField, + EncodedDatabaseIdField, +) +from galaxy.schema.jobs import ( + JobAssociation, + JobOutputCollectionAssociation, +) +from galaxy.schema.schema import ( + AsyncTaskResultSummary, + JobIndexQueryPayload, +) +from galaxy.schema.tasks import ( + QueueJobs, + ToolSource, +) from galaxy.security.idencoding import IdEncodingHelper -from galaxy.webapps.galaxy.services.base import ServiceBase +from galaxy.tool_util.parameters import ( + decode, + RequestToolState, +) +from galaxy.webapps.galaxy.services.base import ( + async_task_summary, + ServiceBase, +) +from .tools import validate_tool_for_running if TYPE_CHECKING: from galaxy.model import ( @@ -31,6 +68,26 @@ if TYPE_CHECKING: LibraryDatasetDatasetAssociation, ) +log = logging.getLogger(__name__) + + +class JobRequest(BaseModel): + tool_id: Optional[str] = Field(default=None, title="tool_id", description="TODO") + tool_uuid: Optional[str] = Field(default=None, title="tool_uuid", description="TODO") + tool_version: Optional[str] = Field(default=None, title="tool_version", description="TODO") + history_id: Optional[DecodedDatabaseIdField] = Field(default=None, title="history_id", description="TODO") + inputs: Optional[dict[str, Any]] = Field(default_factory=lambda: {}, title="Inputs", description="TODO") + use_cached_jobs: Optional[bool] = Field(default=None, title="use_cached_jobs") + rerun_remap_job_id: Optional[DecodedDatabaseIdField] = Field( + default=None, title="rerun_remap_job_id", description="TODO" + ) + send_email_notification: bool = Field(default=False, title="Send Email Notification", description="TODO") + + +class JobCreateResponse(BaseModel): + tool_request_id: EncodedDatabaseIdField + task_result: AsyncTaskResultSummary + class JobIndexViewEnum(str, Enum): collection = "collection" @@ -45,6 +102,7 @@ class JobsService(ServiceBase): job_manager: JobManager job_search: JobSearch hda_manager: hdas.HDAManager + history_manager: HistoryManager def __init__( self, @@ -52,11 +110,13 @@ class JobsService(ServiceBase): job_manager: JobManager, job_search: JobSearch, hda_manager: hdas.HDAManager, + history_manager: HistoryManager, ): super().__init__(security=security) self.job_manager = job_manager self.job_search = job_search self.hda_manager = hda_manager + self.history_manager = history_manager def show( self, @@ -158,3 +218,61 @@ class JobsService(ServiceBase): else: dataset_dict = {"src": "ldda", "id": dataset.id} return JobAssociation(name=job_dataset_association.name, dataset=dataset_dict) + + def dictify_output_collection_associations(self, trans, job: model.Job) -> list[JobOutputCollectionAssociation]: + output_associations: list[JobOutputCollectionAssociation] = [] + for job_output_collection_association in job.output_dataset_collection_instances: + ref_dict = {"src": "hdca", "id": job_output_collection_association.dataset_collection_id} + output_associations.append( + JobOutputCollectionAssociation( + name=job_output_collection_association.name, + dataset_collection_instance=ref_dict, + ) + ) + return output_associations + + def create(self, trans: ProvidesHistoryContext, job_request: JobRequest) -> JobCreateResponse: + tool_run_reference = ToolRunReference(job_request.tool_id, job_request.tool_uuid, job_request.tool_version) + tool = validate_tool_for_running(trans, tool_run_reference) + history_id = job_request.history_id + target_history = None + if history_id is not None: + target_history = self.history_manager.get_owned(history_id, trans.user, current_history=trans.history) + inputs = job_request.inputs + request_state = RequestToolState(inputs or {}) + request_state.validate(tool, f"{tool.id} (request model)") + request_internal_state = decode(request_state, tool, trans.security.decode_id) + tool_request = ToolRequest() + # TODO: hash and such... + tool_source_model = ToolSourceModel( + source=[p.model_dump() for p in tool.parameters], + hash="TODO", + ) + tool_request.request = request_internal_state.input_state + tool_request.tool_source = tool_source_model + tool_request.state = ToolRequest.states.NEW + tool_request.history = target_history + sa_session = trans.sa_session + sa_session.add(tool_source_model) + sa_session.add(tool_request) + sa_session.commit() + tool_request_id = tool_request.id + tool_source = ToolSource( + raw_tool_source=tool.tool_source.to_string(), + tool_dir=tool.tool_dir, + ) + task_request = QueueJobs( + user=trans.async_request_user, + history_id=target_history and target_history.id, + tool_source=tool_source, + tool_request_id=tool_request_id, + use_cached_jobs=job_request.use_cached_jobs or False, + rerun_remap_job_id=job_request.rerun_remap_job_id, + ) + result = queue_jobs.delay(request=task_request) + return JobCreateResponse( + **{ + "tool_request_id": tool_request_id, + "task_result": async_task_summary(result), + } + ) diff --git a/lib/galaxy/webapps/galaxy/services/tools.py b/lib/galaxy/webapps/galaxy/services/tools.py index 377cd8f23a6..a2c8ad7a2d8 100644 --- a/lib/galaxy/webapps/galaxy/services/tools.py +++ b/lib/galaxy/webapps/galaxy/services/tools.py @@ -6,7 +6,6 @@ from json import dumps from typing import ( Any, cast, - Literal, Optional, Union, ) @@ -51,6 +50,7 @@ from galaxy.schema.fetch_data import ( ) from galaxy.schema.schema import CreateToolLandingRequestPayload from galaxy.security.idencoding import IdEncodingHelper +from galaxy.tool_util.parameters import ToolParameterT from galaxy.tool_util_models.parameters import ( CollectionElementCollectionRequestUri, CollectionElementDataRequestUri, @@ -59,6 +59,7 @@ from galaxy.tool_util_models.parameters import ( FileRequestUri, ) from galaxy.tools import Tool +from galaxy.tools._types import InputFormatT from galaxy.tools.search import ToolBoxSearch from galaxy.util.path import safe_contains from galaxy.webapps.galaxy.services._fetch_util import validate_and_normalize_targets @@ -236,6 +237,14 @@ class ToolsService(ServiceBase): public=data_landing_payload.public, ) + def inputs( + self, + trans: ProvidesHistoryContext, + tool_ref: ToolRunReference, + ) -> list[ToolParameterT]: + tool = get_tool(trans, tool_ref) + return tool.parameters + def create_fetch( self, trans: ProvidesHistoryContext, @@ -328,7 +337,7 @@ class ToolsService(ServiceBase): input_format = str(payload.get("input_format", "legacy")) if input_format not in ("legacy", "21.01"): raise exceptions.RequestParameterInvalidException(f"input_format invalid {input_format}") - input_format = cast(Literal["legacy", "21.01"], input_format) + input_format = cast(InputFormatT, input_format) if "data_manager_mode" in payload: incoming["__data_manager_mode"] = payload["data_manager_mode"] vars = tool.handle_input( diff --git a/lib/galaxy_test/api/conftest.py b/lib/galaxy_test/api/conftest.py index 5c9460410bc..0cae9e6426a 100644 --- a/lib/galaxy_test/api/conftest.py +++ b/lib/galaxy_test/api/conftest.py @@ -147,7 +147,7 @@ def required_tool(dataset_populator: DatasetPopulator, history_id: str, required return tool -@pytest.fixture(params=["legacy", "21.01"]) +@pytest.fixture(params=["legacy", "21.01", "request"]) def tool_input_format(request) -> Iterator[DescribeToolInputs]: yield DescribeToolInputs(request.param) diff --git a/lib/galaxy_test/api/test_tool_execute.py b/lib/galaxy_test/api/test_tool_execute.py index 7ddb66c72f8..78e0f2d34d7 100644 --- a/lib/galaxy_test/api/test_tool_execute.py +++ b/lib/galaxy_test/api/test_tool_execute.py @@ -27,16 +27,25 @@ def test_multidata_param( ): hda1 = target_history.with_dataset("1\t2\t3").src_dict hda2 = target_history.with_dataset("4\t5\t6").src_dict - inputs = tool_input_format.when.flat( - { - "f1": {"batch": False, "values": [hda1, hda2]}, - "f2": {"batch": False, "values": [hda2, hda1]}, - } - ).when.nested( - { - "f1": {"batch": False, "values": [hda1, hda2]}, - "f2": {"batch": False, "values": [hda2, hda1]}, - } + inputs = ( + tool_input_format.when.flat( + { + "f1": {"batch": False, "values": [hda1, hda2]}, + "f2": {"batch": False, "values": [hda2, hda1]}, + } + ) + .when.nested( + { + "f1": {"batch": False, "values": [hda1, hda2]}, + "f2": {"batch": False, "values": [hda2, hda1]}, + } + ) + .when.request( + { + "f1": [hda1, hda2], + "f2": [hda2, hda1], + } + ) ) execution = required_tool.execute.with_inputs(inputs) execution.assert_has_job(0).with_output("out1").with_contents("1\t2\t3\n4\t5\t6\n") @@ -139,18 +148,29 @@ def test_map_over_with_output_format_actions( ): hdca = target_history.with_pair() for use_action in ["do", "dont"]: - inputs = tool_input_format.when.flat( - { - "input_cond|dispatch": use_action, - "input_cond|input": {"batch": True, "values": [hdca.src_dict]}, - } - ).when.nested( - { - "input_cond": { - "dispatch": use_action, - "input": {"batch": True, "values": [hdca.src_dict]}, + inputs = ( + tool_input_format.when.flat( + { + "input_cond|dispatch": use_action, + "input_cond|input": {"batch": True, "values": [hdca.src_dict]}, } - } + ) + .when.nested( + { + "input_cond": { + "dispatch": use_action, + "input": {"batch": True, "values": [hdca.src_dict]}, + } + } + ) + .when.request( + { + "input_cond": { + "dispatch": use_action, + "input": {"__class__": "Batch", "values": [hdca.src_dict]}, + } + } + ) ) execute = required_tool.execute.with_inputs(inputs) execute.assert_has_n_jobs(2).assert_creates_n_implicit_collections(1) @@ -256,20 +276,33 @@ def test_multi_run_in_repeat( multi_run_in_repeat_datasets: MultiRunInRepeatFixtures, tool_input_format: DescribeToolInputs, ): - inputs = tool_input_format.when.flat( - { - "input1": {"batch": False, "values": [multi_run_in_repeat_datasets.common_dataset]}, - "queries_0|input2": {"batch": True, "values": multi_run_in_repeat_datasets.repeat_datasets}, - } - ).when.nested( - { - "input1": {"batch": False, "values": [multi_run_in_repeat_datasets.common_dataset]}, - "queries": [ - { - "input2": {"batch": True, "values": multi_run_in_repeat_datasets.repeat_datasets}, - } - ], - } + inputs = ( + tool_input_format.when.flat( + { + "input1": {"batch": False, "values": [multi_run_in_repeat_datasets.common_dataset]}, + "queries_0|input2": {"batch": True, "values": multi_run_in_repeat_datasets.repeat_datasets}, + } + ) + .when.nested( + { + "input1": {"batch": False, "values": [multi_run_in_repeat_datasets.common_dataset]}, + "queries": [ + { + "input2": {"batch": True, "values": multi_run_in_repeat_datasets.repeat_datasets}, + } + ], + } + ) + .when.request( + { + "input1": multi_run_in_repeat_datasets.common_dataset, + "queries": [ + { + "input2": {"__class__": "Batch", "values": multi_run_in_repeat_datasets.repeat_datasets}, + } + ], + } + ) ) execute = required_tool.execute.with_inputs(inputs) _check_multi_run_in_repeat(execute) @@ -282,20 +315,33 @@ def test_multi_run_in_repeat_mismatch( tool_input_format: DescribeToolInputs, ): """Same test as above but without the batch wrapper around the common dataset shared between multirun.""" - inputs = tool_input_format.when.flat( - { - "input1": multi_run_in_repeat_datasets.common_dataset, - "queries_0|input2": {"batch": True, "values": multi_run_in_repeat_datasets.repeat_datasets}, - } - ).when.nested( - { - "input1": multi_run_in_repeat_datasets.common_dataset, - "queries": [ - { - "input2": {"batch": True, "values": multi_run_in_repeat_datasets.repeat_datasets}, - } - ], - } + inputs = ( + tool_input_format.when.flat( + { + "input1": multi_run_in_repeat_datasets.common_dataset, + "queries_0|input2": {"batch": True, "values": multi_run_in_repeat_datasets.repeat_datasets}, + } + ) + .when.nested( + { + "input1": multi_run_in_repeat_datasets.common_dataset, + "queries": [ + { + "input2": {"batch": True, "values": multi_run_in_repeat_datasets.repeat_datasets}, + } + ], + } + ) + .when.request( + { + "input1": multi_run_in_repeat_datasets.common_dataset, + "queries": [ + { + "input2": {"__class__": "Batch", "values": multi_run_in_repeat_datasets.repeat_datasets}, + } + ], + } + ) ) execute = required_tool.execute.with_inputs(inputs) _check_multi_run_in_repeat(execute) @@ -328,18 +374,31 @@ def test_multirun_on_multiple_inputs( two_multi_run_datasets: TwoMultiRunsFixture, tool_input_format: DescribeToolInputs, ): - inputs = tool_input_format.when.flat( - { - "input1": {"batch": True, "values": two_multi_run_datasets.first_two_datasets}, - "queries_0|input2": {"batch": True, "values": two_multi_run_datasets.second_two_datasets}, - } - ).when.nested( - { - "input1": {"batch": True, "values": two_multi_run_datasets.first_two_datasets}, - "queries": [ - {"input2": {"batch": True, "values": two_multi_run_datasets.second_two_datasets}}, - ], - } + inputs = ( + tool_input_format.when.flat( + { + "input1": {"batch": True, "values": two_multi_run_datasets.first_two_datasets}, + "queries_0|input2": {"batch": True, "values": two_multi_run_datasets.second_two_datasets}, + } + ) + .when.nested( + { + "input1": {"batch": True, "values": two_multi_run_datasets.first_two_datasets}, + "queries": [ + {"input2": {"batch": True, "values": two_multi_run_datasets.second_two_datasets}}, + ], + } + ) + .when.request( + { + "input1": {"__class__": "Batch", "values": two_multi_run_datasets.first_two_datasets}, + "queries": [ + { + "input2": {"__class__": "Batch", "values": two_multi_run_datasets.second_two_datasets}, + } + ], + } + ) ) execute = required_tool.execute.with_inputs(inputs) execute.assert_has_n_jobs(2) @@ -353,18 +412,39 @@ def test_multirun_on_multiple_inputs_unlinked( two_multi_run_datasets: TwoMultiRunsFixture, tool_input_format: DescribeToolInputs, ): - inputs = tool_input_format.when.flat( - { - "input1": {"batch": True, "linked": False, "values": two_multi_run_datasets.first_two_datasets}, - "queries_0|input2": {"batch": True, "linked": False, "values": two_multi_run_datasets.second_two_datasets}, - } - ).when.nested( - { - "input1": {"batch": True, "linked": False, "values": two_multi_run_datasets.first_two_datasets}, - "queries": [ - {"input2": {"batch": True, "linked": False, "values": two_multi_run_datasets.second_two_datasets}}, - ], - } + inputs = ( + tool_input_format.when.flat( + { + "input1": {"batch": True, "linked": False, "values": two_multi_run_datasets.first_two_datasets}, + "queries_0|input2": { + "batch": True, + "linked": False, + "values": two_multi_run_datasets.second_two_datasets, + }, + } + ) + .when.nested( + { + "input1": {"batch": True, "linked": False, "values": two_multi_run_datasets.first_two_datasets}, + "queries": [ + {"input2": {"batch": True, "linked": False, "values": two_multi_run_datasets.second_two_datasets}}, + ], + } + ) + .when.request( + { + "input1": {"__class__": "Batch", "values": two_multi_run_datasets.first_two_datasets, "linked": False}, + "queries": [ + { + "input2": { + "__class__": "Batch", + "values": two_multi_run_datasets.second_two_datasets, + "linked": False, + }, + } + ], + } + ) ) execute = required_tool.execute.with_inputs(inputs) execute.assert_has_n_jobs(4) @@ -379,7 +459,9 @@ def test_map_over_collection( target_history: TargetHistory, required_tool: RequiredTool, tool_input_format: DescribeToolInputs ): hdca = target_history.with_pair(["123", "456"]) - inputs = tool_input_format.when.any({"input1": {"batch": True, "values": [hdca.src_dict]}}) + legacy = {"input1": {"batch": True, "values": [hdca.src_dict]}} + request = {"input1": {"__class__": "Batch", "values": [hdca.src_dict]}} + inputs = tool_input_format.when.flat(legacy).when.nested(legacy).when.request(request) execute = required_tool.execute.with_inputs(inputs) execute.assert_has_n_jobs(2).assert_creates_n_implicit_collections(1) output_collection = execute.assert_creates_implicit_collection(0) @@ -544,7 +626,11 @@ def test_select_on_null_errors(required_tools: list[RequiredTool], tool_input_fo # is passed, an error (rightfully) occurs. This test verifies that. null_parameter = tool_input_format.when.any({"parameter": None}) for required_tool in required_tools: - required_tool.execute.with_inputs(null_parameter).assert_fails.with_error_containing("an invalid option") + fails = required_tool.execute.with_inputs(null_parameter).assert_fails + if tool_input_format.is_request: + fails.with_error_containing("Input should be") + else: + fails.with_error_containing("an invalid option") @requires_tool_id("gx_select_dynamic_empty") @@ -555,7 +641,11 @@ def test_select_empty_causes_error_regardless( # despite selects otherwise selecting defaults - nothing can be done if the select option list is empty empty = tool_input_format.when.any({}) for required_tool in required_tools: - required_tool.execute.with_inputs(empty).assert_fails.with_error_containing("an invalid option") + failure = required_tool.execute.with_inputs(empty).assert_fails + if tool_input_format.is_request: + failure.with_error_containing("validation error") + else: + failure.with_error_containing("an invalid option") @requires_tool_id("gx_select_optional") diff --git a/lib/galaxy_test/api/test_tool_execution.py b/lib/galaxy_test/api/test_tool_execution.py new file mode 100644 index 00000000000..d21ba61e54b --- /dev/null +++ b/lib/galaxy_test/api/test_tool_execution.py @@ -0,0 +1,134 @@ +""" """ + +from typing import ( + Any, + Dict, +) + +import requests + +from galaxy_test.base.api_asserts import assert_status_code_is_ok +from galaxy_test.base.populators import ( + DatasetPopulator, + skip_without_tool, +) +from ._framework import ApiTestCase + + +class TestToolExecution(ApiTestCase): + dataset_populator: DatasetPopulator + + def setUp(self): + super().setUp() + self.dataset_populator = DatasetPopulator(self.galaxy_interactor) + + @skip_without_tool("gx_int") + def test_validation(self): + with self.dataset_populator.test_history() as history_id: + self._assert_request_validates("gx_int", history_id, {"parameter": 5}) + self._assert_request_invalid("gx_int", history_id, {"parameter": None}) + self._assert_request_invalid("gx_int", history_id, {"parameter": "5"}) + + @skip_without_tool("gx_int") + def test_execution(self): + with self.dataset_populator.test_history() as history_id: + response = self._run("gx_int", history_id, {"parameter": 5}) + assert_status_code_is_ok(response) + response_json = response.json() + tool_request_id = response_json.get("tool_request_id") + task_result = response_json["task_result"] + history_tool_requests = self.dataset_populator.get_history_tool_requests(history_id) + assert tool_request_id in [tr["id"] for tr in history_tool_requests] + self.dataset_populator.wait_on_task_object(task_result) + state = self.dataset_populator.wait_on_tool_request(tool_request_id) + assert state + jobs = self.galaxy_interactor.jobs_for_tool_request(tool_request_id) + self.dataset_populator.wait_for_jobs(jobs, assert_ok=True) + + @skip_without_tool("gx_data") + def test_execution_with_src_urls(self): + with self.dataset_populator.test_history() as history_id: + response = self._run( + "gx_data", + history_id, + { + "parameter": { + "src": "url", + "url": "https://raw.githubusercontent.com/galaxyproject/planemo/7be1bf5b3971a43eaa73f483125bfb8cabf1c440/tests/data/hello.txt", + "ext": "txt", + } + }, + ) + assert_status_code_is_ok(response) + response_json = response.json() + tool_request_id = response_json.get("tool_request_id") + task_result = response_json["task_result"] + self.dataset_populator.wait_on_task_object(task_result) + state = self.dataset_populator.wait_on_tool_request(tool_request_id) + assert state, str(self.dataset_populator.get_tool_request(tool_request_id)) + jobs = self.galaxy_interactor.jobs_for_tool_request(tool_request_id) + self.dataset_populator.wait_for_jobs(jobs, assert_ok=True) + if len(jobs) != 1: + raise Exception(f"Found incorrect number of jobs for tool request - was expecting a single job {jobs}") + assert len(jobs) == 1, jobs + job_id = jobs[0]["id"] + job_outputs = self.galaxy_interactor.job_outputs(job_id) + assert len(job_outputs) == 1 + job_output = job_outputs[0] + assert job_output["name"] == "output" + content = self.dataset_populator.get_history_dataset_content(history_id, dataset=job_output["dataset"]) + assert content == "Hello World!" + + # verify input was not left deferred and materialized before the job started + input_dataset_details = self.dataset_populator.get_history_dataset_details(history_id, hid=1) + assert input_dataset_details["state"] == "ok", input_dataset_details + + @skip_without_tool("gx_data") + def test_execution_with_deferred_src_urls(self): + with self.dataset_populator.test_history() as history_id: + response = self._run( + "gx_data", + history_id, + { + "parameter": { + "src": "url", + "url": "https://raw.githubusercontent.com/galaxyproject/planemo/7be1bf5b3971a43eaa73f483125bfb8cabf1c440/tests/data/hello.txt", + "ext": "txt", + "deferred": True, + } + }, + ) + assert_status_code_is_ok(response) + response_json = response.json() + tool_request_id = response_json.get("tool_request_id") + task_result = response_json["task_result"] + self.dataset_populator.wait_on_task_object(task_result) + state = self.dataset_populator.wait_on_tool_request(tool_request_id) + assert state, str(self.dataset_populator.get_tool_request(tool_request_id)) + jobs = self.galaxy_interactor.jobs_for_tool_request(tool_request_id) + self.dataset_populator.wait_for_jobs(jobs, assert_ok=True) + if len(jobs) != 1: + raise Exception(f"Found incorrect number of jobs for tool request - was expecting a single job {jobs}") + assert len(jobs) == 1, jobs + job_id = jobs[0]["id"] + job_outputs = self.galaxy_interactor.job_outputs(job_id) + assert len(job_outputs) == 1 + job_output = job_outputs[0] + assert job_output["name"] == "output" + content = self.dataset_populator.get_history_dataset_content(history_id, dataset=job_output["dataset"]) + assert content == "Hello World!" + + # verify input was left deferred and infer must have been materialized just for the job + input_dataset_details = self.dataset_populator.get_history_dataset_details(history_id, hid=1) + assert input_dataset_details["state"] == "deferred", input_dataset_details + + def _assert_request_validates(self, tool_id: str, history_id: str, inputs: Dict[str, Any]): + response = self._run(tool_id, history_id, inputs) + assert response.status_code == 200 + + def _assert_request_invalid(self, tool_id: str, history_id: str, inputs: Dict[str, Any]): + response = self._run(tool_id, history_id, inputs) + assert response.status_code == 400 + + def _run(self, tool_id: str, history_id: str, inputs: Dict[str, Any]) -> requests.Response: + return self.dataset_populator.tool_request_raw(tool_id, inputs, history_id) diff --git a/lib/galaxy_test/base/populators.py b/lib/galaxy_test/base/populators.py index ef01abad9fa..02e6c66d92f 100644 --- a/lib/galaxy_test/base/populators.py +++ b/lib/galaxy_test/base/populators.py @@ -174,6 +174,7 @@ TOOL_WITH_SHELL_COMMAND = { DEFAULT_TIMEOUT = 60 # Secs to wait for state to turn ok SKIP_FLAKEY_TESTS_ON_ERROR = os.environ.get("GALAXY_TEST_SKIP_FLAKEY_TESTS_ON_ERROR", None) +INPUT_FORMAT_T = Literal["legacy", "21.01", "request"] PRIVATE_ROLE_TYPE = "private" @@ -1210,6 +1211,15 @@ class BaseDatasetPopulator(BasePopulator): payload = self.run_tool_payload(tool_id, inputs, history_id, **kwds) return self.tools_post(payload) + def tool_request_raw(self, tool_id: str, inputs: dict[str, Any], history_id: str) -> Response: + payload = { + "tool_id": tool_id, + "history_id": history_id, + "inputs": inputs, + } + response = self._post("jobs", data=payload, json=True) + return response + def run_tool(self, tool_id: str, inputs: dict, history_id: str, **kwds): tool_response = self.run_tool_raw(tool_id, inputs, history_id, **kwds) api_asserts.assert_status_code_is(tool_response, 200) @@ -1752,8 +1762,38 @@ class BaseDatasetPopulator(BasePopulator): wait_on(is_ready, "waiting for download to become ready") assert is_ready() + def wait_on_tool_request(self, tool_request_id: str): + # should this to defer to interactor's copy of this method? + + def state(): + state_response = self._get(f"tool_requests/{tool_request_id}/state") + state_response.raise_for_status() + return state_response.json() + + def is_ready(): + is_complete = state() in ["submitted", "failed"] + return True if is_complete else None + + wait_on(is_ready, "waiting for tool request to submit") + return state() == "submitted" + + def get_tool_request(self, tool_request_id: str) -> dict[str, Any]: + response = self._get(f"tool_requests/{tool_request_id}") + api_asserts.assert_status_code_is_ok(response) + return response.json() + + def get_history_tool_requests(self, history_id: str) -> list[dict[str, Any]]: + response = self._get(f"histories/{history_id}/tool_requests") + api_asserts.assert_status_code_is_ok(response) + return response.json() + def wait_on_task(self, async_task_response: Response): - task_id = async_task_response.json()["id"] + response_json = async_task_response.json() + self.wait_on_task_object(response_json) + + def wait_on_task_object(self, async_task_json: dict[str, Any]): + assert "id" in async_task_json, f"Task response {async_task_json} does not contain expected 'id' field." + task_id = async_task_json["id"] return self.wait_on_task_id(task_id) def wait_on_task_id(self, task_id: str): @@ -3969,8 +4009,9 @@ class DescribeJob: class DescribeFailure: - def __init__(self, response: Response): + def __init__(self, response: Response, tool_request: Optional[dict[str, Any]] = None): self._response = response + self._tool_request = tool_request def __call__(self) -> Self: return self @@ -3980,7 +4021,14 @@ class DescribeFailure: return self def with_error_containing(self, message: str) -> Self: - assert message in self._response.text + actual_text = self._response.text + if message not in actual_text: + if self._tool_request: + state_message = self._tool_request["state_message"] + if message not in state_message: + raise AssertionError(f"'{message}' not found in '{state_message}'") + else: + raise AssertionError(f"'{message}' not found in '{actual_text}'") return self @@ -4000,10 +4048,10 @@ class RequiredTool: class DescribeToolInputs: - _input_format: str = "legacy" + _input_format: INPUT_FORMAT_T = "legacy" _inputs: Optional[dict[str, Any]] - def __init__(self, input_format: str): + def __init__(self, input_format: INPUT_FORMAT_T): self._input_format = input_format self._inputs = None @@ -4017,10 +4065,19 @@ class DescribeToolInputs: return self def nested(self, inputs: dict[str, Any]) -> Self: - if self._input_format == "21.01": + if self._input_format in ["21.01", "request"]: self._inputs = inputs return self + def request(self, inputs: dict[str, Any]) -> Self: + if self._input_format in ["request"]: + self._inputs = inputs + return self + + @property + def is_request(self) -> bool: + return self._input_format == "request" + # aliases for self to create silly little English sentense... inputs.when.flat().when.legacy() @property def when(self) -> Self: @@ -4030,8 +4087,9 @@ class DescribeToolInputs: class DescribeToolExecution: _history_id: Optional[str] = None _execute_response: Optional[Response] = None - _input_format: Optional[str] = None + _input_format: Optional[INPUT_FORMAT_T] = None _inputs: dict[str, Any] + _tool_request_id: Optional[str] = None # if input_format == "request" request ID def __init__(self, dataset_populator: BaseDatasetPopulator, tool_id: str): self._dataset_populator = dataset_populator @@ -4059,14 +4117,27 @@ class DescribeToolExecution: self._input_format = "21.01" return self + def with_request(self, inputs: dict[str, Any]) -> Self: + self._inputs = inputs + self._input_format = "request" + return self + def _execute(self): kwds = {} if self._input_format is not None: kwds["input_format"] = self._input_format history_id = self._ensure_history_id - self._execute_response = self._dataset_populator.run_tool_raw( - self._tool_id, self._inputs, history_id, assert_ok=False, **kwds - ) + if self._input_format == "request": + execute_response = self._dataset_populator.tool_request_raw(self._tool_id, self._inputs, history_id) + if execute_response.status_code == 200: + response_json = execute_response.json() + tool_request_id = response_json.get("tool_request_id") + self._dataset_populator.wait_on_tool_request(tool_request_id) + self._execute_response = execute_response + else: + self._execute_response = self._dataset_populator.run_tool_raw( + self._tool_id, self._inputs, history_id, assert_ok=False, **kwds + ) @property def _ensure_history_id(self) -> str: @@ -4084,25 +4155,58 @@ class DescribeToolExecution: execute_response = self._execute_response assert execute_response is not None api_asserts.assert_status_code_is_ok(execute_response) + if self._input_format == "request": + response_json = execute_response.json() + tool_request_id = response_json.get("tool_request_id") + task_result = response_json["task_result"] + self._dataset_populator.wait_on_task_object(task_result) + self._tool_request_id = tool_request_id + return execute_response.json() + @property + def _jobs(self) -> list[dict[str, Any]]: + if self._input_format == "request": + tool_request_id = self._tool_request_id + assert tool_request_id, "request not exected" + jobs = self._dataset_populator.galaxy_interactor.jobs_for_tool_request(tool_request_id) + else: + response = self._assert_executed_ok() + jobs = response["jobs"] + return jobs + def assert_has_n_jobs(self, n: int) -> Self: - response = self._assert_executed_ok() - jobs = response["jobs"] - if len(jobs) != n: - raise AssertionError(f"Expected tool execution to produce {n} jobs but it produced {len(jobs)}") + self._assert_executed_ok() + jobs = self._jobs + num_jobs = len(jobs) + if num_jobs != n: + raise AssertionError(f"Expected tool execution to produce {n} jobs but it produced {num_jobs}") return self - def assert_creates_n_implicit_collections(self, n: int) -> Self: + @property + def _tool_request(self): + tool_request_id = self._tool_request_id + assert tool_request_id, "No tool request set - was tool executed as a request" + return self._dataset_populator.get_tool_request(tool_request_id) + + @property + def _implicit_collections(self) -> list: response = self._assert_executed_ok() - collections = response["implicit_collections"] + if self._input_format == "request": + tool_request = self._tool_request + collections = tool_request["implicit_collections"] + else: + collections = response["implicit_collections"] + return collections + + def assert_creates_n_implicit_collections(self, n: int) -> Self: + collections = self._implicit_collections if len(collections) != n: raise AssertionError(f"Expected tool execution to produce {n} implicit but it produced {len(collections)}") return self def assert_creates_implicit_collection(self, index: Union[str, int]) -> "DescribeToolExecutionOutputCollection": - response = self._assert_executed_ok() - collections = response["implicit_collections"] + collections = self._implicit_collections assert isinstance(index, int) # TODO: implement and then prefer str. history_id = self._ensure_history_id return DescribeToolExecutionOutputCollection(self._dataset_populator, history_id, collections[index]["id"]) @@ -4112,8 +4216,8 @@ class DescribeToolExecution: return self.assert_has_n_jobs(1).assert_has_job(0) def assert_has_job(self, job_index: int = 0) -> DescribeJob: - response = self._assert_executed_ok() - job = response["jobs"][job_index] + self._assert_executed_ok() + job = self._jobs[job_index] history_id = self._ensure_history_id return DescribeJob(self._dataset_populator, history_id, job["id"]) @@ -4122,15 +4226,18 @@ class DescribeToolExecution: self._ensure_executed() execute_response = self._execute_response assert execute_response is not None + tool_request = None if execute_response.status_code != 200: return DescribeFailure(execute_response) else: - response = self._assert_executed_ok() - jobs = response["jobs"] + self._assert_executed_ok() + jobs = self._jobs for job in jobs: final_state = self._dataset_populator.wait_for_job(job["id"]) assert final_state == "error" - return DescribeFailure(execute_response) + if self._tool_request_id: + tool_request = self._tool_request + return DescribeFailure(execute_response, tool_request) # alternative assert_ syntax for cases where it reads better. @property diff --git a/test/functional/test_toolbox_pytest.py b/test/functional/test_toolbox_pytest.py index 19c8c6b8c6e..82915fba890 100644 --- a/test/functional/test_toolbox_pytest.py +++ b/test/functional/test_toolbox_pytest.py @@ -1,10 +1,15 @@ import os from typing import ( + cast, NamedTuple, ) import pytest +from galaxy.tool_util.verify.interactor import ( + DEFAULT_USE_LEGACY_API, + UseLegacyApiT, +) from galaxy_test.api._framework import ApiTestCase from galaxy_test.driver.driver_util import GalaxyTestDriver @@ -60,4 +65,7 @@ class TestFrameworkTools(ApiTestCase): @pytest.mark.parametrize("testcase", cases(), ids=idfn) def test_tool(self, testcase: ToolTest): - self._test_driver.run_tool_test(testcase.tool_id, testcase.test_index, tool_version=testcase.tool_version) + use_legacy_api = cast(UseLegacyApiT, os.environ.get("GALAXY_TEST_USE_LEGACY_TOOL_API", DEFAULT_USE_LEGACY_API)) + self._test_driver.run_tool_test( + testcase.tool_id, testcase.test_index, tool_version=testcase.tool_version, use_legacy_api=use_legacy_api + ) diff --git a/test/functional/tools/parameters/gx_conditional_select_dynamic.xml b/test/functional/tools/parameters/gx_conditional_select_dynamic.xml new file mode 100644 index 00000000000..8837a48ebff --- /dev/null +++ b/test/functional/tools/parameters/gx_conditional_select_dynamic.xml @@ -0,0 +1,52 @@ + + > '$output' + ]]> + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/test/functional/tools/parameters/gx_repeat_select_dynamic.xml b/test/functional/tools/parameters/gx_repeat_select_dynamic.xml new file mode 100644 index 00000000000..ecbfff242fc --- /dev/null +++ b/test/functional/tools/parameters/gx_repeat_select_dynamic.xml @@ -0,0 +1,42 @@ + + > '$output' + ]]> + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/test/functional/tools/parameters/gx_section_select_dynamic.xml b/test/functional/tools/parameters/gx_section_select_dynamic.xml new file mode 100644 index 00000000000..531264f5f00 --- /dev/null +++ b/test/functional/tools/parameters/gx_section_select_dynamic.xml @@ -0,0 +1,42 @@ + + > '$output' + ]]> + + +
+ + + + +
+ + + + + + +
+ + +
+ + + + + + + + +
+ +
+ + + + + + + + diff --git a/test/unit/tool_util/parameter_specification.yml b/test/unit/tool_util/parameter_specification.yml index 7f011a59f60..f0e537ab520 100644 --- a/test/unit/tool_util/parameter_specification.yml +++ b/test/unit/tool_util/parameter_specification.yml @@ -263,8 +263,9 @@ gx_text: - parameter: moocow - parameter: 'some spaces' - parameter: '' + # The legacy API behavior is to make these non-null request parameters into implicit + # "" values for text parameters. - {} - # need to explicitly mark these as non-optional - parameter: null request_invalid: &gx_text_request_invalid - parameter: 5 @@ -627,10 +628,16 @@ gx_genomebuild_optional: gx_genomebuild_multiple: request_valid: - parameter: ["hg18", hg19"] + # I don't love this but this is how the legacy API works and the tool parameters + # wrappers work this way also. Not sure how I would change this. + - parameter: null request_invalid: - parameter: 6 - # DIVERGES_FROM_CURRENT_API + job_internal_valid: + - parameter: ["hg18", hg19"] - parameter: null + job_internal_invalid: + - {} gx_directory_uri: request_valid: diff --git a/test/unit/tool_util/test_parameter_convert.py b/test/unit/tool_util/test_parameter_convert.py index add0875b616..f9265a3d2c2 100644 --- a/test/unit/tool_util/test_parameter_convert.py +++ b/test/unit/tool_util/test_parameter_convert.py @@ -43,7 +43,19 @@ def test_decode_data(): assert decoded_state.input_state["parameter"]["id"] == EXAMPLE_ID_1 -def test_encode_collection(): +def test_decode_data_batch(): + tool_source = tool_source_for("parameters/gx_data") + bundle = input_models_for_tool_source(tool_source) + request_state = RequestToolState( + {"parameter": {"__class__": "Batch", "values": [{"src": "hda", "id": EXAMPLE_ID_1_ENCODED}]}} + ) + request_state.validate(bundle) + decoded_state = decode(request_state, bundle, _fake_decode) + assert decoded_state.input_state["parameter"]["values"][0]["src"] == "hda" + assert decoded_state.input_state["parameter"]["values"][0]["id"] == EXAMPLE_ID_1 + + +def test_decode_collection(): tool_source = tool_source_for("parameters/gx_data_collection") bundle = input_models_for_tool_source(tool_source) request_state = RequestToolState({"parameter": {"src": "hdca", "id": EXAMPLE_ID_1_ENCODED}}) @@ -119,6 +131,22 @@ def test_landing_encode_data(): assert encoded_state.input_state["parameter"]["id"] == EXAMPLE_ID_1_ENCODED +def test_landing_encode_data_batch(): + tool_source = tool_source_for("parameters/gx_data") + bundle = input_models_for_tool_source(tool_source) + request_state = LandingRequestToolState( + {"parameter": {"__class__": "Batch", "values": [{"src": "hda", "id": EXAMPLE_ID_1_ENCODED}]}} + ) + request_state.validate(bundle) + decoded_state = landing_decode(request_state, bundle, _fake_decode) + assert decoded_state.input_state["parameter"]["values"][0]["src"] == "hda" + assert decoded_state.input_state["parameter"]["values"][0]["id"] == EXAMPLE_ID_1 + + encoded_state = landing_encode(decoded_state, bundle, _fake_encode) + assert encoded_state.input_state["parameter"]["values"][0]["src"] == "hda" + assert encoded_state.input_state["parameter"]["values"][0]["id"] == EXAMPLE_ID_1_ENCODED + + def test_dereference(): tool_source = tool_source_for("parameters/gx_data") bundle = input_models_for_tool_source(tool_source) diff --git a/test/unit/tool_util/test_parameter_test_cases.py b/test/unit/tool_util/test_parameter_test_cases.py index 0a3a7034947..4925edf276b 100644 --- a/test/unit/tool_util/test_parameter_test_cases.py +++ b/test/unit/tool_util/test_parameter_test_cases.py @@ -288,13 +288,15 @@ def test_convert_to_requests(): def _validate_path(tool_path: str): tool_source = get_tool_source(tool_path) + tool_id = tool_source.parse_id() + model_name = f"{tool_id} (test case model)" parsed_tool = parse_tool(tool_source) profile = tool_source.parse_profile() test_cases: List[ToolSourceTest] = tool_source.parse_tests_to_dict()["tests"] for test_case in test_cases: if test_case.get("expect_failure"): continue - test_case_state_and_warnings = case_state(test_case, parsed_tool.inputs, profile) + test_case_state_and_warnings = case_state(test_case, parsed_tool.inputs, profile, name=model_name) tool_state = test_case_state_and_warnings.tool_state assert tool_state.state_representation == "test_case_xml"