-
Notifications
You must be signed in to change notification settings - Fork 83
/
interface.py
119 lines (99 loc) · 3.22 KB
/
interface.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
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()
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]