vertical attributes storage

dictlike-polymorphic.py

"""
example from sqlalchemy/examples/vertical ported to one column storage
with casts.
"""

from sqlalchemy.orm.interfaces import PropComparator
from sqlalchemy.ext.hybrid import hybrid_property


import logging
logging.basicConfig(filename="log")
logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO)

import operator

# Using the VerticalPropertyDictMixin from the base example
from dictlike import VerticalPropertyDictMixin


class PolymorphicVerticalProperty(object):
    """A key/value pair with polymorphic value storage.
    Subclasses must provide a 'type_map' and value_column class attributes::

      type_map = {
         <python type> : ('type column value', sqlalchemy type class),
         # ...
      }

      value_column = "value"

    """
    type_map = {type(None): (None, None), }
    value_column = "val"

    def __init__(self, key, value=None):
        self.key = key
        self.value = value

    @hybrid_property
    def value(self):
        return getattr(self, self.value_column)

    @value.setter
    def value_setter(self, value):
        py_type = type(value)
        if py_type not in self.type_map:
            raise TypeError(py_type)

        for field_type in self.type_map:
            discriminator, field = self.type_map[field_type]
            if py_type == field_type:
                self.type = discriminator
        setattr(self, self.value_column, value)

    @value.deleter
    def value_deleter(self):
        self._set_value(None)

    @value.comparator
    class value(PropComparator):
        def __init__(self, cls):
            self.cls = cls

        def _op(self, other, operation):
            other_type_name, other_type_class = \
                self.cls.type_map.get(type(other), (None, None))
            return and_(
                operation(
                    cast(getattr(self.cls, self.cls.value_column),
                         other_type_class),
                    other),
                self.cls.type == other_type_name)

        def __eq__(self, other):
            return self._op(other, operator.eq)

        def __ne__(self, other):
            return self._op(other, operator.ne)

        def __le__(self, other):
            return self._op(other, operator.le)

        def __lt__(self, other):
            return self._op(other, operator.lt)

        def __ge__(self, other):
            return self._op(other, operator.ge)

        def __gt__(self, other):
            return self._op(other, operator.gt)

        def in_(self, other):
            other_type_name, other_type_class = \
                self.cls.type_map.get(type(other[0]), (None, None))
            return and_(
                or_(*[cast(getattr(self.cls, self.cls.value_column),
                           other_type_class) == o for o in other]),
                self.cls.type == other_type_name)

    def __repr__(self):
        return '<%s %r=%r>' % (self.__class__.__name__, self.key, self.value)


if __name__ == '__main__':
    from sqlalchemy.sql.expression import cast
    from sqlalchemy import (MetaData, Table, Column, Integer, Unicode,
                            ForeignKey, UnicodeText, and_, not_, or_,
                            Boolean, Float, create_engine)
    from sqlalchemy.orm import mapper, relationship, Session
    from sqlalchemy.orm.collections import attribute_mapped_collection

    metadata = MetaData()

    animals = Table('animal', metadata,
                    Column('id', Integer, primary_key=True),
                    Column('name', Unicode(100)))

    chars = Table('facts', metadata,
                  Column('animal_id', Integer, ForeignKey('animal.id'),
                         primary_key=True),
                  Column('key', Unicode(64), primary_key=True),
                  Column('type', Unicode(16), default=None),
                  Column('val', UnicodeText, default=None))

    class AnimalFact(PolymorphicVerticalProperty):
        type_map = {
            int: (u'integer', Integer),
            unicode: (u'text', UnicodeText),
            str: (u'text', UnicodeText),
            bool: (u'boolean', Boolean),
            float: (u'float', Float),
            type(None): (None, None)
            }
        value_column = "val"

    class Animal(VerticalPropertyDictMixin):
        """An animal.

        Animal facts are available via the 'facts' property or by using
        dict-like accessors on an Animal instance::

          cat['color'] = 'calico'
          # or, equivalently:
          cat.facts['color'] = AnimalFact('color', 'calico')
        """

        _property_type = AnimalFact
        _property_mapping = 'facts'

        def __init__(self, name):
            self.name = name

        def __repr__(self):
            return '<%s %r>' % (self.__class__.__name__, self.name)

    mapper(Animal, animals, properties=
           {'facts': relationship(
                AnimalFact, backref='animal',
                collection_class=attribute_mapped_collection('key')),
            })

    mapper(AnimalFact, chars)

    engine = create_engine('postgresql://alexp:#@10.0.1.204/test')

    metadata.create_all(engine)
    session = Session(engine)

    stoat = Animal(u'stoat')
    stoat[u'color'] = u'red'
    stoat[u'cuteness'] = 7
    stoat[u'weasel-like'] = True

    session.add(stoat)
    session.commit()

    critter = session.query(Animal).filter(Animal.name == u'stoat').one()
    print critter[u'color']
    print type(critter[u'color'])
    print critter[u'cuteness']
    print type(critter[u'cuteness'])

    print "changing cuteness value and type:"
    critter[u'cuteness'] = u'very cute'

    session.commit()

    marten = Animal(u'marten')
    marten[u'cuteness'] = 5
    marten[u'weasel-like'] = True
    marten[u'poisonous'] = False
    session.add(marten)

    shrew = Animal(u'shrew')
    shrew[u'cuteness'] = 4
    shrew[u'weasel-like'] = False
    shrew[u'poisonous'] = True

    session.add(shrew)
    session.commit()

    q = (session.query(Animal).
         filter(Animal.facts.any(
                and_(AnimalFact.key == u'weasel-like',
                     AnimalFact.value == True))))
    print 'weasel-like animals', q.all()

    # Save some typing by wrapping that up in a function:
    with_characteristic = lambda key, value: and_(AnimalFact.key == key,
                                                  AnimalFact.value == value)

    q = (session.query(Animal).
         filter(Animal.facts.any(
                with_characteristic(u'weasel-like', True))))
    print 'weasel-like animals again', q.all()

    q = (session.query(Animal).
         filter(Animal.facts.any(with_characteristic(u'poisonous', False))))
    print 'animals with poisonous=False', q.all()

    q = (session.query(Animal).
         filter(or_(Animal.facts.any(
                    with_characteristic(u'poisonous', False)),
                    not_(Animal.facts.any(AnimalFact.key == u'poisonous')))))
    print 'non-poisonous animals', q.all()

    q = (session.query(Animal).
         filter(Animal.facts.any(AnimalFact.value == 5)))
    print 'any animal with a .value of 5', q.all()

    q = (session.query(Animal).
         filter(Animal.facts.any(
                with_characteristic(u'weasel-like', u'satan'))))
    print 'error-like animals again', q.all()

    q = (session.query(Animal).
         filter(Animal.facts.any(AnimalFact.value > 4)))
    print 'any animal with a .value greater then 4', q.all()

    # Facts can be queried as well.
    q = (session.query(AnimalFact).
         filter(with_characteristic(u'cuteness', u'very cute')))
    print q.all()

    q = (session.query(Animal).
         filter(Animal.facts.any(AnimalFact.value.in_((5, 4)))))
    print 'any animal with a .value in  5, 4', q.all()

    session.close()
    metadata.drop_all(engine)