Source code for galaxy.webapps.galaxy.services.tools

import logging
import shutil
import tempfile
from json import dumps
from typing import (
    Any,
    Dict,
    List,
    Optional,
    Union,
)

from starlette.datastructures import UploadFile

from galaxy import (
    exceptions,
    util,
)
from galaxy.config import GalaxyAppConfiguration
from galaxy.managers.collections_util import dictify_dataset_collection_instance
from galaxy.managers.context import (
    ProvidesHistoryContext,
    ProvidesUserContext,
)
from galaxy.managers.histories import HistoryManager
from galaxy.model import (
    LibraryDatasetDatasetAssociation,
    PostJobAction,
)
from galaxy.model.base import transaction
from galaxy.schema.fetch_data import (
    FetchDataFormPayload,
    FetchDataPayload,
    FilesPayload,
)
from galaxy.security.idencoding import IdEncodingHelper
from galaxy.tools import Tool
from galaxy.tools.search import ToolBoxSearch
from galaxy.webapps.galaxy.services._fetch_util import validate_and_normalize_targets
from galaxy.webapps.galaxy.services.base import ServiceBase

log = logging.getLogger(__name__)


[docs]class ToolsService(ServiceBase):
[docs]    def __init__(
        self,
        config: GalaxyAppConfiguration,
        toolbox_search: ToolBoxSearch,
        security: IdEncodingHelper,
        history_manager: HistoryManager,
    ):
        super().__init__(security)
        self.config = config
        self.toolbox_search = toolbox_search
        self.history_manager = history_manager

[docs]    def create_fetch(
        self,
        trans: ProvidesHistoryContext,
        fetch_payload: Union[FetchDataFormPayload, FetchDataPayload],
        files: Optional[List[UploadFile]] = None,
    ):
        payload = fetch_payload.model_dump(exclude_unset=True)
        request_version = "1"
        history_id = payload.pop("history_id")
        clean_payload = {}
        files_payload = {}
        if files:
            for i, upload_file in enumerate(files):
                with tempfile.NamedTemporaryFile(
                    dir=trans.app.config.new_file_path, prefix="upload_file_data_", delete=False
                ) as dest:
                    shutil.copyfileobj(upload_file.file, dest)  # type: ignore[misc]  # https://github.com/python/mypy/issues/15031
                upload_file.file.close()
                files_payload[f"files_{i}|file_data"] = FilesPayload(
                    filename=upload_file.filename, local_filename=dest.name
                )
        for key, value in payload.items():
            if key == "key":
                continue
            if key.startswith("files_") or key.startswith("__files_"):
                files_payload[key] = value
                continue
            clean_payload[key] = value
        clean_payload["check_content"] = self.config.check_upload_content
        validate_and_normalize_targets(trans, clean_payload)
        request = dumps(clean_payload)
        create_payload = {
            "tool_id": "__DATA_FETCH__",
            "history_id": history_id,
            "inputs": {
                "request_version": request_version,
                "request_json": request,
                "file_count": str(len(files_payload)),
            },
        }
        create_payload.update(files_payload)
        return self._create(trans, create_payload)

    def _create(self, trans: ProvidesHistoryContext, payload, **kwd):
        if trans.user_is_bootstrap_admin:
            raise exceptions.RealUserRequiredException("Only real users can execute tools or run jobs.")
        action = payload.get("action")
        if action == "rerun":
            raise Exception("'rerun' action has been deprecated")

        # Get tool.
        tool_version = payload.get("tool_version")
        tool_id = payload.get("tool_id")
        tool_uuid = payload.get("tool_uuid")
        get_kwds = dict(
            tool_id=tool_id,
            tool_uuid=tool_uuid,
            tool_version=tool_version,
        )
        if tool_id is None and tool_uuid is None:
            raise exceptions.RequestParameterMissingException("Must specify either a tool_id or a tool_uuid.")

        tool = trans.app.toolbox.get_tool(**get_kwds)
        if not tool:
            log.debug(f"Not found tool with kwds [{get_kwds}]")
            raise exceptions.ToolMissingException("Tool not found.")
        if not tool.allow_user_access(trans.user):
            raise exceptions.ItemAccessibilityException("Tool not accessible.")
        if self.config.user_activation_on:
            if not trans.user:
                log.warning("Anonymous user attempts to execute tool, but account activation is turned on.")
            elif not trans.user.active:
                log.warning(
                    f'User "{trans.user.email}" attempts to execute tool, but account activation is turned on and user account is not active.'
                )

        # Set running history from payload parameters.
        # History not set correctly as part of this API call for
        # dataset upload.
        if history_id := payload.get("history_id"):
            history_id = trans.security.decode_id(history_id) if isinstance(history_id, str) else history_id
            target_history = self.history_manager.get_mutable(history_id, trans.user, current_history=trans.history)
        else:
            target_history = None

        # Set up inputs.
        inputs = payload.get("inputs", {})
        if not isinstance(inputs, dict):
            raise exceptions.RequestParameterInvalidException(f"inputs invalid {inputs}")

        # Find files coming in as multipart file data and add to inputs.
        for k, v in payload.items():
            if k.startswith("files_") or k.startswith("__files_"):
                inputs[k] = v

        # for inputs that are coming from the Library, copy them into the history
        self._patch_library_inputs(trans, inputs, target_history)

        # TODO: encode data ids and decode ids.
        # TODO: handle dbkeys
        params = util.Params(inputs, sanitize=False)
        incoming = params.__dict__

        # use_cached_job can be passed in via the top-level payload or among the tool inputs.
        # I think it should be a top-level parameter, but because the selector is implemented
        # as a regular tool parameter we accept both.
        use_cached_job = payload.get("use_cached_job", False) or util.string_as_bool(
            inputs.get("use_cached_job", "false")
        )
        preferred_object_store_id = payload.get("preferred_object_store_id")
        input_format = str(payload.get("input_format", "legacy"))
        if "data_manager_mode" in payload:
            incoming["__data_manager_mode"] = payload["data_manager_mode"]
        vars = tool.handle_input(
            trans,
            incoming,
            history=target_history,
            use_cached_job=use_cached_job,
            input_format=input_format,
            preferred_object_store_id=preferred_object_store_id,
        )

        new_pja_flush = False
        for job in vars.get("jobs", []):
            if inputs.get("send_email_notification", False):
                # Unless an anonymous user is invoking this via the API it
                # should never be an option, but check and enforce that here
                if trans.user is None:
                    raise exceptions.ToolExecutionError("Anonymously run jobs cannot send an email notification.")
                else:
                    job_email_action = PostJobAction("EmailAction")
                    job.add_post_job_action(job_email_action)
                    new_pja_flush = True

        if new_pja_flush:
            with transaction(trans.sa_session):
                trans.sa_session.commit()

        return self._handle_inputs_output_to_api_response(trans, tool, target_history, vars)

    def _handle_inputs_output_to_api_response(self, trans, tool, target_history, vars):
        # TODO: check for errors and ensure that output dataset(s) are available.
        output_datasets = vars.get("out_data", [])
        rval: Dict[str, Any] = {"outputs": [], "output_collections": [], "jobs": [], "implicit_collections": []}
        rval["produces_entry_points"] = tool.produces_entry_points
        if job_errors := vars.get("job_errors", []):
            # If we are here - some jobs were successfully executed but some failed.
            rval["errors"] = job_errors

        outputs = rval["outputs"]
        # TODO:?? poss. only return ids?
        for output_name, output in output_datasets:
            output_dict = output.to_dict()
            # add the output name back into the output data structure
            # so it's possible to figure out which newly created elements
            # correspond with which tool file outputs
            output_dict["output_name"] = output_name
            outputs.append(output_dict)

        for job in vars.get("jobs", []):
            rval["jobs"].append(job.to_dict(view="collection"))

        for output_name, collection_instance in vars.get("output_collections", []):
            history = target_history or trans.history
            output_dict = dictify_dataset_collection_instance(
                collection_instance,
                security=trans.security,
                url_builder=trans.url_builder,
                parent=history,
            )
            output_dict["output_name"] = output_name
            rval["output_collections"].append(output_dict)

        for output_name, collection_instance in vars.get("implicit_collections", {}).items():
            history = target_history or trans.history
            output_dict = dictify_dataset_collection_instance(
                collection_instance,
                security=trans.security,
                url_builder=trans.url_builder,
                parent=history,
            )
            output_dict["output_name"] = output_name
            rval["implicit_collections"].append(output_dict)

        trans.security.encode_all_ids(rval, recursive=True)
        return rval

    def _search(self, q, view):
        """
        Perform the search on the given query.
        Boosts and numer of results are configurable in galaxy.ini file.

        :param q: the query to search with
        :type  q: str

        :return:      Dictionary containing the tools' ids of the best hits.
        :return type: dict
        """
        panel_view = view or self.config.default_panel_view

        results = self.toolbox_search.search(
            q=q,
            panel_view=panel_view,
            config=self.config,
        )
        return results

    def _patch_library_inputs(self, trans: ProvidesHistoryContext, inputs, target_history):
        """
        Transform inputs from the data library to history items.
        """
        for k, v in inputs.items():
            new_value = self._patch_library_dataset(trans, v, target_history)
            if new_value:
                v = new_value
            elif isinstance(v, dict) and "values" in v:
                for index, value in enumerate(v["values"]):
                    patched = self._patch_library_dataset(trans, value, target_history)
                    if patched:
                        v["values"][index] = patched
            inputs[k] = v

    def _patch_library_dataset(self, trans: ProvidesHistoryContext, v, target_history):
        if isinstance(v, dict) and "id" in v and v.get("src") == "ldda":
            ldda = trans.sa_session.get(LibraryDatasetDatasetAssociation, self.decode_id(v["id"]))
            if trans.user_is_admin or trans.app.security_agent.can_access_dataset(
                trans.get_current_user_roles(), ldda.dataset
            ):
                return ldda.to_history_dataset_association(target_history, add_to_history=True)

    #
    # -- Helper methods --
    #
    def _get_tool(self, trans, id, tool_version=None, user=None) -> Tool:
        tool = trans.app.toolbox.get_tool(id, tool_version)
        if not tool:
            raise exceptions.ObjectNotFound(f"Could not find tool with id '{id}'.")
        if not tool.allow_user_access(user):
            raise exceptions.AuthenticationFailed(f"Access denied, please login for tool with id '{id}'.")
        return tool

    def _detect(self, trans: ProvidesUserContext, tool_id):
        """
        Detect whether the tool with the given id is installed.

        :param tool_id: exact id of the tool
        :type tool_id:  str

        :return:      list with available versions
        "return type: list
        """
        tools = trans.app.toolbox.get_tool(tool_id, get_all_versions=True)
        detected_versions = []
        if tools:
            for tool in tools:
                if tool and tool.allow_user_access(trans.user):
                    detected_versions.append(tool.version)
        return detected_versions