Mercurial > repos > shellac > guppy_basecaller
diff env/lib/python3.7/site-packages/planemo/shed/interface.py @ 5:9b1c78e6ba9c draft default tip
"planemo upload commit 6c0a8142489327ece472c84e558c47da711a9142"
author | shellac |
---|---|
date | Mon, 01 Jun 2020 08:59:25 -0400 (2020-06-01) |
parents | 79f47841a781 |
children |
line wrap: on
line diff
--- a/env/lib/python3.7/site-packages/planemo/shed/interface.py Thu May 14 16:47:39 2020 -0400 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,121 +0,0 @@ -"""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]