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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
|
"""
Contains possible interaction dealing with Galaxy users.
Most of these methods must be executed by a registered Galaxy admin user.
"""
from typing import (
Any,
Dict,
List,
Optional,
TYPE_CHECKING,
)
from bioblend import ConnectionError
from bioblend.galaxy.client import Client
if TYPE_CHECKING:
from bioblend.galaxy import GalaxyInstance
class UserClient(Client):
module = "users"
def __init__(self, galaxy_instance: "GalaxyInstance") -> None:
super().__init__(galaxy_instance)
def get_users(
self,
deleted: bool = False,
f_email: Optional[str] = None,
f_name: Optional[str] = None,
f_any: Optional[str] = None,
) -> List[Dict[str, Any]]:
"""
Get a list of all registered users. If ``deleted`` is set to ``True``,
get a list of deleted users.
:type deleted: bool
:param deleted: Whether to include deleted users
:type f_email: str
:param f_email: filter for user emails. The filter will be active for
non-admin users only if the Galaxy instance has the
``expose_user_email`` option set to ``true`` in the
``config/galaxy.yml`` configuration file.
:type f_name: str
:param f_name: filter for user names. The filter will be active for
non-admin users only if the Galaxy instance has the
``expose_user_name`` option set to ``true`` in the
``config/galaxy.yml`` configuration file.
:type f_any: str
:param f_any: filter for user email or name. Each filter will be active
for non-admin users only if the Galaxy instance has the
corresponding ``expose_user_*`` option set to ``true`` in the
``config/galaxy.yml`` configuration file.
:rtype: list
:return: a list of dicts with user details.
For example::
[{'email': 'a_user@example.org',
'id': 'dda47097d9189f15',
'url': '/api/users/dda47097d9189f15'}]
"""
params: Dict[str, Any] = {}
if f_email:
params["f_email"] = f_email
if f_name:
params["f_name"] = f_name
if f_any:
params["f_any"] = f_any
return self._get(deleted=deleted, params=params)
def show_user(self, user_id: str, deleted: bool = False) -> Dict[str, Any]:
"""
Display information about a user.
:type user_id: str
:param user_id: encoded user ID
:type deleted: bool
:param deleted: whether to return results for a deleted user
:rtype: dict
:return: a dictionary containing information about the user
"""
return self._get(id=user_id, deleted=deleted)
def create_remote_user(self, user_email: str) -> Dict[str, Any]:
"""
Create a new Galaxy remote user.
.. note::
This method works only if the Galaxy instance has the
``allow_user_creation`` and ``use_remote_user`` options set to
``true`` in the ``config/galaxy.yml`` configuration file. Also
note that setting ``use_remote_user`` will require an upstream
authentication proxy server; however, if you do not have one, access
to Galaxy via a browser will not be possible.
:type user_email: str
:param user_email: email of the user to be created
:rtype: dict
:return: a dictionary containing information about the created user
"""
payload = {
"remote_user_email": user_email,
}
return self._post(payload)
def create_local_user(self, username: str, user_email: str, password: str) -> Dict[str, Any]:
"""
Create a new Galaxy local user.
.. note::
This method works only if the Galaxy instance has the
``allow_user_creation`` option set to ``true`` and
``use_remote_user`` option set to ``false`` in the
``config/galaxy.yml`` configuration file.
:type username: str
:param username: username of the user to be created
:type user_email: str
:param user_email: email of the user to be created
:type password: str
:param password: password of the user to be created
:rtype: dict
:return: a dictionary containing information about the created user
"""
payload = {
"username": username,
"email": user_email,
"password": password,
}
return self._post(payload)
def get_current_user(self) -> Dict[str, Any]:
"""
Display information about the user associated with this Galaxy
connection.
:rtype: dict
:return: a dictionary containing information about the current user
"""
url = self._make_url() + "/current"
return self._get(url=url)
def create_user_apikey(self, user_id: str) -> str:
"""
Create a new API key for a given user.
:type user_id: str
:param user_id: encoded user ID
:rtype: str
:return: the API key for the user
"""
url = self._make_url(user_id) + "/api_key"
payload = {
"user_id": user_id,
}
return self._post(payload, url=url)
def delete_user(self, user_id: str, purge: bool = False) -> Dict[str, Any]:
"""
Delete a user.
.. note::
This method works only if the Galaxy instance has the
``allow_user_deletion`` option set to ``true`` in the
``config/galaxy.yml`` configuration file.
:type user_id: str
:param user_id: encoded user ID
:type purge: bool
:param purge: if ``True``, also purge (permanently delete) the history
:rtype: dict
:return: a dictionary containing information about the deleted user
"""
params = {}
if purge is True:
params["purge"] = purge
return self._delete(id=user_id, params=params)
def get_user_apikey(self, user_id: str) -> str:
"""
Get the current API key for a given user.
:type user_id: str
:param user_id: encoded user ID
:rtype: str
:return: the API key for the user, or 'Not available.' if it doesn't
exist yet.
"""
try:
url = self._make_url(user_id) + "/api_key/detailed"
return self._get(url=url)["key"]
except ConnectionError as e:
if e.status_code == 204:
return "Not available."
elif e.status_code != 404:
raise
# Galaxy 22.05 or earlier
url = self._make_url(user_id) + "/api_key/inputs"
return self._get(url=url)["inputs"][0]["value"]
def get_or_create_user_apikey(self, user_id: str) -> str:
"""
Get the current API key for a given user, creating one if it doesn't
exist yet.
:type user_id: str
:param user_id: encoded user ID
:rtype: str
:return: the API key for the user
.. note::
This method works only on Galaxy 21.01 or later.
"""
url = self._make_url(user_id) + "/api_key"
return self._get(url=url)
def update_user(self, user_id: str, user_data: Optional[Dict] = None, **kwargs: Any) -> Dict[str, Any]:
"""
Update user information. You can either pass the attributes you want to
change in the user_data dictionary, or provide them separately as
keyword arguments.
For attributes that cannot be expressed as keywords (e.g.
extra_user_preferences use a `|` sign), pass them in user_data.
:type user_id: str
:param user_id: encoded user ID
:type user_data: dict
:param user_data: a dict containing the values to be updated, eg. { "username" : "newUsername", "email": "new@email" }
:type username: str
:param username: Replace user name with the given string
:type email: str
:param email: Replace user email with the given string
:rtype: dict
:return: details of the updated user
"""
if user_data is None:
user_data = {}
user_data.update(kwargs)
url = self._make_url(user_id) + "/information/inputs"
return self._put(url=url, payload=user_data, id=user_id)
|
