ENH: New location for functions for high level API

src/caosdb/high_level_api.py

+# -*- coding: utf-8 -*-
+#
+# This file is a part of the CaosDB Project.
+#
+# Copyright (C) 2018 Research Group Biomedical Physics,
+# Max-Planck-Institute for Dynamics and Self-Organization Göttingen
+# Copyright (C) 2020 Timm Fitschen <t.fitschen@indiscale.com>
+# Copyright (C) 2020-2022 IndiScale GmbH <info@indiscale.com>
+# Copyright (C) 2022 Alexander Schlemmer <alexander.schlemmer@ds.mpg.de>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see <https://www.gnu.org/licenses/>.
+#
+# ** end header
+#
+
+"""
+A high level API for accessing CaosDB entities from within python.
+
+This is refactored from apiutils.
+"""
+
+import sys
+
+from caosdb.common.datatype import (BOOLEAN, DATETIME, DOUBLE, FILE, INTEGER,
+                                    REFERENCE, TEXT, is_reference)
+from caosdb.common.models import (Container, Entity, File, Property, Query,
+                                  Record, RecordType, execute_query,
+                                  get_config)
+
+from .apiutils import get_type_of_entity_with
+
+class CaosDBPythonEntity(object):
+
+    _last_id = 0
+
+    def __init__(self):
+        # Save a copy of the dry state
+        # of this object in order to be
+        # able to detect conflicts.
+        self.do_not_expand = False
+        self._parents = []
+        self._id = CaosDBPythonEntity._get_id()
+        self._path = None
+        self._file = None
+        
+        # TODO:
+        # 3.) resolve references up to a specific depth (including infinity)
+        # 4.) resolve parents function -> partially implemented by
+        # get_parent_names
+        
+        self._references = {}
+        self._properties = set()
+        self._datatypes = {}
+        self._forbidden = dir(self)
+
+    @staticmethod
+    def _get_id():
+        CaosDBPythonEntity._last_id -= 1
+
+        return CaosDBPythonEntity._last_id
+
+    def _set_property_from_entity(self, ent):
+        name = ent.name
+        val = ent.value
+        pr = ent.datatype
+        val, reference = self._type_converted_value(val, pr)
+        self.set_property(name, val, reference, datatype=pr)
+
+    def set_property(self, name, value, is_reference=False,
+                     overwrite=False, datatype=None):
+        """
+        overwrite: Use this if you definitely only want one property with that name (set to True).
+        """
+        self._datatypes[name] = datatype
+
+        if isinstance(name, Entity):
+            name = name.name
+
+        if name in self._forbidden:
+            raise RuntimeError("Entity cannot be converted to a corresponding "
+                               "Python representation. Name of property " +
+                               name + " is forbidden!")
+        already_exists = (name in dir(self))
+
+        if already_exists and not overwrite:
+            # each call to _set_property checks first if it already exists
+            #        if yes: Turn the attribute into a list and
+            #                place all the elements into that list.
+            att = self.__getattribute__(name)
+
+            if isinstance(att, list):
+                pass
+            else:
+                old_att = self.__getattribute__(name)
+                self.__setattr__(name, [old_att])
+
+                if is_reference:
+                    self._references[name] = [
+                        self._references[name]]
+            att = self.__getattribute__(name)
+            att.append(value)
+
+            if is_reference:
+                self._references[name].append(int(value))
+        else:
+            if is_reference:
+                self._references[name] = value
+            self.__setattr__(name, value)
+
+        if not (already_exists and overwrite):
+            self._properties.add(name)
+
+    add_property = set_property
+
+    def set_id(self, idx):
+        self._id = idx
+
+    def _type_converted_list(self, val, pr):
+        """Convert a list to a python list of the correct type."""
+        prrealpre = pr.replace("&lt;", "<").replace("&gt;", ">")
+        prreal = prrealpre[prrealpre.index("<") + 1:prrealpre.rindex(">")]
+        lst = [self._type_converted_value(i, prreal) for i in val]
+
+        return ([i[0] for i in lst], lst[0][1])
+
+    def _type_converted_value(self, val, pr):
+        """Convert val to the correct type which is indicated by the database
+        type string in pr.
+
+        Returns a tuple with two entries:
+        - the converted value
+        - True if the value has to be interpreted as an id acting as a reference
+
+        If datatype is None return the tuple:
+        (value, False)
+        """
+
+        if val is None:
+            return (None, False)
+        elif pr is None:
+            return (val, False)
+        elif pr == DOUBLE:
+            return (float(val), False)
+        elif pr == BOOLEAN:
+            return (bool(val), False)
+        elif pr == INTEGER:
+            return (int(val), False)
+        elif pr == TEXT:
+            return (val, False)
+        elif pr == FILE:
+            return (int(val), False)
+        elif pr == REFERENCE:
+            return (int(val), True)
+        elif pr == DATETIME:
+            return (val, False)
+        elif pr[0:4] == "LIST":
+            return self._type_converted_list(val, pr)
+        elif isinstance(val, Entity):
+            return (convert_to_python_object(val), False)
+        else:
+            return (int(val), True)
+
+    def attribute_as_list(self, name):
+        """This is a workaround for the problem that lists containing only one
+        element are indistinguishable from simple types in this
+        representation."""
+        att = self.__getattribute__(name)
+
+        if isinstance(att, list):
+            return att
+        else:
+            return [att]
+
+    def _add_parent(self, parent):
+        """
+        TODO
+        """
+        self._parents.append(parent.id)
+
+    def add_parent(self, parent_id=None, parent_name=None):
+        """
+        TODO
+        """
+        if parent_id is not None:
+            self._parents.append(parent_id)
+        elif parent_name is not None:
+            self._parents.append(parent_name)
+        else:
+            raise ValueError("no parent identifier supplied")
+
+    def get_parent_names(self):
+        new_plist = []
+
+        for p in self._parents:
+            obj_type = get_type_of_entity_with(p)
+            ent = obj_type(id=p).retrieve()
+            new_plist.append(ent.name)
+
+        return new_plist
+
+    def resolve_references(self, deep=False, visited=dict()):
+        for i in self._references:
+            if isinstance(self._references[i], list):
+                for j in range(len(self._references[i])):
+                    new_id = self._references[i][j]
+                    obj_type = get_type_of_entity_with(new_id)
+
+                    if new_id in visited:
+                        new_object = visited[new_id]
+                    else:
+                        ent = obj_type(id=new_id).retrieve()
+                        new_object = convert_to_python_object(ent)
+                        visited[new_id] = new_object
+
+                        if deep:
+                            new_object.resolve_references(deep, visited)
+                    self.__getattribute__(i)[j] = new_object
+            else:
+                new_id = self._references[i]
+                obj_type = get_type_of_entity_with(new_id)
+
+                if new_id in visited:
+                    new_object = visited[new_id]
+                else:
+                    ent = obj_type(id=new_id).retrieve()
+                    new_object = convert_to_python_object(ent)
+                    visited[new_id] = new_object
+
+                    if deep:
+                        new_object.resolve_references(deep, visited)
+                self.__setattr__(i, new_object)
+
+    def __str__(self, indent=1, name=None):
+        if name is None:
+            result = str(self.__class__.__name__) + "\n"
+        else:
+            result = name + "\n"
+
+        for p in self._properties:
+            value = self.__getattribute__(p)
+
+            if isinstance(value, CaosDBPythonEntity):
+                result += indent * "\t" + \
+                    value.__str__(indent=indent + 1, name=p)
+            else:
+                result += indent * "\t" + p + "\n"
+
+        return result
+
+
+class CaosDBPythonRecord(CaosDBPythonEntity):
+    pass
+
+
+class CaosDBPythonRecordType(CaosDBPythonEntity):
+    pass
+
+
+class CaosDBPythonProperty(CaosDBPythonEntity):
+    pass
+
+
+class CaosDBPythonFile(CaosDBPythonEntity):
+    def get_File(self, target=None):
+        f = File(id=self._id).retrieve()
+        self._file = f.download(target)
+
+
+def _single_convert_to_python_object(robj, entity):
+    robj._id = entity.id
+
+    for i in entity.properties:
+        robj._set_property_from_entity(i)
+
+    for i in entity.parents:
+        robj._add_parent(i)
+
+    if entity.path is not None:
+        robj._path = entity.path
+
+    if entity.file is not None:
+        robj._file = entity.file
+
+    return robj
+
+
+def _single_convert_to_entity(entity, robj, recursive_depth, **kwargs):
+    """
+    recursive_depth: disabled if 0
+    """
+    if robj._id is not None:
+        entity.id = robj._id
+
+    if robj._path is not None:
+        entity.path = robj._path
+
+    if robj._file is not None:
+        entity.file = robj._file
+
+    children = []
+
+    for parent in robj._parents:
+        if sys.version_info[0] < 3:
+            if hasattr(parent, "encode"):
+                entity.add_parent(name=parent)
+            else:
+                entity.add_parent(id=parent)
+        else:
+            if hasattr(parent, "encode"):
+                entity.add_parent(name=parent)
+            else:
+                entity.add_parent(id=parent)
+
+    def add_property(entity, prop, name, _recursive=False, datatype=None):
+        if datatype is None:
+            raise RuntimeError("Datatype must not be None.")
+
+        if isinstance(prop, CaosDBPythonEntity):
+            entity.add_property(name=name, value=str(
+                prop._id), datatype=datatype)
+
+            if _recursive and not prop.do_not_expand:
+                return convert_to_entity(prop, recursive=_recursive)
+            else:
+                return []
+        else:
+            if isinstance(prop, float) or isinstance(prop, int):
+                prop = str(prop)
+            entity.add_property(name=name, value=prop, datatype=datatype)
+
+            return []
+
+    if recursive_depth == 0:
+        recursive = False
+    else:
+        recursive = True
+
+    for prop in robj._properties:
+        value = robj.__getattribute__(prop)
+
+        if isinstance(value, list):
+            if robj._datatypes[prop][0:4] == "LIST":
+                lst = []
+
+                for v in value:
+                    if isinstance(v, CaosDBPythonEntity):
+                        lst.append(v._id)
+
+                        if recursive and not v.do_not_expand:
+                            children.append(convert_to_entity(
+                                v, recursive=recursive_depth-1))
+                    else:
+                        if isinstance(v, float) or isinstance(v, int):
+                            lst.append(str(v))
+                        else:
+                            lst.append(v)
+                entity.add_property(name=prop, value=lst,
+                                    datatype=robj._datatypes[prop])
+            else:
+                for v in value:
+                    children.extend(
+                        add_property(
+                            entity,
+                            v,
+                            prop,
+                            datatype=robj._datatypes[prop],
+                            **kwargs))
+        else:
+            children.extend(
+                add_property(
+                    entity,
+                    value,
+                    prop,
+                    datatype=robj._datatypes[prop],
+                    **kwargs))
+
+    return [entity] + children
+
+
+def convert_to_entity(python_object, **kwargs):
+    if isinstance(python_object, Container):
+        # Create a list of objects:
+
+        return [convert_to_python_object(i, **kwargs) for i in python_object]
+    elif isinstance(python_object, CaosDBPythonRecord):
+        return _single_convert_to_entity(Record(), python_object, **kwargs)
+    elif isinstance(python_object, CaosDBPythonFile):
+        return _single_convert_to_entity(File(), python_object, **kwargs)
+    elif isinstance(python_object, CaosDBPythonRecordType):
+        return _single_convert_to_entity(RecordType(), python_object, **kwargs)
+    elif isinstance(python_object, CaosDBPythonProperty):
+        return _single_convert_to_entity(Property(), python_object, **kwargs)
+    elif isinstance(python_object, CaosDBPythonEntity):
+        return _single_convert_to_entity(Entity(), python_object, **kwargs)
+    else:
+        raise ValueError("Cannot convert an object of this type.")
+
+
+def convert_to_python_object(entity):
+    """"""
+
+    if isinstance(entity, Container):
+        # Create a list of objects:
+
+        return [convert_to_python_object(i) for i in entity]
+    elif isinstance(entity, Record):
+        return _single_convert_to_python_object(CaosDBPythonRecord(), entity)
+    elif isinstance(entity, RecordType):
+        return _single_convert_to_python_object(
+            CaosDBPythonRecordType(), entity)
+    elif isinstance(entity, File):
+        return _single_convert_to_python_object(CaosDBPythonFile(), entity)
+    elif isinstance(entity, Property):
+        return _single_convert_to_python_object(CaosDBPythonProperty(), entity)
+    elif isinstance(entity, Entity):
+        return _single_convert_to_python_object(CaosDBPythonEntity(), entity)
+    else:
+        raise ValueError("Cannot convert an object of this type.")
+

unittests/test_high_level_api.py

+# -*- encoding: utf-8 -*-
+#
+# This file is a part of the CaosDB Project.
+#
+# Copyright (C) 2018 Research Group Biomedical Physics,
+# Max-Planck-Institute for Dynamics and Self-Organization Göttingen
+# Copyright (C) 2022 Alexander Schlemmer <alexander.schlemmer@ds.mpg.de>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see <https://www.gnu.org/licenses/>.
+#
+# ** end header
+#
+# Test high level api module
+# A. Schlemmer, 02/2022
+
+
+
+import caosdb as db
+from caosdb.high_level_api import (convert_to_entity, convert_to_python_object)
+import pytest
+from lxml import etree
+import os
+import tempfile
+import pickle
+
+@pytest.fixture
+def testrecord():
+    parser = etree.XMLParser(remove_comments=True)
+    testrecord = db.Record._from_xml(
+        db.Record(),
+        etree.parse(os.path.join(os.path.dirname(__file__), "test_record.xml"),
+                    parser).getroot())
+    return testrecord
+
+
+
+
+def test_convert_object(testrecord):
+    r2 = convert_to_python_object(testrecord)
+    assert r2.species == "Rabbit"
+
+
+def test_pickle_object(testrecord):
+    r2 = convert_to_python_object(testrecord)
+    with tempfile.TemporaryFile() as f:
+        pickle.dump(r2, f)
+        f.seek(0)
+        rn2 = pickle.load(f)
+    assert r2.date == rn2.date
+
+
+def test_convert_record():
+    """
+    Test the high level python API.
+    """
+    r = db.Record()
+    r.add_parent("bla")
+    r.add_property(name="a", value=42)
+    r.add_property(name="b", value="test")
+
+    obj = convert_to_python_object(r)
+    assert obj.a == 42
+    assert obj.b == "test"
+    breakpoint()