Mercurial > repos > shellac > guppy_basecaller

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

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit c699937486c35866861690329de38ec1a5d9f783"
author shellac
date Sat, 02 May 2020 07:14:21 -0400
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:26e78fe6e8c4
1 """Interface over bioblend and direct access to ToolShed API via requests."""
2
3 import json
4
5 from galaxy.util import unicodify
6
7 from planemo.bioblend import (
8 ensure_module,
9 toolshed,
10 )
11 from planemo.io import untar_to
12
13 REPOSITORY_DOWNLOAD_TEMPLATE = (
14 "%srepository/download?repository_id=%s"
15 "&changeset_revision=default&file_type=gz"
16 )
17
18
19 def tool_shed_instance(url, key, email, password):
20 ensure_module()
21 tsi = toolshed.ToolShedInstance(
22 url=url,
23 key=key,
24 email=email,
25 password=password
26 )
27 return tsi
28
29
30 def find_repository(tsi, owner, name):
31 """ Find repository information for given owner and repository
32 name.
33 """
34 repos = tsi.repositories.get_repositories()
35
36 def matches(r):
37 return r["owner"] == owner and r["name"] == name
38
39 matching_repos = list(filter(matches, repos))
40 if not matching_repos:
41 return None
42 else:
43 return matching_repos[0]
44
45
46 def latest_installable_revision(tsi, repository_id):
47 info = tsi.repositories.show_repository(repository_id)
48 owner = info["owner"]
49 name = info["name"]
50 revisions = tsi.repositories.get_ordered_installable_revisions(
51 name, owner
52 )
53 if len(revisions) == 0:
54 msg = "Failed to find installable revisions for [{0}, {1}].".format(
55 owner,
56 name,
57 )
58 raise Exception(msg)
59 else:
60 return revisions[-1]
61
62
63 def username(tsi):
64 """ Fetch current username from shed given API key/auth.
65 """
66 user = _user(tsi)
67 return user["username"]
68
69
70 def api_exception_to_message(e):
71 """ Convert API exception to human digestable error message - parsing
72 out the shed generate message if possible.
73 """
74 message = unicodify(e)
75 if hasattr(e, "read"):
76 message = e.read()
77 try:
78 # Galaxy passes nice JSON messages as their errors, which bioblend
79 # blindly returns. Attempt to parse those.
80 upstream_error = json.loads(message)
81 message = upstream_error['err_msg']
82 except Exception:
83 pass
84 return message
85
86
87 def find_category_ids(tsi, categories):
88 """ Translate human readable category names into their associated IDs.
89 """
90 category_list = tsi.categories.get_categories()
91
92 category_ids = []
93 for cat in categories:
94 matching_cats = [x for x in category_list if x['name'] == cat]
95 if not matching_cats:
96 message = "Failed to find category %s" % cat
97 raise Exception(message)
98 category_ids.append(matching_cats[0]['id'])
99 return category_ids
100
101
102 def download_tar(tsi, repo_id, destination, to_directory):
103 base_url = tsi.base_url
104 if not base_url.endswith("/"):
105 base_url += "/"
106 download_url = REPOSITORY_DOWNLOAD_TEMPLATE % (base_url, repo_id)
107 if to_directory:
108 tar_args = ['-xzf', '-', '--strip-components=1']
109 untar_to(download_url, tar_args=tar_args, dest_dir=destination)
110 else:
111 untar_to(download_url, path=destination)
112
113
114 def _user(tsi):
115 """ Fetch user information from the ToolShed API for given
116 key.
117 """
118 # TODO: this should be done with an actual bioblend method,
119 # see https://github.com/galaxyproject/bioblend/issues/130.
120 response = tsi.make_get_request(tsi.url + "/users")
121 return response.json()[0]