diff env/lib/python3.7/site-packages/bioblend/galaxyclient.py @ 5:9b1c78e6ba9c draft default tip

"planemo upload commit 6c0a8142489327ece472c84e558c47da711a9142"
author shellac
date Mon, 01 Jun 2020 08:59:25 -0400
parents 79f47841a781
children
line wrap: on
line diff
--- a/env/lib/python3.7/site-packages/bioblend/galaxyclient.py	Thu May 14 16:47:39 2020 -0400
+++ /dev/null	Thu Jan 01 00:00:00 1970 +0000
@@ -1,266 +0,0 @@
-"""
-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
-import six
-from requests_toolbelt import MultipartEncoder
-from six.moves.urllib.parse import urljoin, urlparse
-
-from bioblend import ConnectionError
-from bioblend.util import FileStream
-
-
-class GalaxyClient(object):
-
-    def __init__(self, url, key=None, email=None, password=None, verify=True, timeout=None):
-        """
-        :param verify: Whether to verify the server's TLS certificate
-        :type verify: bool
-        :param timeout: Timeout for requests operations, set to None for no timeout (the default).
-        :type timeout: float
-        """
-        # 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 = verify
-        self.timeout = timeout
-
-    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 = '/'.join([self.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.
-
-        :rtype: requests.Response
-        :return: the response object.
-        """
-        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)
-        kwargs.setdefault('timeout', self.timeout)
-        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.
-
-        :return: The decoded response.
-        """
-
-        def my_dumps(d):
-            """
-            Apply ``json.dumps()`` to the values of the dict ``d`` if they are
-            not of type ``FileStream``.
-            """
-            for k, v in d.items():
-                if not isinstance(v, FileStream):
-                    d[k] = json.dumps(v)
-            return d
-
-        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 = my_dumps(payload)
-            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,
-                          timeout=self.timeout)
-        if r.status_code == 200:
-            try:
-                return r.json()
-            except Exception as e:
-                raise ConnectionError("Request was successful, but cannot decode the response content: %s" %
-                                      e, body=r.content, status_code=r.status_code)
-        # @see self.body for HTTP response body
-        raise ConnectionError("Unexpected HTTP status code: %s" % r.status_code,
-                              body=r.text, status_code=r.status_code)
-
-    def make_delete_request(self, url, payload=None, params=None):
-        """
-        Make a DELETE request using the provided ``url`` and the optional
-        arguments.
-
-        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.
-
-        :type payload: dict
-        :param payload: a JSON-serializable dictionary
-
-        :rtype: requests.Response
-        :return: the response object.
-        """
-        if params is not None and params.get('key', False) is False:
-            params['key'] = self.key
-        else:
-            params = self.default_params
-        if payload is not None:
-            payload = json.dumps(payload)
-        headers = self.json_headers
-        r = requests.delete(url, verify=self.verify, data=payload, params=params,
-                            headers=headers, timeout=self.timeout)
-        return r
-
-    def make_put_request(self, url, payload=None, params=None):
-        """
-        Make a PUT request using the provided ``url`` with required payload.
-
-        :type payload: dict
-        :param payload: a JSON-serializable dictionary
-
-        :return: The decoded response.
-        """
-        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)
-        headers = self.json_headers
-        r = requests.put(url, data=payload, params=params, headers=headers,
-                         verify=self.verify, timeout=self.timeout)
-        if r.status_code == 200:
-            try:
-                return r.json()
-            except Exception as e:
-                raise ConnectionError("Request was successful, but cannot decode the response content: %s" %
-                                      e, body=r.content, status_code=r.status_code)
-        # @see self.body for HTTP response body
-        raise ConnectionError("Unexpected HTTP status code: %s" % r.status_code,
-                              body=r.text, status_code=r.status_code)
-
-    def make_patch_request(self, url, payload=None, params=None):
-        """
-        Make a PATCH request using the provided ``url`` with required payload.
-
-        :type payload: dict
-        :param payload: a JSON-serializable dictionary
-
-        :return: The decoded response.
-        """
-        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)
-        headers = self.json_headers
-        r = requests.patch(url, data=payload, params=params, headers=headers,
-                           verify=self.verify, timeout=self.timeout)
-        if r.status_code == 200:
-            try:
-                return r.json()
-            except Exception as e:
-                raise ConnectionError("Request was successful, but cannot decode the response content: %s" %
-                                      e, body=r.content, status_code=r.status_code)
-        # @see self.body for HTTP response body
-        raise ConnectionError("Unexpected HTTP status code: %s" % r.status_code,
-                              body=r.text, status_code=r.status_code)
-
-    @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.encode())
-            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}