-
Notifications
You must be signed in to change notification settings - Fork 0
/
molgenis.py
242 lines (209 loc) · 9.96 KB
/
molgenis.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
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
import requests
import json
import urllib
import pycurl
import os
try:
from urllib.parse import quote_plus
except ImportError:
# Python 2
from urllib import quote_plus
class Session():
'''Representation of a session with the MOLGENIS REST API.
Usage:
>>> session = molgenis.Session('http://localhost:8080/api/')
>>> session.login('user', 'password')
>>> session.get('Person')
'''
def __init__(self, url="https://molgenis01.gcc.rug.nl:443/api/"):
'''Constructs a new Session.
Args:
url -- URL of the REST API. Should be of form 'http[s]://<molgenis server>[:port]/api/'
Examples:
>>> connection = molgenis.Session('http://localhost:8080/api/')
'''
self.url = url
self.session = requests.Session()
def login(self, username, password):
'''Logs in a user and stores the acquired session token in this Session object.
Args:
username -- username for a registered molgenis user
password -- password for the user
'''
self.session.cookies.clear()
response = self.session.post(self.url + "v1/login",
data=json.dumps({"username": username, "password": password}),
headers={"Content-Type": "application/json"})
if response.status_code == 200:
self.token = response.json()["token"]
response.raise_for_status();
return response;
def logout(self):
'''Logs out the current session token.'''
response = self.session.post(self.url + "v1/logout",
headers=self._get_token_header())
if response.status_code == 200:
self.token = None
response.raise_for_status();
return response;
def getById(self, entity, id, attributes=None, expand=None):
'''Retrieves a single entity row from an entity repository.
Args:
entity -- fully qualified name of the entity
id -- the value for the idAttribute of the entity
attributes -- The list of attributes to retrieve
expand -- the attributes to expand
Examples:
session.get('Person', 'John')
'''
response = self.session.get(self.url + "v1/" + quote_plus(entity) + '/' + quote_plus(id),
headers=self._get_token_header(),
params={"attributes": attributes, "expand": expand})
if response.status_code == 200:
return response.json()
response.raise_for_status();
return response;
def get_total(self, entity, q=None):
response = self.session.get(self.url + "v2/" + quote_plus(entity),
headers=self._get_token_header())
if response.status_code == 200:
return response.json()["total"]
response.raise_for_status()
return response
def get(self, entity, q=None, attributes=None, expand=None, num=100, start=0, sortColumn=None, sortOrder=None):
'''Retrieves entity rows from an entity repository.
Args:
entity -- fully qualified name of the entity
q -- query in json form, see the MOLGENIS REST API v1 documentation for details
attributes -- The list of attributes to retrieve
expand -- the attributes to expand
num -- the amount of entity rows to retrieve
start -- the index of the first row to retrieve (zero indexed)
sortColumn -- the attribute to sort on
sortOrder -- the order to sort in
Examples:
session.get('Person')
'''
if q:
response = self.session.post(self.url + "v2/" + quote_plus(entity),
headers=self._get_token_header_with_content_type(),
params={"_method": "GET"},
data=json.dumps(
{"attributes": attributes, "expand": expand, "num": num,
"start": start,
"sortColumn": sortColumn, "sortOrder": sortOrder}))
else:
response = self.session.get(self.url + "v2/" + quote_plus(entity),
headers=self._get_token_header_with_content_type(),
params={"q": q, "attributes": attributes, "expand": expand, "num": num, "start": start,
"sortColumn": sortColumn, "sortOrder":
sortOrder})
if response.status_code == 200:
return response.json()["items"]
response.raise_for_status();
return response;
def add(self, entity, data={}, files={}, **kwargs):
'''Adds a single entity row to an entity repository.
Args:
entity -- fully qualified name of the entity
files -- dictionary containing file attribute values for the entity row.
The dictionary should for each file attribute map the attribute name to a tuple containing the file name and an
input stream.
data -- dictionary mapping attribute name to non-file attribute value for the entity row, gets merged with the
kwargs argument
**kwargs -- keyword arguments get merged with the data argument
Examples:
>>> session.add('Person', firstName='Jan', lastName='Klaassen')
>>> session.add('Person', {'firstName': 'Jan', 'lastName':'Klaassen'})
You can have multiple file type attributes.
>>> session.add('Plot', files={'image': ('expression.jpg', open('~/first-plot.jpg','rb')),
'image2': ('expression-large.jpg', open('/Users/me/second-plot.jpg', 'rb'))},
data={'name':'IBD-plot'})
'''
response = self.session.post(self.url + "v1/" + quote_plus(entity),
headers=self._get_token_header(),
data=self._merge_two_dicts(data, kwargs),
files=files)
if response.status_code == 201:
return response.headers["Location"].split("/")[-1]
response.raise_for_status()
return response
def update_one(self, entity, id, attr, value):
'''Updates one attribute of a given entity in a table with a given value'''
response = self.session.put(self.url + "v1/" + quote_plus(entity)+"/"+id+"/"+attr,
headers=self._get_token_header_with_content_type(),
data=json.dumps(value))
response.raise_for_status()
return response
def add_all(self, entity, entities):
'''Adds multiple entity rows to an entity repository.'''
response = self.session.post(self.url + "v2/" + quote_plus(entity),
headers=self._get_token_header_with_content_type(),
data=json.dumps({"entities": entities}))
if response.status_code == 201:
return [resource["href"].split("/")[-1] for resource in response.json()["resources"]]
response.raise_for_status();
print(response.json())
return response;
def delete_list(self, entity, entities):
'''Deletes multiple entity rows to an entity repository, given a list of id's.'''
headers = self._get_token_header_with_content_type()
response = self.session.delete(self.url + "v2/" + quote_plus(entity),
headers=headers,
data=json.dumps({"entityIds": entities}))
response.raise_for_status();
return response;
def delete(self, entity, id):
'''Deletes a single entity row from an entity repository.'''
response = self.session.delete(self.url + "v1/" + quote_plus(entity) + "/" + quote_plus(id), headers=
self._get_token_header())
response.raise_for_status();
return response;
def get_entity_meta_data(self, entity):
'''Retrieves the metadata for an entity repository.'''
response = self.session.get(self.url + "v1/" + quote_plus(entity) + "/meta?expand=attributes", headers=
self._get_token_header())
response.raise_for_status();
return response.json();
def get_attribute_meta_data(self, entity, attribute):
'''Retrieves the metadata for a single attribute of an entity repository.'''
response = self.session.get(self.url + "v1/" + quote_plus(entity) + "/meta/" + quote_plus(attribute), headers=
self._get_token_header())
response.raise_for_status();
return response.json();
def _get_token_header(self):
'''Creates an 'x-molgenis-token' header for the current session.'''
try:
return {"x-molgenis-token": self.token}
except AttributeError:
return {}
def _get_token_header_with_content_type(self):
'''Creates an 'x-molgenis-token' header for the current session and a 'Content-Type: application/json' header'''
headers = self._get_token_header()
headers.update({"Content-Type": "application/json"})
return headers
def get_molgenis_version(self):
response = self.session.get(self.url + "v2/version",
headers=self._get_token_header(),
params={"_method": "GET"},
)
if response.status_code == 200:
return response.json()
response.raise_for_status();
return response;
def upload_with_meta_data(self, meta_data_zip):
header = self._get_token_header()
files = {'file': open(os.path.abspath(meta_data_zip), 'rb')}
url = self.url.strip('/api/')+'/plugin/importwizard/importFile'
response = requests.post(url, headers=header, files=files)
print(response)
if response.status_code == 201:
return response.json();
response.raise_for_status();
return response;
@staticmethod
def _merge_two_dicts(x, y):
'''Given two dicts, merge them into a new dict as a shallow copy.'''
z = x.copy()
z.update(y)
return z