Warning
This document is for an in-development version of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.
Source code for galaxy.webapps.galaxy.services.workflows
import logging
import re
from typing import (
Any,
)
from pydantic import UUID4
from galaxy import (
exceptions,
web,
)
from galaxy.managers.context import (
ProvidesHistoryContext,
ProvidesUserContext,
)
from galaxy.managers.jobs import JobManager
from galaxy.managers.workflows import (
RefactorRequest,
RefactorResponse,
WorkflowContentsManager,
WorkflowSerializer,
WorkflowsManager,
)
from galaxy.model import (
ImplicitCollectionJobs,
LandingRequestToWorkflowInvocationAssociation,
StoredWorkflow,
WorkflowInvocation,
WorkflowLandingRequest,
)
from galaxy.schema.fields import DecodedDatabaseIdField
from galaxy.schema.invocation import WorkflowInvocationResponse
from galaxy.schema.schema import (
InvocationsStateCounts,
WorkflowIndexPayload,
)
from galaxy.schema.workflows import (
InvokeWorkflowPayload,
StoredWorkflowDetailed,
WorkflowExtractionByIdsPayload,
WorkflowExtractionPayload,
WorkflowExtractionResult,
)
from galaxy.util.tool_shed.tool_shed_registry import Registry
from galaxy.webapps.galaxy.services.base import ServiceBase
from galaxy.webapps.galaxy.services.notifications import NotificationService
from galaxy.webapps.galaxy.services.sharable import ShareableService
from galaxy.workflow.extract import (
collect_output_label_targets,
extract_workflow,
extract_workflow_by_ids,
normalize_output_label_key,
)
from galaxy.workflow.run import queue_invoke
from galaxy.workflow.run_request import build_workflow_run_configs
log = logging.getLogger(__name__)
def _to_extraction_result(stored_workflow: StoredWorkflow) -> WorkflowExtractionResult:
return WorkflowExtractionResult.model_validate({"id": stored_workflow.id})
def _sanitize_output_label(label: str) -> str:
label = re.sub(r"\s+", " ", label.strip())
if not label:
raise exceptions.RequestParameterInvalidException("output_labels contains an empty label")
return label[:255]
def _validate_input_names(
dataset_names: list[str] | None,
dataset_collection_names: list[str] | None,
) -> None:
"""Validate user-supplied workflow input names (step labels).
Dataset and collection input names share one namespace (the single
``step_labels`` set in ``extract_steps``), so uniqueness is checked across
the combined list. Only inspects names that were actually supplied — the
no-names default path (the ``"Input Dataset"`` constants) is untouched.
Names are kept raw: limits are enforced by rejection, never truncation.
"""
provided = (dataset_names or []) + (dataset_collection_names or [])
seen: set[str] = set()
for name in provided:
if not name.strip():
raise exceptions.RequestParameterInvalidException("workflow input names must not be empty")
if len(name) > 255:
raise exceptions.RequestParameterInvalidException(f"workflow input name exceeds 255 characters: {name!r}")
if name in seen:
raise exceptions.RequestParameterInvalidException(f"workflow input names must be unique: {name!r}")
seen.add(name)
[docs]
class WorkflowsService(ServiceBase):
[docs]
def __init__(
self,
workflows_manager: WorkflowsManager,
workflow_contents_manager: WorkflowContentsManager,
serializer: WorkflowSerializer,
tool_shed_registry: Registry,
notification_service: NotificationService,
job_manager: JobManager,
):
self._workflows_manager = workflows_manager
self._workflow_contents_manager = workflow_contents_manager
self._serializer = serializer
self.shareable_service = ShareableService(workflows_manager, serializer, notification_service)
self._tool_shed_registry = tool_shed_registry
self._job_manager = job_manager
[docs]
def index(
self,
trans: ProvidesUserContext,
payload: WorkflowIndexPayload,
include_total_count: bool = False,
) -> tuple[list[dict[str, Any]], int | None]:
user = trans.user
missing_tools = payload.missing_tools
query, total_matches = self._workflows_manager.index_query(trans, payload, include_total_count)
rval = []
for wf in query.all():
item = wf.to_dict(
value_mapper={"id": trans.security.encode_id, "latest_workflow_id": trans.security.encode_id}
)
encoded_id = trans.security.encode_id(wf.id)
item["annotations"] = [x.annotation for x in wf.annotations]
item["url"] = web.url_for("workflow", id=encoded_id)
item["owner"] = wf.user.username
item["source_metadata"] = wf.latest_workflow.source_metadata
if not payload.skip_step_counts:
item["number_of_steps"] = wf.latest_workflow.step_count
item["show_in_tool_panel"] = False
if user is not None:
item["show_in_tool_panel"] = wf.show_in_tool_panel(user_id=user.id)
rval.append(item)
if missing_tools:
workflows_missing_tools = []
workflows = []
workflows_by_toolshed = {}
for value in rval:
stored_workflow = self._workflows_manager.get_stored_workflow(trans, value["id"], by_stored_id=True)
tools = self._workflow_contents_manager.get_all_tools(stored_workflow.latest_workflow)
missing_tool_ids = [
tool["tool_id"] for tool in tools if trans.app.toolbox.is_missing_shed_tool(tool["tool_id"])
]
if len(missing_tool_ids) > 0:
value["missing_tools"] = missing_tool_ids
workflows_missing_tools.append(value)
for workflow in workflows_missing_tools:
for tool_id in workflow["missing_tools"]:
toolshed, _, owner, name, tool, version = tool_id.split("/")
shed_url = self.__get_full_shed_url(toolshed)
repo_identifier = "/".join((toolshed, owner, name))
if repo_identifier not in workflows_by_toolshed:
workflows_by_toolshed[repo_identifier] = dict(
shed=shed_url.rstrip("/"),
repository=name,
owner=owner,
tools=[tool_id],
workflows=[workflow["name"]],
)
else:
if tool_id not in workflows_by_toolshed[repo_identifier]["tools"]:
workflows_by_toolshed[repo_identifier]["tools"].append(tool_id)
if workflow["name"] not in workflows_by_toolshed[repo_identifier]["workflows"]:
workflows_by_toolshed[repo_identifier]["workflows"].append(workflow["name"])
for repo_tag in workflows_by_toolshed:
workflows.append(workflows_by_toolshed[repo_tag])
return workflows, total_matches
return rval, total_matches
[docs]
def invoke_workflow(
self,
trans,
workflow_id,
payload: InvokeWorkflowPayload,
) -> WorkflowInvocationResponse | list[WorkflowInvocationResponse]:
if trans.anonymous:
raise exceptions.AuthenticationRequired("You need to be logged in to run workflows.")
trans.check_user_activation()
# Get workflow + accessibility check.
by_stored_id = not payload.instance
stored_workflow = self._workflows_manager.get_stored_accessible_workflow(trans, workflow_id, by_stored_id)
version = payload.version
if version is None and payload.instance:
workflow = stored_workflow.get_internal_version_by_id(workflow_id)
else:
workflow = stored_workflow.get_internal_version(version)
run_configs = build_workflow_run_configs(trans, workflow, payload.model_dump(exclude_unset=True))
is_batch = payload.batch
if not is_batch and len(run_configs) != 1:
raise exceptions.RequestParameterInvalidException("Must specify 'batch' to use batch parameters.")
require_exact_tool_versions = payload.require_exact_tool_versions
tools = self._workflow_contents_manager.get_all_tools(workflow)
missing_tools = [
tool
for tool in tools
if not trans.app.toolbox.has_tool(
tool["tool_id"],
tool_version=tool["tool_version"],
tool_uuid=tool["tool_uuid"],
exact=require_exact_tool_versions,
user=trans.user,
)
]
if missing_tools:
missing_tools_message = "Workflow was not invoked; the following required tools are not installed: "
if require_exact_tool_versions:
missing_tools_message += ", ".join(
[f"{tool['tool_id']} (version {tool['tool_version']})" for tool in missing_tools]
)
else:
missing_tools_message += ", ".join([tool["tool_id"] for tool in missing_tools])
raise exceptions.MessageException(missing_tools_message)
invocations = []
for run_config in run_configs:
workflow_scheduler_id = payload.scheduler
# TODO: workflow scheduler hints
work_request_params = dict(scheduler=workflow_scheduler_id)
workflow_invocation = queue_invoke(
trans=trans,
workflow=workflow,
workflow_run_config=run_config,
request_params=work_request_params,
flush=False,
)
invocations.append(workflow_invocation)
# Create landing request association if provided
if payload.landing_uuid:
self._create_landing_request_association(trans, payload.landing_uuid, invocations)
trans.sa_session.commit()
encoded_invocations = [WorkflowInvocationResponse(**invocation.to_dict()) for invocation in invocations]
if is_batch:
return encoded_invocations
else:
return encoded_invocations[0]
[docs]
def extract_from_history(
self,
trans: ProvidesHistoryContext,
history,
payload: WorkflowExtractionPayload,
) -> WorkflowExtractionResult:
if trans.user is None:
raise exceptions.AuthenticationRequired("Workflow extraction requires an authenticated user.")
_validate_input_names(payload.dataset_names, payload.dataset_collection_names)
stored_workflow = extract_workflow(
trans,
user=trans.user,
history=history,
job_ids=payload.job_ids,
dataset_ids=payload.dataset_hids,
dataset_collection_ids=payload.dataset_collection_hids,
workflow_name=payload.workflow_name,
dataset_names=payload.dataset_names,
dataset_collection_names=payload.dataset_collection_names,
)
return _to_extraction_result(stored_workflow)
[docs]
def extract_by_ids(
self,
trans: ProvidesHistoryContext,
payload: WorkflowExtractionByIdsPayload,
) -> WorkflowExtractionResult:
if trans.user is None:
raise exceptions.AuthenticationRequired("Workflow extraction requires an authenticated user.")
self._validate_extract_by_ids_payload(trans, payload)
stored_workflow = extract_workflow_by_ids(
trans,
user=trans.user,
workflow_name=payload.workflow_name,
job_manager=self._job_manager,
job_ids=payload.job_ids,
implicit_collection_jobs_ids=payload.implicit_collection_jobs_ids,
hda_ids=payload.hda_ids,
hdca_ids=payload.hdca_ids,
dataset_names=payload.dataset_names,
dataset_collection_names=payload.dataset_collection_names,
output_labels=payload.output_labels,
)
return _to_extraction_result(stored_workflow)
def _validate_extract_by_ids_payload(
self,
trans: ProvidesHistoryContext,
payload: WorkflowExtractionByIdsPayload,
) -> None:
"""Cross-payload checks that need DB access. Pydantic handles per-field
shape; this enforces semantic rules across job_ids /
implicit_collection_jobs_ids that depend on the loaded Job / ICJ rows
so extract_workflow_by_ids can trust its input.
"""
for field in ("job_ids", "implicit_collection_jobs_ids", "hda_ids", "hdca_ids"):
ids = getattr(payload, field)
if len(set(ids)) != len(ids):
raise exceptions.RequestParameterInvalidException(f"{field} contains duplicates")
for job_id in payload.job_ids:
job = self._job_manager.get_accessible_job(trans, job_id)
icj_assoc = job.implicit_collection_jobs_association
if icj_assoc is not None:
raise exceptions.RequestParameterInvalidException(
f"job_ids[{job_id}] is part of implicit collection jobs "
f"{icj_assoc.implicit_collection_jobs_id} - pass via "
"implicit_collection_jobs_ids instead."
)
sa_session = trans.sa_session
dataset_collection_manager = trans.app.dataset_collection_manager
for icj_id in payload.implicit_collection_jobs_ids:
icj = sa_session.get(ImplicitCollectionJobs, icj_id)
if icj is None:
raise exceptions.ObjectNotFound(f"ImplicitCollectionJobs {icj_id} not found")
if icj.populated_state != ImplicitCollectionJobs.populated_states.OK:
raise exceptions.RequestParameterInvalidException(
f"ImplicitCollectionJobs {icj_id} is in populated_state "
f"{icj.populated_state!r}; only 'ok' is extractable"
)
output_hdcas = icj.output_dataset_collection_instances
if not output_hdcas:
raise exceptions.RequestParameterInvalidException(
f"ImplicitCollectionJobs {icj_id} has no output collections to extract"
)
for hdca in output_hdcas:
dataset_collection_manager.get_dataset_collection_instance(trans, "history", hdca.id)
_validate_input_names(payload.dataset_names, payload.dataset_collection_names)
output_targets = collect_output_label_targets(
trans,
job_manager=self._job_manager,
job_ids=payload.job_ids,
implicit_collection_jobs_ids=payload.implicit_collection_jobs_ids,
)
seen_output_ids = set()
seen_resolved_outputs = set()
seen_labels = set()
for output_label in payload.output_labels:
sanitized_label = _sanitize_output_label(output_label.label)
output_label.label = sanitized_label
output_key = normalize_output_label_key(trans, output_label.kind, output_label.id)
output_label.id = output_key[1]
if output_key in seen_output_ids:
raise exceptions.RequestParameterInvalidException(
f"output_labels contains duplicate {output_label.kind} id {output_label.id}"
)
seen_output_ids.add(output_key)
output_target = output_targets.get(output_key)
if output_target is None:
raise exceptions.RequestParameterInvalidException(
f"output_labels includes {output_label.kind} id {output_label.id} "
"that is not produced by a selected extraction step"
)
if output_target.step_key in seen_resolved_outputs:
raise exceptions.RequestParameterInvalidException(
f"output_labels contains multiple labels for output {output_target.output_name!r}"
)
seen_resolved_outputs.add(output_target.step_key)
if sanitized_label in seen_labels:
raise exceptions.RequestParameterInvalidException(
f"output_labels contains duplicate workflow output label {sanitized_label!r}"
)
seen_labels.add(sanitized_label)
[docs]
def delete(self, trans, workflow_id):
workflow_to_delete = self._workflows_manager.get_stored_workflow(trans, workflow_id)
self._workflows_manager.check_security(trans, workflow_to_delete)
self._workflows_manager.delete(workflow_to_delete)
[docs]
def undelete(self, trans, workflow_id):
workflow_to_undelete = self._workflows_manager.get_stored_workflow(trans, workflow_id)
self._workflows_manager.check_security(trans, workflow_to_undelete)
self._workflows_manager.undelete(workflow_to_undelete)
[docs]
def get_versions(self, trans, workflow_id, instance: bool):
stored_workflow: StoredWorkflow = self._workflows_manager.get_stored_accessible_workflow(
trans, workflow_id, by_stored_id=not instance
)
return [
{"version": i, "update_time": w.update_time.isoformat(), "steps": len(w.steps)}
for i, w in enumerate(reversed(stored_workflow.workflows))
]
[docs]
def invocation_counts(self, trans, workflow_id, instance: bool) -> InvocationsStateCounts:
stored_workflow: StoredWorkflow = self._workflows_manager.get_stored_accessible_workflow(
trans, workflow_id, by_stored_id=not instance
)
return stored_workflow.invocation_counts()
[docs]
def refactor(
self,
trans: ProvidesUserContext,
workflow_id: DecodedDatabaseIdField,
payload: RefactorRequest,
instance: bool,
) -> RefactorResponse:
stored_workflow = self._workflows_manager.get_stored_workflow(trans, workflow_id, by_stored_id=not instance)
return self._workflow_contents_manager.refactor(trans, stored_workflow, payload)
[docs]
def show_workflow(self, trans, workflow_id, instance, legacy, version) -> StoredWorkflowDetailed:
stored_workflow = self._workflows_manager.get_stored_workflow(trans, workflow_id, by_stored_id=not instance)
if stored_workflow.importable is False and stored_workflow.user != trans.user and not trans.user_is_admin:
wf_count = 0 if not trans.user else trans.user.count_stored_workflow_user_assocs(stored_workflow)
if wf_count == 0:
message = "Workflow is neither importable, nor owned by or shared with current user"
raise exceptions.ItemAccessibilityException(message)
if legacy:
style = "legacy"
else:
style = "instance"
if version is None and instance:
# A Workflow instance may not be the latest workflow version attached to StoredWorkflow.
# This figures out the correct version so that we return the correct Workflow and version.
for i, workflow in enumerate(reversed(stored_workflow.workflows)):
if workflow.id == workflow_id:
version = i
break
detailed_workflow = StoredWorkflowDetailed(
**self._workflow_contents_manager.workflow_to_dict(trans, stored_workflow, style=style, version=version)
)
return detailed_workflow
def _get_workflows_list(
self,
trans: ProvidesUserContext,
payload,
):
workflows, _ = self.index(trans, payload)
return workflows
def __get_full_shed_url(self, url):
for shed_url in self._tool_shed_registry.tool_sheds.values():
if url in shed_url:
return shed_url
return None
def _create_landing_request_association(
self, trans: ProvidesUserContext, landing_uuid: UUID4 | None, invocations: list[WorkflowInvocation]
):
"""Create association between landing request and workflow invocations."""
# Look up the workflow landing request by UUID
workflow_landing_request = (
trans.sa_session.query(WorkflowLandingRequest).where(WorkflowLandingRequest.uuid == landing_uuid).first()
)
if not workflow_landing_request:
raise exceptions.ObjectNotFound(f"WorkflowLandingRequest with UUID {landing_uuid} not found")
# Create associations for each invocation
for invocation in invocations:
association = LandingRequestToWorkflowInvocationAssociation(
landing_request=workflow_landing_request, workflow_invocation=invocation
)
trans.sa_session.add(association)