Mercurial > repos > bcclaywell > argo_navis

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

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