pybis.py

            project=project,
            data=None,
            props=props,
            code=code,
            **kwargs,
        )

    new_collection = new_experiment  # Alias

    def create_external_data_management_system(
            self, code, label, address, address_type="FILE_SYSTEM"
    ):
        """Create an external DMS.
        :param code: An openBIS code for the external DMS.
        :param label: A human-readable label.
        :param address: The address for accessing the external DMS. E.g., a URL.
        :param address_type: One of OPENBIS, URL, or FILE_SYSTEM
        :return:
        """
        request = {
            "method": "createExternalDataManagementSystems",
            "params": [
                self.token,
                [
                    {
                        "code": code,
                        "label": label,
                        "addressType": address_type,
                        "address": address,
                        "@type": "as.dto.externaldms.create.ExternalDmsCreation",
                    }
                ],
            ],
        }
        resp = self._post_request(self.as_v3, request)
        return self.get_external_data_management_system(resp[0]["permId"])

    def delete_entity(self, entity, id, reason, id_name="permId"):
        """Deletes Spaces, Projects, Experiments, Samples and DataSets"""

        type = get_type_for_entity(entity, "delete")
        method = get_method_for_entity(entity, "delete")
        request = {
            "method": method,
            "params": [
                self.token,
                [{id_name: id, "@type": type}],
                {"reason": reason, "@type": type},
            ],
        }
        self._post_request(self.as_v3, request)

    def delete_openbis_entity(self, entity, objectId, reason="No reason given"):
        method = get_method_for_entity(entity, "delete")
        delete_options = get_type_for_entity(entity, "delete")
        delete_options["reason"] = reason

        request = {"method": method, "params": [self.token, [objectId], delete_options]}
        return self._post_request(self.as_v3, request)

    def confirm_deletions(self, deletion_ids):
        request = {
            "method": "confirmDeletions",
            "params": [
                self.token,
                deletion_ids,
            ],
        }
        self._post_request(self.as_v3, request)

    def get_deletions(self, start_with=None, count=None):
        search_criteria = {"@type": "as.dto.deletion.search.DeletionSearchCriteria"}
        fetchopts = get_fetchoption_for_entity("deletion")
        fetchoptsDeleted = get_fetchoption_for_entity("deletedObjects")
        fetchoptsDeleted["from"] = start_with
        fetchoptsDeleted["count"] = count
        fetchopts["deletedObjects"] = fetchoptsDeleted

        request = {
            "method": "searchDeletions",
            "params": [
                self.token,
                search_criteria,
                fetchopts,
            ],
        }
        resp = self._post_request(self.as_v3, request)
        objects = resp["objects"]
        parse_jackson(objects)

        new_objs = []
        for value in objects:
            del_objs = extract_deletion(value)
            if len(del_objs) > 0:
                new_objs.append(*del_objs)

        return DataFrame(new_objs)

    def new_project(self, space, code, description=None, **kwargs):
        return Project(
            self, None, space=space, code=code, description=description, **kwargs
        )

    def _gen_fetchoptions(self, options, foType):
        fo = {"@type": foType}
        for option in options:
            fo[option] = get_fetchoption_for_entity(option)
        return fo

    def get_project(self, projectId, only_data=False, use_cache=True):
        """Returns a Project object for a given identifier, code or permId."""

        project = (
                not only_data
                and use_cache
                and self._object_cache(entity="project", code=projectId)
        )
        if project:
            return project

        options = ["space", "registrator", "modifier", "attachments"]
        if is_identifier(projectId) or is_permid(projectId):
            request = self._create_get_request(
                "getProjects",
                "project",
                projectId,
                options,
                "as.dto.project.fetchoptions.ProjectFetchOptions",
            )
            resp = self._post_request(self.as_v3, request)
            if len(resp) == 0:
                raise ValueError("No such project: %s" % projectId)
            if only_data:
                return resp[projectId]
            project = Project(openbis_obj=self, type=None, data=resp[projectId])
            if self.use_cache:
                self._object_cache(entity="project", code=projectId, value=project)
            return project

        else:
            search_criteria = _gen_search_criteria(
                {"project": "Project", "operator": "AND", "code": projectId}
            )
            fo = self._gen_fetchoptions(
                options, foType="as.dto.project.fetchoptions.ProjectFetchOptions"
            )
            request = {
                "method": "searchProjects",
                "params": [self.token, search_criteria, fo],
            }
            resp = self._post_request(self.as_v3, request)
            if len(resp["objects"]) == 0:
                raise ValueError("No such project: %s" % projectId)
            if only_data:
                return resp["objects"][0]

            project = Project(openbis_obj=self, type=None, data=resp["objects"][0])
            if self.use_cache:
                self._object_cache(entity="project", code=projectId, value=project)
            return project

    def get_projects(
            self,
            space=None,
            code=None,
            start_with=None,
            count=None,
    ):
        """Get a list of all available projects (DataFrame object)."""

        sub_criteria = []
        if space:
            sub_criteria.append(_subcriteria_for_code(space, "space"))
        if code:
            sub_criteria.append(_criteria_for_code(code))

        criteria = {
            "criteria": sub_criteria,
            "@type": "as.dto.project.search.ProjectSearchCriteria",
            "operator": "AND",
        }

        fetchopts = {"@type": "as.dto.project.fetchoptions.ProjectFetchOptions"}
        fetchopts["from"] = start_with
        fetchopts["count"] = count
        for option in ["registrator", "modifier", "leader"]:
            fetchopts[option] = get_fetchoption_for_entity(option)

        request = {
            "method": "searchProjects",
            "params": [
                self.token,
                criteria,
                fetchopts,
            ],
        }
        resp = self._post_request(self.as_v3, request)

        def create_data_frame(attrs, props, response):
            attrs = [
                "code",
                "identifier",
                "permId",
                "description",
                "leader",
                "registrator",
                "registrationDate",
                "modifier",
                "modificationDate",
                "frozen",
                "frozenForExperiments",
                "frozenForSamples",
            ]
            objects = response["objects"]
            if len(objects) == 0:
                projects = DataFrame(columns=attrs)
            else:
                parse_jackson(objects)

                projects = DataFrame(objects)

                projects["registrationDate"] = projects["registrationDate"].map(
                    format_timestamp
                )
                projects["modificationDate"] = projects["modificationDate"].map(
                    format_timestamp
                )
                projects["leader"] = projects["leader"].map(extract_person)
                projects["registrator"] = projects["registrator"].map(extract_person)
                projects["modifier"] = projects["modifier"].map(extract_person)
                projects["permId"] = projects["permId"].map(extract_permid)
                projects["identifier"] = projects["identifier"].map(extract_identifier)
            return projects[attrs]

        return Things(
            openbis_obj=self,
            entity="project",
            identifier_name="identifier",
            start_with=start_with,
            count=count,
            totalCount=resp.get("totalCount"),
            response=resp,
            df_initializer=create_data_frame,
        )

    def _create_get_request(self, method_name, entity, permids, options, foType):

        if not isinstance(permids, list):
            permids = [permids]

        type = f"as.dto.{entity.lower()}.id.{entity.capitalize()}"
        search_params = []
        for permid in permids:
            # decide if we got a permId or an identifier
            match = re.match("/", permid)
            if match:
                search_params.append(
                    {"identifier": permid, "@type": type + "Identifier"}
                )
            else:
                search_params.append({"permId": permid, "@type": type + "PermId"})

        fo = {"@type": foType}
        for option in options:
            fo[option] = get_fetchoption_for_entity(option)

        request = {
            "method": method_name,
            "params": [self.token, search_params, fo],
        }
        return request

    def clear_cache(self, entity=None):
        """Empty the internal object cache
        If you do not specify any entity, the complete cache is cleared.
        As entity, you can specify either:
        space, project, vocabulary, term, sampleType, experimentType, dataSetType
        """
        if entity:
            self.cache[entity] = {}
        else:
            self.cache = {}

    def _object_cache(self, entity=None, code=None, value=None):

        # return the value, if no value provided
        if value is None:
            if entity in self.cache:
                return self.cache[entity].get(code)
        else:
            if entity not in self.cache:
                self.cache[entity] = {}

            self.cache[entity][code] = value

    def get_terms(self, vocabulary=None, start_with=None, count=None, use_cache=True):
        """Returns information about existing vocabulary terms.
        If a vocabulary code is provided, it only returns the terms of that vocabulary.
        """

        if (
                use_cache
                and self.use_cache
                and vocabulary is not None
                and start_with is None
                and count is None
        ):
            voc = self._object_cache(entity="term", code=vocabulary)
            if voc:
                return voc

        search_request = {}
        if vocabulary is not None:
            search_request = _gen_search_criteria(
                {
                    "vocabulary": "VocabularyTerm",
                    "criteria": [{"vocabulary": "Vocabulary", "code": vocabulary}],
                }
            )
        search_request[
            "@type"
        ] = "as.dto.vocabulary.search.VocabularyTermSearchCriteria"

        fetchopts = get_fetchoption_for_entity("vocabularyTerm")
        fetchopts["from"] = start_with
        fetchopts["count"] = count

        request = {
            "method": "searchVocabularyTerms",
            "params": [self.token, search_request, fetchopts],
        }
        resp = self._post_request(self.as_v3, request)

        def create_data_frame(attrs, props, response):
            attrs = "code vocabularyCode label description registrationDate modificationDate official ordinal".split()

            objects = response["objects"]
            if len(objects) == 0:
                terms = DataFrame(columns=attrs)
            else:
                parse_jackson(objects)
                terms = DataFrame(objects)
                terms["vocabularyCode"] = terms["permId"].map(
                    extract_attr("vocabularyCode")
                )
                terms["registrationDate"] = terms["registrationDate"].map(
                    format_timestamp
                )
                terms["modificationDate"] = terms["modificationDate"].map(
                    format_timestamp
                )
            return terms[attrs]

        things = Things(
            openbis_obj=self,
            entity="term",
            identifier_name="code",
            additional_identifier="vocabularyCode",
            start_with=start_with,
            count=count,
            totalCount=resp.get("totalCount"),
            response=resp,
            df_initializer=create_data_frame,
        )
        if (
                self.use_cache
                and vocabulary is not None
                and start_with is None
                and count is None
        ):
            self._object_cache(entity="term", code=vocabulary, value=things)

        return things

    def new_term(self, code, vocabularyCode, label=None, description=None):
        return VocabularyTerm(
            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,
            "@type": "as.dto.vocabulary.id.VocabularyTermPermId",
        }
        fetchopts = get_fetchoption_for_entity("vocabularyTerm")
        for opt in ["registrator"]:
            fetchopts[opt] = get_fetchoption_for_entity(opt)

        request = {
            "method": "getVocabularyTerms",
            "params": [self.token, [search_request], fetchopts],
        }
        resp = self._post_request(self.as_v3, request)

        if resp is None or len(resp) == 0:
            raise ValueError(
                f"no VocabularyTerm found with code='{code}' and vocabularyCode='{vocabularyCode}'"
            )
        else:
            parse_jackson(resp)
            for ident in resp:
                if only_data:
                    return resp[ident]
                else:
                    return VocabularyTerm(self, resp[ident])

    def get_vocabularies(self, code=None, start_with=None, count=None):
        """Returns information about vocabulary"""

        sub_criteria = []
        if code:
            sub_criteria.append(_criteria_for_code(code))
        criteria = {
            "criteria": sub_criteria,
            "@type": "as.dto.vocabulary.search.VocabularySearchCriteria",
            "operator": "AND",
        }

        fetchopts = get_fetchoption_for_entity("vocabulary")
        fetchopts["from"] = start_with
        fetchopts["count"] = count
        for option in ["registrator"]:
            fetchopts[option] = get_fetchoption_for_entity(option)

        request = {
            "method": "searchVocabularies",
            "params": [self.token, criteria, fetchopts],
        }
        resp = self._post_request(self.as_v3, request)

        def create_data_frame(attrs, props, response):
            attrs = "code description managedInternally chosenFromList urlTemplate registrator registrationDate modificationDate".split()

            objects = response["objects"]
            if len(objects) == 0:
                vocs = DataFrame(columns=attrs)
            else:
                parse_jackson(response)
                vocs = DataFrame(objects)
                vocs["registrationDate"] = vocs["registrationDate"].map(
                    format_timestamp
                )
                vocs["modificationDate"] = vocs["modificationDate"].map(
                    format_timestamp
                )
                vocs["registrator"] = vocs["registrator"].map(extract_person)
            return vocs[attrs]

        return Things(
            openbis_obj=self,
            entity="vocabulary",
            identifier_name="code",
            start_with=start_with,
            count=count,
            totalCount=resp.get("totalCount"),
            response=resp,
            df_initializer=create_data_frame,
        )

    def get_vocabulary(self, code, only_data=False, use_cache=True):
        """Returns the details of a given vocabulary (including vocabulary terms)"""

        code = str(code).upper()
        voc = (
                not only_data
                and use_cache
                and self._object_cache(entity="vocabulary", code=code)
        )
        if voc:
            return voc

        entity = "vocabulary"
        method_name = get_method_for_entity(entity, "get")
        objectIds = _type_for_id(code.upper(), entity)
        fetchopts = get_fetchoption_for_entity(entity)

        request = {
            "method": method_name,
            "params": [self.token, [objectIds], fetchopts],
        }
        resp = self._post_request(self.as_v3, request)

        if len(resp) == 0:
            raise ValueError(f"no {entity} found with identifier: {code}")
        else:
            parse_jackson(resp)
            for ident in resp:
                data = resp[ident]
                if only_data:
                    return data
                vocabulary = Vocabulary(openbis_obj=self, data=data)
                if self.use_cache:
                    self._object_cache(entity="vocabulary", code=code, value=vocabulary)
                return vocabulary

    def new_tag(self, code, description=None):
        """Creates a new tag (for this user)"""
        return Tag(self, code=code, description=description)

    def get_tags(self, code=None, start_with=None, count=None):
        """Returns a DataFrame of all tags"""

        search_criteria = get_search_type_for_entity("tag", "AND")

        criteria = []
        fetchopts = get_fetchoption_for_entity("tag")
        fetchopts["from"] = start_with
        fetchopts["count"] = count
        for option in ["owner"]:
            fetchopts[option] = get_fetchoption_for_entity(option)
        if code:
            criteria.append(_criteria_for_code(code))
        search_criteria["criteria"] = criteria
        request = {
            "method": "searchTags",
            "params": [self.token, search_criteria, fetchopts],
        }

        resp = self._post_request(self.as_v3, request)
        return self._tag_list_for_response(
            response=resp["objects"], totalCount=resp["totalCount"]
        )

    def get_tag(self, permId, only_data=False, use_cache=True):
        """Returns a specific tag"""

        just_one = True
        identifiers = []
        if isinstance(permId, list):
            just_one = False
            for ident in permId:
                identifiers.append(_type_for_id(ident, "tag"))
        else:
            tag = (
                    not only_data
                    and use_cache
                    and self._object_cache(entity="tag", code=permId)
            )
            if tag:
                return tag
            identifiers.append(_type_for_id(permId, "tag"))

        fetchopts = get_fetchoption_for_entity("tag")
        for option in ["owner"]:
            fetchopts[option] = get_fetchoption_for_entity(option)
        request = {
            "method": "getTags",
            "params": [self.token, identifiers, fetchopts],
        }

        resp = self._post_request(self.as_v3, request)

        if just_one:
            if len(resp) == 0:
                raise ValueError(f"no such tag found: {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:
                        self._object_cache(entity="tag", code=permId, value=tag)
                    return tag
        else:
            return self._tag_list_for_response(response=list(resp.values()))

    def _tag_list_for_response(self, response, totalCount=0):
        def create_data_frame(attrs, props, response):
            parse_jackson(response)
            attrs = [
                "permId",
                "code",
                "description",
                "owner",
                "private",
                "registrationDate",
            ]
            if len(response) == 0:
                tags = DataFrame(columns=attrs)
            else:
                tags = DataFrame(response)
                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)
            return tags[attrs]

        return Things(
            openbis_obj=self,
            entity="tag",
            identifier_name="permId",
            totalCount=totalCount,
            response=response,
            df_initializer=create_data_frame,
        )

    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:
            criteria.append(
                {
                    "@type": "as.dto.common.search.PermIdSearchCriteria",
                    "fieldValue": {
                        "@type": "as.dto.common.search.StringEqualToValue",
                        "value": permId,
                    },
                }
            )

        if entityType is not None:
            typeCriteria.append(
                {
                    "@type": "as.dto.entitytype.search.EntityTypeSearchCriteria",
                    "criteria": [_criteria_for_code(entityType)],
                }
            )

        if propertyType is not None:
            typeCriteria.append(
                {
                    "@type": "as.dto.property.search.PropertyTypeSearchCriteria",
                    "criteria": [_criteria_for_code(propertyType)],
                }
            )

        if entityType is not None and propertyType is not None:
            criteria.append(
                {
                    "@type": "as.dto.property.search.PropertyAssignmentSearchCriteria",
                    "criteria": typeCriteria,
                }
            )
        else:
            criteria += typeCriteria

        saCriteria = {
            "@type": "as.dto.semanticannotation.search.SemanticAnnotationSearchCriteria",
            "criteria": criteria,
        }

        objects = self._search_semantic_annotations(saCriteria)

        if only_data:
            return objects

        def create_data_frame(attrs, props, response):
            attrs = [
                "permId",
                "entityType",
                "propertyType",
                "predicateOntologyId",
                "predicateOntologyVersion",
                "predicateAccessionId",
                "descriptorOntologyId",
                "descriptorOntologyVersion",
                "descriptorAccessionId",
                "creationDate",
            ]
            if len(response) == 0:
                annotations = DataFrame(columns=attrs)
            else:
                annotations = DataFrame(response)
            return annotations[attrs]

        return Things(
            openbis_obj=self,
            entity="semantic_annotation",
            identifier_name="permId",
            response=objects,
            df_initializer=create_data_frame,
        )

    def _search_semantic_annotations(self, criteria):

        fetch_options = {
            "@type": "as.dto.semanticannotation.fetchoptions.SemanticAnnotationFetchOptions",
            "entityType": {
                "@type": "as.dto.entitytype.fetchoptions.EntityTypeFetchOptions"
            },
            "propertyType": {
                "@type": "as.dto.property.fetchoptions.PropertyTypeFetchOptions"
            },
            "propertyAssignment": {
                "@type": "as.dto.property.fetchoptions.PropertyAssignmentFetchOptions",
                "entityType": {
                    "@type": "as.dto.entitytype.fetchoptions.EntityTypeFetchOptions"
                },
                "propertyType": {
                    "@type": "as.dto.property.fetchoptions.PropertyTypeFetchOptions"
                },
            },
        }

        request = {
            "method": "searchSemanticAnnotations",
            "params": [self.token, criteria, fetch_options],
        }

        resp = self._post_request(self.as_v3, request)
        if len(resp["objects"]) == 0:
            return []
        else:
            objects = resp["objects"]
            parse_jackson(objects)

            for obj in objects:
                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"])
            return objects

    def get_semantic_annotations(self):
        """Get a list of all available semantic annotations (DataFrame object)."""

        objects = self._search_semantic_annotations(
            {
                "@type": "as.dto.semanticannotation.search.SemanticAnnotationSearchCriteria"
            }
        )

        def create_data_frame(attrs, props, response):
            attrs = [
                "permId",
                "entityType",
                "propertyType",
                "predicateOntologyId",
                "predicateOntologyVersion",
                "predicateAccessionId",
                "descriptorOntologyId",
                "descriptorOntologyVersion",
                "descriptorAccessionId",
                "creationDate",
            ]
            if len(response) == 0:
                annotations = DataFrame(columns=attrs)
            else:
                annotations = DataFrame(response)
            return annotations[attrs]

        return Things(
            openbis_obj=self,
            entity="semantic_annotation",
            identifier_name="permId",
            response=objects,
            df_initializer=create_data_frame,
        )

    def get_semantic_annotation(self, permId, only_data=False):
        objects = self.search_semantic_annotations(permId=permId, only_data=True)
        if len(objects) == 0:
            raise ValueError(
                "Semantic annotation with permId " + permId + " not found."
            )
        obj = objects[0]
        if only_data:
            return obj
        else:
            return SemanticAnnotation(self, isNew=False, **obj)

    def get_plugins(self, start_with=None, count=None):

        criteria = []
        search_criteria = get_search_type_for_entity("plugin", "AND")
        search_criteria["criteria"] = criteria

        fetchopts = get_fetchoption_for_entity("plugin")
        for option in ["registrator"]:
            fetchopts[option] = get_fetchoption_for_entity(option)
        fetchopts["from"] = start_with
        fetchopts["count"] = count

        request = {
            "method": "searchPlugins",
            "params": [
                self.token,
                search_criteria,
                fetchopts,
            ],
        }
        resp = self._post_request(self.as_v3, request)

        def create_data_frame(attrs, props, response):
            attrs = [
                "name",
                "description",
                "pluginType",
                "pluginKind",
                "entityKinds",
                "registrator",
                "registrationDate",
                "permId",
            ]

            objects = response["objects"]
            if len(objects) == 0:
                plugins = DataFrame(columns=attrs)
            else:
                parse_jackson(objects)

                plugins = DataFrame(objects)
                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
                )
            return plugins[attrs]

        return Things(
            openbis_obj=self,
            entity="plugin",
            identifier_name="name",
            start_with=start_with,
            count=count,
            totalCount=resp.get("totalCount"),
            response=resp,
            df_initializer=create_data_frame,
        )

    def get_plugin(self, permId, only_data=False, with_script=True):
        search_request = _type_for_id(permId, "plugin")
        fetchopts = get_fetchoption_for_entity("plugin")
        options = ["registrator"]
        if with_script:
            options.append("script")

        for option in options:
            fetchopts[option] = get_fetchoption_for_entity(option)

        request = {
            "method": "getPlugins",
            "params": [self.token, [search_request], fetchopts],
        }

        resp = self._post_request(self.as_v3, request)
        parse_jackson(resp)

        if resp is None or len(resp) == 0:
            raise ValueError("no such plugin found: " + permId)
        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):
        """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   --
        """
        return Plugin(self, name=name, pluginType=pluginType, **kwargs)

    def new_property_type(
            self,
            code,
            label,
            description,
            dataType,
            managedInternally=False,
            vocabulary=None,
            materialType=None,
            schema=None,
            transformation=None,
            metaData=None,
    ):
        """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)
        """

        if isinstance(vocabulary, Vocabulary):
            vocabulary = vocabulary.code

        return PropertyType(
            openbis_obj=self,
            code=code,
            label=label,
            description=description,
            dataType=dataType,
            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

        identifiers = []
        only_one = False
        if not isinstance(code, list):
            code = [code]
            only_one = True

        for c in code:
            identifiers.append(
                {"permId": c.upper(), "@type": "as.dto.property.id.PropertyTypePermId"}
            )

        fetchopts = get_fetchoption_for_entity("propertyType")
        options = ["vocabulary", "materialType", "semanticAnnotations", "registrator"]
        for option in options:
            fetchopts[option] = get_fetchoption_for_entity(option)

        request = {
            "method": "getPropertyTypes",
            "params": [self.token, identifiers, fetchopts],
        }

        resp = self._post_request(self.as_v3, request)
        parse_jackson(resp)

        if only_one:
            if len(resp) == 0:
                raise ValueError(f"no such propertyType: {code}")
            for ident in resp:
                if only_data:
                    return resp[ident]
                else:
                    pt = PropertyType(openbis_obj=self, data=resp[ident])
                    if self.use_cache: