def get_attribute_history(self, state, key, passive=True):
        hashkey = ("history", state, key)

        # cache the objects, not the states; the strong reference here
        # prevents newly loaded objects from being dereferenced during the 
        # flush process
        if hashkey in self.attributes:
            (added, unchanged, deleted, cached_passive) = self.attributes[hashkey]
            # if the cached lookup was "passive" and now we want non-passive, do a non-passive
            # lookup and re-cache
            if cached_passive and not passive:
                (added, unchanged, deleted) = attributes.get_history(state, key, passive=False)
                self.attributes[hashkey] = (added, unchanged, deleted, passive)
        else:
            (added, unchanged, deleted) = attributes.get_history(state, key, passive=passive)
            self.attributes[hashkey] = (added, unchanged, deleted, passive)

        if added is None:
            return (added, unchanged, deleted)
        else:
            return (
                [getattr(c, '_state', c) for c in added],
                [getattr(c, '_state', c) for c in unchanged],
                [getattr(c, '_state', c) for c in deleted],
                )

    def test_lazy_backref_collections(self):
        class Foo(_base.BasicEntity):
            pass
        class Bar(_base.BasicEntity):
            pass

        lazy_load = []
        def lazyload(instance):
            def load():
                return lazy_load
            return load

        attributes.register_class(Foo)
        attributes.register_class(Bar)
        attributes.register_attribute(Foo, 'bars', uselist=True, extension=attributes.GenericBackrefExtension('foo'), trackparent=True, callable_=lazyload, useobject=True)
        attributes.register_attribute(Bar, 'foo', uselist=False, extension=attributes.GenericBackrefExtension('bars'), trackparent=True, useobject=True)

        bar1, bar2, bar3, bar4 = [Bar(id=1), Bar(id=2), Bar(id=3), Bar(id=4)]
        lazy_load = [bar1, bar2, bar3]

        f = Foo()
        bar4 = Bar()
        bar4.foo = f
        eq_(attributes.get_history(attributes.instance_state(f), 'bars'), ([bar4], [bar1, bar2, bar3], []))

        lazy_load = None
        f = Foo()
        bar4 = Bar()
        bar4.foo = f
        eq_(attributes.get_history(attributes.instance_state(f), 'bars'), ([bar4], [], []))

        lazy_load = [bar1, bar2, bar3]
        attributes.instance_state(f).expire_attributes(['bars'])
        eq_(attributes.get_history(attributes.instance_state(f), 'bars'), ((), [bar1, bar2, bar3], ()))

    def test_many_to_one_cascade(self):
        mapper(Address, addresses, properties={
            'user':relationship(User)
        })
        mapper(User, users)

        u1 = User(id=1, name="u1")
        a1 =Address(id=1, email_address="a1", user=u1)
        u2 = User(id=2, name="u2")

        sess = create_session()
        sess.add_all([a1, u2])
        sess.flush()

        a1.user = u2

        sess2 = create_session()
        a2 = sess2.merge(a1)
        eq_(
            attributes.get_history(a2, 'user'), 
            ([u2], (), [attributes.PASSIVE_NO_RESULT])
        )
        assert a2 in sess2.dirty

        sess.refresh(a1)

        sess2 = create_session()
        a2 = sess2.merge(a1, load=False)
        eq_(
            attributes.get_history(a2, 'user'), 
            ((), [u1], ())
        )
        assert a2 not in sess2.dirty

def edit_task(name, goal, strategy, task):
    project = models.Projects.query.filter_by(id=name).first()
    pgoal = models.Goals.query.filter_by(id=goal).first()
    pstrat = models.Strategies.query.filter_by(id=strategy).first()
    ptask = models.Tasks.query.filter_by(id=task).first()
    form = task_form(obj=ptask)
    form.populate_obj(ptask)
    form.deadline.data = ptask.deadline.strftime("%m/%d/%Y")
    tform = task_form(request.values)
    if request.method == "POST" and form.validate_on_submit():
        # if it changed from True to false, set complete date to None
        if get_history(ptask, "complete")[0] == [True] and get_history(ptask, "complete")[2] == [False]:
            print "changed from false to true"
            ptask.completeDate = datetime.datetime.utcnow()
        if get_history(ptask, "complete")[0] == [False] and get_history(ptask, "complete")[2] == [True]:
            print "changed from true to false"
            ptask.completeDate = None
        # task=tform.task.data
        # strat=pstrat
        # note = tform.note.data
        # staff=tform.staff.data
        # deadline=tform.deadline.data
        # complete=tform.complete.data
        # created=datetime.datetime.utcnow()
        db.session.commit()
        return redirect(url_for("task_outline", name=name, goal=goal, strategy=strategy))
    return render_template(
        "edit_task.html", tform=tform, form=form, project=project, pgoal=pgoal, pstrat=pstrat, ptask=ptask
    )

    def test_dict_collections(self):
        class Foo(fixtures.Base):
            pass
        class Bar(fixtures.Base):
            pass

        from sqlalchemy.orm.collections import attribute_mapped_collection

        attributes.register_class(Foo)
        attributes.register_attribute(Foo, 'someattr', uselist=True, useobject=True, typecallable=attribute_mapped_collection('name'))

        hi = Bar(name='hi')
        there = Bar(name='there')
        old = Bar(name='old')
        new = Bar(name='new')

        f = Foo()
        self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [], []))

        f.someattr['hi'] = hi
        self.assertEquals(attributes.get_history(f._state, 'someattr'), ([hi], [], []))

        f.someattr['there'] = there
        self.assertEquals(tuple([set(x) for x in attributes.get_history(f._state, 'someattr')]), (set([hi, there]), set([]), set([])))

        f._state.commit(['someattr'])
        self.assertEquals(tuple([set(x) for x in attributes.get_history(f._state, 'someattr')]), (set([]), set([hi, there]), set([])))

    def test_dict_collections(self):
        class Foo(_base.BasicEntity):
            pass
        class Bar(_base.BasicEntity):
            pass

        from sqlalchemy.orm.collections import attribute_mapped_collection

        attributes.register_class(Foo)
        attributes.register_attribute(Foo, 'someattr', uselist=True, useobject=True, typecallable=attribute_mapped_collection('name'))

        hi = Bar(name='hi')
        there = Bar(name='there')
        old = Bar(name='old')
        new = Bar(name='new')

        f = Foo()
        eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ((), [], ()))

        f.someattr['hi'] = hi
        eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ([hi], [], []))

        f.someattr['there'] = there
        eq_(tuple([set(x) for x in attributes.get_history(attributes.instance_state(f), 'someattr')]), (set([hi, there]), set(), set()))

        attributes.instance_state(f).commit(['someattr'])
        eq_(tuple([set(x) for x in attributes.get_history(attributes.instance_state(f), 'someattr')]), (set(), set([hi, there]), set()))

def channel_visibility_change(mapper, connection, target):
    if (
        get_history(target, "public").has_changes()
        or get_history(target, "deleted").has_changes()
        or get_history(target, "visible").has_changes()
    ):

        instances = list(VideoInstance.query.filter(VideoInstance.channel == target.id).values("id"))
        if instances:
            update_video_instance_date_updated([i[0] for i in instances], visible=_channel_is_public(target))

def leasing_force_expiration(mapper, connection, target):

    expired = False

    added, unchanged, deleted = get_history(target, 'static_ip')
    expired = expired or added or deleted

    added, unchanged, deleted = get_history(target, 'pool_subnet')
    expired = expired or added or deleted

    if expired:
       Lease.query.with_parent(target).update({'force_expire': True})

    def _assert_history(self, obj, compare, compare_passive=None):
        if isinstance(obj, self.classes.User):
            attrname = "addresses"
        elif isinstance(obj, self.classes.Order):
            attrname = "items"

        eq_(attributes.get_history(obj, attrname), compare)

        if compare_passive is None:
            compare_passive = compare

        eq_(attributes.get_history(obj, attrname, attributes.LOAD_AGAINST_COMMITTED), compare_passive)

    def test_get_history(self):
        Edge = self.classes.Edge
        Point = self.classes.Point
        from sqlalchemy.orm.attributes import get_history

        e1 = Edge()
        e1.start = Point(1, 2)
        eq_(
            get_history(e1, "start"),
            ([Point(x=1, y=2)], (), [Point(x=None, y=None)]),
        )

        eq_(get_history(e1, "end"), ((), [Point(x=None, y=None)], ()))

    def test_object_collections_set(self):
        class Foo(_base.BasicEntity):
            pass
        class Bar(_base.BasicEntity):
            def __nonzero__(self):
                assert False

        attributes.register_class(Foo)
        attributes.register_attribute(Foo, 'someattr', uselist=True, useobject=True)

        hi = Bar(name='hi')
        there = Bar(name='there')
        old = Bar(name='old')
        new = Bar(name='new')

        # case 1.  new object
        f = Foo()
        eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ((), [], ()))

        f.someattr = [hi]
        eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ([hi], [], []))

        attributes.instance_state(f).commit(['someattr'])
        eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ((), [hi], ()))

        f.someattr = [there]

        eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ([there], [], [hi]))
        attributes.instance_state(f).commit(['someattr'])

        eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ((), [there], ()))

        f.someattr = [hi]
        eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ([hi], [], [there]))

        f.someattr = [old, new]
        eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ([old, new], [], [there]))

        # case 2.  object with direct settings (similar to a load operation)
        f = Foo()
        collection = attributes.init_collection(attributes.instance_state(f), 'someattr')
        collection.append_without_event(new)
        attributes.instance_state(f).commit_all()
        eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ((), [new], ()))

        f.someattr = [old]
        eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ([old], [], [new]))

        attributes.instance_state(f).commit(['someattr'])
        eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ((), [old], ()))

    def test_object_collections_set(self):
        class Foo(fixtures.Base):
            pass
        class Bar(fixtures.Base):
            def __nonzero__(self):
                assert False

        attributes.register_class(Foo)
        attributes.register_attribute(Foo, 'someattr', uselist=True, useobject=True)

        hi = Bar(name='hi')
        there = Bar(name='there')
        old = Bar(name='old')
        new = Bar(name='new')

        # case 1.  new object
        f = Foo()
        self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [], []))

        f.someattr = [hi]
        self.assertEquals(attributes.get_history(f._state, 'someattr'), ([hi], [], []))

        f._state.commit(['someattr'])
        self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [hi], []))

        f.someattr = [there]

        self.assertEquals(attributes.get_history(f._state, 'someattr'), ([there], [], [hi]))
        f._state.commit(['someattr'])

        self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [there], []))

        f.someattr = [hi]
        self.assertEquals(attributes.get_history(f._state, 'someattr'), ([hi], [], [there]))

        f.someattr = [old, new]
        self.assertEquals(attributes.get_history(f._state, 'someattr'), ([old, new], [], [there]))

        # case 2.  object with direct settings (similar to a load operation)
        f = Foo()
        collection = attributes.init_collection(f, 'someattr')
        collection.append_without_event(new)
        f._state.commit_all()
        self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [new], []))

        f.someattr = [old]
        self.assertEquals(attributes.get_history(f._state, 'someattr'), ([old], [], [new]))

        f._state.commit(['someattr'])
        self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [old], []))

    def test_lazyhistory(self):
        """tests that history functions work with lazy-loading attributes"""

        class Foo(_base.BasicEntity):
            pass
        class Bar(_base.BasicEntity):
            pass

        attributes.register_class(Foo)
        attributes.register_class(Bar)

        bar1, bar2, bar3, bar4 = [Bar(id=1), Bar(id=2), Bar(id=3), Bar(id=4)]
        def func1():
            return "this is func 1"
        def func2():
            return [bar1, bar2, bar3]

        attributes.register_attribute(Foo, 'col1', uselist=False, callable_=lambda o:func1, useobject=True)
        attributes.register_attribute(Foo, 'col2', uselist=True, callable_=lambda o:func2, useobject=True)
        attributes.register_attribute(Bar, 'id', uselist=False, useobject=True)

        x = Foo()
        attributes.instance_state(x).commit_all()
        x.col2.append(bar4)
        eq_(attributes.get_history(attributes.instance_state(x), 'col2'), ([bar4], [bar1, bar2, bar3], []))

    def on_model_change(self, form, model, is_created):
        if form.password.data is None or len(form.password.data) < 3:
            prev_hash = get_history(model, 'password_hash')[2][0]
            model.password_hash = prev_hash
            db.session.commit()

        pass

 def after_update(self):
     client_id_hist = get_history(self.client_id)
     if client_id_hist.deleted:
         session = orm.Session.object_session(self)
         session.pipe.delete(self._cache_name_client_id.format(
             client_id_hist.deleted[0]
         ))

    def test_basic(self):
        mapper(User, users, properties={
            'addresses':dynamic_loader(mapper(Address, addresses))
        })
        sess = create_session()
        u1 = User(name='jack')
        u2 = User(name='ed')
        u2.addresses.append(Address(email_address='[email protected]'))
        u1.addresses.append(Address(email_address='[email protected]'))
        sess.add_all((u1, u2))
        sess.flush()
        
        from sqlalchemy.orm import attributes
        self.assertEquals(attributes.get_history(attributes.instance_state(u1), 'addresses'), ([], [Address(email_address='[email protected]')], []))
        
        sess.clear()

        # test the test fixture a little bit
        assert User(name='jack', addresses=[Address(email_address='wrong')]) != sess.query(User).first()
        assert User(name='jack', addresses=[Address(email_address='[email protected]')]) == sess.query(User).first()

        assert [
            User(name='jack', addresses=[Address(email_address='[email protected]')]),
            User(name='ed', addresses=[Address(email_address='[email protected]')])
        ] == sess.query(User).all()

    def _record(self, mapper, model, operation):
        pk = tuple(mapper.primary_key_from_instance(model))
        #orm.object_session(model)._model_changes[pk] = (model, operation)
        changes = {}
        
        for prop in object_mapper(model).iterate_properties:
            if not isinstance(prop, RelationshipProperty):
                try:
                    history = attributes.get_history(model, prop.key)
                except:
                    continue

                added, unchanged, deleted = history

                newvalue = added[0] if added else None

                if operation=='delete':
                    oldvalue = unchanged[0] if unchanged else None
                else:
                    oldvalue = deleted[0] if deleted else None

                if newvalue or oldvalue:
                    changes[prop.key] = (oldvalue, newvalue)
        
        orm.object_session(model)._model_changes[pk] = (model.__tablename__, pk[0], changes, operation)
        return EXT_CONTINUE

def attrs_changed(obj, *attrs):
    """Checks if the given fields have been changed since the last flush

    :param obj: SQLAlchemy-mapped object
    :param attrs: attribute names
    """
    return any(get_history(obj, attr).has_changes() for attr in attrs)

def create_version(obj, session, deleted = False):
    obj_mapper = object_mapper(obj)
    history_mapper = obj.__history_mapper__
    history_cls = history_mapper.class_
    
    obj_state = attributes.instance_state(obj)
    
    attr = {}

    obj_changed = False
    
    for om, hm in zip(obj_mapper.iterate_to_root(), history_mapper.iterate_to_root()):
        if hm.single:
            continue
    
        for hist_col in hm.local_table.c:
            if hist_col.key == 'version':
                continue
                
            obj_col = om.local_table.c[hist_col.key]

            # get the value of the
            # attribute based on the MapperProperty related to the
            # mapped column.  this will allow usage of MapperProperties
            # that have a different keyname than that of the mapped column.
            try:
                prop = obj_mapper.get_property_by_column(obj_col)
            except UnmappedColumnError:
                # in the case of single table inheritance, there may be 
                # columns on the mapped table intended for the subclass only.
                # the "unmapped" status of the subclass column on the 
                # base class is a feature of the declarative module as of sqla 0.5.2.
                continue
                
            # expired object attributes and also deferred cols might not be in the
            # dict.  force it to load no matter what by using getattr().
            if prop.key not in obj_state.dict:
                getattr(obj, prop.key)

            a, u, d = attributes.get_history(obj, prop.key)

            if d:
                attr[hist_col.key] = d[0]
                obj_changed = True
            elif u:
                attr[hist_col.key] = u[0]
            else:
                # if the attribute had no value.
                attr[hist_col.key] = a[0]
                obj_changed = True
                
    if not obj_changed and not deleted:            
        return

    attr['version'] = obj.version
    hist = history_cls()
    for key, value in attr.iteritems():
        setattr(hist, key, value)
    session.add(hist)
    obj.version += 1

def changeset(obj):
    """
    Returns a humanized changeset for given SQLAlchemy declarative object.

    :param obj: SQLAlchemy declarative model object
    """
    data = {}
    session = sa.orm.object_session(obj)
    if session and obj in session.deleted:
        for prop in obj.__mapper__.iterate_properties:
            if isinstance(prop, sa.orm.ColumnProperty):
                if not prop.columns[0].primary_key:
                    value = getattr(obj, prop.key)
                    if value is not None:
                        data[prop.key] = [None, getattr(obj, prop.key)]
    else:
        for prop in obj.__mapper__.iterate_properties:
            history = get_history(obj, prop.key)
            if history.has_changes():
                old_value = history.deleted[0] if history.deleted else None
                new_value = history.added[0] if history.added else None

                if new_value:
                    data[prop.key] = [new_value, old_value]
    return data