Skip to content
Snippets Groups Projects
pybis.py 186 KiB
Newer Older
  • Learn to ignore specific revisions
  • 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):
    
            fetchopts = get_fetchoption_for_entity("propertyType")
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            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
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        ):
            """takes a list of objects and returns a Things object"""
    
            def create_data_frame(attrs, props, response):
                attrs = openbis_definitions("propertyType")["attrs"]
                if len(response) == 0:
                    df = DataFrame(columns=attrs)
                else:
                    df = DataFrame(response)
                    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
                    )
                return df[attrs]
    
    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,
                start_with=start_with,
                count=count,
                totalCount=totalCount,
    
                df_initializer=create_data_frame,
    
    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
    
                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)
    
    
            def create_data_frame(attrs, props, response):
                parse_jackson(response)
                entity_types = []
                defs = get_definition_for_entity(entity)
                attrs = defs["attrs"]
                objects = response["objects"]
                if len(objects) == 0:
                    entity_types = DataFrame(columns=attrs)
                else:
                    parse_jackson(objects)
                    entity_types = DataFrame(objects)
                    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
                    )
                return entity_types[attrs]
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                openbis_obj=self,
                entity=entity,
                start_with=start_with,
    
                single_item_method=getattr(self, cls._single_item_method_name),
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                count=count,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                totalCount=resp.get("totalCount"),
    
                df_initializer=create_data_frame,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def get_entity_type(
    
                self,
                entity,
                identifier,
                cls,
                method=None,
                only_data=False,
                with_vocabulary=False,
                use_cache=True,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        ):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            et = (
    
                    not only_data
                    and not isinstance(identifier, list)
                    and use_cache
                    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:
    
                    raise ValueError(f"no such {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(
    
                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 = {
    
                "@type": f"as.dto.{entity.lower()}.search.{entity}TypeSearchCriteria"
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            }
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            fetch_options = {
    
                "@type": f"as.dto.{entity.lower()}.fetchoptions.{entity}TypeFetchOptions"
    
    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"] = get_fetchoption_for_entity(
                    "propertyAssignments"
                )
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                if self.get_server_information().api_version > "3.3":
    
                    fetch_options["validationPlugin"] = get_fetchoption_for_entity("plugin")
    
            request = {
                "method": method_name,
    
                "params": [self.token, search_request, fetch_options],
    
            }
            resp = self._post_request(self.as_v3, request)
    
            def create_data_frame(attrs, props, response):
                parse_jackson(response)
    
                if type_name is not None:
                    if len(response["objects"]) == 1:
                        return EntityType(openbis_obj=self, data=response["objects"][0])
                    elif len(response["objects"]) == 0:
    
                        raise ValueError(f"No such {entity} type: {type_name}")
    
                            f"There is more than one entry for entity={entity} and type={type_name}"
    
                types = []
                attrs = self._get_attributes(
                    type_name, types, additional_attributes, optional_attributes
                )
                objects = response["objects"]
                if len(objects) == 0:
                    types = DataFrame(columns=attrs)
                else:
                    parse_jackson(objects)
                    types = DataFrame(objects)
    
                    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
                start_with=start_with,
                count=count,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                totalCount=resp.get("totalCount"),
    
                df_initializer=create_data_frame,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def _get_attributes(
    
                self, type_name, types, additional_attributes, optional_attributes
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        ):
            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."""
    
        def is_token_valid(self, token: str = None):
    
            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",
    
            resp = self._post_request(self.as_v3, request)
    
            return resp
    
        def get_session_info(self, token=None):
            if token is None:
                token = self.token
    
            if token is None:
    
    
            request = {"method": "getSessionInformation", "params": [token]}
            try:
                resp = self._post_request(self.as_v3, request)
    
                parse_jackson(resp)
            except Exception as exc:
                return None
            return SessionInformation(openbis_obj=self, data=resp)
    
        def set_token(self, token, save_token=False):
    
            """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 token:
                return
    
            if not self.is_token_valid(token):
    
    Juan Fuentes's avatar
    Juan Fuentes committed
                raise ValueError("Session is no longer valid. Please log in again.")
    
                self.__dict__["token"] = token
    
            if save_token:
                self._save_token_to_disk()
    
        def get_dataset(self, permIds, only_data=False, props=None, **kvals):
    
            """fetch a dataset and some metadata attached to it:
            - properties
            - sample
            - parents
            - children
            - containers
            - dataStore
            - physicalData
            - linkedData
            :return: a DataSet object
            """
    
            just_one = True
            identifiers = []
            if isinstance(permIds, list):
                just_one = False
                for permId in permIds:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    identifiers.append(_type_for_id(permId, "dataset"))
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                identifiers.append(_type_for_id(permIds, "dataset"))
    
            fetchopts = get_fetchoption_for_entity("dataSet")
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            for option in [
                "tags",
                "properties",
                "dataStore",
                "physicalData",
                "linkedData",
                "experiment",
                "sample",
                "registrator",
                "modifier",
            ]:
    
                fetchopts[option] = get_fetchoption_for_entity(option)
    
                "method": "getDataSets",
    
            resp = self._post_request(self.as_v3, request)
    
                    raise ValueError(f"no such dataset found: {permIds}")
    
                parse_jackson(resp)
    
                for permId in resp:
                    if only_data:
                        return resp[permId]
                    else:
                        return DataSet(
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                            openbis_obj=self,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                            type=self.get_dataset_type(resp[permId]["type"]["code"]),
                            data=resp[permId],
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                return self._dataset_list_for_response(
    
                    response=list(resp.values()), props=props, parsed=False
    
        def _dataset_list_for_response(
    
                self,
                response,
                attrs=None,
                props=None,
                start_with=None,
                count=None,
                totalCount=0,
                objects=None,
                parsed=False,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """returns a Things object, containing a DataFrame plus some additional information"""
    
    
            def extract_attribute(attribute_to_extract):
                def return_attribute(obj):
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    if obj is None:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        return ""
                    return obj.get(attribute_to_extract, "")
    
    
                return return_attribute
    
            if not parsed:
                parse_jackson(response)
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            if attrs is None:
                attrs = []
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                entity, _, attr = attr.partition(".")
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    try:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                            return obj["project"][attr]
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                            return obj["project"]["identifier"]["identifier"]
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        return ""
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                entity, _, attr = attr.partition(".")
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    try:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                            return obj["project"]["space"][attr]
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                            return obj["project"]["space"]["code"]
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                        return ""
    
    
            def create_data_frame(attrs, props, response):
                default_attrs = [
                    "permId",
                    "type",
                    "experiment",
                    "sample",
                    "registrationDate",
                    "modificationDate",
                    "location",
                    "status",
                    "presentInArchive",
                    "size",
                ]
                display_attrs = default_attrs + attrs
    
                if props is None:
                    props = []
                else:
                    if isinstance(props, str):
                        props = [props]
    
                if len(response) == 0:
                    for prop in props:
                        if prop == "*":
                            continue
                        display_attrs.append(prop)
                    datasets = DataFrame(columns=display_attrs)
                else:
                    datasets = DataFrame(response)
                    for attr in attrs:
                        if "project" in attr:
    
                            datasets[attr] = datasets["experiment"].map(
                                extract_project(attr)
                            )
    
                        elif "space" in attr:
                            datasets[attr] = datasets["experiment"].map(extract_space(attr))
                        elif "." in attr:
                            entity, attribute_to_extract = attr.split(".")
                            datasets[attr] = datasets[entity].map(
                                extract_attribute(attribute_to_extract)
                            )
                    for attr in attrs:
                        # if no dot supplied, just display the code of the space, project or experiment
                        if any(entity == attr for entity in ["experiment", "sample"]):
                            datasets[attr] = datasets[attr].map(extract_nested_identifier)
    
                    datasets["registrationDate"] = datasets["registrationDate"].map(
                        format_timestamp
                    )
                    datasets["modificationDate"] = datasets["modificationDate"].map(
                        format_timestamp
                    )
                    datasets["experiment"] = datasets["experiment"].map(
                        extract_nested_identifier
                    )
                    datasets["sample"] = datasets["sample"].map(extract_nested_identifier)
                    datasets["type"] = datasets["type"].map(extract_code)
                    datasets["permId"] = datasets["code"]
                    for column in ["parents", "children", "components", "containers"]:
                        if column in datasets:
                            datasets[column] = datasets[column].map(extract_identifiers)
                    datasets["size"] = datasets["physicalData"].map(
                        lambda x: x.get("size") if x else ""
                    )
                    datasets["status"] = datasets["physicalData"].map(
                        lambda x: x.get("status") if x else ""
                    )
                    datasets["presentInArchive"] = datasets["physicalData"].map(
                        lambda x: x.get("presentInArchive") if x else ""
                    )
                    datasets["location"] = datasets["physicalData"].map(
                        lambda x: x.get("location") if x else ""
                    )
    
                    for prop in props:
                        if prop == "*":
                            # include all properties in dataFrame.
                            # expand the dataFrame by adding new columns
                            columns = []
                            for i, dataSet in enumerate(response):
                                for prop_name, val in dataSet.get("properties", {}).items():
                                    datasets.loc[i, prop_name.upper()] = val
                                    columns.append(prop_name.upper())
    
                            display_attrs += set(columns)
                            continue
    
                        else:
                            # property name is provided
                            for i, dataSet in enumerate(response):
                                val = dataSet.get("properties", {}).get(
                                    prop, ""
                                ) or dataSet.get("properties", {}).get(prop.upper(), "")
                                datasets.loc[i, prop.upper()] = val
                            display_attrs.append(prop.upper())
                return datasets[display_attrs]
    
            def create_objects(response):
                return objects
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                openbis_obj=self,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                entity="dataset",
                identifier_name="permId",
    
                start_with=start_with,
                count=count,
    
                totalCount=totalCount,
    
                response=response,
                df_initializer=create_data_frame,
    
                objects_initializer=create_objects,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def get_sample(
    
                self, sample_ident, only_data=False, withAttachments=False, props=None, **kvals
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        ):
    
            """Retrieve metadata for the sample.
            Get metadata for the sample and any directly connected parents of the sample to allow access
            to the same information visible in the ELN UI. The metadata will be on the file system.
            :param sample_identifiers: A list of sample identifiers to retrieve.
            """
    
            only_one = True
    
                only_one = False
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    identifiers.append(_type_for_id(ident, "sample"))
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                identifiers.append(_type_for_id(sample_ident, "sample"))
    
    
            fetchopts = get_fetchoption_for_entity("sample")
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            options = [
                "tags",
                "properties",
                "attachments",
                "space",
                "experiment",
                "registrator",
                "modifier",
                "dataSets",
            ]
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            if self.get_server_information().project_samples_enabled:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                options.append("project")
    
                fetchopts[option] = get_fetchoption_for_entity(option)
    
                fetchopts["attachments"] = get_fetchoption_for_entity(
                    "attachmentsWithContent"
                )
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            for key in ["parents", "children", "container", "components"]:
                fetchopts[key] = {"@type": "as.dto.sample.fetchoptions.SampleFetchOptions"}
    
                "method": "getSamples",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                "params": [self.token, identifiers, fetchopts],
    
            resp = self._post_request(self.as_v3, request)
    
            if only_one:
    
                    raise ValueError(f"no such sample found: {sample_ident}")
    
                for sample_ident in resp:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                            openbis_obj=self,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                            type=self.get_sample_type(resp[sample_ident]["type"]["code"]),
                            data=resp[sample_ident],
    
                return self._sample_list_for_response(
                    response=list(resp.values()), props=props, parsed=False
                )
    
                self,
                response,
                attrs=None,
                props=None,
                start_with=None,
                count=None,
                totalCount=0,
                parsed=False,
    
            def create_data_frame(attrs, props, response):
                """returns a Things object, containing a DataFrame plus additional information"""
    
                def extract_attribute(attribute_to_extract):
                    def return_attribute(obj):
                        if obj is None:
                            return ""
                        return obj.get(attribute_to_extract, "")
    
                    return return_attribute
    
    
                if attrs is None:
                    attrs = []
                default_attrs = [
                    "identifier",
                    "permId",
                    "type",
                    "registrator",
                    "registrationDate",
                    "modifier",
                    "modificationDate",
                ]
                display_attrs = default_attrs + attrs
                if props is None:
                    props = []
                else:
                    if isinstance(props, str):
                        props = [props]
                if len(response) == 0:
                    for prop in props:
                        if prop == "*":
                            continue
                        display_attrs.append(prop)
                    samples = DataFrame(columns=display_attrs)
                else:
    
                    samples = DataFrame(response)
                    for attr in attrs:
                        if "." in attr:
                            entity, attribute_to_extract = attr.split(".")
                            samples[attr] = samples[entity].map(
                                extract_attribute(attribute_to_extract)
                            )
                        # if no dot supplied, just display the code of the space, project or experiment
                        elif attr in ["project", "experiment"]:
                            samples[attr] = samples[attr].map(extract_nested_identifier)
                        elif attr in ["space"]:
                            samples[attr] = samples[attr].map(extract_code)
    
                    samples["registrationDate"] = samples["registrationDate"].map(
                        format_timestamp
                    )
                    samples["modificationDate"] = samples["modificationDate"].map(
                        format_timestamp
                    )
                    samples["registrator"] = samples["registrator"].map(extract_person)
                    samples["modifier"] = samples["modifier"].map(extract_person)
                    samples["identifier"] = samples["identifier"].map(extract_identifier)
    
                    samples["container"] = samples["container"].map(
                        extract_nested_identifier
                    )
    
                    for column in ["parents", "children", "components"]:
                        if column in samples:
                            samples[column] = samples[column].map(extract_identifiers)
                    samples["permId"] = samples["permId"].map(extract_permid)
                    samples["type"] = samples["type"].map(extract_nested_permid)
    
                    for prop in props:
                        if prop == "*":
                            # include all properties in dataFrame.
                            # expand the dataFrame by adding new columns
                            columns = []
                            for i, sample in enumerate(response):
                                for prop_name, val in sample.get("properties", {}).items():
                                    samples.loc[i, prop_name.upper()] = val
                                    columns.append(prop_name.upper())
    
                            display_attrs += set(columns)
                            continue
                        else:
                            # property name is provided
                            for i, sample in enumerate(response):
                                if "properties" in sample:
                                    properties = sample["properties"]
    
                                    val = properties.get(prop, "") or properties.get(
                                        prop.upper(), ""
                                    )
    
                                    samples.loc[i, prop.upper()] = val
                                else:
                                    samples.loc[i, prop.upper()] = ""
                            display_attrs.append(prop.upper())
    
    
                return list(
                    map(
                        lambda obj: Sample(
                            openbis_obj=self,
                            type=self.get_sample_type(obj["type"]["code"]),
                            data=obj,
    
                        ),
                        response,
                    )
                )
    
            result = Things(
                openbis_obj=self,
                entity="sample",
                identifier_name="identifier",
                start_with=start_with,
                count=count,
                totalCount=totalCount,
                response=response,
                df_initializer=create_data_frame,
                objects_initializer=create_objects,
                attrs=attrs,
                props=props,
            )
    
        @staticmethod
        def decode_attribute(entity, attribute):
            params = {}
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            attribute, *alias = re.split(r"\s+AS\s+", attribute, flags=re.IGNORECASE)
    
            alias = alias[0] if alias else attribute
    
            regex = re.compile(
                r"""^                         # beginning of the string
                    (?P<requested_entity>\w+) # the entity itself
                    (\.(?P<attribute>\w+))?   # capture an optional .attribute
                    $                         # end of string
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            """,
                re.X,
            )
    
            match = re.search(regex, attribute)
            params = match.groupdict()
    
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            if params["requested_entity"] == "object":
                params["entity"] = "sample"
            elif params["requested_entity"] == "collection":
                params["entity"] = "experiment"
            elif params["requested_entity"] in ["space", "project"]:
                params["entity"] = params["requested_entity"]
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            else:
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                params["entity"] = params["requested_entity"]
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            if not params["attribute"]:
                params["attribute"] = "code"
            params["alias"] = alias
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            del params["requested_entity"]
    
        def _decode_property(self, entity, property):
    
            # match something like: property_name.term.label AS label_alias
            regex = re.compile(
                r"""^
                    (?P<alias_alternative>
                    (?P<property>[^\.]*  )
                    (?:
                        \.
                        (?P<subentity>term|pa) \.
                        (?P<field>code|vocabularyCode|label|description|ordinal|dataType)
                    )?
                    )
                    (
                    \s+(?i)AS\s+
                    (?P<alias>\w+)
                    )?
                    \s*
                    $
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                """,
                re.X,
    
            )
            match = re.search(regex, property)
            if not match:
                try:
    
                    params = self.decode_attribute(entity, property)
    
                    return params
                except ValueError:
                    raise ValueError(f"unable to parse property: {property}")
            params = match.groupdict()
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            if not params["alias"]:
                params["alias"] = params["alias_alternative"]
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        get_object = get_sample  # Alias
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        def get_external_data_management_systems(
    
                self, start_with=None, count=None, only_data=False
    
    Swen Vermeul's avatar
    Swen Vermeul committed
        ):
            entity = "externalDms"
    
    Swen Vermeul's avatar
    Swen Vermeul committed
            criteria = get_type_for_entity(entity, "search")
    
            fetchopts = get_fetchoption_for_entity(entity)
            request = {
                "method": "searchExternalDataManagementSystems",
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                "params": [
                    self.token,
                    criteria,
                    fetchopts,
                ],
    
            }
            response = self._post_request(self.as_v3, request)
    
    
            def create_data_frame(attrs, props, response):
                parse_jackson(response)
                attrs = "code label address addressType urlTemplate openbis".split()
    
                if len(response["objects"]) == 0:
                    entities = DataFrame(columns=attrs)
                else:
                    objects = response["objects"]
                    parse_jackson(objects)
                    entities = DataFrame(objects)
                    entities["permId"] = entities["permId"].map(extract_permid)
                return entities[attrs]
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                openbis_obj=self,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                entity="externalDms",
                identifier_name="permId",
    
                start_with=start_with,
                count=count,
    
                totalCount=response.get("totalCount"),
                response=response,
    
                df_initializer=create_data_frame,
    
        def get_external_data_management_system(self, permId, only_data=False):
    
            """Retrieve metadata for the external data management system.
    
            :param permId: A permId for an external DMS.
    
            :param only_data: Return the result data as a hash-map, not an object.
            """
    
            request = {
                "method": "getExternalDataManagementSystems",
                "params": [
                    self.token,
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    [
                        {
                            "@type": "as.dto.externaldms.id.ExternalDmsPermId",
                            "permId": permId,
                        }
                    ],
    
    Swen Vermeul's avatar
    Swen Vermeul committed
                    {
                        "@type": "as.dto.externaldms.fetchoptions.ExternalDmsFetchOptions",
                    },
    
                ],
            }
    
            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 external DMS found: " + permId)
    
            else:
                for ident in resp:
                    if only_data:
                        return resp[ident]
                    else:
                        return ExternalDMS(self, resp[ident])
    
    
        get_externalDms = get_external_data_management_system  # alias
    
    
        def new_space(self, **kwargs):