diff venv/lib/python2.7/site-packages/planemo/shed/interface.py @ 0:d67268158946 draft

planemo upload commit a3f181f5f126803c654b3a66dd4e83a48f7e203b
author bcclaywell
date Mon, 12 Oct 2015 17:43:33 -0400
parents
children
line wrap: on
line diff
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/venv/lib/python2.7/site-packages/planemo/shed/interface.py	Mon Oct 12 17:43:33 2015 -0400
@@ -0,0 +1,119 @@
+""" Interface over bioblend and direct access to ToolShed
+API via requests.
+"""
+import json
+from planemo.io import untar_to
+from planemo.bioblend import (
+    toolshed,
+    ensure_module,
+)
+
+REPOSITORY_DOWNLOAD_TEMPLATE = (
+    "%srepository/download?repository_id=%s"
+    "&changeset_revision=default&file_type=gz"
+)
+
+
+def tool_shed_instance(url, key, email, password):
+    ensure_module(toolshed)
+    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 = str(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.repositories.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:
+        untar_args = "-xzf - -C %s --strip-components 1" % destination
+    else:
+        untar_args = None
+    untar_to(download_url, destination, untar_args)
+
+
+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]