attributes.py

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

        attributes.register_class(Foo)
        attributes.register_attribute(Foo, 'element', uselist=False, useobject=False)
        x = Foo()
        x.element = ['one', 'two', 'three']
        x._state.commit_all()
        x.element[1] = 'five'
        assert not x._state.is_modified()

    def test_descriptorattributes(self):
        """changeset: 1633 broke ability to use ORM to map classes with unusual
        descriptor attributes (for example, classes that inherit from ones
        implementing zope.interface.Interface).
        This is a simple regression test to prevent that defect.
        """
        class des(object):
            def __get__(self, instance, owner): raise AttributeError('fake attribute')

        class Foo(object):
            A = des()


        attributes.unregister_class(Foo)

    def test_collectionclasses(self):

        class Foo(object):pass
        attributes.register_class(Foo)
        attributes.register_attribute(Foo, "collection", uselist=True, typecallable=set, useobject=True)
        assert isinstance(Foo().collection, set)

        attributes.unregister_attribute(Foo, "collection")

        try:
            attributes.register_attribute(Foo, "collection", uselist=True, typecallable=dict, useobject=True)
            assert False
        except exceptions.ArgumentError, e:
            assert str(e) == "Type InstrumentedDict must elect an appender method to be a collection class"

        class MyDict(dict):
            @collection.appender
            def append(self, item):
                self[item.foo] = item
            @collection.remover
            def remove(self, item):
                del self[item.foo]
        attributes.register_attribute(Foo, "collection", uselist=True, typecallable=MyDict, useobject=True)
        assert isinstance(Foo().collection, MyDict)

        attributes.unregister_attribute(Foo, "collection")

        class MyColl(object):pass
        try:
            attributes.register_attribute(Foo, "collection", uselist=True, typecallable=MyColl, useobject=True)
            assert False
        except exceptions.ArgumentError, e:
            assert str(e) == "Type MyColl must elect an appender method to be a collection class"

        class MyColl(object):
            @collection.iterator
            def __iter__(self):
                return iter([])
            @collection.appender
            def append(self, item):
                pass
            @collection.remover
            def remove(self, item):
                pass
        attributes.register_attribute(Foo, "collection", uselist=True, typecallable=MyColl, useobject=True)
        try:
            Foo().collection
            assert True
        except exceptions.ArgumentError, e:
            assert False


class BackrefTest(TestBase):

    def test_manytomany(self):
        class Student(object):pass
        class Course(object):pass

        attributes.register_class(Student)
        attributes.register_class(Course)
        attributes.register_attribute(Student, 'courses', uselist=True, extension=attributes.GenericBackrefExtension('students'), useobject=True)
        attributes.register_attribute(Course, 'students', uselist=True, extension=attributes.GenericBackrefExtension('courses'), useobject=True)

        s = Student()
        c = Course()
        s.courses.append(c)
        self.assert_(c.students == [s])
        s.courses.remove(c)
        self.assert_(c.students == [])

        (s1, s2, s3) = (Student(), Student(), Student())

        c.students = [s1, s2, s3]
        self.assert_(s2.courses == [c])
        self.assert_(s1.courses == [c])
        s1.courses.remove(c)
        self.assert_(c.students == [s2,s3])

    def test_onetomany(self):
        class Post(object):pass
        class Blog(object):pass

        attributes.register_class(Post)
        attributes.register_class(Blog)
        attributes.register_attribute(Post, 'blog', uselist=False, extension=attributes.GenericBackrefExtension('posts'), trackparent=True, useobject=True)
        attributes.register_attribute(Blog, 'posts', uselist=True, extension=attributes.GenericBackrefExtension('blog'), trackparent=True, useobject=True)
        b = Blog()
        (p1, p2, p3) = (Post(), Post(), Post())
        b.posts.append(p1)
        b.posts.append(p2)
        b.posts.append(p3)
        self.assert_(b.posts == [p1, p2, p3])
        self.assert_(p2.blog is b)

        p3.blog = None
        self.assert_(b.posts == [p1, p2])
        p4 = Post()
        p4.blog = b
        self.assert_(b.posts == [p1, p2, p4])

        p4.blog = b
        p4.blog = b
        self.assert_(b.posts == [p1, p2, p4])

        # assert no failure removing None
        p5 = Post()
        p5.blog = None
        del p5.blog

    def test_onetoone(self):
        class Port(object):pass
        class Jack(object):pass
        attributes.register_class(Port)
        attributes.register_class(Jack)
        attributes.register_attribute(Port, 'jack', uselist=False, extension=attributes.GenericBackrefExtension('port'), useobject=True)
        attributes.register_attribute(Jack, 'port', uselist=False, extension=attributes.GenericBackrefExtension('jack'), useobject=True)
        p = Port()
        j = Jack()
        p.jack = j
        self.assert_(j.port is p)
        self.assert_(p.jack is not None)

        j.port = None
        self.assert_(p.jack is None)

class DeferredBackrefTest(TestBase):
    def setUp(self):
        global Post, Blog, called, lazy_load

        class Post(object):
            def __init__(self, name):
                self.name = name
            def __eq__(self, other):
                return other.name == self.name

        class Blog(object):
            def __init__(self, name):
                self.name = name
            def __eq__(self, other):
                return other.name == self.name

        called = [0]

        lazy_load = []
        def lazy_posts(instance):
            def load():
                called[0] += 1
                return lazy_load
            return load

        attributes.register_class(Post)
        attributes.register_class(Blog)
        attributes.register_attribute(Post, 'blog', uselist=False, extension=attributes.GenericBackrefExtension('posts'), trackparent=True, useobject=True)
        attributes.register_attribute(Blog, 'posts', uselist=True, extension=attributes.GenericBackrefExtension('blog'), callable_=lazy_posts, trackparent=True, useobject=True)

    def test_lazy_add(self):
        global lazy_load

        p1, p2, p3 = Post("post 1"), Post("post 2"), Post("post 3")
        lazy_load = [p1, p2, p3]

        b = Blog("blog 1")
        p = Post("post 4")
        p.blog = b
        p = Post("post 5")
        p.blog = b
        # setting blog doesnt call 'posts' callable
        assert called[0] == 0

        # calling backref calls the callable, populates extra posts
        assert b.posts == [p1, p2, p3, Post("post 4"), Post("post 5")]
        assert called[0] == 1

    def test_lazy_remove(self):
        global lazy_load
        called[0] = 0
        lazy_load = []

        b = Blog("blog 1")
        p = Post("post 1")
        p.blog = b
        assert called[0] == 0

        lazy_load = [p]

        p.blog = None
        p2 = Post("post 2")
        p2.blog = b
        assert called[0] == 0
        assert b.posts == [p2]
        assert called[0] == 1

    def test_normal_load(self):
        global lazy_load
        lazy_load = (p1, p2, p3) = [Post("post 1"), Post("post 2"), Post("post 3")]
        called[0] = 0

        b = Blog("blog 1")

        # assign without using backref system
        p2.__dict__['blog'] = b

        assert b.posts == [Post("post 1"), Post("post 2"), Post("post 3")]
        assert called[0] == 1
        p2.blog = None
        p4 = Post("post 4")
        p4.blog = b
        assert b.posts == [Post("post 1"), Post("post 3"), Post("post 4")]
        assert called[0] == 1

        called[0] = 0
        lazy_load = (p1, p2, p3) = [Post("post 1"), Post("post 2"), Post("post 3")]

class HistoryTest(TestBase):
    def test_scalar(self):
        class Foo(fixtures.Base):
            pass

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

        # 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'], []))

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

        # case 2.  object with direct dictionary settings (similar to a load operation)
        f = Foo()
        f.__dict__['someattr'] = 'new'
        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'], []))

        # setting None on uninitialized is currently a change for a scalar attribute
        # no lazyload occurs so this allows overwrite operation to proceed
        f = Foo()
        self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [], []))
        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_mutable_scalar(self):
        class Foo(fixtures.Base):
            pass

        attributes.register_class(Foo)
        attributes.register_attribute(Foo, 'someattr', uselist=False, useobject=False, mutable_scalars=True, copy_function=dict)

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

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

        f._state.commit(['someattr'])
        self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [{'foo':'hi'}], []))
        self.assertEquals(f._state.committed_state['someattr'], {'foo':'hi'})

        f.someattr['foo'] = 'there'
        self.assertEquals(f._state.committed_state['someattr'], {'foo':'hi'})

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

        self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [{'foo':'there'}], []))

        # case 2.  object with direct dictionary settings (similar to a load operation)
        f = Foo()
        f.__dict__['someattr'] = {'foo':'new'}
        self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [{'foo':'new'}], []))

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

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


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

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

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

        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'], []))

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

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

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