Source code for

import json
import logging
from collections import namedtuple

from galaxy import (
from import params_to_strings
from import ValidationContext
from galaxy.web import expose_api_raw_anonymous_and_sessionless
from galaxy.webapps.base.controller import BaseAPIController
from tool_shed.dependencies.repository import relation_builder
from import tool_validator
from tool_shed.util import (
from tool_shed.util import shed_util_common as suc
from tool_shed.utility_containers import ToolShedUtilityContainerManager
from import ToolSearch

log = logging.getLogger(__name__)

[docs]class ToolsController(BaseAPIController): """RESTful controller for interactions with tools in the Tool Shed."""
[docs] @expose_api_raw_anonymous_and_sessionless def index(self, trans, **kwd): """ GET /api/tools Displays a collection of tools with optional criteria. :param q: (optional)if present search on the given query will be performed :type q: str :param page: (optional)requested page of the search :type page: int :param page_size: (optional)requested page_size of the search :type page_size: int :param jsonp: (optional)flag whether to use jsonp format response, defaults to False :type jsonp: bool :param callback: (optional)name of the function to wrap callback in used only when jsonp is true, defaults to 'callback' :type callback: str :returns dict: object containing list of results and metadata Examples: GET http://localhost:9009/api/tools GET http://localhost:9009/api/tools?q=fastq """ q = kwd.get("q", "") if not q: raise exceptions.NotImplemented( 'Listing of all the tools is not implemented. Provide parameter "q" to search instead.' ) else: page = kwd.get("page", 1) page_size = kwd.get("page_size", 10) try: page = int(page) page_size = int(page_size) except ValueError: raise exceptions.RequestParameterInvalidException('The "page" and "page_size" have to be integers.') return_jsonp = util.asbool(kwd.get("jsonp", False)) callback = kwd.get("callback", "callback") search_results = self._search(trans, q, page, page_size) if return_jsonp: response = str(f"{callback}({json.dumps(search_results)});") else: response = json.dumps(search_results) return response
def _search(self, trans, q, page=1, page_size=10): """ Perform the search over TS tools index. Note that search works over the Whoosh index which you have to pre-create with scripts/tool_shed/ manually. Also TS config option toolshed_search_on has to be True and whoosh_index_dir has to be specified. """ conf = if not conf.toolshed_search_on: raise exceptions.ConfigDoesNotAllowException( "Searching the TS through the API is turned off for this instance." ) if not conf.whoosh_index_dir: raise exceptions.ConfigDoesNotAllowException( "There is no directory for the search index specified. Please contact the administrator." ) search_term = q.strip() if len(search_term) < 1: raise exceptions.RequestParameterInvalidException("The search term has to be at least one character long.") tool_search = ToolSearch() Boosts = namedtuple( "Boosts", ["tool_name_boost", "tool_description_boost", "tool_help_boost", "tool_repo_owner_username_boost"] ) boosts = Boosts( float(conf.get("tool_name_boost", 1.2)), float(conf.get("tool_description_boost", 0.6)), float(conf.get("tool_help_boost", 0.4)), float(conf.get("tool_repo_owner_username_boost", 0.3)), ) results =, search_term, page, page_size, boosts) results["hostname"] = web.url_for("/", qualified=True) return results
[docs] @expose_api_raw_anonymous_and_sessionless def json(self, trans, **kwd): """ GET /api/tools/json Get the tool form JSON for a tool in a repository. :param guid: the GUID of the tool :param guid: str :param tsr_id: the ID of the repository :param tsr_id: str :param changeset: the changeset at which to load the tool json :param changeset: str """ guid = kwd.get("guid", None) tsr_id = kwd.get("tsr_id", None) changeset = kwd.get("changeset", None) if None in [changeset, tsr_id, guid]: message = "Changeset, repository ID, and tool GUID are all required parameters." trans.response.status = 400 return {"status": "error", "message": message} tsucm = ToolShedUtilityContainerManager( repository = repository_util.get_repository_in_tool_shed(, tsr_id) repository_clone_url = common_util.generate_clone_url_for_repository_in_tool_shed(repository.user, repository) repository_metadata = metadata_util.get_repository_metadata_by_changeset_revision(, tsr_id, changeset) toolshed_base_url = str(web.url_for("/", qualified=True)).rstrip("/") rb = relation_builder.RelationBuilder(, repository, repository_metadata, toolshed_base_url) repository_dependencies = rb.get_repository_dependencies_for_changeset_revision() containers_dict = tsucm.build_repository_containers( repository, changeset, repository_dependencies, repository_metadata ) found_tool = None for folder in containers_dict["valid_tools"].folders: if hasattr(folder, "valid_tools"): for tool in folder.valid_tools: = tool.tool_id tool_guid = suc.generate_tool_guid(repository_clone_url, tool) if tool_guid == guid: found_tool = tool break if found_tool is None: message = f"Unable to find tool with guid {guid} in repository {}." trans.response.status = 404 return {"status": "error", "message": message} with ValidationContext.from_app( as validation_context: tv = tool_validator.ToolValidator(validation_context) repository, tool, valid, message = tv.load_tool_from_changeset_revision( tsr_id, changeset, found_tool.tool_config ) if message or not valid: status = "error" return dict(message=message, status=status) tool_help = "" if tool_help ="/static"), host_url=web.url_for("/", qualified=True)) tool_help = util.unicodify(tool_help, "utf-8") tool_dict = tool.to_dict(trans) tool_dict["inputs"] = {} tool.populate_model(trans, tool.inputs, {}, tool_dict["inputs"]) tool_dict.update( { "help": tool_help, "citations": bool(tool.citations), "requirements": [{"name":, "version": r.version} for r in tool.requirements], "state_inputs": params_to_strings(tool.inputs, {},, "display": tool.display_interface, "action": web.url_for(tool.action), "method": tool.method, "enctype": tool.enctype, } ) return json.dumps(tool_dict)