Added some EAV helpers

sqlalchemy_utils/__init__.py

@@ -1,5 +1,6 @@
 from .aggregates import aggregated
 from .decorators import generates
+from .eav import MetaValue, MetaType
 from .exceptions import ImproperlyConfigured
 from .functions import (
     batch_fetch,
@@ -86,6 +87,8 @@ __all__ = (
     IPAddressType,
     LocaleType,
     Merger,
+    MetaType,
+    MetaValue,
     NumberRange,
     NumberRangeException,
     NumberRangeRawType,

sqlalchemy_utils/eav.py

@@ -0,0 +1,100 @@
+from inspect import isclass
+import sqlalchemy as sa
+from sqlalchemy.ext.hybrid import hybrid_property
+import six
+
+
+class MetaType(sa.types.TypeDecorator):
+    impl = sa.types.UnicodeText
+
+    def __init__(self, data_types):
+        sa.types.TypeDecorator.__init__(self)
+        self.data_types = data_types
+
+    def type_key(self, data_type):
+        for key, type_ in six.iteritems(self.data_types):
+            if (
+                isinstance(data_type, type_) or
+                (isclass(data_type) and issubclass(data_type, type_))
+            ):
+                return six.text_type(key)
+
+    def process_bind_param(self, value, dialect):
+        if value is None:
+            return
+        return self.type_key(value)
+
+    def process_result_value(self, value, dialect):
+        if value is not None:
+            return self.data_types[value]
+
+
+class MetaValue(object):
+    def __init__(self, attr, type_column):
+        self.attr = attr
+        self.type_column = type_column
+
+
+def coalesce(*args):
+    for arg in args:
+        if arg is not None:
+            return arg
+    return None
+
+
+@sa.event.listens_for(sa.orm.mapper, 'mapper_configured')
+def instrument_meta_values(mapper, class_):
+    operations = []
+    for key, attr_value in six.iteritems(class_.__dict__):
+        if isinstance(attr_value, MetaValue):
+            attr = attr_value.attr
+            type_column = attr_value.type_column
+
+            parent_class = getattr(class_, attr).mapper.class_
+            type_prop = getattr(parent_class, type_column).property
+            type_ = type_prop.columns[0].type
+            generated_keys = []
+            for type_key, data_type in six.iteritems(type_.data_types):
+                generated_key = key + '_' + type_key
+                operations.append((
+                    class_,
+                    generated_key,
+                    sa.Column(generated_key, data_type)
+                ))
+                generated_keys.append(generated_key)
+
+            def getter(self):
+                return coalesce(
+                    *map(lambda key: getattr(self, key), generated_keys)
+                )
+
+            def setter(self, value):
+                setattr(
+                    self,
+                    key + '_' +
+                    type_.type_key(
+                        getattr(
+                            getattr(self, attr),
+                            type_column
+                        )
+                    ),
+                    value
+                )
+
+            def expression(self):
+                return sa.func.coalesce(
+                    *map(lambda key: getattr(self, key), generated_keys)
+                )
+
+            operations.append((
+                class_,
+                key,
+                hybrid_property(
+                    getter,
+                    setter,
+                    expr=expression
+                )
+            ))
+
+    for operation in operations:
+        setattr(*operation)

tests/__init__.py

@@ -37,6 +37,7 @@ class TestCase(object):
 
         Session = sessionmaker(bind=self.connection)
         self.session = Session()
+        sa.orm.configure_mappers()
 
     def teardown_method(self, method):
         aggregates.manager.reset()

tests/test_eav.py

@@ -0,0 +1,54 @@
+import sqlalchemy as sa
+from sqlalchemy_utils import MetaType, MetaValue
+from tests import TestCase
+
+
+class TestMetaModel(TestCase):
+    def create_models(self):
+        class Question(self.Base):
+            __tablename__ = 'question'
+            id = sa.Column(sa.Integer, primary_key=True)
+            data_type = sa.Column(
+                MetaType({
+                    'str': sa.String,
+                    'unicode': sa.UnicodeText,
+                    'int': sa.Integer,
+                    'datetime': sa.DateTime
+                })
+            )
+
+        class Answer(self.Base):
+            __tablename__ = 'answer'
+            id = sa.Column(sa.Integer, primary_key=True)
+            value = MetaValue('question', 'data_type')
+
+            question_id = sa.Column(sa.Integer, sa.ForeignKey(Question.id))
+            question = sa.orm.relationship(Question)
+
+        self.Question = Question
+        self.Answer = Answer
+
+    def test_meta_type_conversion(self):
+        question = self.Question(data_type=sa.String(200))
+        self.session.add(question)
+        self.session.commit()
+
+        self.session.refresh(question)
+        assert question.data_type.__name__ == 'String'
+
+    def test_auto_generates_meta_value_columns(self):
+        assert hasattr(self.Answer, 'value_str')
+        assert hasattr(self.Answer, 'value_int')
+        assert hasattr(self.Answer, 'value_datetime')
+
+    def test_meta_value_setting(self):
+        question = self.Question(data_type=sa.String)
+        answer = self.Answer(question=question)
+        answer.value = 'some answer'
+        assert answer.value == answer.value_str
+
+    def test_meta_value_as_expression(self):
+        assert str(self.Answer.value) == (
+            'coalesce(answer.value_int, answer.value_unicode'
+            ', answer.value_str, answer.value_datetime)'
+        )