Clone wiki

sqlalchemy / UsageRecipes / VersionedMap


This is a variant of the UsageRecipes/VersionedRows recipe. Here we store a dictionary of key/value pairs, storing the k/v's in a "vertical" fashion where each key gets a row. The value is split out into two separate datatypes, string and int - the range of datatype storage can be adjusted for individual needs.

Changes to the "data" attribute of a ConfigData object result in the ConfigData object being copied into a new one, and new associations to its data are created. Values which aren't changed between versions are referenced by both the former and the newer ConfigData object. Overall, only INSERT statements are emitted - no rows are UPDATed or DELETEd.

An optional feature is also illustrated which associates individual key/value pairs with the ConfigData object in which it first originated. Since a new row is only persisted when a new value is created for a particular key, the recipe provides a way to query among the full series of changes which occurred for any particular key in the dictionary.

The set of all ConfigData in a particular table represents a single series of versions. By adding additional columns to ConfigData, the system can be made to store multiple version streams distinguished by those additional values.

"""Illustrate a persisted map of key/value pairs.  All history
of changes is preserved, and only INSERT statements are emitted
when changes occur.  Values are never duplicated if no net
change is present.


from sqlalchemy import Column, String, Integer, ForeignKey, \
from sqlalchemy.orm.interfaces import SessionExtension
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import attributes, relationship, backref, \
    sessionmaker, make_transient, validates
from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.orm.collections import attribute_mapped_collection

class VersionExtension(SessionExtension):
    """Apply the new_version() method of objects which are 
    marked as dirty during a flush.


    def before_flush(self, session, flush_context, instances):
        for instance in session.dirty:
            if hasattr(instance, 'new_version') and \
                session.is_modified(instance, passive=True):

                # make it transient

                # re-add

Base = declarative_base()

class ConfigData(Base):
    """Represent a series of key/value pairs.

    ConfigData will generate a new version of itself
    upon change.

    The "data" dictionary provides access via 
    string name mapped to a string/int value.

    __tablename__ = 'config'

    id = Column(Integer, primary_key=True)
    """Primary key column of this ConfigData."""

    elements = relationship("ConfigValueAssociation", 
    """Dictionary-backed collection of ConfigValueAssociation objects,
    keyed to the name of the associated ConfigValue.

    Note there's no "cascade" here.  ConfigValueAssociation objects
    are never deleted or changed.

    def _new_value(name, value):
        """Create a new entry for usage in the 'elements' dictionary."""
        return ConfigValueAssociation(ConfigValue(name, value))

    data = association_proxy("elements", "value", creator=_new_value)
    """Proxy to the 'value' elements of each related ConfigValue,
    via the 'elements' dictionary.

    def __init__(self, data): = data

    def _associate_with_element(self, key, element):
        """Associate incoming ConfigValues with this 
        ConfigData, if not already associated.

        This is an optional feature which allows
        more comprehensive history tracking.

        if element.config_value.originating_config is None:
            element.config_value.originating_config = self
        return element

    def new_version(self, session):
        # convert to an INSERT
        make_transient(self) = None

        # history of the 'elements' collecton.
        # this is a tuple of groups: (added, unchanged, deleted)
        hist = attributes.get_history(self, 'elements')

        # rewrite the 'elements' collection
        # from scratch, removing all history
        attributes.set_committed_value(self, 'elements', {})

        # new elements in the "added" group
        # are moved to our new collection.
        for elem in hist.added:
            self.elements[] = elem

        # copy elements in the 'unchanged' group.
        # the new ones associate with the new ConfigData,
        # the old ones stay associated with the old ConfigData
        for elem in hist.unchanged:
            self.elements[] = ConfigValueAssociation(elem.config_value)

        # we also need to expire changes on each ConfigValueAssociation
        # that is to remain associated with the old ConfigData.
        # Here, each one takes care of that in its new_version()
        # method, though we could do that here as well.

class ConfigValueAssociation(Base):
    """Relate ConfigData objects to associated ConfigValue objects."""

    __tablename__ = 'config_value_association'

    config_id = Column(ForeignKey(''), primary_key=True)
    """Reference the primary key of the ConfigData object."""

    config_value_id = Column(ForeignKey(''), primary_key=True)
    """Reference the primary key of hte ConfigValue object."""

    config_value = relationship("ConfigValue", lazy="joined", innerjoin=True)
    """Reference the related ConfigValue object."""

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

    def new_version(self, session):
        """Expire all pending state, as ConfigValueAssociation is immutable."""


    def name(self):

    def value(self):
        return self.config_value.value

    def value(self, value):
        """Intercept set events.

        Create a new ConfigValueAssociation upon change, 
        replacing this one in the parent ConfigData's dictionary.

        If no net change, do nothing.

        if value != self.config_value.value:
            self.config_data.elements[] = \
                        ConfigValue(, value)

class ConfigValue(Base):
    """Represent an individual key/value pair at a given point in time.

    ConfigValue is immutable.

    __tablename__ = 'config_value'

    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    originating_config_id = Column(Integer, ForeignKey(''), 
    int_value = Column(Integer)
    string_value = Column(String(255))

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

    originating_config = relationship("ConfigData")
    """Reference to the originating ConfigData.

    This is optional, and allows history tracking of
    individual values.


    def new_version(self, session):
        raise NotImplementedError("ConfigValue is immutable.")

    def value(self):
        for k in ('int_value', 'string_value'):
            v = getattr(self, k)
            if v is not None:
                return v
            return None

    def value(self, value):
        if isinstance(value, int):
            self.int_value = value
            self.string_value = None
            self.string_value = str(value)
            self.int_value = None

if __name__ == '__main__':
    engine = create_engine('sqlite://', echo=True)
    Session = sessionmaker(bind=engine, extension=VersionExtension())

    sess = Session()

    config = ConfigData({

    version_one =['user_name'] = 'yahoo'

    version_two =

    assert version_one != version_two

    # two versions have been created.

    assert == {

    old_config = sess.query(ConfigData).get(version_one)
    assert == {

    # the history of any key can be acquired using
    # the originating_config_id attribute
    history = sess.query(ConfigValue).\

    assert [(h.value, h.originating_config_id) for h in history] == \
            [('twitter', version_one), ('yahoo', version_two)]