/usr/lib/python3/dist-packages/bioblend/galaxyclient.py is in python3-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 | """
Helper class for Galaxy and ToolShed Instance object
This class is primarily a helper for the library and user code
should not use it directly.
A base representation of an instance
"""
import base64
import json
import requests
from requests_toolbelt import MultipartEncoder
import six
from six.moves.urllib.parse import urljoin, urlparse
from .galaxy.client import ConnectionError
class GalaxyClient(object):
def __init__(self, url, key=None, email=None, password=None):
# Make sure the url scheme is defined (otherwise requests will not work)
if not urlparse(url).scheme:
url = "http://" + url
# All of Galaxy's and ToolShed's API's are rooted at <url>/api so make that the url
self.base_url = url
self.url = urljoin(url, 'api')
# If key has been supplied, use it; otherwise just set email and
# password and grab user's key before first request.
if key:
self._key = key
else:
self._key = None
self.email = email
self.password = password
self.json_headers = {'Content-Type': 'application/json'}
self.verify = True # Should SSL verification be done
def _make_url(self, module, module_id=None, deleted=False, contents=False):
"""
Compose a URL based on the provided arguments.
:type module: :class:`~.galaxy.Client` subclass
:param module: The base module for which to make the URL. For
example: an object of class LibraryClient, WorkflowClient,
HistoryClient, ToolShedClient
:type module_id: str
:param module_id: The encoded ID for a specific module (eg, library ID)
:type deleted: bool
:param deleted: If ``True``, include ``deleted`` in the URL, after the module
name (eg, ``<base_url>/api/libraries/deleted``)
:type contents: bool
:param contents: If ``True``, include 'contents' in the URL, after the module ID:
``<base_url>/api/libraries/<encoded_library_id>/contents``
"""
c_url = self.url
c_url = '/'.join([c_url, module.module])
if deleted is True:
c_url = '/'.join([c_url, 'deleted'])
if module_id is not None:
c_url = '/'.join([c_url, module_id])
if contents is True:
c_url = '/'.join([c_url, 'contents'])
return c_url
def make_get_request(self, url, **kwargs):
"""
Make a GET request using the provided ``url``.
Keyword arguments are the same as in requests.request.
If ``verify`` is not provided, ``self.verify`` will be used.
If the ``params`` are not provided, use ``default_params`` class field.
If params are provided and the provided dict does not have ``key`` key,
the default ``self.key`` value will be included in what's passed to
the server via the request.
"""
params = kwargs.get('params')
if params is not None and params.get('key', False) is False:
params['key'] = self.key
else:
params = self.default_params
kwargs['params'] = params
kwargs.setdefault('verify', self.verify)
r = requests.get(url, **kwargs)
return r
def make_post_request(self, url, payload, params=None, files_attached=False):
"""
Make a POST request using the provided ``url`` and ``payload``.
The ``payload`` must be a dict that contains the request values.
The payload dict may contain file handles (in which case the files_attached
flag must be set to true).
If the ``params`` are not provided, use ``default_params`` class field.
If params are provided and the provided dict does not have ``key`` key,
the default ``self.key`` value will be included in what's passed to
the server via the request.
The return value will contain the response body as a JSON object.
"""
if params is not None and params.get('key', False) is False:
params['key'] = self.key
else:
params = self.default_params
# Compute data, headers, params arguments for request.post,
# leveraging the requests-toolbelt library if any files have
# been attached.
if files_attached:
payload.update(params)
payload = MultipartEncoder(fields=payload)
headers = self.json_headers.copy()
headers['Content-Type'] = payload.content_type
post_params = {}
else:
payload = json.dumps(payload)
headers = self.json_headers
post_params = params
r = requests.post(url, data=payload, headers=headers,
verify=self.verify, params=post_params)
if r.status_code == 200:
return r.json()
# @see self.body for HTTP response body
raise ConnectionError("Unexpected response from galaxy: %s" %
r.status_code, body=r.text)
def make_delete_request(self, url, payload=None, params=None):
"""
Make a DELETE request using the provided ``url`` and the optional
arguments.
The ``payload`` must be a dict that can be converted into a JSON
object (via ``json.dumps``)
If the ``params`` are not provided, use ``default_params`` class field.
If params are provided and the provided dict does not have ``key`` key,
the default ``self.key`` value will be included in what's passed to
the server via the request.
"""
if params is not None and params.get('key', False) is False:
params['key'] = self.key
else:
params = self.default_params
r = requests.delete(url, verify=self.verify, data=payload, params=params)
return r
def make_put_request(self, url, payload=None, params=None):
"""
Make a PUT request using the provided ``url`` with required payload.
The ``payload`` must be a dict that can be converted into a JSON
object (via ``json.dumps``)
"""
if params is not None and params.get('key', False) is False:
params['key'] = self.key
else:
params = self.default_params
payload = json.dumps(payload)
r = requests.put(url, verify=self.verify, data=payload, params=params)
return r
@property
def key(self):
if not self._key and self.email is not None and self.password is not None:
unencoded_credentials = "%s:%s" % (self.email, self.password)
authorization = base64.b64encode(unencoded_credentials)
headers = self.json_headers.copy()
headers["Authorization"] = authorization
auth_url = "%s/authenticate/baseauth" % self.url
# make_post_request uses default_params, which uses this and
# sets wrong headers - so using lower level method.
r = requests.get(auth_url, verify=self.verify, headers=headers)
if r.status_code != 200:
raise Exception("Failed to authenticate user.")
response = r.json()
if isinstance(response, (six.string_types, six.text_type)):
# bug in Tool Shed
response = json.loads(response)
self._key = response["api_key"]
return self._key
@property
def default_params(self):
return {'key': self.key}
|