/usr/lib/python2.7/dist-packages/bioblend/galaxy/tools/__init__.py is in python-bioblend 0.7.0-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 | """
Contains possible interaction dealing with Galaxy tools.
"""
from bioblend.galaxy.client import Client
from bioblend.util import attach_file
from os.path import basename
from json import dumps
class ToolClient(Client):
def __init__(self, galaxy_instance):
self.module = 'tools'
super(ToolClient, self).__init__(galaxy_instance)
def get_tools(self, tool_id=None, name=None, trackster=None):
"""
Get all tools or filter the specific one(s) via the provided ``name``
or ``tool_id``. Provide only one argument, ``name`` or ``tool_id``,
but not both.
If ``name`` is set and multiple names match the given name, all the
tools matching the argument will be returned.
:type tool_id: str
:param tool_id: id of the requested tool
:type name: str
:param name: name of the requested tool(s)
:type trackster: bool
:param trackster: if True, only tools that are compatible with
Trackster are returned
:rtype: list
:return: List of tool descriptions.
.. seealso:: bioblend.galaxy.toolshed.get_repositories()
"""
if tool_id is not None and name is not None:
raise ValueError('Provide only one argument between name or tool_id, but not both')
tools = self._raw_get_tool(in_panel=False, trackster=trackster)
if tool_id is not None:
tool = next((_ for _ in tools if _['id'] == tool_id), None)
tools = [tool] if tool is not None else []
elif name is not None:
tools = [_ for _ in tools if _['name'] == name]
return tools
def get_tool_panel(self):
"""
Get a list of available tool elements in Galaxy's configured toolbox.
:rtype: list
:return: List containing tools (if not in sections) or tool sections
with nested tool descriptions.
.. seealso:: bioblend.galaxy.toolshed.get_repositories()
"""
return self._raw_get_tool(in_panel=True)
def _raw_get_tool(self, in_panel=None, trackster=None):
params = {}
params['in_panel'] = in_panel
params['trackster'] = trackster
return Client._get(self, params=params)
def show_tool(self, tool_id, io_details=False, link_details=False):
"""
Get details of a given tool.
:type tool_id: str
:param tool_id: id of the requested tool
:type io_details: bool
:param io_details: if True, get also input and output details
:type link_details: bool
:param link_details: if True, get also link details
"""
params = {}
params['io_details'] = io_details
params['link_details'] = link_details
return Client._get(self, id=tool_id, params=params)
def run_tool(self, history_id, tool_id, tool_inputs):
"""
Runs tool specified by ``tool_id`` in history indicated
by ``history_id`` with inputs from ``dict`` ``tool_inputs``.
:type history_id: str
:param history_id: encoded ID of the history in which to run the tool
:type tool_id: str
:param tool_id: ID of the tool to be run
:type tool_inputs: dict
:param tool_inputs: dictionary of input datasets and parameters
for the tool (see below)
The ``tool_inputs`` dict should contain input datasets and parameters
in the (largely undocumented) format used by the Galaxy API.
Some examples can be found in https://bitbucket.org/galaxy/galaxy-central/src/tip/test/api/test_tools.py .
"""
payload = {}
payload["history_id"] = history_id
payload["tool_id"] = tool_id
try:
payload["inputs"] = tool_inputs.to_dict()
except AttributeError:
payload["inputs"] = tool_inputs
return self._tool_post(payload)
def upload_file(self, path, history_id, **keywords):
"""
Upload the file specified by ``path`` to the history specified by
``history_id``.
:type path: str
:param path: path of the file to upload
:type history_id: str
:param history_id: id of the history where to upload the file
:type file_name: str
:param file_name: (optional) name of the new history dataset
:type file_type: str
:param file_type: Galaxy datatype for the new dataset, default is auto
:type dbkey: str
:param dbkey: (optional) genome dbkey
:type to_posix_lines: bool
:param to_posix_lines: if True, convert universal line endings to POSIX
line endings. Default is True. Set to False if you upload a gzip, bz2
or zip archive containing a binary file
:type space_to_tab: bool
:param space_to_tab: whether to convert spaces to tabs. Default is
False. Applicable only if to_posix_lines is True
"""
if "file_name" not in keywords:
keywords["file_name"] = basename(path)
payload = self._upload_payload(history_id, **keywords)
payload["files_0|file_data"] = attach_file(path, name=keywords["file_name"])
try:
return self._tool_post(payload, files_attached=True)
finally:
payload["files_0|file_data"].close()
def upload_from_ftp(self, path, history_id, **keywords):
"""
Upload the file specified by ``path`` from the user's FTP directory to
the history specified by ``history_id``.
:type path: str
:param path: path of the file in the user's FTP directory
:type history_id: str
:param history_id: id of the history where to upload the file
See :meth:`upload_file` for the optional parameters.
"""
payload = self._upload_payload(history_id, **keywords)
payload['files_0|ftp_files'] = path
return self._tool_post(payload)
def paste_content(self, content, history_id, **kwds):
"""
Upload a string to a new dataset in the history specified by
``history_id``.
:type content: str
:param content: content of the new dataset to upload or a list of URLs
(one per line) to upload
:type history_id: str
:param history_id: id of the history where to upload the content
See :meth:`upload_file` for the optional parameters (except file_name).
"""
payload = self._upload_payload(history_id, **kwds)
payload["files_0|url_paste"] = content
return self._tool_post(payload, files_attached=False)
put_url = paste_content
def _upload_payload(self, history_id, **keywords):
payload = {}
payload["history_id"] = history_id
payload["tool_id"] = keywords.get("tool_id", "upload1")
tool_input = {}
tool_input["file_type"] = keywords.get('file_type', 'auto')
tool_input["dbkey"] = keywords.get("dbkey", "?")
if not keywords.get('to_posix_lines', True):
tool_input['files_0|to_posix_lines'] = False
elif keywords.get('space_to_tab', False):
tool_input['files_0|space_to_tab'] = 'Yes'
if 'file_name' in keywords:
tool_input["files_0|NAME"] = keywords['file_name']
tool_input["files_0|type"] = "upload_dataset"
payload["inputs"] = tool_input
return payload
def _tool_post(self, payload, files_attached=False):
if files_attached:
# If files_attached - this will be posted as multi-part form data
# and so each individual parameter needs to be encoded so can be
# decoded as JSON by Galaxy (hence dumping complex parameters).
# If no files are attached, the whole thing is posted as
# application/json and dumped/loaded all at once by requests and
# Galaxy.
complex_payload_params = ["inputs"]
for key in complex_payload_params:
if key in payload:
payload[key] = dumps(payload[key])
return Client._post(self, payload, files_attached=files_attached)
|