Newer
Older
# put the properties in the self.p namespace (without checking them)
for key, value in data['properties'].items():
self.p.__dict__[key.lower()] = value
Swen Vermeul
committed
def __dir__(self):
'props', 'get_parents()', 'get_children()',
'add_parents()', 'add_children()', 'del_parents()', 'del_children()',
'space', 'project', 'experiment', 'tags',
'set_tags()', 'add_tags()', 'del_tags()',
Swen Vermeul
committed
'add_attachment()', 'get_attachments()', 'download_attachments()',
'save()', 'delete()'
Swen Vermeul
committed
@property
def props(self):
return self.__dict__['p']
Swen Vermeul
committed
@property
def type(self):
@type.setter
def type(self, type_name):
sample_type = self.openbis.get_sample_type(type_name)
self.p.__dict__['_type'] = sample_type
self.a.__dict__['_type'] = sample_type
def __getattr__(self, name):
return getattr(self.__dict__['a'], name)
def __setattr__(self, name, value):
if name in ['set_properties', 'set_tags', 'add_tags']:
raise ValueError("These are methods which should not be overwritten")
setattr(self.__dict__['a'], name, value)
def _repr_html_(self):
return self.a._repr_html_()
def __repr__(self):
return self.a.__repr__()
def set_properties(self, properties):
self.openbis.update_sample(self.permId, properties=properties)
def save(self):
props = self.p._all_props()
Swen Vermeul
committed
if self.is_new:
request["params"][1][0]["properties"] = props
resp = self.openbis._post_request(self.openbis.as_v3, request)
if VERBOSE: print("Sample successfully created.")
new_sample_data = self.openbis.get_sample(resp[0]['permId'], only_data=True)
self._set_data(new_sample_data)
return self
else:
request["params"][1][0]["properties"] = props
self.openbis._post_request(self.openbis.as_v3, request)
if VERBOSE: print("Sample successfully updated.")
Swen Vermeul
committed
new_sample_data = self.openbis.get_sample(self.permId, only_data=True)
self._set_data(new_sample_data)
self.openbis.delete_entity(entity='Sample',id=self.permId, reason=reason)
if VERBOSE: print("Sample {} has been sucessfully deleted.".format(self.permId))
def get_datasets(self, **kwargs):
return self.openbis.get_datasets(sample=self.permId, **kwargs)
def get_projects(self, **kwargs):
return self.openbis.get_project(withSamples=[self.permId], **kwargs)
Swen Vermeul
committed
def get_experiment(self):
Swen Vermeul
committed
return self.openbis.get_experiment(self._experiment['identifier'])
except Exception:
pass
@property
def experiment(self):
return self.openbis.get_experiment(self._experiment['identifier'])
except Exception:
pass
Swen Vermeul
committed
4093
4094
4095
4096
4097
4098
4099
4100
4101
4102
4103
4104
4105
4106
4107
4108
4109
4110
4111
4112
4113
4114
4115
4116
4117
4118
4119
4120
4121
4122
4123
4124
4125
4126
4127
4128
class RoleAssignment(OpenBisObject):
""" managing openBIS role assignments
"""
def __init__(self, openbis_obj, data=None, **kwargs):
self.__dict__['openbis'] = openbis_obj
self.__dict__['a'] = AttrHolder(openbis_obj, 'RoleAssignment' )
if data is not None:
self.a(data)
self.__dict__['data'] = data
if kwargs is not None:
for key in kwargs:
setattr(self, key, kwargs[key])
def __dir__(self):
"""all the available methods and attributes that should be displayed
when using the autocompletion feature (TAB) in Jupyter
"""
return [
'id', 'role', 'roleLevel', 'space', 'project','group'
]
def __str__(self):
return "{}".format(self.get('role'))
def delete(self, reason):
self.openbis.delete_entity(
entity='RoleAssignment', id=self.id,
reason=reason, id_name='techId'
)
if VERBOSE: print("RoleAssignment has been sucessfully deleted.".format(self.permId))
4129
4130
4131
4132
4133
4134
4135
4136
4137
4138
4139
4140
4141
4142
4143
4144
4145
4146
4147
4148
4149
4150
4151
class Person(OpenBisObject):
""" managing openBIS persons
"""
def __init__(self, openbis_obj, data=None, **kwargs):
self.__dict__['openbis'] = openbis_obj
self.__dict__['a'] = AttrHolder(openbis_obj, 'Person' )
if data is not None:
self.a(data)
self.__dict__['data'] = data
if kwargs is not None:
for key in kwargs:
setattr(self, key, kwargs[key])
def __dir__(self):
"""all the available methods and attributes that should be displayed
when using the autocompletion feature (TAB) in Jupyter
"""
return [
'permId', 'userId', 'firstName', 'lastName', 'email',
Swen Vermeul
committed
'registrator', 'registrationDate','space',
'get_roles()', 'assign_role()', 'revoke_role(techId)',
Swen Vermeul
committed
def get_roles(self, **search_args):
""" Get all roles that are assigned to this person.
Provide additional search arguments to refine your search.
Usage::
person.get_roles()
person.get_roles(space='TEST_SPACE')
"""
return self.openbis.get_role_assignments(person=self, **search_args)
def assign_role(self, role, **kwargs):
self.openbis.assign_role(role=role, person=self, **kwargs)
if VERBOSE:
print(
"Role {} successfully assigned to person {}".format(role, self.userId)
)
Swen Vermeul
committed
def revoke_role(self, role, reason='no specific reason'):
""" Revoke a role from this person.
"""
if isinstance(role, int):
ra = self.openbis.get_role_assignment(role)
ra.delete(reason)
if VERBOSE:
print(
"Role {} successfully revoked from person {}".format(role, self.code)
)
return
else:
raise ValueError("Please provide the techId of the role assignment you want to revoke")
def __str__(self):
return "{} {}".format(self.get('firstName'), self.get('lastName'))
def delete(self, reason):
raise ValueError("Persons cannot be deleted")
def save(self):
if self.is_new:
request = self._new_attrs()
# for new and updated objects, the parameter is
# unfortunately called homeSpaceId, spaceId throws no error
# but makes no change either
if "spaceId" in request['params'][1][0]:
request['params'][1][0]['homeSpaceId'] = request['params'][1][0]['spaceId']
del(request['params'][1][0]['spaceId'])
resp = self.openbis._post_request(self.openbis.as_v3, request)
if VERBOSE: print("Person successfully created.")
new_person_data = self.openbis.get_person(resp[0]['permId'], only_data=True)
self._set_data(new_person_data)
return self
else:
request = self._up_attrs()
# for new and updated objects, the parameter is
# unfortunately called homeSpaceId, spaceId throws no error
# but makes no change either
if "spaceId" in request['params'][1][0]:
request['params'][1][0]['homeSpaceId'] = request['params'][1][0]['spaceId']
del(request['params'][1][0]['spaceId'])
return json.dumps(request)
self.openbis._post_request(self.openbis.as_v3, request)
if VERBOSE: print("Person successfully updated.")
4222
4223
4224
4225
4226
4227
4228
4229
4230
4231
4232
4233
4234
4235
4236
4237
4238
4239
4240
4241
4242
4243
new_person_data = self.openbis.get_person(self.permId, only_data=True)
self._set_data(new_person_data)
class Group(OpenBisObject):
""" Managing openBIS authorization groups
"""
def __init__(self, openbis_obj, data=None, **kwargs):
self.__dict__['openbis'] = openbis_obj
self.__dict__['a'] = AttrHolder(openbis_obj, 'AuthorizationGroup')
if data is not None:
self.a(data)
self.__dict__['data'] = data
if kwargs is not None:
for key in kwargs:
setattr(self, key, kwargs[key])
def __dir__(self):
return [
'code','description','users','roleAssignments',
'get_users()', 'set_users()', 'add_users()', 'del_users()',
Swen Vermeul
committed
'get_roles()', 'assign_role()', 'revoke_role(techId)'
def get_persons(self):
Swen Vermeul
committed
""" Returns a Things object wich contains all Persons (Users)
that belong to this group.
"""
persons = DataFrame(self._users)
persons['permId'] = persons['permId'].map(extract_permid)
persons['registrationDate'] = persons['registrationDate'].map(format_timestamp)
persons['space'] = persons['space'].map(extract_nested_permid)
p = Things(
self.openbis, entity='person',
df=persons[['permId', 'userId', 'firstName', 'lastName', 'email', 'space', 'registrationDate', 'active']],
identifier_name='permId'
)
return p
get_users = get_persons # Alias
def get_roles(self, **search_args):
""" Get all roles that are assigned to this group.
Provide additional search arguments to refine your search.
Usage::
group.get_roles()
group.get_roles(space='TEST_SPACE')
"""
Swen Vermeul
committed
return self.openbis.get_role_assignments(group=self, **search_args)
def assign_role(self, role, **kwargs):
""" Assign a role to this group. If no additional attribute is provided,
roleLevel will default to INSTANCE. If a space attribute is provided,
the roleLevel will be SPACE. If a project attribute is provided,
roleLevel will be PROJECT.
Usage::
group.assign_role(role='ADMIN')
group.assign_role(role='ADMIN', space='TEST_SPACE')
"""
self.openbis.assign_role(role=role, group=self, **kwargs)
if VERBOSE:
print(
"Role {} successfully assigned to group {}".format(role, self.code)
)
Swen Vermeul
committed
def revoke_role(self, role, reason='no specific reason'):
""" Revoke a role from this group.
"""
Swen Vermeul
committed
if isinstance(role, int):
ra = self.openbis.get_role_assignment(role)
ra.delete(reason)
if VERBOSE:
print(
"Role {} successfully revoked from group {}".format(role, self.code)
)
return
else:
raise ValueError("Please provide the techId of the role assignment you want to revoke")
""" creates a nice table in Jupyter notebooks when the object itself displayed
"""
4313
4314
4315
4316
4317
4318
4319
4320
4321
4322
4323
4324
4325
4326
4327
4328
4329
4330
4331
4332
4333
4334
4335
4336
4337
4338
4339
4340
def nvl(val, string=''):
if val is None:
return string
return val
html = """
<table border="1" class="dataframe">
<thead>
<tr style="text-align: right;">
<th>attribute</th>
<th>value</th>
</tr>
</thead>
<tbody>
"""
for attr in self._allowed_attrs:
if attr in ['users','roleAssignments']:
continue
html += "<tr> <td>{}</td> <td>{}</td> </tr>".format(
attr, nvl(getattr(self, attr, ''), '')
)
html += """
</tbody>
</table>
"""
4341
4342
4343
4344
4345
4346
4347
4348
4349
4350
4351
4352
4353
4354
4355
4356
4357
4358
4359
4360
4361
4362
4363
4364
4365
4366
4367
4368
4369
4370
if getattr(self, '_users') is not None:
html += """
<br/>
<b>Users</b>
<table border="1" class="dataframe">
<thead>
<tr style="text-align: right;">
<th>userId</th>
<th>FirstName</th>
<th>LastName</th>
<th>Email</th>
<th>Space</th>
<th>active</th>
</tr>
</thead>
<tbody>
"""
for user in self._users:
html += "<tr><td>{}</td><td>{}</td><td>{}</td><td>{}</td><td>{}</td><td>{}</td></tr>".format(
user.get('userId'),
user.get('firstName'),
user.get('lastName'),
user.get('email'),
user.get('space').get('code') if user.get('space') is not None else '',
user.get('active'),
)
html += """
</tbody>
</table>
"""
def save(self):
if self.is_new:
request = self._new_attrs()
resp = self.openbis._post_request(self.openbis.as_v3, request)
if VERBOSE: print("Group successfully created.")
# re-fetch group from openBIS
new_data = self.openbis.get_person(resp[0]['permId'], only_data=True)
self._set_data(new_data)
return self
else:
request = self._up_attrs()
self.openbis._post_request(self.openbis.as_v3, request)
if VERBOSE: print("Group successfully updated.")
# re-fetch group from openBIS
new_data = self.openbis.get_group(self.permId, only_data=True)
self._set_data(new_data)
""" managing openBIS spaces
"""
def __init__(self, openbis_obj, data=None, **kwargs):
self.__dict__['openbis'] = openbis_obj
self.__dict__['a'] = AttrHolder(openbis_obj, 'Space' )
if data is not None:
self.a(data)
self.__dict__['data'] = data
if kwargs is not None:
for key in kwargs:
setattr(self, key, kwargs[key])
def __dir__(self):
"""all the available methods and attributes that should be displayed
when using the autocompletion feature (TAB) in Jupyter
"""
return [
'code', 'permId', 'description', 'registrator', 'registrationDate', 'modificationDate',
'get_projects()',
"new_project(code='', description='', attachments)",
'get_samples()',
'delete()'
]
def __str__(self):
return self.data.get('code', None)
def get_samples(self, **kwargs):
return self.openbis.get_samples(space=self.code, **kwargs)
get_objects = get_samples #Alias
def get_sample(self, sample_code):
if is_identifier(sample_code) or is_permid(sample_code):
return self.openbis.get_sample(sample_code)
else:
# we assume we just got the code
return self.openbis.get_sample('/{}/{}'.format(self.code,sample_code) )
def get_projects(self, **kwargs):
return self.openbis.get_projects(space=self.code, **kwargs)
def new_project(self, code, description=None, **kwargs):
return self.openbis.new_project(self.code, code, description, **kwargs)
def new_sample(self, **kwargs):
return self.openbis.new_sample(space=self, **kwargs)
def delete(self, reason):
self.openbis.delete_entity('Space', self.permId, reason)
if VERBOSE: print("Space {} has been sucessfully deleted.".format(self.permId))
def save(self):
if self.is_new:
request = self._new_attrs()
resp = self.openbis._post_request(self.openbis.as_v3, request)
if VERBOSE: print("Space successfully created.")
new_space_data = self.openbis.get_space(resp[0]['permId'], only_data=True)
self._set_data(new_space_data)
return self
else:
request = self._up_attrs()
self.openbis._post_request(self.openbis.as_v3, request)
if VERBOSE: print("Space successfully updated.")
new_space_data = self.openbis.get_space(self.permId, only_data=True)
self._set_data(new_space_data)
Chandrasekhar Ramakrishnan
committed
4467
4468
4469
4470
4471
4472
4473
4474
4475
4476
4477
4478
4479
4480
4481
4482
4483
4484
4485
4486
4487
4488
4489
4490
4491
4492
4493
class ExternalDMS():
""" managing openBIS external data management systems
"""
def __init__(self, openbis_obj, data=None, **kwargs):
self.__dict__['openbis'] = openbis_obj
if data is not None:
self.__dict__['data'] = data
if kwargs is not None:
for key in kwargs:
setattr(self, key, kwargs[key])
def __getattr__(self, name):
return self.__dict__['data'].get(name)
def __dir__(self):
"""all the available methods and attributes that should be displayed
when using the autocompletion feature (TAB) in Jupyter
"""
return ['code', 'label', 'urlTemplate', 'address', 'addressType', 'openbis']
def __str__(self):
return self.data.get('code', None)
class Things():
"""An object that contains a DataFrame object about an entity available in openBIS.
"""
def __init__(self, openbis_obj, entity, df, identifier_name='code'):
self.df = df
self.identifier_name = identifier_name
return tabulate(self.df, headers=list(self.df))
def __len__(self):
return len(self.df)
def _repr_html_(self):
return self.df._repr_html_()
if self.entity not in ['sample', 'dataset']:
Swen Vermeul
committed
raise ValueError("{}s do not have parents".format(self.entity))
if self.df is not None and len(self.df) > 0:
dfs = []
for ident in self.df[self.identifier_name]:
# get all objects that have this object as a child == parent
try:
parents = getattr(self.openbis, 'get_' + self.entity.lower() + 's')(withChildren=ident, **kwargs)
Swen Vermeul
committed
dfs.append(parents.df)
except ValueError:
pass
if len(dfs) > 0:
return Things(self.openbis, self.entity, pd.concat(dfs), self.identifier_name)
else:
return Things(self.openbis, self.entity, DataFrame(), self.identifier_name)
if self.entity not in ['sample', 'dataset']:
Swen Vermeul
committed
raise ValueError("{}s do not have children".format(self.entity))
if self.df is not None and len(self.df) > 0:
dfs = []
for ident in self.df[self.identifier_name]:
# get all objects that have this object as a child == parent
try:
parents = getattr(self.openbis, 'get_' + self.entity.lower() + 's')(withParent=ident, **kwargs)
Swen Vermeul
committed
dfs.append(parents.df)
except ValueError:
pass
if len(dfs) > 0:
return Things(self.openbis, self.entity, pd.concat(dfs), self.identifier_name)
else:
return Things(self.openbis, self.entity, DataFrame(), self.identifier_name)
Swen Vermeul
committed
if self.entity not in ['space', 'project', 'experiment']:
raise ValueError("{}s do not have samples".format(self.entity))
if self.df is not None and len(self.df) > 0:
dfs = []
for ident in self.df[self.identifier_name]:
args = {}
args[self.entity.lower()] = ident
try:
samples = self.openbis.get_samples(**args, **kwargs)
Swen Vermeul
committed
dfs.append(samples.df)
except ValueError:
pass
if len(dfs) > 0:
return Things(self.openbis, 'sample', pd.concat(dfs), 'identifier')
else:
return Things(self.openbis, 'sample', DataFrame(), 'identifier')
get_objects = get_samples # Alias
if self.entity not in ['sample', 'experiment']:
Swen Vermeul
committed
raise ValueError("{}s do not have datasets".format(self.entity))
if self.df is not None and len(self.df) > 0:
dfs = []
for ident in self.df[self.identifier_name]:
args = {}
args[self.entity.lower()] = ident
try:
datasets = self.openbis.get_datasets(**args, **kwargs)
Swen Vermeul
committed
dfs.append(datasets.df)
except ValueError:
pass
if len(dfs) > 0:
return Things(self.openbis, 'dataset', pd.concat(dfs), 'permId')
else:
return Things(self.openbis, 'dataset', DataFrame(), 'permId')
def __getitem__(self, key):
if self.df is not None and len(self.df) > 0:
row = None
if isinstance(key, int):
# get thing by rowid
row = self.df.loc[[key]]
elif isinstance(key, list):
# treat it as a normal dataframe
return self.df[key]
row = self.df[self.df[self.identifier_name] == key.upper()]
# invoke the openbis.get_<entity>() method
return getattr(self.openbis, 'get_' + self.entity)(row[self.identifier_name].values[0])
Swen Vermeul
committed
def __iter__(self):
for item in self.df[[self.identifier_name]][self.identifier_name].iteritems():
yield getattr(self.openbis, 'get_' + self.entity)(item[1])
Swen Vermeul
committed
# return self.df[[self.identifier_name]].to_dict()[self.identifier_name]
Swen Vermeul
committed
Swen Vermeul
committed
"""
def __init__(self, openbis_obj, type, project=None, data=None, props=None, code=None, **kwargs):
Swen Vermeul
committed
self.__dict__['openbis'] = openbis_obj
self.__dict__['type'] = type
self.__dict__['p'] = PropertyHolder(openbis_obj, type)
self.__dict__['a'] = AttrHolder(openbis_obj, 'Experiment', type)
Swen Vermeul
committed
if data is not None:
self._set_data(data)
if project is not None:
setattr(self, 'project', project)
Swen Vermeul
committed
if props is not None:
for key in props:
setattr(self.p, key, props[key])
if code is not None:
self.code = code
Swen Vermeul
committed
if kwargs is not None:
Swen Vermeul
committed
defs = _definitions('Experiment')
Swen Vermeul
committed
for key in kwargs:
Swen Vermeul
committed
if key in defs['attrs']:
setattr(self, key, kwargs[key])
else:
raise ValueError("{attr} is not a known attribute for an Experiment".format(attr=key))
Swen Vermeul
committed
def _set_data(self, data):
# assign the attribute data to self.a by calling it
# (invoking the AttrHolder.__call__ function)
self.a(data)
self.__dict__['data'] = data
Swen Vermeul
committed
# put the properties in the self.p namespace (without checking them)
for key, value in data['properties'].items():
self.p.__dict__[key.lower()] = value
Swen Vermeul
committed
def __str__(self):
return self.data['code']
Swen Vermeul
committed
def __dir__(self):
# the list of possible methods/attributes displayed
# when invoking TAB-completition
return [
'code', 'permId', 'identifier',
'type', 'project',
'props.',
'project', 'tags', 'attachments', 'data',
'get_datasets()', 'get_samples()',
'set_tags()', 'add_tags()', 'del_tags()',
Swen Vermeul
committed
'add_attachment()', 'get_attachments()', 'download_attachments()',
'save()'
Swen Vermeul
committed
]
@property
def props(self):
return self.__dict__['p']
@property
def type(self):
Swen Vermeul
committed
@type.setter
def type(self, type_name):
experiment_type = self.openbis.get_experiment_type(type_name)
self.p.__dict__['_type'] = experiment_type
self.a.__dict__['_type'] = experiment_type
Swen Vermeul
committed
def __getattr__(self, name):
return getattr(self.__dict__['a'], name)
def __setattr__(self, name, value):
if name in ['set_properties', 'add_tags()', 'del_tags()', 'set_tags()']:
Swen Vermeul
committed
raise ValueError("These are methods which should not be overwritten")
setattr(self.__dict__['a'], name, value)
def _repr_html_(self):
html = self.a._repr_html_()
return html
def set_properties(self, properties):
self.openbis.update_experiment(self.permId, properties=properties)
Swen Vermeul
committed
def save(self):
if self.is_new:
request = self._new_attrs()
props = self.p._all_props()
request["params"][1][0]["properties"] = props
resp = self.openbis._post_request(self.openbis.as_v3, request)
if VERBOSE: print("Experiment successfully created.")
new_exp_data = self.openbis.get_experiment(resp[0]['permId'], only_data=True)
self._set_data(new_exp_data)
return self
Swen Vermeul
committed
else:
request = self._up_attrs()
props = self.p._all_props()
request["params"][1][0]["properties"] = props
self.openbis._post_request(self.openbis.as_v3, request)
if VERBOSE: print("Experiment successfully updated.")
new_exp_data = self.openbis.get_experiment(resp[0]['permId'], only_data=True)
self._set_data(new_exp_data)
def delete(self, reason):
if self.permId is None:
return None
self.openbis.delete_entity(entity='Experiment', id=self.permId, reason=reason)
if VERBOSE: print("Experiment {} has been sucessfully deleted.".format(self.permId))
if self.permId is None:
return None
return self.openbis.get_datasets(experiment=self.permId, **kwargs)
if self.permId is None:
return None
return self.openbis.get_project(experiment=self.permId, **kwargs)
Swen Vermeul
committed
if self.permId is None:
return None
return self.openbis.get_samples(experiment=self.permId, **kwargs)
Swen Vermeul
committed
get_objects = get_samples # Alias
Swen Vermeul
committed
4750
4751
4752
4753
4754
4755
4756
4757
4758
4759
4760
4761
4762
4763
4764
4765
4766
4767
4768
4769
4770
4771
4772
4773
4774
def add_samples(self, *samples):
for sample in samples:
if isinstance(sample, str):
obj = self.openbis.get_sample(sample)
else:
# we assume we got a sample object
obj = sample
# a sample can only belong to exactly one experiment
if obj.experiment is not None:
raise ValueError(
"sample {} already belongs to experiment {}".format(
obj.code, obj.experiment
)
)
else:
if self.is_new:
raise ValueError("You need to save this experiment first before you can assign any samples to it")
else:
# update the sample directly
obj.experiment = self.identifier
obj.save()
self.a.__dict__['_samples'].append(obj._identifier)
add_objects = add_samples # Alias
Swen Vermeul
committed
def del_samples(self, samples):
if not isinstance(samples, list):
samples = [samples]
for sample in samples:
if isinstance(sample, str):
obj = self.openbis.get_sample(sample)
objects.append(obj)
else:
# we assume we got an object
objects.append(obj)
self.samples = objects
del_objects = del_samples # Alias
class Attachment():
def __init__(self, filename, title=None, description=None):
if not os.path.exists(filename):
raise ValueError("File not found: {}".format(filename))
self.fileName = filename
self.title = title
self.description = description
Swen Vermeul
committed
"fileName": self.fileName,
"title": self.title,
"description": self.description,
Swen Vermeul
committed
def get_data(self):
with open(self.fileName, 'rb') as att:
content = att.read()
contentb64 = base64.b64encode(content).decode()
return {
"fileName": self.fileName,
"title": self.title,
"description": self.description,
"content": contentb64,
"@type": "as.dto.attachment.create.AttachmentCreation",
Swen Vermeul
committed
def __init__(self, openbis_obj, data=None, **kwargs):
self.__dict__['openbis'] = openbis_obj
self.__dict__['a'] = AttrHolder(openbis_obj, 'Project')
if data is not None:
self.a(data)
self.__dict__['data'] = data
if kwargs is not None:
for key in kwargs:
setattr(self, key, kwargs[key])
def __dir__(self):
"""all the available methods and attributes that should be displayed
when using the autocompletion feature (TAB) in Jupyter
"""
return ['code', 'permId', 'identifier', 'description', 'space', 'registrator',
'registrationDate', 'modifier', 'modificationDate', 'add_attachment()',
'get_attachments()', 'download_attachments()',
'get_experiments()', 'get_samples()', 'get_datasets()',
Swen Vermeul
committed
'save()', 'delete()'
def get_samples(self, **kwargs):
return self.openbis.get_samples(project=self.permId, **kwargs)
def get_sample(self, sample_code):
if is_identifier(sample_code) or is_permid(sample_code):
return self.openbis.get_sample(sample_code)
else:
# we assume we just got the code
return self.openbis.get_sample(project=self, code=sample_code)
get_objects = get_samples # Alias
def get_experiments(self):
return self.openbis.get_experiments(project=self.permId)
def get_datasets(self):
return self.openbis.get_datasets(project=self.permId)
def delete(self, reason):
self.openbis.delete_entity(entity='Project', id=self.permId, reason=reason)
if VERBOSE: print("Project {} has been sucessfully deleted.".format(self.permId))
if self.is_new:
request = self._new_attrs()
resp = self.openbis._post_request(self.openbis.as_v3, request)
self.a.__dict__['_is_new'] = False
if VERBOSE: print("Project successfully created.")
new_project_data = self.openbis.get_project(resp[0]['permId'], only_data=True)
self._set_data(new_project_data)
return self
request = self._up_attrs()
self.openbis._post_request(self.openbis.as_v3, request)
if VERBOSE: print("Project successfully updated.")
class SemanticAnnotation():
def __init__(self, openbis_obj, isNew=True, **kwargs):
self._openbis = openbis_obj
self._isNew = isNew;
self.permId = kwargs.get('permId')
self.entityType = kwargs.get('entityType')
self.propertyType = kwargs.get('propertyType')
self.predicateOntologyId = kwargs.get('predicateOntologyId')
self.predicateOntologyVersion = kwargs.get('predicateOntologyVersion')
self.predicateAccessionId = kwargs.get('predicateAccessionId')
self.descriptorOntologyId = kwargs.get('descriptorOntologyId')
self.descriptorOntologyVersion = kwargs.get('descriptorOntologyVersion')
self.descriptorAccessionId = kwargs.get('descriptorAccessionId')
self.creationDate = kwargs.get('creationDate')
def __dir__(self):
return [
'permId', 'entityType', 'propertyType',
'predicateOntologyId', 'predicateOntologyVersion',
'predicateAccessionId', 'descriptorOntologyId',
'descriptorOntologyVersion', 'descriptorAccessionId',
'creationDate',
'save()', 'delete()'
]
4911
4912
4913
4914
4915
4916
4917
4918
4919
4920
4921
4922
4923
4924
4925
4926
4927
4928
4929
4930
4931
4932
4933
4934
4935
4936
4937
4938
4939
4940
4941
4942
4943
4944
4945
4946
4947
4948
4949
4950
4951
4952
4953
4954
4955
4956
4957
4958
4959
4960
4961
4962
def save(self):
if self._isNew:
self._create()
else:
self._update()
def _create(self):
creation = {
"@type": "as.dto.semanticannotation.create.SemanticAnnotationCreation"
}
if self.entityType is not None and self.propertyType is not None:
creation["propertyAssignmentId"] = {
"@type": "as.dto.property.id.PropertyAssignmentPermId",
"entityTypeId" : {
"@type": "as.dto.entitytype.id.EntityTypePermId",
"permId" : self.entityType,
"entityKind" : "SAMPLE"
},
"propertyTypeId" : {
"@type" : "as.dto.property.id.PropertyTypePermId",
"permId" : self.propertyType
}
}
elif self.entityType is not None:
creation["entityTypeId"] = {
"@type": "as.dto.entitytype.id.EntityTypePermId",
"permId" : self.entityType,
"entityKind" : "SAMPLE"
}
elif self.propertyType is not None:
creation["propertyTypeId"] = {
"@type" : "as.dto.property.id.PropertyTypePermId",
"permId" : self.propertyType
}
for attr in ['predicateOntologyId', 'predicateOntologyVersion', 'predicateAccessionId', 'descriptorOntologyId', 'descriptorOntologyVersion', 'descriptorAccessionId']:
creation[attr] = getattr(self, attr)
request = {
"method": "createSemanticAnnotations",
"params": [
self._openbis.token,
[creation]
]
}
self._openbis._post_request(self._openbis.as_v3, request)
self._isNew = False
if VERBOSE: print("Semantic annotation successfully created.")
4964
4965
4966
4967
4968
4969
4970
4971
4972
4973
4974
4975
4976
4977
4978
4979
4980
4981
4982
4983
4984
4985
4986
4987
4988
4989
4990
def _update(self):
update = {
"@type": "as.dto.semanticannotation.update.SemanticAnnotationUpdate",
"semanticAnnotationId" : {
"@type" : "as.dto.semanticannotation.id.SemanticAnnotationPermId",
"permId" : self.permId
}
}
for attr in ['predicateOntologyId', 'predicateOntologyVersion', 'predicateAccessionId', 'descriptorOntologyId', 'descriptorOntologyVersion', 'descriptorAccessionId']:
update[attr] = {
"@type" : "as.dto.common.update.FieldUpdateValue",
"isModified" : True,
"value" : getattr(self, attr)
}
request = {
"method": "updateSemanticAnnotations",
"params": [
self._openbis.token,
[update]
]
}
self._openbis._post_request(self._openbis.as_v3, request)
if VERBOSE: print("Semantic annotation successfully updated.")
def delete(self, reason):
self._openbis.delete_entity(entity='SemanticAnnotation', id=self.permId, reason=reason)
if VERBOSE: print("Semantic annotation successfully deleted.")