Skip to content
Snippets Groups Projects
attribute.py 39 KiB
Newer Older
  • Learn to ignore specific revisions
  • #   Copyright ETH 2018 - 2023 Zürich, Scientific IT Services
    # 
    #   Licensed under the Apache License, Version 2.0 (the "License");
    #   you may not use this file except in compliance with the License.
    #   You may obtain a copy of the License at
    # 
    #        http://www.apache.org/licenses/LICENSE-2.0
    #   
    #   Unless required by applicable law or agreed to in writing, software
    #   distributed under the License is distributed on an "AS IS" BASIS,
    #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    #   See the License for the specific language governing permissions and
    #   limitations under the License.
    #
    
    from pandas import DataFrame, Series
    from tabulate import tabulate
    
    Swen Vermeul's avatar
    Swen Vermeul committed
    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
    
    Swen Vermeul's avatar
    Swen Vermeul committed
    
    class AttrHolder:
        """General class for both samples and experiments that hold all common attributes, such as:
    
        - space
        - project
        - experiment (sample)
        - samples (experiment)
        - dataset
        - tags
        """
    
        def __init__(self, openbis_obj, entity, type=None):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            self.__dict__["_openbis"] = openbis_obj
            self.__dict__["_entity"] = entity
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                self.__dict__["_type"] = type.data
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            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.
            Instead of invoking a special method we «call» the object with the data
               self(data)
            which automatically invokes this method.
            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.
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                sample._space = {
    
                    '@type': 'as.dto.space.id.SpacePermId',
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    '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
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            self.__dict__["_is_new"] = False
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            for attr in self._defs["attrs"]:
    
                if attr in ["code", "permId", "identifier", "type"]:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    self.__dict__["_" + attr] = data.get(attr, None)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    if isinstance(self.__dict__["_" + attr], dict):
                        self.__dict__["_" + attr].pop("@id", None)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                elif attr in ["vocabularyCode"]:
                    self.__dict__["_" + attr] = data.get("permId", {}).get(attr, None)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                elif attr in ["validationPlugin"]:
    
                    d = data.get(attr, None)
                    if d is not None:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        d = d["permId"]
                    self.__dict__["_" + attr] = d
    
                elif attr in ["space"]:
                    d = data.get(attr, None)
                    if d is not None:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        d = d["permId"]
                    self.__dict__["_" + attr] = d
    
                elif attr in ["sample", "experiment", "project", "container"]:
    
                    d = data.get(attr, None)
                    if d is not None:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        d = d["identifier"]
                    self.__dict__["_" + attr] = d
    
                elif attr in ["parents", "children", "samples", "components", "containers"]:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    self.__dict__["_" + attr] = []
    
                    if data[attr] is not None:
                        for item in data[attr]:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                                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
    
                    self.add_tags(data[attr])
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                elif attr.endswith("Date"):
                    self.__dict__["_" + attr] = format_timestamp(data.get(attr))
    
                elif attr in ["registrator", "modifier", "dataProducer", "owner"]:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    self.__dict__["_" + attr] = extract_person(data.get(attr))
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    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.
            """
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            attr2ids = openbis_definitions("attr2ids")
            new_obj = get_type_for_entity(self.entity, "create")
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            for attr in self._defs["attrs_new"]:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                if attr == "type":
                    new_obj["typeId"] = self._type["permId"]
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                elif attr == "kind":
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    # when creating a new dataset, the attribute «kind» is called «dataSetKind»
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    new_obj["dataSetKind"] = self._kind
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                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
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                elif attr == "userIds":
                    if "_changed_users" not in self.__dict__:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    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"}
                            )
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    items = getattr(self, "_" + attr)
    
    
                    key = None
                    if attr in attr2ids:
                        # translate parents into parentIds, children into childIds etc.
                        key = attr2ids[attr]
                    else:
                        key = attr
    
                    new_obj[key] = items
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            # if method_name is not defined: guess the method name for creating a new entity
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                method_name = get_method_for_entity(self.entity, "create")
            request = {"method": method_name, "params": [self.openbis.token, [new_obj]]}
    
        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.
            """
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            # defs = openbis_definitions(self._entity)
            attr2ids = openbis_definitions("attr2ids")
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            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...
    
            identifier_name = self._defs["identifier"]
    
            if permId:
                up_obj[identifier_name] = permId
            else:
                up_obj[identifier_name] = self._permId
    
    
            # look at all attributes available for that entity
            # that can be updated
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            for attr in self._defs["attrs_up"]:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                if attr == "attachments":
    
                    # v3 API currently only supports adding attachments
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    attachments = self.__dict__.get("_new_attachments", None)
    
                    if attachments is None:
                        continue
                    atts_data = [attachment.get_data() for attachment in attachments]
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    up_obj["attachments"] = {
                        "actions": [
                            {
                                "items": atts_data,
                                "@type": "as.dto.common.update.ListUpdateActionAdd",
                            }
                        ],
                        "@type": "as.dto.attachment.update.AttachmentListUpdateValue",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                elif attr == "tags":
    
                    items = []
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    for tag in self.__dict__["_tags"]:
                        items.append(
                            {"permId": tag["permId"], "@type": "as.dto.tag.id.TagPermId"}
                        )
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    up_obj["tagIds"] = {
    
                        "actions": [
                            {
                                "items": items,
                                "@type": "as.dto.common.update.ListUpdateActionSet",
                            }
                        ],
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        "@type": "as.dto.common.update.IdListUpdateValue",
    
                elif attr == "metaData":
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    # 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}],
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        "@type": "as.dto.common.update.ListUpdateMapValues",
                    }
                elif attr == "userIds":
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    if "_changed_users" not in self.__dict__:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    for userId in self.__dict__["_changed_users"]:
                        actions.append(
                            {
                                "items": [
                                    {
                                        "permId": userId,
                                        "@type": "as.dto.person.id.PersonPermId",
                                    }
                                ],
    
                                "@type": f'as.dto.common.update.ListUpdateAction{self.__dict__["_changed_users"][userId]["action"]}',
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    up_obj["userIds"] = {
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        "@type": "as.dto.common.update.IdListUpdateValue",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                elif (
                    attr
                    in "description label official ordinal autoGeneratedCode subcodeUnique listable showContainer showParents showParentMetadata disallowDeletion validationPlugin".split()
                ):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    # alway update common fields
    
                    key = attr2ids.get(attr, attr)
                    up_obj[key] = {
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        "value": self.__dict__["_" + attr],
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        "@type": "as.dto.common.update.FieldUpdateValue",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                elif "_" + attr in self.__dict__:
    
                    # handle multivalue attributes (parents, children, tags etc.)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    # we only cover the Set mechanism, which means we always update
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    if "multi" in self._defs and attr in self._defs["multi"]:
                        items = self.__dict__.get("_" + attr, [])
    
                        if items is None:
                            # 'None' multivalue attributes should be omitted during update
                            continue
    
                        up_obj[attr2ids[attr]] = {
                            "actions": [
                                {
                                    "items": items,
                                    "@type": "as.dto.common.update.ListUpdateActionSet",
                                }
                            ],
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                            "@type": "as.dto.common.update.IdListUpdateValue",
    
                        }
                    else:
                        # handle single attributes (space, experiment, project, container, etc.)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        value = self.__dict__.get("_" + attr, {})
    
                        elif isinstance(value, bool):
    
                            # for boolean values where no type is needed
    
                            up_obj[attr] = value
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        elif isinstance(value, dict) and len(value) == 0:
    
                            # value is {}: it means that we want this attribute to be
                            # deleted, not updated.
                            up_obj[attr2ids[attr]] = {
                                "@type": "as.dto.common.update.FieldUpdateValue",
                                "isModified": True,
                            }
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        elif "isModified" in value and value["isModified"] == True:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                            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,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                                "value": val,
    
                            }
    
            # update an existing entity
            if method_name is None:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                method_name = get_method_for_entity(self.entity, "update")
            request = {"method": method_name, "params": [self.openbis.token, [up_obj]]}
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """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
                roleAssignments are returned as an array of dictionaries.
            """
    
            name_map = {
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                "group": "authorizationGroup",
                "roles": "roleAssignments",
                "permid": "permId",
                "collection": "experiment",
                "object": "sample",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            int_name = "_" + name
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                if int_name == "_attachments":
    
                    attachments = []
                    for att in self._attachments:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        attachments.append(
                            {
                                "fileName": att.get("fileName"),
                                "title": att.get("title"),
                                "description": att.get("description"),
                                "version": att.get("version"),
                            }
                        )
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                elif int_name == "_users":
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        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,
                            }
                        )
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                elif int_name == "_roleAssignments":
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        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"),
                            }
                        )
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                # 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:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                            values.append(item["identifier"])
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                            values.append(item["code"])
    
                        elif "userId" in item:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                            values.append(item["userId"])
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                            values.append(item["permId"])
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                            values.append(item)
    
                    return values
    
                # attribute contains a dictionary: same procedure as above.
                elif isinstance(self.__dict__[int_name], dict):
                    if "identifier" in self.__dict__[int_name]:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        return self.__dict__[int_name]["identifier"]
    
                    elif "code" in self.__dict__[int_name]:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        return self.__dict__[int_name]["code"]
    
                    elif "name" in self.__dict__[int_name]:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        return self.__dict__[int_name]["name"]
    
                    elif "userId" in self.__dict__[int_name]:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        return self.__dict__[int_name]["userId"]
    
                    elif "permId" in self.__dict__[int_name]:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        return self.__dict__[int_name]["permId"]
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        return self.__dict__[int_name]["id"]
    
                    else:
                        return self.__dict__[int_name]
    
    
                else:
                    return self.__dict__[int_name]
            else:
                return None
    
        def __setattr__(self, name, value):
            """This method is always invoked whenever we assign an attribute to an
            object, e.g.
                new_sample.space = 'MATERIALS'
                new_sample.parents = ['/MATERIALS/YEAST747']
            """
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                "collection": "experiment",
                "object": "sample",
                "parent": "parents",
                "child": "children",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                if name not in self._defs["attrs_new"]:
    
                    raise ValueError(
    
                        f'No such attribute: «{name}» for entity: {self.entity}. Allowed attributes are: {self._defs["attrs_new"]}'
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                if name not in self._defs["attrs_up"]:
    
                    raise ValueError(
    
                        f'No such attribute: «{name}» for entity: {self.entity}. Allowed attributes are: {self._defs["attrs_up"]}'
    
    
                if not isinstance(value, list):
                    value = [value]
                objs = []
                for val in value:
                    if isinstance(val, str):
                        # fetch objects in openBIS, make sure they actually exists
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        obj = getattr(self._openbis, "get_" + self._entity.lower())(val)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    elif getattr(val, "_permId"):
    
                        # we got an existing object
                        objs.append(val)
    
                permids = []
                for item in objs:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    if getattr(item, "_identifier") is not None:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    elif getattr(item, "_permId") is not None:
    
                    # remove any existing @id keys to prevent jackson parser errors
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    id.pop("@id", None)
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                self.__dict__["_" + name] = permids
    
            elif name in ["tags"]:
                self.set_tags(value)
    
            elif name in ["users"]:
                self.set_users(value)
    
    
            elif name in ["vocabulary"]:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                if value is None or value == "":
                    self.__dict__["_vocabulary"] = None
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    self.__dict__["_vocabulary"] = {
    
                        "@type": "as.dto.vocabulary.id.VocabularyPermId",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        "permId": value.upper(),
    
            elif name in ["validationPlugin"]:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                if value is None or value == "":
                    self.__dict__["_validationPlugin"] = None
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    self.__dict__["_validationPlugin"] = {
    
                        "@type": "as.dto.plugin.id.PluginPermId",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        "permId": value,
    
    
            elif name in ["materialType"]:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                if value is None or value == "":
                    self.__dict__["_materialType"] = None
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    self.__dict__["_materialType"] = {
    
                        "@type": "as.dto.entitytype.id.EntityTypePermId",
                        "permId": value.upper(),
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        "entityKind": "MATERIAL",
    
            elif name in ["attachments"]:
                if isinstance(value, list):
                    for item in value:
                        if isinstance(item, dict):
                            self.add_attachment(**item)
                        else:
                            self.add_attachment(item)
    
                else:
                    self.add_attachment(value)
    
            elif name in ["space"]:
                obj = None
                if value is None:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    self.__dict__["_" + name] = None
    
                    return
    
                if isinstance(value, str):
                    # fetch object in openBIS, make sure it actually exists
                    obj = getattr(self._openbis, "get_" + name)(value)
                else:
                    obj = value
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                self.__dict__["_" + name] = obj.data["permId"]
    
    
                # mark attribute as modified, if it's an existing entity
                if self.is_new:
                    pass
                else:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    self.__dict__["_" + name]["isModified"] = True
    
    
            elif name in ["sample", "experiment", "project"]:
                obj = None
                if isinstance(value, str):
                    # fetch object in openBIS, make sure it actually exists
                    obj = getattr(self._openbis, "get_" + name)(value)
                elif value is None:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    self.__dict__["_" + name] = {}
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                self.__dict__["_" + name] = obj.data["identifier"]
    
    
                # mark attribute as modified, if it's an existing entity
                if self.is_new:
                    pass
                else:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    self.__dict__["_" + name]["isModified"] = True
    
                raise KeyError(f"you can not modify the {name}")
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    if self._type["autoGeneratedCode"]:
                        raise KeyError(
    
                            f"This {self.entity}Type has auto-generated code. You cannot set a code"
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                self.__dict__["_code"] = value
    
    
            elif name in ["userId", "description"]:
                # values that are directly assigned
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                self.__dict__["_" + name] = value
    
            elif name in self._defs:
                # enum: check whether value is a valid constant
                if value and value not in self._defs[name]:
                    raise ValueError(
    
                        f"Allowed values for enum {name} are: {self._defs[name]}"
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    self.__dict__["_" + name] = value
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                self.__dict__["_" + name] = value
    
    
        def get_type(self):
            return self._type
    
        def _ident_for_whatever(self, whatever):
            if isinstance(whatever, str):
                # fetch parent in openBIS, we are given an identifier
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                obj = getattr(self._openbis, "get_" + self._entity.lower())(whatever)
    
            else:
                # we assume we got an object
                obj = whatever
    
            ident = None
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            if getattr(obj, "_identifier"):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            elif getattr(obj, "_permId"):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            ident.pop("@id", None)
    
        def get_container(self, **kwargs):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            return getattr(self._openbis, "get_" + self._entity.lower())(
                self.container, **kwargs
            )
    
    
        def get_containers(self, **kwargs):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """get the containers and return them as a list (Things/DataFrame)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """
            return getattr(self._openbis, "get_" + self._entity.lower())(
                self.containers, **kwargs
            )
    
    
        def set_containers(self, containers_to_set):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """set the new _containers list"""
            self.__dict__["_containers"] = []
    
            self.add_containers(containers_to_set)
    
        def add_containers(self, containers_to_add):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """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)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                if ident not in self.__dict__["_containers"]:
                    self.__dict__["_containers"].append(ident)
    
    
        def del_containers(self, containers_to_remove):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """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)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                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):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """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)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """
            return getattr(self._openbis, "get_" + self._entity.lower())(
                self.components, **kwargs
            )
    
        def set_components(self, components_to_set):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """Samples and DataSets may contain other DataSets and Samples. This function sets the
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """
            self.__dict__["_components"] = []
    
            self.add_components(components_to_set)
    
    
        def add_components(self, components_to_add):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """Samples and DataSets may contain other DataSets and Samples. This function adds
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """
    
            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)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                if ident not in self.__dict__["_components"]:
                    self.__dict__["_components"].append(ident)
    
        def del_components(self, components_to_remove):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """Samples and DataSets may contain other DataSets and Samples. This function removes
    
            additional Samples/DataSets from the current object.
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """
    
            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)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                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)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """get the current parents and return them as a list (Things/DataFrame)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """
            return getattr(self._openbis, "get_" + self._entity.lower())(
                self.parents, **kwargs
            )
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """set the new _parents list"""
            self.__dict__["_parents"] = []
    
            self.add_parents(parents_to_set)
    
        def add_parents(self, parents_to_add):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """add parent to _parents list"""
    
            if not isinstance(parents_to_add, list):
                parents_to_add = [parents_to_add]
    
            if self.__dict__["_parents"] is None:
                self.__dict__["_parents"] = []
    
            for parent in parents_to_add:
                ident = self._ident_for_whatever(parent)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                if ident not in self.__dict__["_parents"]:
                    self.__dict__["_parents"].append(ident)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """remove parent from _parents list"""
    
            if not isinstance(parents_to_remove, list):
                parents_to_remove = [parents_to_remove]
    
            if self.__dict__["_parents"] is None:
                self.__dict__["_parents"] = []
    
                ident = self._ident_for_whatever(parent)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                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)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """get the current children and return them as a list (Things/DataFrame)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """
            return getattr(self._openbis, "get_" + self._entity.lower())(
                self.children, **kwargs
            )
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """set the new _children list"""
            self.__dict__["_children"] = []
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """add children to _children list"""
            if getattr(self, "_children") is None:
                self.__dict__["_children"] = []
    
            if self.__dict__["_children"] is None:
                self.__dict__["_children"] = []
    
            if not isinstance(children, list):
                children = [children]
            for child in children:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                self.__dict__["_children"].append(self._ident_for_whatever(child))
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """remove children from _children list"""
            if getattr(self, "_children") is None:
    
                return
            if not isinstance(children, list):
                children = [children]
    
            if self.__dict__["_children"] is None:
                self.__dict__["_children"] = []
    
            for child in children:
                ident = self._ident_for_whatever(child)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                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)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            if getattr(self, "_tags") is not None:
                return [x["code"] for x in self._tags]
    
        def get_tags(self):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            if getattr(self, "_tags") is not None:
                return self._openbis.get_tag([x["permId"] for x in self._tags])
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """set _tags list"""
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            self.__dict__["_tags"] = []
    
            self.add_tags(tags)
    
        def add_tags(self, tags):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """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)
                    tag_dict = {
                        "code": tag_obj.code,
                        "permId": tag_obj.permId,
                    }
                else:
                    tag_dict = {
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        "code": tag["code"],
                        "permId": tag["permId"]["permId"],
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                if tag_dict not in self.__dict__["_tags"]:
                    self.__dict__["_tags"].append(tag_dict)
    
    
        def del_tags(self, tags):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """remove tags from _tags list"""
    
            if not isinstance(tags, list):
                tags = [tags]
    
            for tag in tags:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                for i, tag_dict in enumerate(self.__dict__["_tags"]):
                    if tag in tag_dict[i]["code"] or tag in tag_dict[i]["permId"]:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        tag_dict.pop(i, None)
    
    
        def set_users(self, userIds):
            if userIds is None:
                return
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            if getattr(self, "_userIds") is None:
                self.__dict__["_userIds"] = []
    
            if not isinstance(userIds, list):
                userIds = [userIds]
            for userId in userIds:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                self.__dict__["_userIds"].append(
                    {"permId": userId, "@type": "as.dto.person.id.PersonPermId"}
                )
    
    
        def add_users(self, userIds):
            if userIds is None:
                return
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            if getattr(self, "_changed_users") is None:
                self.__dict__["_changed_users"] = {}
    
    
            if not isinstance(userIds, list):
                userIds = [userIds]
            for userId in userIds:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                self.__dict__["_changed_users"][userId] = {"action": "Add"}
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        add_members = add_users  # Alias
    
    
        def del_users(self, userIds):
            if userIds is None:
                return
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            if getattr(self, "_changed_users") is None:
                self.__dict__["_changed_users"] = {}
    
    
            if not isinstance(userIds, list):
                userIds = [userIds]
            for userId in userIds:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                self.__dict__["_changed_users"][userId] = {"action": "Remove"}
    
    
        del_members = del_users  # Alias
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            if getattr(self, "_attachments") is None:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                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)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            if getattr(self, "_attachments") is None:
                self.__dict__["_attachments"] = []
    
            self._attachments.append(att.get_data_short())
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            if getattr(self, "_new_attachments") is None:
                self.__dict__["_new_attachments"] = []
    
        def download_attachments(self, destination_folder=None):
            method = "get" + self.entity.lower().capitalize() + "s"
    
            request = {
                "method": method,
                "params": [
                    self._openbis.token,
                    [self._permId],
    
                    {
                        **fetch_option[self.entity.lower()],
                        "attachments": fetch_option["attachmentsWithContent"],
                    },
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                ],
    
            }
            resp = self._openbis._post_request(self._openbis.as_v3, request)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            attachments = resp[self.permId]["attachments"]
    
    
            if destination_folder is None:
                destination_folder = os.getcwd()
    
                filename = (
                    pathlib.Path(destination_folder)
                    / pathlib.Path(attachment["fileName"]).name
    
                )
                os.makedirs(os.path.dirname(filename), exist_ok=True)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                with open(filename, "wb") as att:
                    content = base64.b64decode(attachment["content"])
    
                file_list.append(str(filename))
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """Return all attributes of an entity in a dict"""
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            for attr in self._defs["attrs"]:
                if attr == "attachments":
    
                    continue
                attrs[attr] = getattr(self, attr)
            return attrs
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            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>
            """
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            attrs = self._defs["attrs_new"] if self.is_new else self._defs["attrs"]