Source code for galaxy.webapps.galaxy.controllers.tool_runner

"""
Controller handles external tool related requests
"""

import logging

from markupsafe import escape

import galaxy.util
from galaxy import web
from galaxy.tools import DataSourceTool
from galaxy.web import (
    error,
    url_for,
)
from galaxy.webapps.base.controller import BaseUIController

log = logging.getLogger(__name__)


[docs]class ToolRunner(BaseUIController):
    # Hack to get biomart to work, ideally, we could pass tool_id to biomart and receive it back
[docs]    @web.expose
    def biomart(self, trans, tool_id="biomart", **kwd):
        """Catches the tool id and redirects as needed"""
        return self.index(trans, tool_id=tool_id, **kwd)

    # test to get hapmap to work, ideally, we could pass tool_id to hapmap biomart and receive it back
[docs]    @web.expose
    def hapmapmart(self, trans, tool_id="hapmapmart", **kwd):
        """Catches the tool id and redirects as needed"""
        return self.index(trans, tool_id=tool_id, **kwd)

[docs]    @web.expose
    def default(self, trans, tool_id=None, **kwd):
        """Catches the tool id and redirects as needed"""
        return self.index(trans, tool_id=tool_id, **kwd)

    def __get_tool(self, tool_id, tool_version=None, get_loaded_tools_by_lineage=False, set_selected=False):
        tool_version_select_field, tools, tool = self.get_toolbox().get_tool_components(
            tool_id, tool_version, get_loaded_tools_by_lineage, set_selected
        )
        return tool

[docs]    @web.expose
    def index(self, trans, tool_id=None, from_noframe=None, **kwd):
        def __tool_404__():
            log.debug("index called with tool id '%s' but no such tool exists", tool_id)
            trans.log_event(f"Tool id '{tool_id}' does not exist")
            trans.response.status = 404
            return trans.show_error_message(f"Tool '{escape(tool_id)}' does not exist.")

        # tool id not available, redirect to main page
        if tool_id is None:
            return trans.response.send_redirect(url_for(controller="root", action="welcome"))
        tool = self.__get_tool(tool_id)
        # tool id is not matching, display an error
        if not tool:
            return __tool_404__()
        if tool.require_login and not trans.user:
            redirect = url_for(controller="tool_runner", action="index", tool_id=tool_id, **kwd)
            return trans.response.send_redirect(
                url_for(
                    controller="user",
                    action="login",
                    cntrller="user",
                    status="info",
                    message="You must be logged in to use this tool.",
                    redirect=redirect,
                )
            )
        if not tool.allow_user_access(trans.user):
            return __tool_404__()
        # FIXME: Tool class should define behavior
        if tool.tool_type in ["default", "interactivetool"]:
            return trans.response.send_redirect(url_for(controller="root", tool_id=tool_id))

        # execute tool without displaying form
        # (used for datasource tools, but note that data_source_async tools
        # are handled separately by the async controller)
        params = galaxy.util.Params(kwd, sanitize=False).__dict__
        if tool.input_translator:
            # perform test translation of the incoming params without affecting originals
            # the actual translation will happen later
            # this is only for checking if we end up with required parameters
            test_params = params.copy()
            tool.input_translator.translate(test_params)
        else:
            test_params = params
        if tool.tool_type == "data_source":
            if "URL" not in test_params:
                error("Execution of `data_source` tools requires a `URL` parameter")
            # preserve original params sent by the remote server as extra dict
            # before in-place translation happens, then clean the incoming params
            params.update({"incoming_request_params": params.copy()})
            if tool.input_translator and tool.wants_params_cleaned:
                for k in list(params.keys()):
                    if k not in tool.input_translator.vocabulary and k not in ("URL", "incoming_request_params"):
                        # the remote server has sent a param
                        # that the tool is not expecting -> drop it
                        del params[k]
        else:
            if "runtool_btn" not in test_params:
                error("Tool execution through the `tool_runner` requires a `runtool_btn` flag")

        # We may be visiting Galaxy for the first time ( e.g., sending data from UCSC ),
        # so make sure to create a new history if we've never had one before.
        history = tool.get_default_history_by_trans(trans, create=True)
        try:
            vars = tool.handle_input(trans, params, history=history)
        except Exception as e:
            error(galaxy.util.unicodify(e))
        if len(params) > 0:
            trans.log_event(f"Tool params: {str(params)}", tool_id=tool_id)
        return trans.fill_template("root/tool_runner.mako", **vars)

[docs]    @web.expose
    def rerun(self, trans, id=None, job_id=None, **kwd):
        """
        Given a HistoryDatasetAssociation id, find the job and that created
        the dataset, extract the parameters, and display the appropriate tool
        form with parameters already filled in.
        """
        if job_id is None:
            if not id:
                error("'id' parameter is required")
            try:
                id = int(id)
            except ValueError:
                # it's not an un-encoded id, try to parse as encoded
                try:
                    id = trans.security.decode_id(id)
                except Exception:
                    error("Invalid value for 'id' parameter")
            # Get the dataset object
            data = trans.sa_session.query(trans.app.model.HistoryDatasetAssociation).get(id)
            # only allow rerunning if user is allowed access to the dataset.
            if not (
                trans.user_is_admin
                or trans.app.security_agent.can_access_dataset(trans.get_current_user_roles(), data.dataset)
            ):
                error("You are not allowed to access this dataset")
            # Get the associated job, if any.
            job = data.creating_job
            if job:
                job_id = trans.security.encode_id(job.id)
            else:
                raise Exception("Failed to get job information for dataset hid %d" % data.hid)
        return trans.response.send_redirect(url_for(controller="root", job_id=job_id))

[docs]    @web.expose
    def data_source_redirect(self, trans, tool_id=None):
        """
        Redirects a user accessing a Data Source tool to its target action link.
        This method will subvert mix-mode content blocking in several browsers when
        accessing non-https data_source tools from an https galaxy server.

        Tested as working on Safari 7.0 and FireFox 26
        Subverting did not work on Chrome 31
        """
        if tool_id is None:
            return trans.response.send_redirect(url_for(controller="root", action="welcome"))
        tool = self.__get_tool(tool_id)
        # No tool matching the tool id, display an error (shouldn't happen)
        if not tool:
            log.error("data_source_redirect called with tool id '%s' but no such tool exists", tool_id)
            trans.log_event(f"Tool id '{tool_id}' does not exist")
            trans.response.status = 404
            return trans.show_error_message(f"Tool '{escape(tool_id)}' does not exist.")

        if isinstance(tool, DataSourceTool):
            link = url_for(tool.action, **tool.get_static_param_values(trans))
        else:
            link = url_for(controller="tool_runner", tool_id=tool.id)
        return trans.response.send_redirect(link)