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)