attributes.py

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

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

        # setting None on uninitialized is currently not a change for an object attribute
        # (this is different than scalar attribute).  a lazyload has occured so if its
        # None, its really None
        f = Foo()
        self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [None], []))
        f.someattr = None
        self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [None], []))

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

    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_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_object_collections_mutate(self):
        class Foo(fixtures.Base):
            pass
        class Bar(fixtures.Base):
            pass

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

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

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

        f.someattr.append(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.append(there)

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

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

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

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

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

        # case 2.  object with direct settings (similar to a load operation)
        f = Foo()
        f.__dict__['id'] = 1
        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.append(old)
        self.assertEquals(attributes.get_history(f._state, 'someattr'), ([old], [new], []))

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

        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.id = 1
        f.someattr.remove(new)
        self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [], [new]))

        # case 3.  mixing appends with sets
        f = Foo()
        f.someattr.append(hi)
        self.assertEquals(attributes.get_history(f._state, 'someattr'), ([hi], [], []))
        f.someattr.append(there)
        self.assertEquals(attributes.get_history(f._state, 'someattr'), ([hi, there], [], []))
        f.someattr = [there]
        self.assertEquals(attributes.get_history(f._state, 'someattr'), ([there], [], []))

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

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

        f1 = Foo()
        b1 = Bar()
        self.assertEquals(attributes.get_history(f1._state, 'bars'), ([], [], []))
        self.assertEquals(attributes.get_history(b1._state, 'foo'), ([], [None], []))

        #b1.foo = f1
        f1.bars.append(b1)
        self.assertEquals(attributes.get_history(f1._state, 'bars'), ([b1], [], []))
        self.assertEquals(attributes.get_history(b1._state, 'foo'), ([f1], [], []))

        b2 = Bar()
        f1.bars.append(b2)
        self.assertEquals(attributes.get_history(f1._state, 'bars'), ([b1, b2], [], []))
        self.assertEquals(attributes.get_history(b1._state, 'foo'), ([f1], [], []))
        self.assertEquals(attributes.get_history(b2._state, 'foo'), ([f1], [], []))

    def test_lazy_backref_collections(self):
        class Foo(fixtures.Base):
            pass
        class Bar(fixtures.Base):
            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
        self.assertEquals(attributes.get_history(f._state, 'bars'), ([bar4], [bar1, bar2, bar3], []))

        lazy_load = None
        f = Foo()
        bar4 = Bar()
        bar4.foo = f
        self.assertEquals(attributes.get_history(f._state, 'bars'), ([bar4], [], []))

        lazy_load = [bar1, bar2, bar3]
        f._state.expire_attributes(['bars'])
        self.assertEquals(attributes.get_history(f._state, 'bars'), ([], [bar1, bar2, bar3], []))

    def test_collections_via_lazyload(self):
        class Foo(fixtures.Base):
            pass
        class Bar(fixtures.Base):
            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, callable_=lazyload, 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()
        f.bars = []
        self.assertEquals(attributes.get_history(f._state, 'bars'), ([], [], [bar1, bar2, bar3]))

        f = Foo()
        f.bars.append(bar4)
        self.assertEquals(attributes.get_history(f._state, 'bars'), ([bar4], [bar1, bar2, bar3], []) )

        f = Foo()
        f.bars.remove(bar2)
        self.assertEquals(attributes.get_history(f._state, 'bars'), ([], [bar1, bar3], [bar2]))
        f.bars.append(bar4)
        self.assertEquals(attributes.get_history(f._state, 'bars'), ([bar4], [bar1, bar3], [bar2]))

        f = Foo()
        del f.bars[1]
        self.assertEquals(attributes.get_history(f._state, 'bars'), ([], [bar1, bar3], [bar2]))

        lazy_load = None
        f = Foo()
        f.bars.append(bar2)
        self.assertEquals(attributes.get_history(f._state, 'bars'), ([bar2], [], []))

    def test_scalar_via_lazyload(self):
        class Foo(fixtures.Base):
            pass

        lazy_load = None
        def lazyload(instance):
            def load():
                return lazy_load
            return load

        attributes.register_class(Foo)
        attributes.register_attribute(Foo, 'bar', uselist=False, callable_=lazyload, useobject=False)
        lazy_load = "hi"

        # with scalar non-object, the lazy callable is only executed on gets, not history
        # operations

        f = Foo()
        self.assertEquals(f.bar, "hi")
        self.assertEquals(attributes.get_history(f._state, 'bar'), ([], ["hi"], []))

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

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

        f = Foo()
        self.assertEquals(f.bar, "hi")
        del f.bar
        self.assertEquals(attributes.get_history(f._state, 'bar'), ([], [], ["hi"]))
        assert f.bar is None
        self.assertEquals(attributes.get_history(f._state, 'bar'), ([None], [], ["hi"]))

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

        lazy_load = None
        def lazyload(instance):
            def load():
                return lazy_load
            return load

        attributes.register_class(Foo)
        attributes.register_class(Bar)
        attributes.register_attribute(Foo, 'bar', uselist=False, callable_=lazyload, trackparent=True, useobject=True)
        bar1, bar2 = [Bar(id=1), Bar(id=2)]
        lazy_load = bar1

        # with scalar object, the lazy callable is only executed on gets and history
        # operations

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

        f = Foo()
        f.bar = None
        self.assertEquals(attributes.get_history(f._state, 'bar'), ([None], [], [bar1]))

        f = Foo()
        f.bar = bar2
        self.assertEquals(attributes.get_history(f._state, 'bar'), ([bar2], [], [bar1]))
        f.bar = bar1
        self.assertEquals(attributes.get_history(f._state, 'bar'), ([], [bar1], []))

        f = Foo()
        self.assertEquals(f.bar, bar1)
        del f.bar
        self.assertEquals(attributes.get_history(f._state, 'bar'), ([None], [], [bar1]))
        assert f.bar is None
        self.assertEquals(attributes.get_history(f._state, 'bar'), ([None], [], [bar1]))

if __name__ == "__main__":
    testenv.main()