diff --git a/lib/galaxy/celery/tasks.py b/lib/galaxy/celery/tasks.py index 51512beb5e4..59bff3b5113 100644 --- a/lib/galaxy/celery/tasks.py +++ b/lib/galaxy/celery/tasks.py @@ -59,6 +59,7 @@ from galaxy.schema.tasks import ( PurgeDatasetsTaskRequest, QueueJobs, SetupHistoryExportJob, + TOOL_SOURCE_CLASS, WriteHistoryContentTo, WriteHistoryTo, WriteInvocationTo, @@ -74,9 +75,14 @@ log = get_logger(__name__) @lru_cache -def cached_create_tool_from_representation(app: MinimalManagerApp, raw_tool_source: str, tool_dir: str = ""): +def cached_create_tool_from_representation( + app: MinimalManagerApp, + raw_tool_source: str, + tool_source_class: TOOL_SOURCE_CLASS, + tool_dir: str = "", +): return create_tool_from_representation( - app=app, raw_tool_source=raw_tool_source, tool_dir=tool_dir, tool_source_class="XmlToolSource" + app=app, raw_tool_source=raw_tool_source, tool_dir=tool_dir, tool_source_class=tool_source_class ) @@ -225,11 +231,14 @@ def setup_fetch_data( self, job_id: int, raw_tool_source: str, + tool_source_class: TOOL_SOURCE_CLASS, app: MinimalManagerApp, sa_session: galaxy_scoped_session, task_user_id: Optional[int] = None, ): - tool = cached_create_tool_from_representation(app=app, raw_tool_source=raw_tool_source) + tool = cached_create_tool_from_representation( + app=app, raw_tool_source=raw_tool_source, tool_source_class=tool_source_class + ) job = sa_session.get(Job, job_id) assert job # self.request.hostname is the actual worker name given by the `-n` argument, not the hostname as you might think. @@ -258,11 +267,14 @@ def setup_fetch_data( def finish_job( job_id: int, raw_tool_source: str, + tool_source_class: TOOL_SOURCE_CLASS, app: MinimalManagerApp, sa_session: galaxy_scoped_session, task_user_id: Optional[int] = None, ): - tool = cached_create_tool_from_representation(app=app, raw_tool_source=raw_tool_source) + tool = cached_create_tool_from_representation( + app=app, raw_tool_source=raw_tool_source, tool_source_class=tool_source_class + ) job = sa_session.get(Job, job_id) assert job # TODO: assert state ? @@ -335,8 +347,12 @@ def fetch_data( @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 + app=app, + raw_tool_source=request.tool_source.raw_tool_source, + tool_dir=request.tool_source.tool_dir, + tool_source_class=request.tool_source.tool_source_class, ) + job_submitter.queue_jobs( tool, request, diff --git a/lib/galaxy/schema/tasks.py b/lib/galaxy/schema/tasks.py index 85b27862267..279ddf2123c 100644 --- a/lib/galaxy/schema/tasks.py +++ b/lib/galaxy/schema/tasks.py @@ -1,5 +1,6 @@ from enum import Enum from typing import ( + Literal, Optional, ) from uuid import UUID @@ -153,9 +154,13 @@ class TaskResult(Model): ) +TOOL_SOURCE_CLASS = Literal["XmlToolSource", "YamlToolSource", "CwlToolSource"] + + class ToolSource(Model): raw_tool_source: str tool_dir: str + tool_source_class: TOOL_SOURCE_CLASS = "XmlToolSource" class QueueJobs(Model): diff --git a/lib/galaxy/tool_util/parser/yaml.py b/lib/galaxy/tool_util/parser/yaml.py index 29b37d064a5..0a2e56e47e4 100644 --- a/lib/galaxy/tool_util/parser/yaml.py +++ b/lib/galaxy/tool_util/parser/yaml.py @@ -332,6 +332,8 @@ def _parse_test(i, test_dict) -> ToolSourceTest: def to_test_assert_list(assertions) -> AssertionList: + assertions = assertions or [] + def expand_dict_form(item): key, value = item new_value = value.copy() diff --git a/lib/galaxy/tools/execute.py b/lib/galaxy/tools/execute.py index 5b5cb396a48..e2e32606143 100644 --- a/lib/galaxy/tools/execute.py +++ b/lib/galaxy/tools/execute.py @@ -314,16 +314,26 @@ def _execute( # task_user_id parameter is used to do task user rate limiting. It is only passed # to first task in chain because it is only necessary to rate limit the first # task in a chain. + tool_source_class = type(tool.tool_source).__name__ async_result = ( setup_fetch_data.s( - job_id, raw_tool_source=raw_tool_source, task_user_id=getattr(trans.user, "id", None) + job_id, + raw_tool_source=raw_tool_source, + task_user_id=getattr(trans.user, "id", None), + tool_source_class=tool_source_class, ) | fetch_data.s(job_id=job_id) | set_job_metadata.s( extended_metadata_collection="extended" in tool.app.config.metadata_strategy, job_id=job_id, - ).set(link_error=finish_job.si(job_id=job_id, raw_tool_source=raw_tool_source)) - | finish_job.si(job_id=job_id, raw_tool_source=raw_tool_source) + ).set( + link_error=finish_job.si( + job_id=job_id, + raw_tool_source=raw_tool_source, + tool_source_class=tool_source_class, + ) + ) + | finish_job.si(job_id=job_id, raw_tool_source=raw_tool_source, tool_source_class=tool_source_class) )() job2.set_runner_external_id(async_result.task_id) continue diff --git a/lib/galaxy/webapps/galaxy/services/jobs.py b/lib/galaxy/webapps/galaxy/services/jobs.py index 249b8544fa9..2cad4ad0391 100644 --- a/lib/galaxy/webapps/galaxy/services/jobs.py +++ b/lib/galaxy/webapps/galaxy/services/jobs.py @@ -273,6 +273,7 @@ class JobsService(ServiceBase): tool_source = ToolSource( raw_tool_source=tool.tool_source.to_string(), tool_dir=tool.tool_dir, + tool_source_class=type(tool.tool_source).__name__, ) task_request = QueueJobs( user=trans.async_request_user,