diff --git a/pybis/src/python/pybis/attribute.py b/pybis/src/python/pybis/attribute.py
index fe796bf76cacf2b77eaeadc2cff3c1e47e0f3d20..e034dffe105c5b2b2be1f359426b636a5c982492 100644
--- a/pybis/src/python/pybis/attribute.py
+++ b/pybis/src/python/pybis/attribute.py
@@ -1,15 +1,30 @@
from pandas import DataFrame, Series
from tabulate import tabulate
-from .definitions import openbis_definitions, fetch_option, get_method_for_entity, get_type_for_entity
-from .utils import parse_jackson, check_datatype, split_identifier, format_timestamp, is_identifier, is_permid, nvl, extract_person
+from .definitions import (
+ openbis_definitions,
+ fetch_option,
+ get_method_for_entity,
+ get_type_for_entity,
+)
+from .utils import (
+ parse_jackson,
+ check_datatype,
+ split_identifier,
+ format_timestamp,
+ is_identifier,
+ is_permid,
+ nvl,
+ extract_person,
+)
from .attachment import Attachment
import copy
import base64
import os
-class AttrHolder():
- """ General class for both samples and experiments that hold all common attributes, such as:
+
+class AttrHolder:
+ """General class for both samples and experiments that hold all common attributes, such as:
- space
- project
- experiment (sample)
@@ -19,19 +34,19 @@ class AttrHolder():
"""
def __init__(self, openbis_obj, entity, type=None):
- self.__dict__['_openbis'] = openbis_obj
- self.__dict__['_entity'] = entity
+ self.__dict__["_openbis"] = openbis_obj
+ self.__dict__["_entity"] = entity
if type is not None:
- self.__dict__['_type'] = type.data
+ self.__dict__["_type"] = type.data
- self.__dict__['_defs'] = openbis_definitions(entity)
- #self.__dict__['_allowed_attrs'] = openbis_definitions(entity)['attrs']
- #self.__dict__['_allowed_attrs_new'] = openbis_definitions(entity)['attrs_new']
- #self.__dict__['_allowed_attrs_up'] = openbis_definitions(entity)['attrs_up']
- self.__dict__['_identifier'] = None
- self.__dict__['_is_new'] = True
- self.__dict__['_tags'] = []
+ self.__dict__["_defs"] = openbis_definitions(entity)
+ # self.__dict__['_allowed_attrs'] = openbis_definitions(entity)['attrs']
+ # self.__dict__['_allowed_attrs_new'] = openbis_definitions(entity)['attrs_new']
+ # self.__dict__['_allowed_attrs_up'] = openbis_definitions(entity)['attrs_up']
+ self.__dict__["_identifier"] = None
+ self.__dict__["_is_new"] = True
+ self.__dict__["_tags"] = []
def __call__(self, data):
"""This internal method is invoked when an existing object is loaded.
@@ -41,54 +56,54 @@ class AttrHolder():
Since the data comes from openBIS, we do not have to check it (hence the
self.__dict__ statements to prevent invoking the __setattr__ method)
Internally data is stored with an underscore, e.g.
- sample._space = {
+ sample._space = {
'@type': 'as.dto.space.id.SpacePermId',
- 'permId': 'MATERIALS'
+ 'permId': 'MATERIALS'
}
but when fetching the attribute without the underscore, we only return
the relevant data for the user:
sample.space # MATERIALS
"""
# entity is read from openBIS, so it is not new anymore
- self.__dict__['_is_new'] = False
+ self.__dict__["_is_new"] = False
- for attr in self._defs['attrs']:
+ for attr in self._defs["attrs"]:
if attr in ["code", "permId", "identifier", "type"]:
- self.__dict__['_' + attr] = data.get(attr, None)
+ self.__dict__["_" + attr] = data.get(attr, None)
# remove the @id attribute
- if isinstance(self.__dict__['_' + attr], dict):
- self.__dict__['_' + attr].pop('@id', None)
+ if isinstance(self.__dict__["_" + attr], dict):
+ self.__dict__["_" + attr].pop("@id", None)
- elif attr in ['vocabularyCode']:
- self.__dict__['_'+attr] = data.get('permId', {}).get(attr, None)
+ elif attr in ["vocabularyCode"]:
+ self.__dict__["_" + attr] = data.get("permId", {}).get(attr, None)
- elif attr in ['validationPlugin']:
+ elif attr in ["validationPlugin"]:
d = data.get(attr, None)
if d is not None:
- d = d['permId']
- self.__dict__['_' + attr] = d
+ d = d["permId"]
+ self.__dict__["_" + attr] = d
elif attr in ["space"]:
d = data.get(attr, None)
if d is not None:
- d = d['permId']
- self.__dict__['_' + attr] = d
+ d = d["permId"]
+ self.__dict__["_" + attr] = d
elif attr in ["sample", "experiment", "project", "container"]:
d = data.get(attr, None)
if d is not None:
- d = d['identifier']
- self.__dict__['_' + attr] = d
+ d = d["identifier"]
+ self.__dict__["_" + attr] = d
elif attr in ["parents", "children", "samples", "components", "containers"]:
- self.__dict__['_' + attr] = []
+ self.__dict__["_" + attr] = []
if data[attr] is not None:
for item in data[attr]:
try:
- if 'identifier' in item:
- self.__dict__['_' + attr].append(item['identifier'])
- elif 'permId' in item:
- self.__dict__['_' + attr].append(item['permId'])
+ if "identifier" in item:
+ self.__dict__["_" + attr].append(item["identifier"])
+ elif "permId" in item:
+ self.__dict__["_" + attr].append(item["permId"])
except Exception:
# TODO: under certain circumstances, openBIS only delivers an integer
pass
@@ -96,55 +111,53 @@ class AttrHolder():
elif attr in ["tags"]:
self.add_tags(data[attr])
- elif attr.endswith('Date'):
- self.__dict__['_'+attr] = format_timestamp(data.get(attr))
+ elif attr.endswith("Date"):
+ self.__dict__["_" + attr] = format_timestamp(data.get(attr))
- elif attr in ['registrator', 'modifier', 'dataProducer']:
- self.__dict__['_'+attr] = extract_person(data.get(attr))
+ elif attr in ["registrator", "modifier", "dataProducer"]:
+ self.__dict__["_" + attr] = extract_person(data.get(attr))
else:
- self.__dict__['_' + attr] = data.get(attr, None)
-
+ self.__dict__["_" + attr] = data.get(attr, None)
def _new_attrs(self, method_name=None):
"""Returns the Python-equivalent JSON request when a new object is created.
It is used internally by the save() method of a newly created object.
"""
- attr2ids = openbis_definitions('attr2ids')
- new_obj = get_type_for_entity(self.entity, 'create')
+ attr2ids = openbis_definitions("attr2ids")
+ new_obj = get_type_for_entity(self.entity, "create")
- for attr in self._defs['attrs_new']:
+ for attr in self._defs["attrs_new"]:
items = None
- if attr == 'type':
- new_obj['typeId'] = self._type['permId']
+ if attr == "type":
+ new_obj["typeId"] = self._type["permId"]
continue
- elif attr == 'kind':
+ elif attr == "kind":
# when creating a new dataset, the attribute 芦kind禄 is called 芦dataSetKind禄
- new_obj['dataSetKind'] = self._kind
+ new_obj["dataSetKind"] = self._kind
- elif attr == 'attachments':
- attachments = getattr(self, '_new_attachments')
+ elif attr == "attachments":
+ attachments = getattr(self, "_new_attachments")
if attachments is None:
continue
atts_data = [attachment.get_data() for attachment in attachments]
items = atts_data
- elif attr == 'userIds':
- if '_changed_users' not in self.__dict__:
+ elif attr == "userIds":
+ if "_changed_users" not in self.__dict__:
continue
- new_obj[attr]=[]
- for userId in self.__dict__['_changed_users']:
- if self.__dict__['_changed_users'][userId]['action'] == 'Add':
- new_obj[attr].append({
- "permId": userId,
- "@type": "as.dto.person.id.PersonPermId"
- })
+ new_obj[attr] = []
+ for userId in self.__dict__["_changed_users"]:
+ if self.__dict__["_changed_users"][userId]["action"] == "Add":
+ new_obj[attr].append(
+ {"permId": userId, "@type": "as.dto.person.id.PersonPermId"}
+ )
else:
- items = getattr(self, '_' + attr)
+ items = getattr(self, "_" + attr)
key = None
if attr in attr2ids:
@@ -155,27 +168,20 @@ class AttrHolder():
new_obj[key] = items
- # if method_name is not defined: guess the method name for creating a new entity
+ # if method_name is not defined: guess the method name for creating a new entity
if method_name is None:
- method_name = get_method_for_entity(self.entity, 'create')
- request = {
- "method": method_name,
- "params": [
- self.openbis.token,
- [new_obj]
- ]
- }
+ method_name = get_method_for_entity(self.entity, "create")
+ request = {"method": method_name, "params": [self.openbis.token, [new_obj]]}
return request
-
def _up_attrs(self, method_name=None, permId=None):
"""Returns the Python-equivalent JSON request when a new object is updated.
It is used internally by the save() method of an object to be updated.
"""
- #defs = openbis_definitions(self._entity)
- attr2ids = openbis_definitions('attr2ids')
+ # defs = openbis_definitions(self._entity)
+ attr2ids = openbis_definitions("attr2ids")
- up_obj = get_type_for_entity(self.entity, 'update')
+ up_obj = get_type_for_entity(self.entity, "update")
# for some weird reasons, the permId is called differently
# for every openBIS entity, but only when updating...
@@ -187,79 +193,105 @@ class AttrHolder():
# look at all attributes available for that entity
# that can be updated
- for attr in self._defs['attrs_up']:
+ for attr in self._defs["attrs_up"]:
items = None
- if attr == 'attachments':
+ if attr == "attachments":
# v3 API currently only supports adding attachments
- attachments = self.__dict__.get('_new_attachments', None)
+ attachments = self.__dict__.get("_new_attachments", None)
if attachments is None:
continue
atts_data = [attachment.get_data() for attachment in attachments]
- up_obj['attachments'] = {
- "actions": [{
- "items": atts_data,
- "@type": "as.dto.common.update.ListUpdateActionAdd"
- }],
- "@type": "as.dto.attachment.update.AttachmentListUpdateValue"
+ up_obj["attachments"] = {
+ "actions": [
+ {
+ "items": atts_data,
+ "@type": "as.dto.common.update.ListUpdateActionAdd",
+ }
+ ],
+ "@type": "as.dto.attachment.update.AttachmentListUpdateValue",
}
- elif attr == 'tags':
+ elif attr == "tags":
items = []
- for tag in self.__dict__['_tags']:
- items.append({
- "permId": tag['permId'],
- "@type" : "as.dto.tag.id.TagPermId"
- })
+ for tag in self.__dict__["_tags"]:
+ items.append(
+ {"permId": tag["permId"], "@type": "as.dto.tag.id.TagPermId"}
+ )
- up_obj['tagIds'] = {
+ up_obj["tagIds"] = {
"actions": [
{
"items": items,
"@type": "as.dto.common.update.ListUpdateActionSet",
}
],
- "@type": "as.dto.common.update.IdListUpdateValue"
+ "@type": "as.dto.common.update.IdListUpdateValue",
}
-
- elif attr == 'userIds':
+ elif attr in "metaData".split():
+ # ListUpdateMapValues
+ metaData = self.__dict__["_" + attr]
+ if metaData:
+ items = [metaData]
+ data_type = "as.dto.common.update.ListUpdateActionSet"
+ elif metaData is not None and len(metaData) == 0:
+ # metaData needs to be set to {} in order to remove it.
+ items = ["custom_widget"]
+ data_type = "as.dto.common.update.ListUpdateActionRemove"
+ up_obj[attr] = {
+ "actions": [
+ {
+ "items": items,
+ "@type": data_type
+ # "@type": "as.dto.common.update.ListUpdateActionSet",
+ # "@type": "as.dto.common.update.ListUpdateActionRemove",
+ }
+ ],
+ "@type": "as.dto.common.update.ListUpdateMapValues",
+ }
+ elif attr == "userIds":
actions = []
- if '_changed_users' not in self.__dict__:
+ if "_changed_users" not in self.__dict__:
continue
- for userId in self.__dict__['_changed_users']:
- actions.append({
- "items": [
- {
- "permId": userId,
- "@type": "as.dto.person.id.PersonPermId"
- }
- ],
- "@type": "as.dto.common.update.ListUpdateAction{}".format(
- self.__dict__['_changed_users'][userId]['action']
- )
- })
+ for userId in self.__dict__["_changed_users"]:
+ actions.append(
+ {
+ "items": [
+ {
+ "permId": userId,
+ "@type": "as.dto.person.id.PersonPermId",
+ }
+ ],
+ "@type": "as.dto.common.update.ListUpdateAction{}".format(
+ self.__dict__["_changed_users"][userId]["action"]
+ ),
+ }
+ )
- up_obj['userIds'] = {
+ up_obj["userIds"] = {
"actions": actions,
- "@type": "as.dto.common.update.IdListUpdateValue"
+ "@type": "as.dto.common.update.IdListUpdateValue",
}
- elif attr in 'description label official ordinal autoGeneratedCode subcodeUnique listable showContainer showParents showParentMetadata disallowDeletion validationPlugin'.split():
+ elif (
+ attr
+ in "description label official ordinal autoGeneratedCode subcodeUnique listable showContainer showParents showParentMetadata disallowDeletion validationPlugin".split()
+ ):
# alway update common fields
key = attr2ids.get(attr, attr)
up_obj[key] = {
- "value": self.__dict__['_'+attr],
+ "value": self.__dict__["_" + attr],
"isModified": True,
- "@type": "as.dto.common.update.FieldUpdateValue"
+ "@type": "as.dto.common.update.FieldUpdateValue",
}
- elif '_' + attr in self.__dict__:
+ elif "_" + attr in self.__dict__:
# handle multivalue attributes (parents, children, tags etc.)
- # we only cover the Set mechanism, which means we always update
+ # we only cover the Set mechanism, which means we always update
# all items in a list
- if 'multi' in self._defs and attr in self._defs['multi']:
- items = self.__dict__.get('_' + attr, [])
+ if "multi" in self._defs and attr in self._defs["multi"]:
+ items = self.__dict__.get("_" + attr, [])
if items == None:
items = []
up_obj[attr2ids[attr]] = {
@@ -269,11 +301,11 @@ class AttrHolder():
"@type": "as.dto.common.update.ListUpdateActionSet",
}
],
- "@type": "as.dto.common.update.IdListUpdateValue"
+ "@type": "as.dto.common.update.IdListUpdateValue",
}
else:
# handle single attributes (space, experiment, project, container, etc.)
- value = self.__dict__.get('_' + attr, {})
+ value = self.__dict__.get("_" + attr, {})
if value is None:
pass
elif isinstance(value, bool):
@@ -286,33 +318,26 @@ class AttrHolder():
"@type": "as.dto.common.update.FieldUpdateValue",
"isModified": True,
}
- elif 'isModified' in value and value['isModified'] == True:
+ elif "isModified" in value and value["isModified"] == True:
val = {}
- for x in ['identifier','permId','@type']:
+ for x in ["identifier", "permId", "@type"]:
if x in value:
val[x] = value[x]
up_obj[attr2ids[attr]] = {
"@type": "as.dto.common.update.FieldUpdateValue",
"isModified": True,
- "value": val
+ "value": val,
}
# update an existing entity
if method_name is None:
- method_name = get_method_for_entity(self.entity, 'update')
- request = {
- "method": method_name,
- "params": [
- self.openbis.token,
- [up_obj]
- ]
- }
+ method_name = get_method_for_entity(self.entity, "update")
+ request = {"method": method_name, "params": [self.openbis.token, [up_obj]]}
return request
-
def __getattr__(self, name):
- """ handles all attribute requests dynamically.
+ """handles all attribute requests dynamically.
Values are returned in a sensible way, for example:
the identifiers of parents, children and components are returned as an
array of values, whereas attachments, users (of groups) and
@@ -320,67 +345,74 @@ class AttrHolder():
"""
name_map = {
- 'group' : 'authorizationGroup',
- 'roles' : 'roleAssignments',
- 'permid' : 'permId',
- 'collection' : 'experiment',
- 'object' : 'sample'
+ "group": "authorizationGroup",
+ "roles": "roleAssignments",
+ "permid": "permId",
+ "collection": "experiment",
+ "object": "sample",
}
if name in name_map:
name = name_map[name]
- int_name = '_' + name
+ int_name = "_" + name
if int_name in self.__dict__:
- if int_name == '_attachments':
+ if int_name == "_attachments":
attachments = []
for att in self._attachments:
- attachments.append({
- "fileName": att.get('fileName'),
- "title": att.get('title'),
- "description": att.get('description'),
- "version": att.get('version'),
- })
+ attachments.append(
+ {
+ "fileName": att.get("fileName"),
+ "title": att.get("title"),
+ "description": att.get("description"),
+ "version": att.get("version"),
+ }
+ )
return attachments
- elif int_name == '_users':
+ elif int_name == "_users":
users = []
for user in self._users:
- users.append({
- "firstName": user.get('firstName'),
- "lastName" : user.get('lastName'),
- "email" : user.get('email'),
- "userId" : user.get('userId'),
- "space" : user.get('space').get('code') if user.get('space') is not None else None,
- })
+ users.append(
+ {
+ "firstName": user.get("firstName"),
+ "lastName": user.get("lastName"),
+ "email": user.get("email"),
+ "userId": user.get("userId"),
+ "space": user.get("space").get("code")
+ if user.get("space") is not None
+ else None,
+ }
+ )
return users
- elif int_name == '_roleAssignments':
+ elif int_name == "_roleAssignments":
ras = []
for ra in self._roleAssignments:
- ras.append({
- "techId": ra.get('id').get('techId'),
- "role": ra.get('role'),
- "roleLevel": ra.get('roleLevel'),
- "space": ra.get('space').get('code'),
- "project": ra.get('role'),
- })
+ ras.append(
+ {
+ "techId": ra.get("id").get("techId"),
+ "role": ra.get("role"),
+ "roleLevel": ra.get("roleLevel"),
+ "space": ra.get("space").get("code"),
+ "project": ra.get("role"),
+ }
+ )
return ras
-
- # if the attribute contains a list,
+ # if the attribute contains a list,
# return a list of either identifiers, codes or
# permIds (whatever is available first)
elif isinstance(self.__dict__[int_name], list):
values = []
for item in self.__dict__[int_name]:
if "identifier" in item:
- values.append(item['identifier'])
+ values.append(item["identifier"])
elif "code" in item:
- values.append(item['code'])
+ values.append(item["code"])
elif "userId" in item:
- values.append(item['userId'])
+ values.append(item["userId"])
elif "permId" in item:
- values.append(item['permId'])
+ values.append(item["permId"])
else:
values.append(item)
return values
@@ -388,17 +420,17 @@ class AttrHolder():
# attribute contains a dictionary: same procedure as above.
elif isinstance(self.__dict__[int_name], dict):
if "identifier" in self.__dict__[int_name]:
- return self.__dict__[int_name]['identifier']
+ return self.__dict__[int_name]["identifier"]
elif "code" in self.__dict__[int_name]:
- return self.__dict__[int_name]['code']
+ return self.__dict__[int_name]["code"]
elif "name" in self.__dict__[int_name]:
- return self.__dict__[int_name]['name']
+ return self.__dict__[int_name]["name"]
elif "userId" in self.__dict__[int_name]:
- return self.__dict__[int_name]['userId']
+ return self.__dict__[int_name]["userId"]
elif "permId" in self.__dict__[int_name]:
- return self.__dict__[int_name]['permId']
+ return self.__dict__[int_name]["permId"]
elif "id" in self.__dict__[int_name]:
- return self.__dict__[int_name]['id']
+ return self.__dict__[int_name]["id"]
else:
return self.__dict__[int_name]
@@ -414,25 +446,27 @@ class AttrHolder():
# experiment aka collection, sample aka object
name_map = {
- 'collection' : 'experiment',
- 'object' : 'sample',
- 'parent' : 'parents',
- 'child' : 'children',
+ "collection": "experiment",
+ "object": "sample",
+ "parent": "parents",
+ "child": "children",
}
if name in name_map:
name = name_map[name]
if self._is_new:
- if name not in self._defs['attrs_new']:
+ if name not in self._defs["attrs_new"]:
raise ValueError(
- "No such attribute: 芦{}禄 for entity: {}. Allowed attributes are: {}"
- .format(name, self.entity, self._defs['attrs_new'])
+ "No such attribute: 芦{}禄 for entity: {}. Allowed attributes are: {}".format(
+ name, self.entity, self._defs["attrs_new"]
+ )
)
else:
- if name not in self._defs['attrs_up']:
+ if name not in self._defs["attrs_up"]:
raise ValueError(
- "No such attribute: 芦{}禄 for entity: {}. Allowed attributes are: {}"
- .format(name, self.entity, self._defs['attrs_up'])
+ "No such attribute: 芦{}禄 for entity: {}. Allowed attributes are: {}".format(
+ name, self.entity, self._defs["attrs_up"]
+ )
)
if name in ["parents", "children", "components"]:
@@ -443,27 +477,27 @@ class AttrHolder():
for val in value:
if isinstance(val, str):
# fetch objects in openBIS, make sure they actually exists
- obj = getattr(self._openbis, 'get_' + self._entity.lower())(val)
+ obj = getattr(self._openbis, "get_" + self._entity.lower())(val)
objs.append(obj)
- elif getattr(val, '_permId'):
+ elif getattr(val, "_permId"):
# we got an existing object
objs.append(val)
permids = []
for item in objs:
- if getattr(item, '_identifier') is not None:
+ if getattr(item, "_identifier") is not None:
id = item._identifier
- elif getattr(item, '_permId') is not None:
+ elif getattr(item, "_permId") is not None:
id = item._permId
else:
return
-
+
# remove any existing @id keys to prevent jackson parser errors
- id.pop('@id', None)
-
+ id.pop("@id", None)
+
permids.append(id)
- self.__dict__['_' + name] = permids
+ self.__dict__["_" + name] = permids
elif name in ["tags"]:
self.set_tags(value)
@@ -471,31 +505,31 @@ class AttrHolder():
self.set_users(value)
elif name in ["vocabulary"]:
- if value is None or value == '':
- self.__dict__['_vocabulary'] = None
+ if value is None or value == "":
+ self.__dict__["_vocabulary"] = None
else:
- self.__dict__['_vocabulary'] = {
+ self.__dict__["_vocabulary"] = {
"@type": "as.dto.vocabulary.id.VocabularyPermId",
- "permId": value.upper()
+ "permId": value.upper(),
}
elif name in ["validationPlugin"]:
- if value is None or value == '':
- self.__dict__['_validationPlugin'] = None
+ if value is None or value == "":
+ self.__dict__["_validationPlugin"] = None
else:
- self.__dict__['_validationPlugin'] = {
+ self.__dict__["_validationPlugin"] = {
"@type": "as.dto.plugin.id.PluginPermId",
- "permId": value
+ "permId": value,
}
elif name in ["materialType"]:
- if value is None or value == '':
- self.__dict__['_materialType'] = None
+ if value is None or value == "":
+ self.__dict__["_materialType"] = None
else:
- self.__dict__['_materialType'] = {
+ self.__dict__["_materialType"] = {
"@type": "as.dto.entitytype.id.EntityTypePermId",
"permId": value.upper(),
- "entityKind": "MATERIAL"
+ "entityKind": "MATERIAL",
}
elif name in ["attachments"]:
@@ -512,7 +546,7 @@ class AttrHolder():
elif name in ["space"]:
obj = None
if value is None:
- self.__dict__['_'+name] = None
+ self.__dict__["_" + name] = None
return
if isinstance(value, str):
@@ -521,13 +555,13 @@ class AttrHolder():
else:
obj = value
- self.__dict__['_' + name] = obj.data['permId']
+ self.__dict__["_" + name] = obj.data["permId"]
# mark attribute as modified, if it's an existing entity
if self.is_new:
pass
else:
- self.__dict__['_' + name]['isModified'] = True
+ self.__dict__["_" + name]["isModified"] = True
elif name in ["sample", "experiment", "project"]:
obj = None
@@ -535,35 +569,39 @@ class AttrHolder():
# fetch object in openBIS, make sure it actually exists
obj = getattr(self._openbis, "get_" + name)(value)
elif value is None:
- self.__dict__['_'+name] = {}
+ self.__dict__["_" + name] = {}
return
else:
obj = value
- self.__dict__['_' + name] = obj.data['identifier']
+ self.__dict__["_" + name] = obj.data["identifier"]
# mark attribute as modified, if it's an existing entity
if self.is_new:
pass
else:
- self.__dict__['_' + name]['isModified'] = True
+ self.__dict__["_" + name]["isModified"] = True
elif name in ["identifier"]:
raise KeyError("you can not modify the {}".format(name))
elif name == "code":
try:
- if self._type['autoGeneratedCode']:
- raise KeyError("This {}Type has auto-generated code. You cannot set a code".format(self.entity))
+ if self._type["autoGeneratedCode"]:
+ raise KeyError(
+ "This {}Type has auto-generated code. You cannot set a code".format(
+ self.entity
+ )
+ )
except KeyError:
pass
except TypeError:
pass
- self.__dict__['_code'] = value
+ self.__dict__["_code"] = value
elif name in ["userId", "description"]:
# values that are directly assigned
- self.__dict__['_' + name] = value
+ self.__dict__["_" + name] = value
elif name in ["userIds"]:
self.add_users(value)
@@ -572,14 +610,12 @@ class AttrHolder():
# enum: check whether value is a valid constant
if value and value not in self._defs[name]:
raise ValueError(
- "Allowed values for enum {} are: {}"
- .format(name, self._defs[name])
+ "Allowed values for enum {} are: {}".format(name, self._defs[name])
)
else:
- self.__dict__['_'+name] = value
+ self.__dict__["_" + name] = value
else:
- self.__dict__['_'+name] = value
-
+ self.__dict__["_" + name] = value
def get_type(self):
return self._type
@@ -587,200 +623,229 @@ class AttrHolder():
def _ident_for_whatever(self, whatever):
if isinstance(whatever, str):
# fetch parent in openBIS, we are given an identifier
- obj = getattr(self._openbis, 'get_'+self._entity.lower())(whatever)
+ obj = getattr(self._openbis, "get_" + self._entity.lower())(whatever)
else:
# we assume we got an object
obj = whatever
ident = None
- if getattr(obj, '_identifier'):
+ if getattr(obj, "_identifier"):
ident = obj._identifier
- elif getattr(obj, '_permId'):
+ elif getattr(obj, "_permId"):
ident = obj._permId
- ident.pop('@id', None)
+ ident.pop("@id", None)
return ident
def get_container(self, **kwargs):
- return getattr(self._openbis, 'get_'+self._entity.lower())( self.container, **kwargs )
+ return getattr(self._openbis, "get_" + self._entity.lower())(
+ self.container, **kwargs
+ )
def get_containers(self, **kwargs):
- '''get the containers and return them as a list (Things/DataFrame)
+ """get the containers and return them as a list (Things/DataFrame)
or return empty list
- '''
- return getattr(self._openbis, 'get_'+self._entity.lower())( self.containers, **kwargs )
+ """
+ return getattr(self._openbis, "get_" + self._entity.lower())(
+ self.containers, **kwargs
+ )
def set_containers(self, containers_to_set):
- '''set the new _containers list
- '''
- self.__dict__['_containers'] = []
+ """set the new _containers list"""
+ self.__dict__["_containers"] = []
self.add_containers(containers_to_set)
def add_containers(self, containers_to_add):
- '''add component to _containers list
- '''
+ """add component to _containers list"""
if not isinstance(containers_to_add, list):
containers_to_add = [containers_to_add]
for component in containers_to_add:
ident = self._ident_for_whatever(component)
- if ident not in self.__dict__['_containers']:
- self.__dict__['_containers'].append(ident)
+ if ident not in self.__dict__["_containers"]:
+ self.__dict__["_containers"].append(ident)
def del_containers(self, containers_to_remove):
- '''remove component from _containers list
- '''
+ """remove component from _containers list"""
if not isinstance(containers_to_remove, list):
containers_to_remove = [containers_to_remove]
for component in containers_to_remove:
ident = self._ident_for_whatever(component)
- for i, item in enumerate(self.__dict__['_containers']):
- if 'identifier' in ident and 'identifier' in item and ident['identifier'] == item['identifier']:
- self.__dict__['_containers'].pop(i, None)
- elif 'permId' in ident and 'permId' in item and ident['permId'] == item['permId']:
- self.__dict__['_containers'].pop(i, None)
+ for i, item in enumerate(self.__dict__["_containers"]):
+ if (
+ "identifier" in ident
+ and "identifier" in item
+ and ident["identifier"] == item["identifier"]
+ ):
+ self.__dict__["_containers"].pop(i, None)
+ elif (
+ "permId" in ident
+ and "permId" in item
+ and ident["permId"] == item["permId"]
+ ):
+ self.__dict__["_containers"].pop(i, None)
def get_components(self, **kwargs):
- '''Samples and DataSets may contain other DataSets and Samples. This function returns the
+ """Samples and DataSets may contain other DataSets and Samples. This function returns the
contained Samples/DataSets (a.k.a. components) as a list (Things/DataFrame)
- '''
- return getattr(self._openbis, 'get_'+self._entity.lower())( self.components, **kwargs )
+ """
+ return getattr(self._openbis, "get_" + self._entity.lower())(
+ self.components, **kwargs
+ )
get_contained = get_components # Alias
def set_components(self, components_to_set):
- '''Samples and DataSets may contain other DataSets and Samples. This function sets the
+ """Samples and DataSets may contain other DataSets and Samples. This function sets the
contained Samples/DataSets (a.k.a. components)
- '''
- self.__dict__['_components'] = []
+ """
+ self.__dict__["_components"] = []
self.add_components(components_to_set)
set_contained = set_components # Alias
def add_components(self, components_to_add):
- '''Samples and DataSets may contain other DataSets and Samples. This function adds
+ """Samples and DataSets may contain other DataSets and Samples. This function adds
additional Samples/DataSets to the current object.
- '''
+ """
if not isinstance(components_to_add, list):
components_to_add = [components_to_add]
for component in components_to_add:
ident = self._ident_for_whatever(component)
- if ident not in self.__dict__['_components']:
- self.__dict__['_components'].append(ident)
+ if ident not in self.__dict__["_components"]:
+ self.__dict__["_components"].append(ident)
add_contained = add_components # Alias
def del_components(self, components_to_remove):
- '''Samples and DataSets may contain other DataSets and Samples. This function removes
+ """Samples and DataSets may contain other DataSets and Samples. This function removes
additional Samples/DataSets from the current object.
- '''
+ """
if not isinstance(components_to_remove, list):
components_to_remove = [components_to_remove]
for component in components_to_remove:
ident = self._ident_for_whatever(component)
- for i, item in enumerate(self.__dict__['_components']):
- if 'identifier' in ident and 'identifier' in item and ident['identifier'] == item['identifier']:
- self.__dict__['_components'].pop(i, None)
- elif 'permId' in ident and 'permId' in item and ident['permId'] == item['permId']:
- self.__dict__['_components'].pop(i, None)
+ for i, item in enumerate(self.__dict__["_components"]):
+ if (
+ "identifier" in ident
+ and "identifier" in item
+ and ident["identifier"] == item["identifier"]
+ ):
+ self.__dict__["_components"].pop(i, None)
+ elif (
+ "permId" in ident
+ and "permId" in item
+ and ident["permId"] == item["permId"]
+ ):
+ self.__dict__["_components"].pop(i, None)
del_contained = del_components # Alias
def get_parents(self, **kwargs):
- '''get the current parents and return them as a list (Things/DataFrame)
+ """get the current parents and return them as a list (Things/DataFrame)
or return empty list
- '''
- return getattr(self._openbis, 'get_'+self._entity.lower())( self.parents, **kwargs )
+ """
+ return getattr(self._openbis, "get_" + self._entity.lower())(
+ self.parents, **kwargs
+ )
def set_parents(self, parents_to_set):
- '''set the new _parents list
- '''
- self.__dict__['_parents'] = []
+ """set the new _parents list"""
+ self.__dict__["_parents"] = []
self.add_parents(parents_to_set)
def add_parents(self, parents_to_add):
- '''add parent to _parents list
- '''
+ """add parent to _parents list"""
if not isinstance(parents_to_add, list):
parents_to_add = [parents_to_add]
for parent in parents_to_add:
ident = self._ident_for_whatever(parent)
- if ident not in self.__dict__['_parents']:
- self.__dict__['_parents'].append(ident)
+ if ident not in self.__dict__["_parents"]:
+ self.__dict__["_parents"].append(ident)
def del_parents(self, parents_to_remove):
- '''remove parent from _parents list
- '''
+ """remove parent from _parents list"""
if not isinstance(parents_to_remove, list):
parents_to_remove = [parents_to_remove]
for parent in parents_to_remove:
ident = self._ident_for_whatever(parent)
- for i, item in enumerate(self.__dict__['_parents']):
- if 'identifier' in ident and 'identifier' in item and ident['identifier'] == item['identifier']:
- self.__dict__['_parents'].pop(i)
- elif 'permId' in ident and 'permId' in item and ident['permId'] == item['permId']:
- self.__dict__['_parents'].pop(i)
-
+ for i, item in enumerate(self.__dict__["_parents"]):
+ if (
+ "identifier" in ident
+ and "identifier" in item
+ and ident["identifier"] == item["identifier"]
+ ):
+ self.__dict__["_parents"].pop(i)
+ elif (
+ "permId" in ident
+ and "permId" in item
+ and ident["permId"] == item["permId"]
+ ):
+ self.__dict__["_parents"].pop(i)
def get_children(self, **kwargs):
- '''get the current children and return them as a list (Things/DataFrame)
+ """get the current children and return them as a list (Things/DataFrame)
or return empty list
- '''
- return getattr(self._openbis, 'get_'+self._entity.lower())( self.children, **kwargs )
+ """
+ return getattr(self._openbis, "get_" + self._entity.lower())(
+ self.children, **kwargs
+ )
def set_children(self, children_to_set):
- '''set the new _children list
- '''
- self.__dict__['_children'] = []
+ """set the new _children list"""
+ self.__dict__["_children"] = []
self.add_children(children_to_set)
def add_children(self, children):
- '''add children to _children list
- '''
- if getattr(self, '_children') is None:
- self.__dict__['_children'] = []
+ """add children to _children list"""
+ if getattr(self, "_children") is None:
+ self.__dict__["_children"] = []
if not isinstance(children, list):
children = [children]
for child in children:
- self.__dict__['_children'].append(self._ident_for_whatever(child))
+ self.__dict__["_children"].append(self._ident_for_whatever(child))
def del_children(self, children):
- '''remove children from _children list
- '''
- if getattr(self, '_children') is None:
+ """remove children from _children list"""
+ if getattr(self, "_children") is None:
return
if not isinstance(children, list):
children = [children]
for child in children:
ident = self._ident_for_whatever(child)
- for i, item in enumerate(self.__dict__['_children']):
- if 'identifier' in ident and 'identifier' in item and ident['identifier'] == item['identifier']:
- self.__dict__['_children'].pop(i, None)
- elif 'permId' in ident and 'permId' in item and ident['permId'] == item['permId']:
- self.__dict__['_children'].pop(i, None)
+ for i, item in enumerate(self.__dict__["_children"]):
+ if (
+ "identifier" in ident
+ and "identifier" in item
+ and ident["identifier"] == item["identifier"]
+ ):
+ self.__dict__["_children"].pop(i, None)
+ elif (
+ "permId" in ident
+ and "permId" in item
+ and ident["permId"] == item["permId"]
+ ):
+ self.__dict__["_children"].pop(i, None)
@property
def tags(self):
- if getattr(self, '_tags') is not None:
- return [x['code'] for x in self._tags]
+ if getattr(self, "_tags") is not None:
+ return [x["code"] for x in self._tags]
def get_tags(self):
- if getattr(self, '_tags') is not None:
- return self._openbis.get_tag([x['permId'] for x in self._tags])
+ if getattr(self, "_tags") is not None:
+ return self._openbis.get_tag([x["permId"] for x in self._tags])
def set_tags(self, tags):
- '''set _tags list
- '''
+ """set _tags list"""
- self.__dict__['_tags'] = []
+ self.__dict__["_tags"] = []
self.add_tags(tags)
-
def add_tags(self, tags):
- '''add tags to _tags list
- '''
+ """add tags to _tags list"""
if not isinstance(tags, list):
tags = [tags]
-
+
for tag in tags:
if isinstance(tag, str):
tag_obj = self._openbis.get_tag(tag)
@@ -790,85 +855,81 @@ class AttrHolder():
}
else:
tag_dict = {
- "code": tag['code'],
- "permId": tag['permId']['permId'],
+ "code": tag["code"],
+ "permId": tag["permId"]["permId"],
}
- if tag_dict not in self.__dict__['_tags']:
- self.__dict__['_tags'].append(tag_dict)
+ if tag_dict not in self.__dict__["_tags"]:
+ self.__dict__["_tags"].append(tag_dict)
def del_tags(self, tags):
- '''remove tags from _tags list
- '''
+ """remove tags from _tags list"""
if not isinstance(tags, list):
tags = [tags]
for tag in tags:
- for i, tag_dict in enumerate(self.__dict__['_tags']):
- if tag in tag_dict[i]['code'] or \
- tag in tag_dict[i]['permId']:
+ for i, tag_dict in enumerate(self.__dict__["_tags"]):
+ if tag in tag_dict[i]["code"] or tag in tag_dict[i]["permId"]:
tag_dict.pop(i, None)
def set_users(self, userIds):
if userIds is None:
return
- if getattr(self, '_userIds') is None:
- self.__dict__['_userIds'] = []
+ if getattr(self, "_userIds") is None:
+ self.__dict__["_userIds"] = []
if not isinstance(userIds, list):
userIds = [userIds]
for userId in userIds:
- self.__dict__['_userIds'].append({
- "permId": userId,
- "@type": "as.dto.person.id.PersonPermId"
- })
-
+ self.__dict__["_userIds"].append(
+ {"permId": userId, "@type": "as.dto.person.id.PersonPermId"}
+ )
+
def add_users(self, userIds):
if userIds is None:
return
- if getattr(self, '_changed_users') is None:
- self.__dict__['_changed_users'] = {}
+ if getattr(self, "_changed_users") is None:
+ self.__dict__["_changed_users"] = {}
if not isinstance(userIds, list):
userIds = [userIds]
for userId in userIds:
- self.__dict__['_changed_users'][userId] = {
- "action": "Add"
- }
- add_members = add_users # Alias
+ self.__dict__["_changed_users"][userId] = {"action": "Add"}
+ add_members = add_users # Alias
def del_users(self, userIds):
if userIds is None:
return
- if getattr(self, '_changed_users') is None:
- self.__dict__['_changed_users'] = {}
+ if getattr(self, "_changed_users") is None:
+ self.__dict__["_changed_users"] = {}
if not isinstance(userIds, list):
userIds = [userIds]
for userId in userIds:
- self.__dict__['_changed_users'][userId] = {
- "action": "Remove"
- }
+ self.__dict__["_changed_users"][userId] = {"action": "Remove"}
+
del_members = del_users # Alias
def get_attachments(self):
- if getattr(self, '_attachments') is None:
+ if getattr(self, "_attachments") is None:
return None
else:
- return DataFrame(self._attachments)[['fileName', 'title', 'description', 'version']]
+ return DataFrame(self._attachments)[
+ ["fileName", "title", "description", "version"]
+ ]
def add_attachment(self, fileName, title=None, description=None):
att = Attachment(filename=fileName, title=title, description=description)
- if getattr(self, '_attachments') is None:
- self.__dict__['_attachments'] = []
+ if getattr(self, "_attachments") is None:
+ self.__dict__["_attachments"] = []
self._attachments.append(att.get_data_short())
- if getattr(self, '_new_attachments') is None:
- self.__dict__['_new_attachments'] = []
+ if getattr(self, "_new_attachments") is None:
+ self.__dict__["_new_attachments"] = []
self._new_attachments.append(att)
def download_attachments(self):
- method = 'get' + self.entity + 's'
+ method = "get" + self.entity + "s"
entity = self.entity.lower()
request = {
"method": method,
@@ -876,40 +937,36 @@ class AttrHolder():
self._openbis.token,
[self._permId],
dict(
- attachments=fetch_option['attachmentsWithContent'],
+ attachments=fetch_option["attachmentsWithContent"],
**fetch_option[entity]
- )
- ]
+ ),
+ ],
}
resp = self._openbis._post_request(self._openbis.as_v3, request)
- attachments = resp[self.permId]['attachments']
+ attachments = resp[self.permId]["attachments"]
file_list = []
for attachment in attachments:
filename = os.path.join(
- self._openbis.hostname,
- self.permId,
- attachment['fileName']
+ self._openbis.hostname, self.permId, attachment["fileName"]
)
os.makedirs(os.path.dirname(filename), exist_ok=True)
- with open(filename, 'wb') as att:
- content = base64.b64decode(attachment['content'])
+ with open(filename, "wb") as att:
+ content = base64.b64decode(attachment["content"])
att.write(content)
file_list.append(filename)
return file_list
-
def all(self):
- """Return all attributes of an entity in a dict
- """
+ """Return all attributes of an entity in a dict"""
attrs = {}
- for attr in self._defs['attrs']:
- if attr == 'attachments':
+ for attr in self._defs["attrs"]:
+ if attr == "attachments":
continue
attrs[attr] = getattr(self, attr)
return attrs
def _repr_html_(self):
- def nvl(val, string=''):
+ def nvl(val, string=""):
if val is None:
return string
return val
@@ -925,16 +982,16 @@ class AttrHolder():
<tbody>
"""
- attrs = self._defs['attrs_new'] if self.is_new else self._defs['attrs']
+ attrs = self._defs["attrs_new"] if self.is_new else self._defs["attrs"]
for attr in attrs:
- if attr == 'attachments':
+ if attr == "attachments":
continue
html += "<tr> <td>{}</td> <td>{}</td> </tr>".format(
- attr, nvl(getattr(self, attr, ''), '')
+ attr, nvl(getattr(self, attr, ""), "")
)
- if getattr(self, '_attachments') is not None:
+ if getattr(self, "_attachments") is not None:
html += "<tr><td>attachments</td><td>"
- html += "<br/>".join(att['fileName'] for att in self._attachments)
+ html += "<br/>".join(att["fileName"] for att in self._attachments)
html += "</td></tr>"
html += """
@@ -944,40 +1001,32 @@ class AttrHolder():
return html
def __repr__(self):
- """ When using IPython, this method displays a nice table
+ """When using IPython, this method displays a nice table
of all attributes and their values when the object is printed.
"""
- headers = ['attribute', 'value']
+ headers = ["attribute", "value"]
lines = []
- attrs = self._defs['attrs_new'] if self.is_new else self._defs['attrs']
+ attrs = self._defs["attrs_new"] if self.is_new else self._defs["attrs"]
for attr in attrs:
- if attr == 'attachments':
+ if attr == "attachments":
continue
- elif attr == 'users' and '_users' in self.__dict__:
- lines.append([
- attr,
- ", ".join(att['userId'] for att in self._users)
- ])
- elif attr == 'roleAssignments' and '_roleAssignments' in self.__dict__:
+ elif attr == "users" and "_users" in self.__dict__:
+ lines.append([attr, ", ".join(att["userId"] for att in self._users)])
+ elif attr == "roleAssignments" and "_roleAssignments" in self.__dict__:
roles = []
for role in self._roleAssignments:
- if role.get('space') is not None:
- roles.append("{} ({})".format(
- role.get('role'),
- role.get('space').get('code')
- ))
+ if role.get("space") is not None:
+ roles.append(
+ "{} ({})".format(
+ role.get("role"), role.get("space").get("code")
+ )
+ )
else:
- roles.append(role.get('role'))
+ roles.append(role.get("role"))
+
+ lines.append([attr, ", ".join(roles)])
- lines.append([
- attr,
- ", ".join(roles)
- ])
-
else:
- lines.append([
- attr,
- nvl(getattr(self, attr, ''))
- ])
+ lines.append([attr, nvl(getattr(self, attr, ""))])
return tabulate(lines, headers=headers)
diff --git a/pybis/src/python/pybis/definitions.py b/pybis/src/python/pybis/definitions.py
index a83646fb23061fe0a1effb610e980dda55c3f601..c52ee7a3a091403148fbffe01ef5a2c5aa841e93 100644
--- a/pybis/src/python/pybis/definitions.py
+++ b/pybis/src/python/pybis/definitions.py
@@ -14,10 +14,10 @@ def openbis_definitions(entity):
"attrs": "code permId description frozen frozenForProjects frozenForSamples registrator registrationDate modificationDate".split(),
"multi": "".split(),
"identifier": "spaceId",
- "create": { "@type": "as.dto.space.create.SpaceCreation"},
- "update": { "@type": "as.dto.space.update.SpaceUpdate"},
- "delete": { "@type": "as.dto.space.delete.SpaceDeletionOptions"},
- "fetch": { "@type": "as.dto.space.fetchoptions.SpaceFetchOptions"},
+ "create": {"@type": "as.dto.space.create.SpaceCreation"},
+ "update": {"@type": "as.dto.space.update.SpaceUpdate"},
+ "delete": {"@type": "as.dto.space.delete.SpaceDeletionOptions"},
+ "fetch": {"@type": "as.dto.space.fetchoptions.SpaceFetchOptions"},
},
"project": {
"attrs_new": "code description space attachments".split(),
@@ -25,8 +25,8 @@ def openbis_definitions(entity):
"attrs": "code description permId identifier space leader registrator registrationDate modifier modificationDate attachments frozen frozenForExperiments frozenForSamples".split(),
"multi": "".split(),
"identifier": "projectId",
- "create": { "@type": "as.dto.project.create.ProjectCreation"},
- "update": { "@type": "as.dto.project.update.ProjectUpdate"},
+ "create": {"@type": "as.dto.project.create.ProjectCreation"},
+ "update": {"@type": "as.dto.project.update.ProjectUpdate"},
},
"experiment": {
"attrs_new": "code type project tags attachments".split(),
@@ -34,30 +34,30 @@ def openbis_definitions(entity):
"attrs": "code permId identifier type project tags registrator registrationDate modifier modificationDate attachments frozen frozenForDataSets frozenForSamples".split(),
"multi": "tags attachments".split(),
"identifier": "experimentId",
- "create": { "@type": "as.dto.experiment.create.ExperimentCreation"},
- "update": { "@type": "as.dto.experiment.update.ExperimentUpdate"},
+ "create": {"@type": "as.dto.experiment.create.ExperimentCreation"},
+ "update": {"@type": "as.dto.experiment.update.ExperimentUpdate"},
},
"externalDms": {
"attrs_new": "code label address addressType creationId".split(),
"attrs_up": "label address".split(),
"attrs": "code permId label address addressType urlTemplate".split(),
"identifier": "externalDmsId",
- "create": { "@type": "as.dto.externaldms.create.ExternalDmsCreation"},
- "update": { "@type": "as.dto.externaldms.update.ExternalDmsUpdate"},
+ "create": {"@type": "as.dto.externaldms.create.ExternalDmsCreation"},
+ "update": {"@type": "as.dto.externaldms.update.ExternalDmsUpdate"},
},
"sample": {
"attrs_new": "code type project parents children container components space experiment tags attachments".split(),
"attrs_up": "project parents children container components space experiment tags attachments freeze freezeForComponents freezeForChildren freezeForParents freezeForDataSets".split(),
"attrs": "code permId identifier type project parents children components space experiment tags registrator registrationDate modifier modificationDate attachments container frozen frozenForComponents frozenForChildren frozenForParents frozenForDataSets".split(),
"ids2type": {
- 'parentIds': {'permId': {'@type': 'as.dto.sample.id.SamplePermId'}},
- 'childIds': {'permId': {'@type': 'as.dto.sample.id.SamplePermId'}},
- 'componentIds': {'permId': {'@type': 'as.dto.sample.id.SamplePermId'}},
+ "parentIds": {"permId": {"@type": "as.dto.sample.id.SamplePermId"}},
+ "childIds": {"permId": {"@type": "as.dto.sample.id.SamplePermId"}},
+ "componentIds": {"permId": {"@type": "as.dto.sample.id.SamplePermId"}},
},
"identifier": "sampleId",
- "create": { "@type": "as.dto.sample.create.SampleCreation"},
- "update": { "@type": "as.dto.sample.update.SampleUpdate"},
- "delete": { "@type": "as.dto.sample.delete.SampleDeletionOptions"},
+ "create": {"@type": "as.dto.sample.create.SampleCreation"},
+ "update": {"@type": "as.dto.sample.update.SampleUpdate"},
+ "delete": {"@type": "as.dto.sample.delete.SampleDeletionOptions"},
"cre_type": "as.dto.sample.create.SampleCreation",
"multi": "parents children components tags attachments".split(),
},
@@ -66,112 +66,79 @@ def openbis_definitions(entity):
"attrs_up": "description autoGeneratedCode generatedCodePrefix subcodeUnique listable showContainer showParents showParentMetadata, validationPlugin".split(),
"attrs": "permId code description autoGeneratedCode generatedCodePrefix subcodeUnique listable showContainer showParents showParentMetadata modificationDate validationPlugin".split(),
"default_attribute_values": {
- "autoGeneratedCode" : False,
- "generatedCodePrefix" : "S",
- "subcodeUnique" : False,
- "description" : "",
- "listable" : True,
- "showContainer" : False,
- "showParents" : True,
- "showParentMetadata" : False
- },
- "search": {
- "@type": "as.dto.sample.search.SampleTypeSearchCriteria"
- },
- "create": {
- "@type": "as.dto.sample.create.SampleTypeCreation"
- },
- "update": {
- "@type": "as.dto.sample.update.SampleTypeUpdate"
- },
- "delete": {
- "@type": "as.dto.sample.delete.SampleTypeDeletionOptions"
+ "autoGeneratedCode": False,
+ "generatedCodePrefix": "S",
+ "subcodeUnique": False,
+ "description": "",
+ "listable": True,
+ "showContainer": False,
+ "showParents": True,
+ "showParentMetadata": False,
},
+ "search": {"@type": "as.dto.sample.search.SampleTypeSearchCriteria"},
+ "create": {"@type": "as.dto.sample.create.SampleTypeCreation"},
+ "update": {"@type": "as.dto.sample.update.SampleTypeUpdate"},
+ "delete": {"@type": "as.dto.sample.delete.SampleTypeDeletionOptions"},
"identifier": "typeId",
},
"materialType": {
- "attrs_new": "code description validationPlugin".split(),
- "attrs_up": "description validationPlugin".split(),
+ "attrs_new": "code description validationPlugin".split(),
+ "attrs_up": "description validationPlugin".split(),
"attrs": "permId code description validationPlugin".split(),
- "search": {
- "@type": "as.dto.material.search.MaterialTypeSearchCriteria"
- },
- "fetch": {
- "@type": "as.dto.material.fetchoptions.MaterialTypeFetchOptions"
- },
- "create": {
- "@type": "as.dto.material.create.MaterialTypeCreation"
- },
- "update": {
- "@type": "as.dto.material.update.MaterialTypeUpdate"
- },
- "delete": {
- "@type": "as.dto.material.delete.MaterialTypeDeletionOptions"
- },
+ "search": {"@type": "as.dto.material.search.MaterialTypeSearchCriteria"},
+ "fetch": {"@type": "as.dto.material.fetchoptions.MaterialTypeFetchOptions"},
+ "create": {"@type": "as.dto.material.create.MaterialTypeCreation"},
+ "update": {"@type": "as.dto.material.update.MaterialTypeUpdate"},
+ "delete": {"@type": "as.dto.material.delete.MaterialTypeDeletionOptions"},
"identifier": "typeId",
},
"dataSetType": {
- "attrs_new": "code description mainDataSetPattern mainDataSetPath disallowDeletion validationPlugin".split(),
- "attrs_up": "description mainDataSetPattern mainDataSetPath disallowDeletion validationPlugin".split(),
+ "attrs_new": "code description mainDataSetPattern mainDataSetPath disallowDeletion validationPlugin".split(),
+ "attrs_up": "description mainDataSetPattern mainDataSetPath disallowDeletion validationPlugin".split(),
"attrs": "permId code description mainDataSetPattern mainDataSetPath disallowDeletion modificationDate validationPlugin".split(),
- "search": {
- "@type": "as.dto.dataset.search.DataSetTypeSearchCriteria"
- },
- "fetch": {
- "@type": "as.dto.dataset.fetchoptions.DataSetTypeFetchOptions"
- },
- "create": {
- "@type": "as.dto.dataset.create.DataSetTypeCreation"
- },
- "update": {
- "@type": "as.dto.dataset.update.DataSetTypeUpdate"
- },
- "delete": {
- "@type": "as.dto.dataset.delete.DataSetTypeDeletionOptions"
- },
+ "search": {"@type": "as.dto.dataset.search.DataSetTypeSearchCriteria"},
+ "fetch": {"@type": "as.dto.dataset.fetchoptions.DataSetTypeFetchOptions"},
+ "create": {"@type": "as.dto.dataset.create.DataSetTypeCreation"},
+ "update": {"@type": "as.dto.dataset.update.DataSetTypeUpdate"},
+ "delete": {"@type": "as.dto.dataset.delete.DataSetTypeDeletionOptions"},
"identifier": "typeId",
},
"experimentType": {
"attrs_new": "code description validationPlugin".split(),
- "attrs_up": "description modificationDate validationPlugin".split(),
- "attrs": "permId code description modificationDate validationPlugin".split(),
+ "attrs_up": "description modificationDate validationPlugin".split(),
+ "attrs": "permId code description modificationDate validationPlugin".split(),
"search": {
"@type": "as.dto.experiment.search.ExperimentTypeSearchCriteria"
},
"fetch": {
"@type": "as.dto.experiment.fetchoptions.ExperimentTypeFetchOptions"
},
- "create": {
- "@type": "as.dto.experiment.create.ExperimentTypeCreation"
- },
- "update": {
- "@type": "as.dto.experiment.update.ExperimentTypeUpdate"
- },
+ "create": {"@type": "as.dto.experiment.create.ExperimentTypeCreation"},
+ "update": {"@type": "as.dto.experiment.update.ExperimentTypeUpdate"},
"delete": {
"@type": "as.dto.experiment.delete.ExperimentTypeDeletionOptions"
},
"identifier": "typeId",
},
"propertyType": {
- "attrs": "code label description managedInternally dataType vocabulary materialType schema transformation semanticAnnotations registrator registrationDate".split(),
+ "attrs": "code label description managedInternally dataType vocabulary materialType schema transformation semanticAnnotations registrator registrationDate metaData".split(),
"attrs_new": "code label description managedInternally dataType vocabulary materialType schema transformation metaData".split(),
"attrs_up": "label description schema transformation metaData".split(),
- "search": {
- "@type": "as.dto.property.search.PropertyTypeSearchCriteria"
- },
- "create": {
- "@type": "as.dto.property.create.PropertyTypeCreation"
- },
- "update": {
- "@type": "as.dto.property.update.PropertyTypeUpdate"
- },
- "delete": {
- "@type": "as.dto.property.delete.PropertyTypeDeletionOptions"
- },
+ "search": {"@type": "as.dto.property.search.PropertyTypeSearchCriteria"},
+ "create": {"@type": "as.dto.property.create.PropertyTypeCreation"},
+ "update": {"@type": "as.dto.property.update.PropertyTypeUpdate"},
+ "delete": {"@type": "as.dto.property.delete.PropertyTypeDeletionOptions"},
"dataType": [
- "INTEGER", "VARCHAR", "MULTILINE_VARCHAR",
- "REAL", "TIMESTAMP", "BOOLEAN", "CONTROLLEDVOCABULARY",
- "MATERIAL", "HYPERLINK", "XML"
+ "INTEGER",
+ "VARCHAR",
+ "MULTILINE_VARCHAR",
+ "REAL",
+ "TIMESTAMP",
+ "BOOLEAN",
+ "CONTROLLEDVOCABULARY",
+ "MATERIAL",
+ "HYPERLINK",
+ "XML",
],
"identifier": "typeId",
},
@@ -180,12 +147,8 @@ def openbis_definitions(entity):
"attrs_up": "entityType propertyType predicateOntologyId predicateOntologyVersion predicateAccessionId descriptorOntologyId descriptorOntologyVersion descriptorAccessionId ".split(),
"attrs": "permId entityType propertyType predicateOntologyId predicateOntologyVersion predicateAccessionId descriptorOntologyId descriptorOntologyVersion descriptorAccessionId creationDate".split(),
"ids2type": {
- "propertyTypeId": {
- "permId": "as.dto.property.id.PropertyTypePermId"
- },
- "entityTypeId": {
- "permId": "as.dto.entity.id.EntityTypePermId"
- },
+ "propertyTypeId": {"permId": "as.dto.property.id.PropertyTypePermId"},
+ "entityTypeId": {"permId": "as.dto.entity.id.EntityTypePermId"},
},
"identifier": "permId",
"cre_type": "as.dto.sample.create.SampleCreation",
@@ -195,12 +158,15 @@ def openbis_definitions(entity):
"attrs_new": "type code kind experiment sample parents children components containers tags".split(),
"attrs_up": "parents children experiment sample components containers tags freeze freezeForChildren freezeForParents freezeForComponents freezeForContainers".split(),
"attrs": "code permId type kind experiment sample parents children components containers tags accessDate dataProducer dataProductionDate registrator registrationDate modifier modificationDate dataStore measured postRegistered frozen frozenForChildren frozenForParents frozenForComponents frozenForContainers".split(),
-
"ids2type": {
- 'parentIds': {'permId': {'@type': 'as.dto.dataset.id.DataSetPermId'}},
- 'childIds': {'permId': {'@type': 'as.dto.dataset.id.DataSetPermId'}},
- 'componentIds': {'permId': {'@type': 'as.dto.dataset.id.DataSetPermId'}},
- 'containerIds': {'permId': {'@type': 'as.dto.dataset.id.DataSetPermId'}},
+ "parentIds": {"permId": {"@type": "as.dto.dataset.id.DataSetPermId"}},
+ "childIds": {"permId": {"@type": "as.dto.dataset.id.DataSetPermId"}},
+ "componentIds": {
+ "permId": {"@type": "as.dto.dataset.id.DataSetPermId"}
+ },
+ "containerIds": {
+ "permId": {"@type": "as.dto.dataset.id.DataSetPermId"}
+ },
},
"multi": "parents children containers components".split(),
"identifier": "dataSetId",
@@ -225,11 +191,11 @@ def openbis_definitions(entity):
"attrs": "code description managedInternally chosenFromList urlTemplate registrator registrationDate modifier modificationDate".split(),
"multi": "".split(),
"identifier": "vocabularyId",
- "search": { "@type": "as.dto.vocabulary.search.VocabularySearchCriteria" },
- "create": { "@type": "as.dto.vocabulary.create.VocabularyCreation"},
- "update": { "@type": "as.dto.vocabulary.update.VocabularyUpdate"},
- "delete": { "@type": "as.dto.vocabulary.delete.VocabularyDeletionOptions"},
- "fetch": { "@type": "as.dto.vocabulary.fetchoptions.VocabularyFetchOptions"},
+ "search": {"@type": "as.dto.vocabulary.search.VocabularySearchCriteria"},
+ "create": {"@type": "as.dto.vocabulary.create.VocabularyCreation"},
+ "update": {"@type": "as.dto.vocabulary.update.VocabularyUpdate"},
+ "delete": {"@type": "as.dto.vocabulary.delete.VocabularyDeletionOptions"},
+ "fetch": {"@type": "as.dto.vocabulary.fetchoptions.VocabularyFetchOptions"},
},
"vocabularyTerm": {
"attrs_new": "code vocabularyCode label description official ordinal".split(),
@@ -237,10 +203,14 @@ def openbis_definitions(entity):
"attrs": "code vocabularyCode label description official ordinal registrator registrationDate modifier modificationDate".split(),
"multi": "".split(),
"identifier": "vocabularyTermId",
- "create": { "@type": "as.dto.vocabulary.create.VocabularyTermCreation"},
- "update": { "@type": "as.dto.vocabulary.update.VocabularyTermUpdate"},
- "delete": { "@type": "as.dto.vocabulary.delete.VocabularyTermDeletionOptions"},
- "fetch": { "@type": "as.dto.vocabulary.fetchoptions.VocabularyTermFetchOptions"},
+ "create": {"@type": "as.dto.vocabulary.create.VocabularyTermCreation"},
+ "update": {"@type": "as.dto.vocabulary.update.VocabularyTermUpdate"},
+ "delete": {
+ "@type": "as.dto.vocabulary.delete.VocabularyTermDeletionOptions"
+ },
+ "fetch": {
+ "@type": "as.dto.vocabulary.fetchoptions.VocabularyTermFetchOptions"
+ },
},
"plugin": {
"attrs_new": "name description pluginType script available entityKind".split(),
@@ -248,8 +218,8 @@ def openbis_definitions(entity):
"attrs": "permId name description registrator registrationDate pluginKind entityKinds pluginType script available".split(),
"multi": "".split(),
"identifier": "pluginId",
- "pluginType": ['DYNAMIC_PROPERTY', 'MANAGED_PROPERTY', 'ENTITY_VALIDATION'],
- "entityKind": ['MATERIAL', 'EXPERIMENT', 'SAMPLE', 'DATA_SET'],
+ "pluginType": ["DYNAMIC_PROPERTY", "MANAGED_PROPERTY", "ENTITY_VALIDATION"],
+ "entityKind": ["MATERIAL", "EXPERIMENT", "SAMPLE", "DATA_SET"],
},
"person": {
"attrs_new": "userId space".split(),
@@ -258,18 +228,25 @@ def openbis_definitions(entity):
"multi": "".split(),
"identifier": "userId",
},
- "authorizationGroup" : {
+ "authorizationGroup": {
"attrs_new": "code description userIds".split(),
"attrs_up": "code description userIds".split(),
"attrs": "permId code description registrator registrationDate modificationDate users".split(),
"multi": "users".split(),
"identifier": "groupId",
},
- "roleAssignment" : {
+ "roleAssignment": {
"attrs": "id user authorizationGroup role roleLevel space project registrator registrationDate".split(),
"attrs_new": "role roleLevel user authorizationGroup role space project".split(),
"attrs_up": "role roleLevel user authorizationGroup role space project".split(),
- "role": ['POWER_USER', 'OBSERVER', 'USER', 'DISABLED', 'ADMIN', 'ETL_SERVER']
+ "role": [
+ "POWER_USER",
+ "OBSERVER",
+ "USER",
+ "DISABLED",
+ "ADMIN",
+ "ETL_SERVER",
+ ],
},
"attr2ids": {
"space": "spaceId",
@@ -297,18 +274,21 @@ def openbis_definitions(entity):
"validationPlugin": "validationPluginId",
},
"ids2type": {
- 'spaceId': {'permId': {'@type': 'as.dto.space.id.SpacePermId'}},
- 'projectId': {'permId': {'@type': 'as.dto.project.id.ProjectPermId'}},
- 'experimentId': {'permId': {'@type': 'as.dto.experiment.id.ExperimentPermId'}},
- 'tagIds': {'code': {'@type': 'as.dto.tag.id.TagCode'}},
+ "spaceId": {"permId": {"@type": "as.dto.space.id.SpacePermId"}},
+ "projectId": {"permId": {"@type": "as.dto.project.id.ProjectPermId"}},
+ "experimentId": {
+ "permId": {"@type": "as.dto.experiment.id.ExperimentPermId"}
+ },
+ "tagIds": {"code": {"@type": "as.dto.tag.id.TagCode"}},
},
"dataSetFile": {
- "search": { "@type": "dss.dto.datasetfile.search.DataSetFileSearchCriteria" }
+ "search": {"@type": "dss.dto.datasetfile.search.DataSetFileSearchCriteria"}
},
}
return entities[entity]
-get_definition_for_entity = openbis_definitions # Alias
+
+get_definition_for_entity = openbis_definitions # Alias
fetch_option = {
@@ -318,20 +298,22 @@ fetch_option = {
"space": {"@type": "as.dto.space.fetchoptions.SpaceFetchOptions"},
},
"person": {"@type": "as.dto.person.fetchoptions.PersonFetchOptions"},
- "users": {"@type": "as.dto.person.fetchoptions.PersonFetchOptions" },
- "user": {"@type": "as.dto.person.fetchoptions.PersonFetchOptions" },
- "owner": {"@type": "as.dto.person.fetchoptions.PersonFetchOptions" },
+ "users": {"@type": "as.dto.person.fetchoptions.PersonFetchOptions"},
+ "user": {"@type": "as.dto.person.fetchoptions.PersonFetchOptions"},
+ "owner": {"@type": "as.dto.person.fetchoptions.PersonFetchOptions"},
"registrator": {"@type": "as.dto.person.fetchoptions.PersonFetchOptions"},
"modifier": {"@type": "as.dto.person.fetchoptions.PersonFetchOptions"},
"leader": {"@type": "as.dto.person.fetchoptions.PersonFetchOptions"},
- "authorizationGroup": {"@type": "as.dto.authorizationgroup.fetchoptions.AuthorizationGroupFetchOptions"},
+ "authorizationGroup": {
+ "@type": "as.dto.authorizationgroup.fetchoptions.AuthorizationGroupFetchOptions"
+ },
"experiment": {
"@type": "as.dto.experiment.fetchoptions.ExperimentFetchOptions",
- "type": {"@type": "as.dto.experiment.fetchoptions.ExperimentTypeFetchOptions"}
+ "type": {"@type": "as.dto.experiment.fetchoptions.ExperimentTypeFetchOptions"},
},
"sample": {
"@type": "as.dto.sample.fetchoptions.SampleFetchOptions",
- "type": {"@type": "as.dto.sample.fetchoptions.SampleTypeFetchOptions"}
+ "type": {"@type": "as.dto.sample.fetchoptions.SampleTypeFetchOptions"},
},
"samples": {"@type": "as.dto.sample.fetchoptions.SampleFetchOptions"},
"sampleType": {
@@ -341,13 +323,11 @@ fetch_option = {
"propertyType": {
"@type": "as.dto.property.fetchoptions.PropertyTypeFetchOptions"
},
- "plugin": {
- "@type": "as.dto.plugin.fetchoptions.PluginFetchOptions"
- }
+ "plugin": {"@type": "as.dto.plugin.fetchoptions.PluginFetchOptions"},
},
"validationPlugin": {
"@type": "as.dto.plugin.fetchoptions.PluginFetchOptions",
- }
+ },
},
"materialType": {
"@type": "as.dto.material.fetchoptions.MaterialTypeFetchOptions",
@@ -356,13 +336,11 @@ fetch_option = {
"propertyType": {
"@type": "as.dto.property.fetchoptions.PropertyTypeFetchOptions",
},
- "plugin": {
- "@type": "as.dto.plugin.fetchoptions.PluginFetchOptions"
- }
+ "plugin": {"@type": "as.dto.plugin.fetchoptions.PluginFetchOptions"},
},
"validationPlugin": {
"@type": "as.dto.plugin.fetchoptions.PluginFetchOptions",
- }
+ },
},
"dataSetType": {
"@type": "as.dto.dataset.fetchoptions.DataSetTypeFetchOptions",
@@ -371,13 +349,11 @@ fetch_option = {
"propertyType": {
"@type": "as.dto.property.fetchoptions.PropertyTypeFetchOptions",
},
- "plugin": {
- "@type": "as.dto.plugin.fetchoptions.PluginFetchOptions"
- }
+ "plugin": {"@type": "as.dto.plugin.fetchoptions.PluginFetchOptions"},
},
"validationPlugin": {
"@type": "as.dto.plugin.fetchoptions.PluginFetchOptions",
- }
+ },
},
"experimentType": {
"@type": "as.dto.experiment.fetchoptions.ExperimentTypeFetchOptions",
@@ -386,19 +362,17 @@ fetch_option = {
"propertyType": {
"@type": "as.dto.property.fetchoptions.PropertyTypeFetchOptions",
},
- "plugin": {
- "@type": "as.dto.plugin.fetchoptions.PluginFetchOptions"
- }
- },
+ "plugin": {"@type": "as.dto.plugin.fetchoptions.PluginFetchOptions"},
+ },
"validationPlugin": {
"@type": "as.dto.plugin.fetchoptions.PluginFetchOptions",
- }
+ },
},
"dataSet": {
"@type": "as.dto.dataset.fetchoptions.DataSetFetchOptions",
- "type": {"@type": "as.dto.dataset.fetchoptions.DataSetTypeFetchOptions"},
- "parents": {"@type": "as.dto.dataset.fetchoptions.DataSetFetchOptions"},
- "children": {"@type": "as.dto.dataset.fetchoptions.DataSetFetchOptions"},
+ "type": {"@type": "as.dto.dataset.fetchoptions.DataSetTypeFetchOptions"},
+ "parents": {"@type": "as.dto.dataset.fetchoptions.DataSetFetchOptions"},
+ "children": {"@type": "as.dto.dataset.fetchoptions.DataSetFetchOptions"},
"containers": {"@type": "as.dto.dataset.fetchoptions.DataSetFetchOptions"},
"components": {"@type": "as.dto.dataset.fetchoptions.DataSetFetchOptions"},
},
@@ -409,8 +383,10 @@ fetch_option = {
},
"physicalData": {"@type": "as.dto.dataset.fetchoptions.PhysicalDataFetchOptions"},
"linkedData": {
- "externalDms": {"@type": "as.dto.externaldms.fetchoptions.ExternalDmsFetchOptions"},
- "@type": "as.dto.dataset.fetchoptions.LinkedDataFetchOptions"
+ "externalDms": {
+ "@type": "as.dto.externaldms.fetchoptions.ExternalDmsFetchOptions"
+ },
+ "@type": "as.dto.dataset.fetchoptions.LinkedDataFetchOptions",
},
"roleAssignment": {
"@type": "as.dto.roleassignment.fetchoptions.RoleAssignmentFetchOptions",
@@ -425,23 +401,21 @@ fetch_option = {
"@type": "as.dto.property.fetchoptions.PropertyTypeFetchOptions",
"vocabulary": {
"@type": "as.dto.vocabulary.fetchoptions.VocabularyFetchOptions",
- }
- }
+ },
+ },
},
"propertyType": {
"@type": "as.dto.property.fetchoptions.PropertyTypeFetchOptions",
"vocabulary": {
"@type": "as.dto.vocabulary.fetchoptions.VocabularyFetchOptions"
},
- "materialType": {
- "@type": "as.dto.material.fetchoptions.MaterialTypeFetchOptions"
+ "materialType": {
+ "@type": "as.dto.material.fetchoptions.MaterialTypeFetchOptions"
},
"semanticAnnotations": {
"@type": "as.dto.semanticannotation.fetchoptions.SemanticAnnotationFetchOptions"
},
- "registrator": {
- "@type": "as.dto.person.fetchoptions.PersonFetchOptions"
- },
+ "registrator": {"@type": "as.dto.person.fetchoptions.PersonFetchOptions"},
},
"semanticAnnotations": {
"@type": "as.dto.semanticannotation.fetchoptions.SemanticAnnotationFetchOptions"
@@ -451,9 +425,7 @@ fetch_option = {
"attachments": {"@type": "as.dto.attachment.fetchoptions.AttachmentFetchOptions"},
"attachmentsWithContent": {
"@type": "as.dto.attachment.fetchoptions.AttachmentFetchOptions",
- "content": {
- "@type": "as.dto.common.fetchoptions.EmptyFetchOptions"
- },
+ "content": {"@type": "as.dto.common.fetchoptions.EmptyFetchOptions"},
},
"script": {
"@type": "as.dto.common.fetchoptions.EmptyFetchOptions",
@@ -464,23 +436,31 @@ fetch_option = {
"vocabulary": {
"@type": "as.dto.vocabulary.fetchoptions.VocabularyFetchOptions",
},
- "vocabularyTerm": {"@type": "as.dto.vocabulary.fetchoptions.VocabularyTermFetchOptions"},
- "deletedObjects": { "@type": "as.dto.deletion.fetchoptions.DeletedObjectFetchOptions" },
- "deletion": { "@type": "as.dto.deletion.fetchoptions.DeletionFetchOptions" },
- "externalDms": { "@type": "as.dto.externaldms.fetchoptions.ExternalDmsFetchOptions"},
- "dataSetFile": { "@type": "dss.dto.datasetfile.fetchoptions.DataSetFileFetchOptions"},
+ "vocabularyTerm": {
+ "@type": "as.dto.vocabulary.fetchoptions.VocabularyTermFetchOptions"
+ },
+ "deletedObjects": {
+ "@type": "as.dto.deletion.fetchoptions.DeletedObjectFetchOptions"
+ },
+ "deletion": {"@type": "as.dto.deletion.fetchoptions.DeletionFetchOptions"},
+ "externalDms": {"@type": "as.dto.externaldms.fetchoptions.ExternalDmsFetchOptions"},
+ "dataSetFile": {
+ "@type": "dss.dto.datasetfile.fetchoptions.DataSetFileFetchOptions"
+ },
}
+
def get_fetchoption_for_entity(entity):
- entity = entity[0].lower() + entity[1:] # make first character lowercase
+ entity = entity[0].lower() + entity[1:] # make first character lowercase
try:
return fetch_option[entity]
except KeyError as e:
return {}
-def get_type_for_entity(entity, action, parents_or_children=''):
+
+def get_type_for_entity(entity, action, parents_or_children=""):
if action not in "create update delete search".split():
- raise ValueError('unknown action: {}'.format(action))
+ raise ValueError("unknown action: {}".format(action))
definition = openbis_definitions(entity)
if action in definition and not parents_or_children:
@@ -494,22 +474,30 @@ def get_type_for_entity(entity, action, parents_or_children=''):
"delete": "DeletionOptions",
"search": "SearchCriteria",
}
-
+
if parents_or_children:
return {
- "@type": "as.dto.{}.{}.{}{}{}"
- .format(entity.lower(), action, cap_entity, parents_or_children, noun[action])
+ "@type": "as.dto.{}.{}.{}{}{}".format(
+ entity.lower(),
+ action,
+ cap_entity,
+ parents_or_children,
+ noun[action],
+ )
}
else:
return {
- "@type": "as.dto.{}.{}.{}{}"
- .format(entity.lower(), action, cap_entity, noun[action])
+ "@type": "as.dto.{}.{}.{}{}".format(
+ entity.lower(), action, cap_entity, noun[action]
+ )
}
+
def get_fetchoptions(entity, including=None):
- if including is None: including = []
- including += ['@type']
- entity = entity[0].lower() + entity[1:] # make first character lowercase
+ if including is None:
+ including = []
+ including += ["@type"]
+ entity = entity[0].lower() + entity[1:] # make first character lowercase
fo = {}
for inc in including:
try:
@@ -525,8 +513,7 @@ def get_method_for_entity(entity, action):
if entity == "vocabulary":
return "{}Vocabularies".format(action)
-
+
cap_entity = entity[:1].upper() + entity[1:]
return "{}{}s".format(action, cap_entity)
-
diff --git a/pybis/src/python/pybis/openbis_object.py b/pybis/src/python/pybis/openbis_object.py
index 252a41dab37a850c61f0ae4488d49939bc938c06..7b59e015c7bebeb19c591e52dd2d8e126e1335c5 100644
--- a/pybis/src/python/pybis/openbis_object.py
+++ b/pybis/src/python/pybis/openbis_object.py
@@ -1,30 +1,30 @@
from .property import PropertyHolder
from .attribute import AttrHolder
from .utils import VERBOSE
-from .definitions import get_definition_for_entity, get_type_for_entity, get_method_for_entity
+from .definitions import (
+ get_definition_for_entity,
+ get_type_for_entity,
+ get_method_for_entity,
+)
from collections import defaultdict
-class OpenBisObject():
- def __init_subclass__(
- cls,
- entity=None,
- single_item_method_name=None
- ):
+class OpenBisObject:
+ def __init_subclass__(cls, entity=None, single_item_method_name=None):
"""create a specialized parent class.
The class that inherits from OpenBisObject does not need
to implement its own __init__ method in order to provide the
entity name. Instead, it can pass the entity name as a param:
class XYZ(OpenBisObject, entity="myEntity")
"""
- cls._entity=entity
- cls._single_item_method_name=single_item_method_name
+ cls._entity = entity
+ cls._single_item_method_name = single_item_method_name
def __init__(self, openbis_obj, type=None, data=None, props=None, **kwargs):
- self.__dict__['openbis'] = openbis_obj
- self.__dict__['type'] = type
- self.__dict__['p'] = PropertyHolder(openbis_obj, type)
- self.__dict__['a'] = AttrHolder(openbis_obj, self._entity, type)
+ self.__dict__["openbis"] = openbis_obj
+ self.__dict__["type"] = type
+ self.__dict__["p"] = PropertyHolder(openbis_obj, type)
+ self.__dict__["a"] = AttrHolder(openbis_obj, self._entity, type)
# existing OpenBIS object
if data is not None:
@@ -38,88 +38,87 @@ class OpenBisObject():
for key in kwargs:
setattr(self, key, kwargs[key])
-
def __dir__(self):
defs = get_definition_for_entity(self.entity)
if self.is_new:
- return defs['attrs_new']
+ return defs["attrs_new"]
else:
- return list(set(defs['attrs'] + defs['attrs_up']))
-
+ return list(set(defs["attrs"] + defs["attrs_up"]))
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
+ self.__dict__["data"] = data
# put the properties in the self.p namespace (without checking them)
- if 'properties' in data:
- for key, value in data['properties'].items():
+ if "properties" in data:
+ for key, value in data["properties"].items():
self.p.__dict__[key.lower()] = value
-
+
# object is already saved to openBIS, so it is not new anymore
- self.a.__dict__['_is_new'] = False
+ self.a.__dict__["_is_new"] = False
@property
def attrs(self):
- return self.__dict__['a']
+ return self.__dict__["a"]
@property
def space(self):
try:
- return self.openbis.get_space(self._space['permId'])
+ return self.openbis.get_space(self._space["permId"])
except Exception:
pass
@property
def project(self):
try:
- return self.openbis.get_project(self._project['identifier'])
+ return self.openbis.get_project(self._project["identifier"])
except Exception:
pass
@property
def experiment(self):
try:
- return self.openbis.get_experiment(self._experiment['identifier'])
+ return self.openbis.get_experiment(self._experiment["identifier"])
except Exception:
pass
+
collection = experiment # Alias
@property
def sample(self):
try:
- return self.openbis.get_sample(self._sample['identifier'])
+ return self.openbis.get_sample(self._sample["identifier"])
except Exception:
pass
- object = sample # Alias
+
+ object = sample # Alias
@property
def _permId(self):
try:
- return self.data['permId']
+ return self.data["permId"]
except Exception:
return ""
@property
def permId(self):
try:
- return self.data['permId']['permId']
+ return self.data["permId"]["permId"]
except Exception:
try:
- return self.a.__dict__['_permId']['permId']
+ return self.a.__dict__["_permId"]["permId"]
except Exception:
return ""
-
def __getattr__(self, name):
- return getattr(self.__dict__['a'], name)
+ return getattr(self.__dict__["a"], name)
def __setattr__(self, name, value):
- if name in ['set_properties', 'set_tags', 'add_tags']:
+ 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)
+ setattr(self.__dict__["a"], name, value)
def _repr_html_(self):
"""Print all the assigned attributes (identifier, tags, etc.) in a nicely formatted table. See
@@ -128,18 +127,17 @@ class OpenBisObject():
return self.a._repr_html_()
def __repr__(self):
- """same thing as _repr_html_() but for IPython
- """
+ """same thing as _repr_html_() but for IPython"""
return self.a.__repr__()
def mark_to_be_deleted(self):
- self.__dict__['mark_to_be_deleted'] = True
+ self.__dict__["mark_to_be_deleted"] = True
def unmark_to_be_deleted(self):
- self.__dict__['mark_to_be_deleted'] = False
+ self.__dict__["mark_to_be_deleted"] = False
def is_marked_to_be_deleted(self):
- return self.__dict__.get('mark_to_be_deleted', False)
+ return self.__dict__.get("mark_to_be_deleted", False)
def delete(self, reason):
"""Delete this openbis entity.
@@ -147,43 +145,38 @@ class OpenBisObject():
"""
if not self.data:
return
-
+
self.openbis.delete_openbis_entity(
- entity=self._entity,
- objectId=self.data['permId'],
- reason=reason
- )
- if VERBOSE: print(
- "{} {} successfully deleted.".format(
- self._entity,
- self.permId
- )
+ entity=self._entity, objectId=self.data["permId"], reason=reason
)
+ if VERBOSE:
+ print("{} {} successfully deleted.".format(self._entity, self.permId))
def _get_single_item_method(self):
single_item_method = None
if self._single_item_method_name:
- single_item_method = getattr(
- self.openbis, self._single_item_method_name
- )
+ single_item_method = getattr(self.openbis, self._single_item_method_name)
else:
# try to guess the method...
- single_item_method = getattr(self.openbis, 'get_' + self.entity)
+ single_item_method = getattr(self.openbis, "get_" + self.entity)
return single_item_method
-
def save(self):
get_single_item = self._get_single_item_method()
# check for mandatory properties before saving the object
props = None
if self.props:
for prop_name, prop in self.props._property_names.items():
- if prop['mandatory']:
- if getattr(self.props, prop_name) is None \
- or getattr(self.props, prop_name) == "":
+ if prop["mandatory"]:
+ if (
+ getattr(self.props, prop_name) is None
+ or getattr(self.props, prop_name) == ""
+ ):
raise ValueError(
- "Property '{}' is mandatory and must not be None".format(prop_name)
+ "Property '{}' is mandatory and must not be None".format(
+ prop_name
+ )
)
props = self.p._all_props()
@@ -191,22 +184,27 @@ class OpenBisObject():
# NEW
if self.is_new:
request = self._new_attrs()
- if props: request["params"][1][0]["properties"] = props
+ if props:
+ request["params"][1][0]["properties"] = props
resp = self.openbis._post_request(self.openbis.as_v3, request)
- if VERBOSE: print("{} successfully created.".format(self.entity))
- new_entity_data = get_single_item(resp[0]['permId'], only_data=True)
+ if VERBOSE:
+ print("{} successfully created.".format(self.entity))
+ new_entity_data = get_single_item(resp[0]["permId"], only_data=True)
self._set_data(new_entity_data)
return self
# UPDATE
else:
request = self._up_attrs(method_name=None, permId=self._permId)
- if props: request["params"][1][0]["properties"] = props
+
+ if props:
+ request["params"][1][0]["properties"] = props
resp = self.openbis._post_request(self.openbis.as_v3, request)
- if VERBOSE: print("{} successfully updated.".format(self.entity))
+ if VERBOSE:
+ print("{} successfully updated.".format(self.entity))
new_entity_data = get_single_item(self.permId, only_data=True)
self._set_data(new_entity_data)
return self
@@ -216,7 +214,7 @@ class Transaction:
def __init__(self, *entities):
self.entities = {}
- self.reason = 'no reason'
+ self.reason = "no reason"
if not entities:
return
@@ -224,7 +222,6 @@ class Transaction:
for entity in entities:
self.add(entity)
-
def add(self, entity_obj):
"""Collect the creations or updates of entities.
self.entities = {
@@ -245,17 +242,16 @@ class Transaction:
if not entity_obj.entity in self.entities:
self.entities[entity_obj.entity] = defaultdict(list)
- mode = 'update'
+ mode = "update"
if entity_obj.is_new:
- mode = 'create'
+ mode = "create"
elif entity_obj.is_marked_to_be_deleted():
- mode = 'delete'
+ mode = "delete"
else:
- mode = 'update'
+ mode = "update"
self.entities[entity_obj.entity][mode].append(entity_obj)
-
def commit(self):
"""Merge the individual requests to Make one single request.
For each entity-type and mode (create, update) an individual request will be sent,
@@ -263,45 +259,52 @@ class Transaction:
"""
import copy
+
for entity_type in self.entities:
for mode in self.entities[entity_type]:
request_coll = []
for entity in self.entities[entity_type][mode]:
if mode == "delete":
- delete_options = get_type_for_entity(entity_type, 'delete')
- delete_options['reason'] = self.reason
- method = get_method_for_entity(entity_type, 'delete')
+ delete_options = get_type_for_entity(entity_type, "delete")
+ delete_options["reason"] = self.reason
+ method = get_method_for_entity(entity_type, "delete")
request = {
"method": method,
"params": [
entity.openbis.token,
- [
- entity.data['permId']
- ],
- delete_options
- ]
+ [entity.data["permId"]],
+ delete_options,
+ ],
}
request_coll.append(request)
continue
props = None
if entity.props:
for prop_name, prop in entity.props._property_names.items():
- if prop['mandatory']:
- if getattr(entity.props, prop_name) is None \
- or getattr(entity.props, prop_name) == "":
+ if prop["mandatory"]:
+ if (
+ getattr(entity.props, prop_name) is None
+ or getattr(entity.props, prop_name) == ""
+ ):
raise ValueError(
- "Property '{}' is mandatory and must not be None".format(prop_name)
+ "Property '{}' is mandatory and must not be None".format(
+ prop_name
+ )
)
props = entity.p._all_props()
if mode == "create":
request = entity._new_attrs()
- if props: request["params"][1][0]["properties"] = props
+ if props:
+ request["params"][1][0]["properties"] = props
elif mode == "update":
- request = entity._up_attrs(method_name=None, permId=entity._permId)
- if props: request["params"][1][0]["properties"] = props
+ request = entity._up_attrs(
+ method_name=None, permId=entity._permId
+ )
+ if props:
+ request["params"][1][0]["properties"] = props
else:
raise ValueError(f"Unkown mode: {mode}")
@@ -313,27 +316,32 @@ class Transaction:
batch_request = copy.deepcopy(request_coll[0])
# merge all requests into one
for i, request in enumerate(request_coll):
- if i == 0: continue
- batch_request['params'][1].append(
- request['params'][1][0]
- )
+ if i == 0:
+ continue
+ batch_request["params"][1].append(request["params"][1][0])
try:
- resp = entity.openbis._post_request(entity.openbis.as_v3, batch_request)
- if VERBOSE: print(f"{i+1} {entity_type}(s) {mode}d.")
+ resp = entity.openbis._post_request(
+ entity.openbis.as_v3, batch_request
+ )
+ if VERBOSE:
+ print(f"{i+1} {entity_type}(s) {mode}d.")
# mark every sample as not being new anymore
# and add the permId attribute received by the response
# we assume the response permIds are the same order as we sent them
if resp:
for i, resp_item in enumerate(resp):
- if mode == 'delete': continue
- self.entities[entity_type][mode][i].a.__dict__['_is_new'] = False
- self.entities[entity_type][mode][i].a.__dict__['_permId'] = resp_item
-
+ if mode == "delete":
+ continue
+ self.entities[entity_type][mode][i].a.__dict__[
+ "_is_new"
+ ] = False
+ self.entities[entity_type][mode][i].a.__dict__[
+ "_permId"
+ ] = resp_item
except ValueError as err:
- if VERBOSE:
+ if VERBOSE:
print(f"ERROR: {mode} of {i+1} {entity_type}(s) FAILED")
raise ValueError(err)
-