Skip to content
Snippets Groups Projects
pybis.py 166 KiB
Newer Older
  • Learn to ignore specific revisions
  • Swen Vermeul's avatar
    Swen Vermeul committed
                self._object_cache(entity="term", code=vocabulary, value=things)
    
    
        def new_term(self, code, vocabularyCode, label=None, description=None):
            return VocabularyTerm(
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                self,
                data=None,
                code=code,
                vocabularyCode=vocabularyCode,
                label=label,
                description=description,
    
        def get_term(self, code, vocabularyCode, only_data=False):
    
            search_request = {
                "code": code,
                "vocabularyCode": vocabularyCode,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                "@type": "as.dto.vocabulary.id.VocabularyTermPermId",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            fetchopts = get_fetchoption_for_entity("vocabularyTerm")
            for opt in ["registrator"]:
    
                fetchopts[opt] = get_fetchoption_for_entity(opt)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                "method": "getVocabularyTerms",
                "params": [self.token, [search_request], fetchopts],
    
            }
            resp = self._post_request(self.as_v3, request)
    
            if resp is None or len(resp) == 0:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                raise ValueError(
                    "no VocabularyTerm found with code='{}' and vocabularyCode='{}'".format(
                        code, vocabularyCode
                    )
                )
    
            else:
                parse_jackson(resp)
                for ident in resp:
                    if only_data:
                        return resp[ident]
                    else:
                        return VocabularyTerm(self, resp[ident])
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def get_vocabularies(self, code=None, start_with=None, count=None):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """Returns information about vocabulary"""
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            sub_criteria = []
            if code:
                sub_criteria.append(_criteria_for_code(code))
            criteria = {
                "criteria": sub_criteria,
                "@type": "as.dto.vocabulary.search.VocabularySearchCriteria",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                "operator": "AND",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            fetchopts = fetch_option["vocabulary"]
            fetchopts["from"] = start_with
            fetchopts["count"] = count
            for option in ["registrator"]:
    
                fetchopts[option] = fetch_option[option]
    
            request = {
                "method": "searchVocabularies",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                "params": [self.token, criteria, fetchopts],
    
            }
            resp = self._post_request(self.as_v3, request)
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            attrs = "code description managedInternally chosenFromList urlTemplate registrator registrationDate modificationDate".split()
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            if len(resp["objects"]) == 0:
    
                vocs = DataFrame(columns=attrs)
            else:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                objects = resp["objects"]
    
                parse_jackson(resp)
                vocs = DataFrame(objects)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                vocs["registrationDate"] = vocs["registrationDate"].map(format_timestamp)
                vocs["modificationDate"] = vocs["modificationDate"].map(format_timestamp)
                vocs["registrator"] = vocs["registrator"].map(extract_person)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                openbis_obj=self,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                entity="vocabulary",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                df=vocs[attrs],
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                identifier_name="code",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                start_with=start_with,
                count=count,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                totalCount=resp.get("totalCount"),
    
        def get_vocabulary(self, code, only_data=False, use_cache=True):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """Returns the details of a given vocabulary (including vocabulary terms)"""
    
            voc = (
                not only_data
                and use_cache
                and self._object_cache(entity="vocabulary", code=code)
            )
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            entity = "vocabulary"
            method_name = get_method_for_entity(entity, "get")
    
            objectIds = _type_for_id(code.upper(), entity)
            fetchopts = fetch_option[entity]
    
            request = {
                "method": method_name,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                "params": [self.token, [objectIds], fetchopts],
    
            }
            resp = self._post_request(self.as_v3, request)
    
            if len(resp) == 0:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                raise ValueError("no {} found with identifier: {}".format(entity, code))
    
            else:
                parse_jackson(resp)
                for ident in resp:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    data = resp[ident]
                    if only_data:
                        return data
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    vocabulary = Vocabulary(openbis_obj=self, data=data)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        self._object_cache(entity="vocabulary", code=code, value=vocabulary)
    
        def new_tag(self, code, description=None):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """Creates a new tag (for this user)"""
    
            return Tag(self, code=code, description=description)
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def get_tags(self, code=None, start_with=None, count=None):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """Returns a DataFrame of all tags"""
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            search_criteria = get_search_type_for_entity("tag", "AND")
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            fetchopts = fetch_option["tag"]
            fetchopts["from"] = start_with
            fetchopts["count"] = count
            for option in ["owner"]:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                fetchopts[option] = fetch_option[option]
    
            if code:
                criteria.append(_criteria_for_code(code))
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            search_criteria["criteria"] = criteria
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                "params": [self.token, search_criteria, fetchopts],
    
            resp = self._post_request(self.as_v3, request)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            return self._tag_list_for_response(
                response=resp["objects"], totalCount=resp["totalCount"]
            )
    
        def get_tag(self, permId, only_data=False, use_cache=True):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """Returns a specific tag"""
    
            just_one = True
            identifiers = []
            if isinstance(permId, list):
                just_one = False
                for ident in permId:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    identifiers.append(_type_for_id(ident, "tag"))
    
                tag = (
                    not only_data
                    and use_cache
                    and self._object_cache(entity="tag", code=permId)
                )
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                identifiers.append(_type_for_id(permId, "tag"))
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            fetchopts = fetch_option["tag"]
            for option in ["owner"]:
    
                fetchopts[option] = fetch_option[option]
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                "params": [self.token, identifiers, fetchopts],
    
            resp = self._post_request(self.as_v3, request)
    
    
            if just_one:
                if len(resp) == 0:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    raise ValueError("no such tag found: {}".format(permId))
    
                parse_jackson(resp)
                for permId in resp:
                    if only_data:
                        return resp[permId]
                    else:
    
                        tag = Tag(self, data=resp[permId])
                        if self.use_cache:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                            self._object_cache(entity="tag", code=permId, value=tag)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                return self._tag_list_for_response(response=list(resp.values()))
    
        def _tag_list_for_response(self, response, totalCount=0):
    
    
            parse_jackson(response)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            attrs = [
                "permId",
                "code",
                "description",
                "owner",
                "private",
                "registrationDate",
            ]
    
            if len(response) == 0:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                tags = DataFrame(columns=attrs)
            else:
    
                tags = DataFrame(response)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                tags["registrationDate"] = tags["registrationDate"].map(format_timestamp)
                tags["permId"] = tags["permId"].map(extract_permid)
                tags["description"] = tags["description"].map(
                    lambda x: "" if x is None else x
                )
                tags["owner"] = tags["owner"].map(extract_person)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                openbis_obj=self,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                entity="tag",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                df=tags[attrs],
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                identifier_name="permId",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                totalCount=totalCount,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def search_semantic_annotations(
            self, permId=None, entityType=None, propertyType=None, only_data=False
        ):
            """Get a list of semantic annotations for permId, entityType, propertyType or
    
            property type assignment (DataFrame object).
            :param permId: permId of the semantic annotation.
            :param entityType: entity (sample) type to search for.
            :param propertyType: property type to search for
            :param only_data: return result as plain data object.
            :return:  Things of DataFrame objects or plain data object
            """
    
    
            criteria = []
            typeCriteria = []
    
            if permId is not None:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                criteria.append(
                    {
                        "@type": "as.dto.common.search.PermIdSearchCriteria",
                        "fieldValue": {
                            "@type": "as.dto.common.search.StringEqualToValue",
                            "value": permId,
                        },
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                typeCriteria.append(
                    {
                        "@type": "as.dto.entitytype.search.EntityTypeSearchCriteria",
                        "criteria": [_criteria_for_code(entityType)],
                    }
                )
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                typeCriteria.append(
                    {
                        "@type": "as.dto.property.search.PropertyTypeSearchCriteria",
                        "criteria": [_criteria_for_code(propertyType)],
                    }
                )
    
    
            if entityType is not None and propertyType is not None:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                criteria.append(
                    {
                        "@type": "as.dto.property.search.PropertyAssignmentSearchCriteria",
                        "criteria": typeCriteria,
                    }
                )
    
            else:
                criteria += typeCriteria
    
            saCriteria = {
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                "@type": "as.dto.semanticannotation.search.SemanticAnnotationSearchCriteria",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                "criteria": criteria,
    
            }
    
            objects = self._search_semantic_annotations(saCriteria)
    
            if only_data:
                return objects
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            attrs = [
                "permId",
                "entityType",
                "propertyType",
                "predicateOntologyId",
                "predicateOntologyVersion",
                "predicateAccessionId",
                "descriptorOntologyId",
                "descriptorOntologyVersion",
                "descriptorAccessionId",
                "creationDate",
            ]
    
            if len(objects) == 0:
                annotations = DataFrame(columns=attrs)
    
            else:
                annotations = DataFrame(objects)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                openbis_obj=self,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                entity="semantic_annotation",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                df=annotations[attrs],
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                identifier_name="permId",
    
        def _search_semantic_annotations(self, criteria):
    
            fetch_options = {
                "@type": "as.dto.semanticannotation.fetchoptions.SemanticAnnotationFetchOptions",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                "entityType": {
                    "@type": "as.dto.entitytype.fetchoptions.EntityTypeFetchOptions"
                },
                "propertyType": {
                    "@type": "as.dto.property.fetchoptions.PropertyTypeFetchOptions"
                },
    
                "propertyAssignment": {
                    "@type": "as.dto.property.fetchoptions.PropertyAssignmentFetchOptions",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    "entityType": {
                        "@type": "as.dto.entitytype.fetchoptions.EntityTypeFetchOptions"
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    "propertyType": {
                        "@type": "as.dto.property.fetchoptions.PropertyTypeFetchOptions"
    
            }
    
            request = {
                "method": "searchSemanticAnnotations",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                "params": [self.token, criteria, fetch_options],
    
            }
    
            resp = self._post_request(self.as_v3, request)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            if len(resp["objects"]) == 0:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                objects = resp["objects"]
    
                for obj in objects:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    obj["permId"] = obj["permId"]["permId"]
                    if obj.get("entityType") is not None:
                        obj["entityType"] = obj["entityType"]["code"]
                    elif obj.get("propertyType") is not None:
                        obj["propertyType"] = obj["propertyType"]["code"]
                    elif obj.get("propertyAssignment") is not None:
                        obj["entityType"] = obj["propertyAssignment"]["entityType"]["code"]
                        obj["propertyType"] = obj["propertyAssignment"]["propertyType"][
                            "code"
                        ]
                    obj["creationDate"] = format_timestamp(obj["creationDate"])
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                return objects
    
    
        def get_semantic_annotations(self):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """Get a list of all available semantic annotations (DataFrame object)."""
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            objects = self._search_semantic_annotations(
                {
                    "@type": "as.dto.semanticannotation.search.SemanticAnnotationSearchCriteria"
                }
            )
            attrs = [
                "permId",
                "entityType",
                "propertyType",
                "predicateOntologyId",
                "predicateOntologyVersion",
                "predicateAccessionId",
                "descriptorOntologyId",
                "descriptorOntologyVersion",
                "descriptorAccessionId",
                "creationDate",
            ]
    
            if len(objects) == 0:
                annotations = DataFrame(columns=attrs)
            else:
                annotations = DataFrame(objects)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                openbis_obj=self,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                entity="semantic_annotation",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                df=annotations[attrs],
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                identifier_name="permId",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def get_semantic_annotation(self, permId, only_data=False):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            objects = self.search_semantic_annotations(permId=permId, only_data=True)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                raise ValueError(
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    "Semantic annotation with permId " + permId + " not found."
                )
    
                return SemanticAnnotation(self, isNew=False, **obj)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def get_plugins(self, start_with=None, count=None):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
    
            criteria = []
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            search_criteria = get_search_type_for_entity("plugin", "AND")
            search_criteria["criteria"] = criteria
    
    Swen Vermeul's avatar
    Swen Vermeul committed
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            fetchopts = fetch_option["plugin"]
            for option in ["registrator"]:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                fetchopts[option] = fetch_option[option]
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            fetchopts["from"] = start_with
            fetchopts["count"] = count
    
    Swen Vermeul's avatar
    Swen Vermeul committed
    
            request = {
                "method": "searchPlugins",
                "params": [
                    self.token,
                    search_criteria,
                    fetchopts,
                ],
            }
            resp = self._post_request(self.as_v3, request)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            attrs = [
                "name",
                "description",
                "pluginType",
                "pluginKind",
                "entityKinds",
                "registrator",
                "registrationDate",
                "permId",
            ]
    
    Swen Vermeul's avatar
    Swen Vermeul committed
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            if len(resp["objects"]) == 0:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                plugins = DataFrame(columns=attrs)
            else:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                objects = resp["objects"]
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                parse_jackson(objects)
    
                plugins = DataFrame(objects)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                plugins["permId"] = plugins["permId"].map(extract_permid)
                plugins["registrator"] = plugins["registrator"].map(extract_person)
                plugins["registrationDate"] = plugins["registrationDate"].map(
                    format_timestamp
                )
                plugins["description"] = plugins["description"].map(
                    lambda x: "" if x is None else x
                )
                plugins["entityKinds"] = plugins["entityKinds"].map(
                    lambda x: "" if x is None else x
                )
    
    Swen Vermeul's avatar
    Swen Vermeul committed
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                openbis_obj=self,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                entity="plugin",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                df=plugins[attrs],
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                identifier_name="name",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                start_with=start_with,
                count=count,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                totalCount=resp.get("totalCount"),
    
    Swen Vermeul's avatar
    Swen Vermeul committed
    
        def get_plugin(self, permId, only_data=False, with_script=True):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            search_request = _type_for_id(permId, "plugin")
            fetchopts = fetch_option["plugin"]
            options = ["registrator"]
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            if with_script:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                options.append("script")
    
    Swen Vermeul's avatar
    Swen Vermeul committed
    
            for option in options:
                fetchopts[option] = fetch_option[option]
    
            request = {
                "method": "getPlugins",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                "params": [self.token, [search_request], fetchopts],
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            }
    
            resp = self._post_request(self.as_v3, request)
            parse_jackson(resp)
    
            if resp is None or len(resp) == 0:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                raise ValueError("no such plugin found: " + permId)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            else:
                for permId in resp:
                    if only_data:
                        return resp[permId]
                    else:
                        return Plugin(self, data=resp[permId])
    
    
        def new_plugin(self, name, pluginType, **kwargs):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """Creates a new Plugin in openBIS.
    
            name        -- name of the plugin
            description --
            pluginType  -- DYNAMIC_PROPERTY, MANAGED_PROPERTY, ENTITY_VALIDATION
            entityKind  -- MATERIAL, EXPERIMENT, SAMPLE, DATA_SET
            script      -- string of the script itself
            available   --
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            return Plugin(self, name=name, pluginType=pluginType, **kwargs)
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def new_property_type(
            self,
            code,
            label,
            description,
            dataType,
            managedInternally=False,
            vocabulary=None,
            materialType=None,
            schema=None,
            transformation=None,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        ):
            """Creates a new property type.
    
    
            code               -- name of the property type
    
            managedInternally  -- must be set to True if code starts with a $
    
            label              -- displayed label of that property
            description        --
            dataType           -- must contain any of these values:
                                  INTEGER VARCHAR MULTILINE_VARCHAR
                                  REAL TIMESTAMP BOOLEAN HYPERLINK
                                  XML CONTROLLEDVOCABULARY MATERIAL
            vocabulary         -- if dataType is CONTROLLEDVOCABULARY, this attribute
                                  must contain the code of the vocabulary object.
            materialType       --
            schema             --
            transformation     --
    
            metaData           -- used to create properties that contain either RichText or tabular, spreadsheet-like data.
                                  use {'custom_widget' : 'Word Processor'} and MULTILINE_VARCHAR for RichText
                                  use {'custom_widget' : 'Spreadhseet'} and XML for tabular data.
    
            PropertyTypes can be assigned to
            - sampleTypes
            - dataSetTypes
            - experimentTypes
            - materialTypes (deprecated)
            """
    
    
            return PropertyType(
                openbis_obj=self,
                code=code,
                label=label,
                description=description,
                dataType=dataType,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                managedInternally=managedInternally,
                vocabulary=vocabulary,
                materialType=materialType,
                schema=schema,
                transformation=transformation,
    
                metaData=metaData,
    
        def get_property_type(
            self, code, only_data=False, start_with=None, count=None, use_cache=True
        ):
    
            if not isinstance(code, list) and start_with is None and count is None:
                code = str(code).upper()
    
                pt = (
                    use_cache
                    and self.use_cache
                    and self._object_cache(entity="property_type", code=code)
    
                if pt:
                    if only_data:
                        return pt.data
                    else:
                        return pt
    
            only_one = False
            if not isinstance(code, list):
                code = [code]
                only_one = True
    
            for c in code:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                identifiers.append(
                    {"permId": c.upper(), "@type": "as.dto.property.id.PropertyTypePermId"}
                )
    
            fetchopts = fetch_option["propertyType"]
            options = ["vocabulary", "materialType", "semanticAnnotations", "registrator"]
    
            for option in options:
                fetchopts[option] = fetch_option[option]
    
            request = {
                "method": "getPropertyTypes",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                "params": [self.token, identifiers, fetchopts],
    
            }
    
            resp = self._post_request(self.as_v3, request)
            parse_jackson(resp)
    
    
            if only_one:
                if len(resp) == 0:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    raise ValueError("no such propertyType: {}".format(code))
    
                for ident in resp:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        pt = PropertyType(openbis_obj=self, data=resp[ident])
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                            self._object_cache(
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                                entity="property_type", code=code[0], value=pt
                            )
    
            # return a list of objects
            else:
                return self._property_type_things(
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    objects=list(resp.values()),
                    start_with=start_with,
                    count=count,
                    totalCount=len(resp),
    
        def get_property_types(self, code=None, start_with=None, count=None):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            fetchopts = fetch_option["propertyType"]
            fetchopts["from"] = start_with
            fetchopts["count"] = count
            search_criteria = get_search_criteria("propertyType", code=code)
    
    
            request = {
                "method": "searchPropertyTypes",
                "params": [
                    self.token,
                    search_criteria,
                    fetchopts,
                ],
            }
    
            resp = self._post_request(self.as_v3, request)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            objects = resp["objects"]
    
            parse_jackson(objects)
            return self._property_type_things(
                objects=objects,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                start_with=start_with,
                count=count,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                totalCount=resp.get("totalCount"),
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def _property_type_things(
            self, objects, start_with=None, count=None, totalCount=None
        ):
            """takes a list of objects and returns a Things object"""
            attrs = openbis_definitions("propertyType")["attrs"]
    
            if len(objects) == 0:
                df = DataFrame(columns=attrs)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                df["registrationDate"] = df["registrationDate"].map(format_timestamp)
                df["registrator"] = df["registrator"].map(extract_person)
                df["vocabulary"] = df["vocabulary"].map(extract_code)
                df["semanticAnnotations"] = df["semanticAnnotations"].map(
                    extract_nested_permids
                )
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                openbis_obj=self,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                entity="propertyType",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                single_item_method=self.get_property_type,
                df=df[attrs],
                start_with=start_with,
                count=count,
                totalCount=totalCount,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
    
    
        def get_material_types(self, type=None, start_with=None, count=None):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """Returns a list of all available material types"""
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                entity="materialType",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                cls=MaterialType,
                type=type,
                start_with=start_with,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                count=count,
    
        def get_material_type(self, type, only_data=False):
            return self.get_entity_type(
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                entity="materialType",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                cls=MaterialType,
                identifier=type,
                method=self.get_material_type,
                only_data=only_data,
    
        def get_experiment_types(self, type=None, start_with=None, count=None):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """Returns a list of all available experiment types"""
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                entity="experimentType",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                cls=ExperimentType,
                type=type,
                start_with=start_with,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                count=count,
    
        get_collection_types = get_experiment_types  # Alias
    
    
        def get_experiment_type(self, type, only_data=False):
            return self.get_entity_type(
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                entity="experimentType",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                cls=ExperimentType,
                identifier=type,
                method=self.get_experiment_type,
                only_data=only_data,
    
        get_collection_type = get_experiment_type  # Alias
    
    
        def get_dataset_types(self, type=None, start_with=None, count=None):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """Returns a list of all available dataSet types"""
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                entity="dataSetType",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                cls=DataSetType,
                type=type,
                start_with=start_with,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                count=count,
    
        def get_dataset_type(self, type, only_data=False):
            return self.get_entity_type(
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                entity="dataSetType",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                identifier=type,
                cls=DataSetType,
                method=self.get_dataset_type,
                only_data=only_data,
    
        def get_sample_types(self, type=None, start_with=None, count=None):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """Returns a list of all available sample types"""
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                entity="sampleType",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                cls=SampleType,
                type=type,
                start_with=start_with,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                count=count,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        get_object_types = get_sample_types  # Alias
    
        def get_sample_type(self, type, only_data=False, with_vocabulary=False):
    
            return self.get_entity_type(
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                entity="sampleType",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                identifier=type,
                cls=SampleType,
                with_vocabulary=with_vocabulary,
                method=self.get_sample_type,
                only_data=only_data,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        get_object_type = get_sample_type  # Alias
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            self, entity, cls, type=None, start_with=None, count=None, with_vocabulary=False
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            method_name = get_method_for_entity(entity, "search")
    
            if type is not None:
                search_request = _subcriteria_for_code(type, entity)
            else:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                search_request = get_type_for_entity(entity, "search")
    
    
            fetch_options = get_fetchoption_for_entity(entity)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            fetch_options["from"] = start_with
            fetch_options["count"] = count
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                "params": [self.token, search_request, fetch_options],
    
            }
            resp = self._post_request(self.as_v3, request)
            parse_jackson(resp)
    
            entity_types = []
            defs = get_definition_for_entity(entity)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            attrs = defs["attrs"]
            if len(resp["objects"]) == 0:
    
                entity_types = DataFrame(columns=attrs)
            else:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                objects = resp["objects"]
    
                parse_jackson(objects)
                entity_types = DataFrame(objects)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                entity_types["permId"] = entity_types["permId"].map(extract_permid)
                entity_types["modificationDate"] = entity_types["modificationDate"].map(
                    format_timestamp
                )
                entity_types["validationPlugin"] = entity_types["validationPlugin"].map(
                    extract_nested_permid
                )
    
    
            single_item_method = getattr(self, cls._single_item_method_name)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                openbis_obj=self,
                entity=entity,
                df=entity_types[attrs],
                start_with=start_with,
                single_item_method=single_item_method,
                count=count,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                totalCount=resp.get("totalCount"),
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def get_entity_type(
            self,
            entity,
            identifier,
            cls,
            method=None,
            only_data=False,
            with_vocabulary=False,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        ):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            et = (
                not only_data
                and not isinstance(identifier, list)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                and self._object_cache(entity=entity, code=identifier)
            )
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            method_name = get_method_for_entity(entity, "get")
    
            fetch_options = get_fetchoption_for_entity(entity)
    
            if with_vocabulary:
                fetch_options["propertyAssignments"]["propertyType"]["vocabulary"] = {
                    "@type": "as.dto.vocabulary.fetchoptions.VocabularyFetchOptions",
                    "terms": {
                        "@type": "as.dto.vocabulary.fetchoptions.VocabularyTermFetchOptions"
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    },
    
    Swen Vermeul's avatar
    Swen Vermeul committed
    
    
            if not isinstance(identifier, list):
                identifier = [identifier]
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            identifiers = []
    
            for ident in identifier:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                identifiers.append(
                    {
                        "permId": ident,
                        "@type": "as.dto.entitytype.id.EntityTypePermId",
                    }
                )
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                "params": [self.token, identifiers, fetch_options],
    
            resp = self._post_request(self.as_v3, request)
            parse_jackson(resp)
            if len(identifiers) == 1:
                if len(resp) == 0:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    raise ValueError("no such {}: {}".format(entity, identifier[0]))
    
            for ident in resp:
                if only_data:
                    return resp[ident]
                else:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        openbis_obj=self,
                        data=resp[ident],
                        method=method,
    
                    if self.use_cache:
                        self._object_cache(entity=entity, code=ident, value=obj)
                    return obj
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def _get_types_of(
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            self,
            method_name,
            entity,
            type_name=None,
            start_with=None,
            count=None,
            additional_attributes=None,
            optional_attributes=None,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        ):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """Returns a list of all available types of an entity.
    
            If the name of the entity-type is given, it returns a PropertyAssignments object
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            if additional_attributes is None:
                additional_attributes = []
    
            if optional_attributes is None:
                optional_attributes = []
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            search_request = {
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                "@type": "as.dto.{}.search.{}TypeSearchCriteria".format(
                    entity.lower(), entity
                )
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            }
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            fetch_options = {
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                "@type": "as.dto.{}.fetchoptions.{}TypeFetchOptions".format(
                    entity.lower(), entity
                )
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            }
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            fetch_options["from"] = start_with
            fetch_options["count"] = count
    
            if type_name is not None:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                search_request = _gen_search_criteria(
                    {entity.lower(): entity + "Type", "operator": "AND", "code": type_name}
                )
                fetch_options["propertyAssignments"] = fetch_option["propertyAssignments"]
                if self.get_server_information().api_version > "3.3":
                    fetch_options["validationPlugin"] = fetch_option["plugin"]
    
            request = {
                "method": method_name,
    
                "params": [self.token, search_request, fetch_options],
    
            }
            resp = self._post_request(self.as_v3, request)
    
            if type_name is not None:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                if len(resp["objects"]) == 1:
                    return EntityType(openbis_obj=self, data=resp["objects"][0])
                elif len(resp["objects"]) == 0:
                    raise ValueError("No such {} type: {}".format(entity, type_name))
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    raise ValueError(
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        "There is more than one entry for entity={} and type={}".format(
                            entity, type_name
                        )
                    )
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            attrs = self._get_attributes(
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                type_name, types, additional_attributes, optional_attributes
            )
            if len(resp["objects"]) == 0:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                objects = resp["objects"]
    
                parse_jackson(objects)
                types = DataFrame(objects)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                types["modificationDate"] = types["modificationDate"].map(format_timestamp)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                openbis_obj=self,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                entity=entity.lower() + "_type",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                df=types[attrs],
                start_with=start_with,
                count=count,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                totalCount=resp.get("totalCount"),
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def _get_attributes(
            self, type_name, types, additional_attributes, optional_attributes
        ):
            attributes = ["code", "description"] + additional_attributes
            attributes += [
                attribute for attribute in optional_attributes if attribute in types
            ]
            attributes += ["modificationDate"]
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                attributes += ["propertyAssignments"]
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """checks whether a session is still active. Returns true or false."""
    
            self._save_token(token)
    
            This method is useful to check if a token is still valid or if it has timed out,
            requiring the user to login again.
    
            :return: Return True if the token is valid, False if it is not valid.
            """
    
            request = {
                "method": "isSessionActive",
    
            try:
                resp = self._post_request(self.as_v1, request)
    
            except Exception:
    
            return resp
    
        def set_token(self, token, save_token=True):
            """Checks the validity of a token, sets it as the current token and (by default) saves it
            to the disk, i.e. in the ~/.pybis directory
            """
            if not self.is_token_valid(token):
                raise ValueError("session token seems not to be valid.")
    
            self._save_token(token=token)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            # TODO: find out what this is good for
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            if os.environ.get("OPENBIS_URL") == self.url:
                os.environ["OPENBIS_TOKEN"] = self.token
    
        def get_dataset(self, permIds, only_data=False, props=None, **kvals):