diff env/lib/python3.7/site-packages/planemo/shed/interface.py @ 0:26e78fe6e8c4 draft

"planemo upload commit c699937486c35866861690329de38ec1a5d9f783"
author shellac
date Sat, 02 May 2020 07:14:21 -0400
parents
children
line wrap: on
line diff
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/env/lib/python3.7/site-packages/planemo/shed/interface.py	Sat May 02 07:14:21 2020 -0400
@@ -0,0 +1,121 @@
+"""Interface over bioblend and direct access to ToolShed API via requests."""
+
+import json
+
+from galaxy.util import unicodify
+
+from planemo.bioblend import (
+    ensure_module,
+    toolshed,
+)
+from planemo.io import untar_to
+
+REPOSITORY_DOWNLOAD_TEMPLATE = (
+    "%srepository/download?repository_id=%s"
+    "&changeset_revision=default&file_type=gz"
+)
+
+
+def tool_shed_instance(url, key, email, password):
+    ensure_module()
+    tsi = toolshed.ToolShedInstance(
+        url=url,
+        key=key,
+        email=email,
+        password=password
+    )
+    return tsi
+
+
+def find_repository(tsi, owner, name):
+    """ Find repository information for given owner and repository
+    name.
+    """
+    repos = tsi.repositories.get_repositories()
+
+    def matches(r):
+        return r["owner"] == owner and r["name"] == name
+
+    matching_repos = list(filter(matches, repos))
+    if not matching_repos:
+        return None
+    else:
+        return matching_repos[0]
+
+
+def latest_installable_revision(tsi, repository_id):
+    info = tsi.repositories.show_repository(repository_id)
+    owner = info["owner"]
+    name = info["name"]
+    revisions = tsi.repositories.get_ordered_installable_revisions(
+        name, owner
+    )
+    if len(revisions) == 0:
+        msg = "Failed to find installable revisions for [{0}, {1}].".format(
+            owner,
+            name,
+        )
+        raise Exception(msg)
+    else:
+        return revisions[-1]
+
+
+def username(tsi):
+    """ Fetch current username from shed given API key/auth.
+    """
+    user = _user(tsi)
+    return user["username"]
+
+
+def api_exception_to_message(e):
+    """ Convert API exception to human digestable error message - parsing
+    out the shed generate message if possible.
+    """
+    message = unicodify(e)
+    if hasattr(e, "read"):
+        message = e.read()
+        try:
+            # Galaxy passes nice JSON messages as their errors, which bioblend
+            # blindly returns. Attempt to parse those.
+            upstream_error = json.loads(message)
+            message = upstream_error['err_msg']
+        except Exception:
+            pass
+    return message
+
+
+def find_category_ids(tsi, categories):
+    """ Translate human readable category names into their associated IDs.
+    """
+    category_list = tsi.categories.get_categories()
+
+    category_ids = []
+    for cat in categories:
+        matching_cats = [x for x in category_list if x['name'] == cat]
+        if not matching_cats:
+            message = "Failed to find category %s" % cat
+            raise Exception(message)
+        category_ids.append(matching_cats[0]['id'])
+    return category_ids
+
+
+def download_tar(tsi, repo_id, destination, to_directory):
+    base_url = tsi.base_url
+    if not base_url.endswith("/"):
+        base_url += "/"
+    download_url = REPOSITORY_DOWNLOAD_TEMPLATE % (base_url, repo_id)
+    if to_directory:
+        tar_args = ['-xzf', '-', '--strip-components=1']
+        untar_to(download_url, tar_args=tar_args, dest_dir=destination)
+    else:
+        untar_to(download_url, path=destination)
+
+
+def _user(tsi):
+    """ Fetch user information from the ToolShed API for given
+    key.
+    """
+    # TODO: this should be done with an actual bioblend method,
+    # see https://github.com/galaxyproject/bioblend/issues/130.
+    response = tsi.make_get_request(tsi.url + "/users")
+    return response.json()[0]