dictlike-polymorphic.py

SQLAlchemy. 经典的Python ORM框架。学习必看。

"""Mapping a polymorphic-valued vertical table as a dictionary.

This example illustrates accessing and modifying a "vertical" (or
"properties", or pivoted) table via a dict-like interface.  The 'dictlike.py'
example explains the basics of vertical tables and the general approach.  This
example adds a twist- the vertical table holds several "value" columns, one
for each type of data that can be stored.  For example::

  Table('properties', metadata
        Column('owner_id', Integer, ForeignKey('owner.id'),
               primary_key=True),
        Column('key', UnicodeText),
        Column('type', Unicode(16)),
        Column('int_value', Integer),
        Column('char_value', UnicodeText),
        Column('bool_value', Boolean),
        Column('decimal_value', Numeric(10,2)))

For any given properties row, the value of the 'type' column will point to the
'_value' column active for that row.

This example approach uses exactly the same dict mapping approach as the
'dictlike' example.  It only differs in the mapping for vertical rows.  Here,
we'll use a Python @property to build a smart '.value' attribute that wraps up
reading and writing those various '_value' columns and keeps the '.type' up to
date.

Note: Something much like 'comparable_property' is slated for inclusion in a
      future version of SQLAlchemy.
"""

from sqlalchemy.orm.interfaces import PropComparator, MapperProperty
from sqlalchemy.orm import session as sessionlib

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

class PolymorphicVerticalProperty(object):
    """A key/value pair with polymorphic value storage.

    Supplies a smart 'value' attribute that provides convenient read/write
    access to the row's current value without the caller needing to worry
    about the 'type' attribute or multiple columns.

    The 'value' attribute can also be used for basic comparisons in queries,
    allowing the row's logical value to be compared without foreknowledge of
    which column it might be in.  This is not going to be a very efficient
    operation on the database side, but it is possible.  If you're mapping to
    an existing database and you have some rows with a value of str('1') and
    others of int(1), then this could be useful.

    Subclasses must provide a 'type_map' class attribute with the following
    form::

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

    For example,::

      type_map = {
        int: ('integer', 'integer_value'),
        str: ('varchar', 'varchar_value'),
      }

    Would indicate that a Python int value should be stored in the
    'integer_value' column and the .type set to 'integer'.  Conversely, if the
    value of '.type' is 'integer, then the 'integer_value' column is consulted
    for the current value.
    """

    type_map = {
        type(None): (None, None),
        }

    class Comparator(PropComparator):
        """A comparator for .value, builds a polymorphic comparison via CASE.

        Optional.  If desired, install it as a comparator in the mapping::

          mapper(..., properties={
            'value': comparable_property(PolymorphicVerticalProperty.Comparator,
                                         PolymorphicVerticalProperty.value)
          })
        """

        def _case(self):
            cls = self.prop.parent.class_
            whens = [(text("'%s'" % p[0]), getattr(cls, p[1]))
                     for p in cls.type_map.values()
                     if p[1] is not None]
            return case(whens, cls.type, null())
        def __eq__(self, other):
            return cast(self._case(), String) == cast(other, String)
        def __ne__(self, other):
            return cast(self._case(), String) != cast(other, String)

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

    def _get_value(self):
        for discriminator, field in self.type_map.values():
            if self.type == discriminator:
                return getattr(self, field)
        return None

    def _set_value(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]
            field_value = None
            if py_type == field_type:
                self.type = discriminator
                field_value = value
            if field is not None:
                setattr(self, field, field_value)

    def _del_value(self):
        self._set_value(None)

    value = property(_get_value, _set_value, _del_value, doc=
                     """The logical value of this property.""")

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


class comparable_property(MapperProperty):
    """Instruments a Python property for use in query expressions."""

    def __init__(self, comparator, property):
        self.property = property
        self.comparator = comparator(self)

    def do_init(self):
        class_ = self.parent.class_
        sessionlib.register_attribute(class_, self.key, uselist=False,
                                      proxy_property=self.property,
                                      useobject=False,
                                      comparator=self.comparator)

    def setup(self, querycontext, **kwargs):
        pass

    def create_row_processor(self, selectcontext, mapper, row):
        return (None, None, None)


if __name__ == '__main__':
    from sqlalchemy import *
    from sqlalchemy.orm import mapper, relation, create_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('int_value', Integer, default=None),
                  Column('char_value', UnicodeText, default=None),
                  Column('boolean_value', Boolean, default=None))

    class AnimalFact(PolymorphicVerticalProperty):
        type_map = {
            int: (u'integer', 'int_value'),
            unicode: (u'char', 'char_value'),
            bool: (u'boolean', 'boolean_value'),
            type(None): (None, None),
            }

    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': relation(
            AnimalFact, backref='animal',
            collection_class=attribute_mapped_collection('key')),
        })

    mapper(AnimalFact, chars, properties={
        'value': comparable_property(AnimalFact.Comparator, AnimalFact.value)
        })

    metadata.bind = 'sqlite:///'
    metadata.create_all()
    session = create_session()

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

    session.save(stoat)
    session.flush()
    session.clear()

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

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

    metadata.bind.echo = True
    session.flush()
    metadata.bind.echo = False

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

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

    session.save(shrew)
    session.flush()

    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()

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


    metadata.drop_all()