pybis.py

        """Creates a new space in the openBIS instance."""
        return Space(self, None, **kwargs)

    def new_git_data_set(
            self,
            data_set_type,
            path,
            commit_id,
            repository_id,
            dms,
            sample=None,
            experiment=None,
            properties={},
            dss_code=None,
            parents=None,
            data_set_code=None,
            contents=[],
    ):
        """Create a link data set.
        :param data_set_type: The type of the data set
        :param data_set_type: The type of the data set
        :param path: The path to the git repository
        :param commit_id: The git commit id
        :param repository_id: The git repository id - same for copies
        :param dms: An external data managment system object or external_dms_id
        :param sample: A sample object or sample id.
        :param dss_code: Code for the DSS -- defaults to the first dss if none is supplied.
        :param properties: Properties for the data set.
        :param parents: Parents for the data set.
        :param data_set_code: A data set code -- used if provided, otherwise generated on the server
        :param contents: A list of dicts that describe the contents:
            {'file_length': [file length],
             'crc32': [crc32 checksum],
             'directory': [is path a directory?]
             'path': [the relative path string]}
        :return: A DataSet object
        """
        return pbds.GitDataSetCreation(
            self,
            data_set_type,
            path,
            commit_id,
            repository_id,
            dms,
            sample,
            experiment,
            properties,
            dss_code,
            parents,
            data_set_code,
            contents,
        ).new_git_data_set()

    def new_content_copy(self, path, commit_id, repository_id, edms_id, data_set_id):
        """
        Create a content copy in an existing link data set.
        :param path: path of the new content copy
        "param commit_id: commit id of the new content copy
        "param repository_id: repository id of the content copy
        "param edms_id: Id of the external data managment system of the content copy
        "param data_set_id: Id of the data set to which the new content copy belongs
        """
        return pbds.GitDataSetUpdate(self, data_set_id).new_content_copy(
            path, commit_id, repository_id, edms_id
        )

    def search_files(self, data_set_id, dss_code=None):
        return pbds.GitDataSetFileSearch(self, data_set_id).search_files()

    def delete_content_copy(self, data_set_id, content_copy):
        """
        Deletes a content copy from a data set.
        :param data_set_id: Id of the data set containing the content copy
        :param content_copy: The content copy to be deleted
        """
        return pbds.GitDataSetUpdate(self, data_set_id).delete_content_copy(
            content_copy
        )

    @staticmethod
    def sample_to_sample_id(sample):
        """Take sample which may be a string or object and return an identifier for it."""
        return Openbis._object_to_object_id(
            sample, "as.dto.sample.id.SampleIdentifier", "as.dto.sample.id.SamplePermId"
        )

    @staticmethod
    def experiment_to_experiment_id(experiment):
        """Take experiment which may be a string or object and return an identifier for it."""
        return Openbis._object_to_object_id(
            experiment,
            "as.dto.experiment.id.ExperimentIdentifier",
            "as.dto.experiment.id.SamplePermId",
        )

    @staticmethod
    def _object_to_object_id(obj, identifierType, permIdType):
        object_id = None
        if isinstance(obj, str):
            if is_identifier(obj):
                object_id = {"identifier": obj, "@type": identifierType}
            else:
                object_id = {"permId": obj, "@type": permIdType}
        else:
            object_id = {"identifier": obj.identifier, "@type": identifierType}
        return object_id

    @staticmethod
    def data_set_to_data_set_id(data_set):
        if isinstance(data_set, str):
            code = data_set
        else:
            code = data_set.permId
        return {"permId": code, "@type": "as.dto.dataset.id.DataSetPermId"}

    def external_data_managment_system_to_dms_id(self, dms):
        if isinstance(dms, str):
            dms_id = {"permId": dms, "@type": "as.dto.externaldms.id.ExternalDmsPermId"}
        else:
            dms_id = {
                "identifier": dms.code,
                "@type": "as.dto.sample.id.SampleIdentifier",
            }
        return dms_id

    def new_sample(self, type, project=None, props=None, **kwargs):
        """Creates a new sample of a given sample type.
        type         -- sampleType code or object: mandatory
        code         -- name/code for the sample, if not generated automatically
        space        -- space code or object
        project      -- project code or object
        experiment   -- experiment code or object
        collection   -- same as above
        props        -- a dictionary containing the properties
        """
        if "collection" in kwargs:
            kwargs["experiment"] = kwargs["collection"]
            kwargs.pop("collection", None)

        if isinstance(type, str):
            sample_type = self.get_sample_type(type)
        else:
            sample_type = type
        return Sample(
            self, type=sample_type, project=project, data=None, props=props, **kwargs
        )

    new_object = new_sample  # Alias

    def new_transaction(self, *entities):
        return Transaction(*entities)

    def new_sample_type(
            self,
            code,
            generatedCodePrefix,
            subcodeUnique=False,
            autoGeneratedCode=False,
            listable=True,
            showContainer=False,
            showParents=True,
            showParentMetadata=False,
            validationPlugin=None,
    ):
        """Creates a new sample type."""

        return SampleType(
            self,
            code=code,
            generatedCodePrefix=generatedCodePrefix,
            autoGeneratedCode=autoGeneratedCode,
            listable=listable,
            showContainer=showContainer,
            showParents=showParents,
            showParentMetadata=showParentMetadata,
            validationPlugin=validationPlugin,
            method=self.get_sample_type,
        )

    new_object_type = new_sample_type

    def new_dataset_type(
            self,
            code,
            description=None,
            mainDataSetPattern=None,
            mainDataSetPath=None,
            disallowDeletion=False,
            validationPlugin=None,
    ):
        """Creates a new dataSet type."""

        return DataSetType(
            self,
            code=code,
            description=description,
            mainDataSetPattern=mainDataSetPattern,
            mainDataSetPath=mainDataSetPath,
            disallowDeletion=disallowDeletion,
            validationPlugin=validationPlugin,
            method=self.get_dataset_type,
        )

    def new_experiment_type(
            self,
            code,
            description=None,
            validationPlugin=None,
    ):
        """Creates a new experiment type (collection type)"""
        return ExperimentType(
            self,
            code=code,
            description=description,
            validationPlugin=validationPlugin,
            method=self.get_experiment_type,
        )

    new_collection_type = new_experiment_type

    def new_material_type(
            self,
            code,
            description=None,
            validationPlugin=None,
    ):
        """Creates a new material type."""
        return MaterialType(
            self,
            code=code,
            description=description,
            validationPlugin=validationPlugin,
            method=self.get_material_type,
        )

    def new_dataset(
            self,
            type=None,
            kind="PHYSICAL",
            files=None,
            file=None,
            props=None,
            folder=None,
            **kwargs,
    ):
        """Creates a new dataset of a given type.

        type         -- sampleType code or object: mandatory
        sample       -- sample code or object
        experiment   -- experiment code or object
        collection   -- same as above
        file         -- path to a single file or a directory
        files        -- list of paths to files. Instead of a file, a directory (or many directories)
                        can be provided, the structure is kept intact in openBIS
        zipfile      -- path to a zipfile, which is unzipped in openBIS
        kind         -- if set to CONTAINER, no files should be provided.
                        Instead, the dataset acts as a container for other datasets.

        props        -- a dictionary containing the properties
        """

        if type is None:
            raise ValueError("Please provide a dataSet type")

        if file:
            files = [file]

        if isinstance(type, str):
            type_obj = self.get_dataset_type(type.upper())
        else:
            type_obj = type

        if "object" in kwargs:
            kwargs["sample"] = kwargs["object"]
            kwargs.pop("object", None)
        if "collection" in kwargs:
            kwargs["experiment"] = kwargs["collection"]
            kwargs.pop("collection", None)

        return DataSet(
            self,
            type=type_obj,
            kind=kind,
            files=files,
            folder=folder,
            props=props,
            **kwargs,
        )

    def new_semantic_annotation(self, entityType=None, propertyType=None, **kwargs):
        """Note: not functional yet."""
        return SemanticAnnotation(
            openbis_obj=self,
            isNew=True,
            entityType=entityType,
            propertyType=propertyType,
            **kwargs,
        )

    def new_vocabulary(
            self, code, terms, managedInternally=False, chosenFromList=True, **kwargs
    ):
        """Creates a new vocabulary
        Usage::
            new_vocabulary(
                code = 'vocabulary_code',
                description = '',
                terms = [
                    { "code": "term1", "label": "label1", "description": "description1" },
                    { "code": "term2", "label": "label2", "description": "description2" },
                ]
            )
        """
        kwargs["code"] = code
        kwargs["managedInternally"] = managedInternally
        kwargs["chosenFromList"] = chosenFromList
        return Vocabulary(self, data=None, terms=terms, **kwargs)

    def _get_dss_url(self, dss_code=None):
        """internal method to get the downloadURL of a datastore."""
        dss = self.get_datastores()
        if dss_code is None:
            return dss["downloadUrl"][0]
        else:
            return dss[dss["code"] == dss_code]["downloadUrl"][0]


class ExternalDMS:
    """managing openBIS external data management systems"""

    def __init__(self, openbis_obj, data=None, **kwargs):
        self.__dict__["openbis"] = openbis_obj

        if data is not None:
            self.__dict__["data"] = data

        if kwargs is not None:
            for key in kwargs:
                setattr(self, key, kwargs[key])

    def __getattr__(self, name):
        return self.__dict__["data"].get(name)

    def __dir__(self):
        """all the available methods and attributes that should be displayed
        when using the autocompletion feature (TAB) in Jupyter
        """
        return ["code", "label", "urlTemplate", "address", "addressType", "openbis"]

    def __str__(self):
        return self.data.get("code", None)


class ServerInformation:
    def __init__(self, info):
        self._info = self._reformat_info(info)
        self.attrs = [
            "api_version",
            "archiving_configured",
            "authentication_service",
            "authentication_service.switch_aai.label",
            "authentication_service.switch_aai.link",
            "create_continuous_sample_codes",
            "enabled_technologies",
            "openbis_version",
            "openbis_support_email",
            "personal_access_tokens_enabled",
            "personal_access_tokens_max_validity_period",
            "personal_access_tokens_validity_warning_period",
            "project_samples_enabled",
        ]

    def _reformat_info(self, info):
        for bool_field in [
            "archiving-configured",
            "project-samples-enabled",
            "personal-access-tokens-enabled",
        ]:
            if bool_field in info:
                info[bool_field] = info[bool_field] == "true"
        for csv_field in ["enabled-technologies"]:
            if csv_field in info:
                info[csv_field] = list(
                    map(lambda item: item.strip(), info[csv_field].split(","))
                )
        for int_field in [
            "personal-access-tokens-max-validity-period",
            "personal-access-tokens-validity-warning-period",
        ]:
            if int_field in info:
                info[int_field] = int(info[int_field])
        info["openbis-support-email"] = info.get("openbis.support.email", "")
        info.pop("openbis.support.email", "")
        return info

    def __dir__(self):
        return self.attrs

    def __getattr__(self, name):
        return self._info.get(name.replace("_", "-"))

    def get_major_version(self):
        return int(self._info["api-version"].split(".")[0])

    def get_minor_version(self):
        return int(self._info["api-version"].split(".")[1])

    def is_openbis_1605(self):
        return (self.get_major_version() == 3) and (self.get_minor_version() <= 2)

    def is_openbis_1806(self):
        return (self.get_major_version() == 3) and (self.get_minor_version() >= 5)

    def _repr_html_(self):
        html = """
            <table border="1" class="dataframe">
            <thead>
                <tr style="text-align: right;">
                <th>attribute</th>
                <th>value</th>
                </tr>
            </thead>
            <tbody>
        """

        for attr in self.attrs:
            html += f"<tr> <td>{attr}</td> <td>{getattr(self, attr, '')}</td> </tr>"

        html += """
            </tbody>
            </table>
        """
        return html


class PropertyType(
    OpenBisObject, entity="propertyType", single_item_method_name="get_property_type"
):
    pass


class Plugin(OpenBisObject, entity="plugin", single_item_method_name="get_plugin"):
    pass


class PersonalAccessToken(
    OpenBisObject,
    entity="personalAccessToken",
    single_item_method_name="get_personal_access_token",
):
    def renew(self, validFrom: datetime = None, validTo: datetime = None):
        """Create a new personal access token (PAT) based on an existing one.
        The same sessionName and validity period will be used, starting from now.
        A new PAT will be created, regardless if there is already an existing
        (and still valid) one.

        Args:
            validFrom (datetime): begin of the validity period (default:now)
            validTo (datetime):   end of the validity period (default: validFrom + maximum validity period, as configured in openBIS)
        """
        if not validFrom:
            validFrom = datetime.now()

        if not validTo:
            validFrom_orig = datetime.strptime(self.validFromDate, "%Y-%m-%d %H:%M:%S")
            validTo_orig = datetime.strptime(self.validToDate, "%Y-%m-%d %H:%M:%S")
            days_delta = abs(validFrom_orig - validTo_orig).days
            validTo = validFrom + relativedelta(days=days_delta)

        new_pat = self.openbis.new_personal_access_token(
            sessionName=self.sessionName,
            validFrom=validFrom,
            validTo=validTo,
            force=True,
        )
        return new_pat


class SessionInformation(
    OpenBisObject,
    entity="sessionInformation",
):
    pass