openstack/deb-python-sqlalchemy-utils
Code Issues Proposed changes

Remove EAV references (maybe in the future I'll reimplement this as postgres specific module using JSON / HSTORE)

Browse Source
This commit is contained in:
Konsta Vesterinen
2013-12-16 09:33:08 +02:00
parent 74417bef29
commit 05cf9c6eb5
4 changed files with 0 additions and 534 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
docs/eav.rst
View File

@@ -1,5 +0,0 @@
EAV helpers
===========
.. automodule:: sqlalchemy_utils.eav

1
docs/index.rst
View File

@@ -13,7 +13,6 @@ SQLAlchemy-Utils provides custom data types and various utility functions for SQ
aggregates aggregates
decorators decorators
generic_relationship generic_relationship
eav
database_helpers database_helpers
model_helpers model_helpers
license license

231
sqlalchemy_utils/eav.py
View File

@@ -1,231 +0,0 @@
"""
.. warning::
This module is *EXPERIMENTAL*. Everything in the API may change in the
future and break backwards compatibility. Use at your own risk.
SQLAlchemy-Utils provides some helpers for defining EAV_ models.
.. _EAV:
http://en.wikipedia.org/wiki/Entity%E2%80%93attribute%E2%80%93value_model
Why?
----
Consider you have a catalog of products with each Product having very different
set of attributes. Clearly adding separate tables for each product with
distinct columns for each table becomes very time consuming task. Not to
mention it would be impossible for anyone but database admin to add new
attributes to each product.
One solution is to store product attributes in a JSON / XML typed column. This
has some benefits:
* Schema is easy to define
* Needs no 'magic'
::
from sqlalchemy_utils import JSONType
class Product(Base):
__tablename__ = 'product'
id = sa.Column(sa.Integer, primary_key=True)
name = sa.Column(sa.Unicode(255))
category = sa.Column(sa.Unicode(255))
attributes = sa.Column(JSONType)
product = Product(
name='Porsche 911',
category='car',
attributes={
'manufactured': '1991',
'color': 'red',
'maxspeed': '300'
}
)
All good? We have easily defined a model for product which is extendable and
supports all kinds of attributes for products. However what if you want to make
queries such as:
* Find all red cars with maximum speed reaching atleast 300 km/h?
* Find all cars that were manufactured before 1990?
This is where JSON / XML columns fall short. You can't make these comparisons
using those column types. You could switch to using some NoSQL database but
those have their own limitations compared to traditional RDBMS.
::
from sqlalchemy_utils import MetaType, MetaValue
class Product(Base):
__tablename__ = 'product'
id = sa.Column(sa.Integer, primary_key=True)
name = sa.Column(sa.Unicode(255))
category_id = sa.Column(
sa.Integer, sa.ForeignKey(Category.id)
)
category = sa.orm.relationship(Category)
class Category(Base):
__tablename__ = 'product'
id = sa.Column(sa.Integer, primary_key=True)
name = sa.Column(sa.Unicode(255))
class Attribute(Base):
__tablename__ = 'product_attribute'
id = sa.Column(sa.Integer, primary_key=True)
data_type = sa.Column(
MetaType({
'unicode': sa.UnicodeText,
'int': sa.Integer,
'datetime': sa.DateTime
})
)
name = sa.Column(sa.Unicode(255))
category_id = sa.Column(
sa.Integer, sa.ForeignKey(Category.id)
)
category = sa.orm.relationship(Category)
class AttributeValue(Base):
__tablename__ = 'attribute_value'
id = sa.Column(sa.Integer, primary_key=True)
product_id = sa.Column(
sa.Integer, sa.ForeignKey(Product.id)
)
product = sa.orm.relationship(Product)
attr_id = sa.Column(
sa.Integer, sa.ForeignKey(ProductAttribute.id)
)
attr = sa.orm.relationship(ProductAttribute)
value = MetaValue('attr', 'data_type')
Now SQLAlchemy-Utils would create these columns for AttributeValue:
* value_unicode
* value_int
* value_datetime
The `value` attribute is set as hybrid_property.
"""
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
value_key = key
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):
typed_key = (
value_key + '_' +
type_.type_key(
getattr(
getattr(self, attr),
type_column
)
)
)
setattr(
self,
typed_key,
value
)
operations.append((
class_,
key,
property(
getter,
setter
)
))
for operation in operations:
setattr(*operation)

297
tests/test_eav.py
View File

@@ -1,297 +0,0 @@
from pytest import raises
import sqlalchemy as sa
import sqlalchemy.ext.associationproxy
from sqlalchemy.ext.associationproxy import (
AssociationProxy, _AssociationDict
)
from sqlalchemy.orm.collections import (
attribute_mapped_collection,
collection,
MappedCollection,
)
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 == 'some answer'
class MetaTypedCollection(MappedCollection):
def __init__(self):
self.keyfunc = lambda value: value.attr.name
def __getitem__(self, key):
if not self.key_exists(key):
raise KeyError(key)
return self.get(key)
def key_exists(self, key):
adapter = self._sa_adapter
obj = adapter.owner_state.object
return obj.category and key in obj.category.attributes
@collection.appender
@collection.internally_instrumented
def set(self, *args, **kwargs):
if len(args) > 1:
if not self.key_exists(args[0]):
raise KeyError(args[0])
arg = args[1]
else:
arg = args[0]
super(MetaTypedCollection, self).set(arg, **kwargs)
@collection.remover
@collection.internally_instrumented
def remove(self, key):
del self[key]
def assoc_dict_factory(lazy_collection, creator, getter, setter, parent):
if isinstance(parent, MetaAssociationProxy):
return MetaAssociationDict(
lazy_collection, creator, getter, setter, parent
)
else:
return _AssociationDict(
lazy_collection, creator, getter, setter, parent
)
sqlalchemy.ext.associationproxy._AssociationDict = assoc_dict_factory
class MetaAssociationDict(_AssociationDict):
def _create(self, key, value):
parent_obj = self.lazy_collection.ref()
class_ = parent_obj.__mapper__.relationships[
self.lazy_collection.target
].mapper.class_
if not parent_obj.category:
raise KeyError(key)
return class_(attr=parent_obj.category.attributes[key], value=value)
def _get(self, object):
if object is None:
return None
return self.getter(object)
class MetaAssociationProxy(AssociationProxy):
pass
class TestProductCatalog(TestCase):
def create_models(self):
class Category(self.Base):
__tablename__ = 'category'
id = sa.Column(sa.Integer, primary_key=True)
name = sa.Column(sa.Unicode(255))
class Product(self.Base):
__tablename__ = 'product'
id = sa.Column(sa.Integer, primary_key=True)
name = sa.Column(sa.Unicode(255))
category_id = sa.Column(
sa.Integer, sa.ForeignKey(Category.id)
)
category = sa.orm.relationship(Category)
attributes = MetaAssociationProxy(
'attribute_objects',
'value',
)
class Attribute(self.Base):
__tablename__ = 'attribute'
id = sa.Column(sa.Integer, primary_key=True)
data_type = sa.Column(
MetaType({
'unicode': sa.UnicodeText,
'int': sa.Integer,
'datetime': sa.DateTime
})
)
name = sa.Column(sa.Unicode(255))
category_id = sa.Column(
sa.Integer, sa.ForeignKey(Category.id)
)
category = sa.orm.relationship(
Category,
backref=sa.orm.backref(
'attributes',
collection_class=attribute_mapped_collection('name')
)
)
class AttributeValue(self.Base):
__tablename__ = 'attribute_value'
id = sa.Column(sa.Integer, primary_key=True)
product_id = sa.Column(
sa.Integer, sa.ForeignKey(Product.id)
)
product = sa.orm.relationship(
Product,
backref=sa.orm.backref(
'attribute_objects',
collection_class=MetaTypedCollection
)
)
attr_id = sa.Column(
sa.Integer, sa.ForeignKey(Attribute.id)
)
attr = sa.orm.relationship(Attribute)
value = MetaValue('attr', 'data_type')
def repr(self):
return self.value
self.Product = Product
self.Category = Category
self.Attribute = Attribute
self.AttributeValue = AttributeValue
def test_attr_value_setting(self):
attr = self.Attribute(data_type=sa.UnicodeText)
value = self.AttributeValue(attr=attr)
value.value = u'some answer'
assert u'some answer' == value.value_unicode
def test_unknown_attribute_key(self):
product = self.Product()
with raises(KeyError):
product.attributes[u'color'] = u'red'
def test_get_value_returns_none_for_existing_attr(self):
category = self.Category(name=u'cars')
category.attributes = {
u'color': self.Attribute(name=u'color', data_type=sa.UnicodeText),
u'maxspeed': self.Attribute(name=u'maxspeed', data_type=sa.Integer)
}
product = self.Product(
name=u'Porsche 911',
category=category
)
self.session.add(product)
self.session.commit()
assert product.attributes[u'color'] is None
def test_product_attribute_setting(self):
category = self.Category(name=u'cars')
category.attributes = {
u'color': self.Attribute(name=u'color', data_type=sa.UnicodeText),
u'maxspeed': self.Attribute(name=u'maxspeed', data_type=sa.Integer)
}
product = self.Product(
name=u'Porsche 911',
category=category
)
self.session.add(product)
self.session.commit()
product.attribute_objects[u'color'] = self.AttributeValue(
attr=category.attributes['color'], value=u'red'
)
product.attribute_objects[u'maxspeed'] = self.AttributeValue(
attr=category.attributes['maxspeed'], value=300
)
assert product.attribute_objects[u'color'].value_unicode == u'red'
assert product.attribute_objects[u'maxspeed'].value_int == 300
self.session.commit()
assert product.attribute_objects[u'color'].value == u'red'
assert product.attribute_objects[u'maxspeed'].value == 300
def test_association_proxies(self):
category = self.Category(name=u'cars')
category.attributes = {
u'color': self.Attribute(name=u'color', data_type=sa.UnicodeText),
u'maxspeed': self.Attribute(name=u'maxspeed', data_type=sa.Integer)
}
product = self.Product(
name=u'Porsche 911',
category=category
)
self.session.add(product)
self.session.commit()
product.attributes[u'color'] = u'red'
product.attributes[u'maxspeed'] = 300
assert product.attributes[u'color'] == u'red'
assert product.attributes[u'maxspeed'] == 300
self.session.commit()
assert product.attributes[u'color'] == u'red'
assert product.attributes[u'maxspeed'] == 300
# def test_dynamic_hybrid_properties(self):
# category = self.Category(name=u'cars')
# category.attributes = {
# u'color': self.Attribute(name=u'color', data_type=sa.UnicodeText),
# u'maxspeed': self.Attribute(name=u'maxspeed', data_type=sa.Integer)
# }
# product = self.Product(
# name=u'Porsche 911',
# category=category
# )
# self.session.add(product)
# product.attributes[u'color'] = u'red'
# product.attributes[u'maxspeed'] = 300
# self.session.commit()
# (
# self.session.query(self.Product)
# .filter(self.Product.attributes['color'].in_([u'red', u'blue']))
# .order_by(self.Product.attributes['color'])
# )