This document is for an old release 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 tool_shed.webapp.api.tools

import json
import logging
from collections import namedtuple

from galaxy import (
from galaxy.tools.parameters import params_to_strings
from galaxy.tools.repositories 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 tool_shed.tools import tool_validator
from tool_shed.util import (
    shed_util_common as suc
from tool_shed.utility_containers import ToolShedUtilityContainerManager
from tool_shed.webapp.search.tool_search 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('{}({});'.format(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/build_ts_whoosh_index.sh manually. Also TS config option toolshed_search_on has to be True and whoosh_index_dir has to be specified. """ conf = self.app.config 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 = tool_search.search(trans, 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(trans.app) repository = repository_util.get_repository_in_tool_shed(self.app, 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(trans.app, tsr_id, changeset) toolshed_base_url = str(web.url_for('/', qualified=True)).rstrip('/') rb = relation_builder.RelationBuilder(trans.app, 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.id = 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 {repository.name}.' trans.response.status = 404 return {'status': 'error', 'message': message} with ValidationContext.from_app(trans.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: tool_help = tool.help.render(static_path=web.url_for('/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' : r.name, 'version' : r.version} for r in tool.requirements], 'state_inputs' : params_to_strings(tool.inputs, {}, trans.app), 'display' : tool.display_interface, 'action' : web.url_for(tool.action), 'method' : tool.method, 'enctype' : tool.enctype }) return json.dumps(tool_dict)