mapper.py

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

                self.id = id
                self.version = version
            def __composite_values__(self):
                # a tuple this time
                return (self.id, self.version)
            def __eq__(self, other):
                return other.id == self.id and other.version == self.version
            def __ne__(self, other):
                return not self.__eq__(other)

        class Graph(object):
            def __init__(self, version):
                self.version = version

        mapper(Graph, graphs, properties={
            'version':composite(Version, graphs.c.id, graphs.c.version_id)
        })

        sess = create_session()
        g = Graph(Version(1, 1))
        sess.save(g)
        sess.flush()

        sess.clear()
        g2 = sess.query(Graph).get([1, 1])
        assert g.version == g2.version
        sess.clear()

        g2 = sess.query(Graph).get(Version(1, 1))
        assert g.version == g2.version



class NoLoadTest(MapperSuperTest):
    def test_basic(self):
        """tests a basic one-to-many lazy load"""
        m = mapper(User, users, properties = dict(
            addresses = relation(mapper(Address, addresses), lazy=None)
        ))
        q = create_session().query(m)
        l = [None]
        def go():
            x = q.filter(users.c.user_id == 7).all()
            x[0].addresses
            l[0] = x
        self.assert_sql_count(testing.db, go, 1)

        self.assert_result(l[0], User,
            {'user_id' : 7, 'addresses' : (Address, [])},
            )

    def test_options(self):
        m = mapper(User, users, properties = dict(
            addresses = relation(mapper(Address, addresses), lazy=None)
        ))
        q = create_session().query(m).options(lazyload('addresses'))
        l = [None]
        def go():
            x = q.filter(users.c.user_id == 7).all()
            x[0].addresses
            l[0] = x
        self.assert_sql_count(testing.db, go, 2)

        self.assert_result(l[0], User,
            {'user_id' : 7, 'addresses' : (Address, [{'address_id' : 1}])},
            )

class MapperExtensionTest(TestBase):
    def setUpAll(self):
        tables.create()

        global methods, Ext

        methods = []

        class Ext(MapperExtension):
            def load(self, query, *args, **kwargs):
                methods.append('load')
                return EXT_CONTINUE

            def get(self, query, *args, **kwargs):
                methods.append('get')
                return EXT_CONTINUE

            def translate_row(self, mapper, context, row):
                methods.append('translate_row')
                return EXT_CONTINUE

            def create_instance(self, mapper, selectcontext, row, class_):
                methods.append('create_instance')
                return EXT_CONTINUE

            def append_result(self, mapper, selectcontext, row, instance, result, **flags):
                methods.append('append_result')
                return EXT_CONTINUE

            def populate_instance(self, mapper, selectcontext, row, instance, **flags):
                methods.append('populate_instance')
                return EXT_CONTINUE

            def before_insert(self, mapper, connection, instance):
                methods.append('before_insert')
                return EXT_CONTINUE

            def after_insert(self, mapper, connection, instance):
                methods.append('after_insert')
                return EXT_CONTINUE

            def before_update(self, mapper, connection, instance):
                methods.append('before_update')
                return EXT_CONTINUE

            def after_update(self, mapper, connection, instance):
                methods.append('after_update')
                return EXT_CONTINUE

            def before_delete(self, mapper, connection, instance):
                methods.append('before_delete')
                return EXT_CONTINUE

            def after_delete(self, mapper, connection, instance):
                methods.append('after_delete')
                return EXT_CONTINUE

    def tearDown(self):
        clear_mappers()
        methods[:] = []
        tables.delete()

    def tearDownAll(self):
        tables.drop()

    def test_basic(self):
        """test that common user-defined methods get called."""
        mapper(User, users, extension=Ext())
        sess = create_session()
        u = User()
        sess.save(u)
        sess.flush()
        u = sess.query(User).load(u.user_id)
        sess.clear()
        u = sess.query(User).get(u.user_id)
        u.user_name = 'foobar'
        sess.flush()
        sess.delete(u)
        sess.flush()
        self.assertEquals(methods, ['before_insert', 'after_insert', 'load', 'translate_row', 'populate_instance', 'get',
            'translate_row', 'create_instance', 'populate_instance', 'before_update', 'after_update', 'before_delete', 'after_delete'])

    def test_inheritance(self):
        # test using inheritance
        class AdminUser(User):
            pass

        mapper(User, users, extension=Ext())
        mapper(AdminUser, addresses, inherits=User)

        sess = create_session()
        am = AdminUser()
        sess.save(am)
        sess.flush()
        am = sess.query(AdminUser).load(am.user_id)
        sess.clear()
        am = sess.query(AdminUser).get(am.user_id)
        am.user_name = 'foobar'
        sess.flush()
        sess.delete(am)
        sess.flush()
        self.assertEquals(methods, ['before_insert', 'after_insert', 'load', 'translate_row', 'populate_instance', 'get',
            'translate_row', 'create_instance', 'populate_instance', 'before_update', 'after_update', 'before_delete', 'after_delete'])

    def test_after_with_no_changes(self):
        # test that after_update is called even if no cols were updated

        mapper(Item, orderitems, extension=Ext() , properties={
            'keywords':relation(Keyword, secondary=itemkeywords)
        })
        mapper(Keyword, keywords, extension=Ext() )

        sess = create_session()
        i1 = Item()
        k1 = Keyword()
        sess.save(i1)
        sess.save(k1)
        sess.flush()
        self.assertEquals(methods, ['before_insert', 'after_insert', 'before_insert', 'after_insert'])

        methods[:] = []
        i1.keywords.append(k1)
        sess.flush()
        self.assertEquals(methods, ['before_update', 'after_update'])


    def test_inheritance_with_dupes(self):
        # test using inheritance, same extension on both mappers
        class AdminUser(User):
            pass

        ext = Ext()
        mapper(User, users, extension=ext)
        mapper(AdminUser, addresses, inherits=User, extension=ext)

        sess = create_session()
        am = AdminUser()
        sess.save(am)
        sess.flush()
        am = sess.query(AdminUser).load(am.user_id)
        sess.clear()
        am = sess.query(AdminUser).get(am.user_id)
        am.user_name = 'foobar'
        sess.flush()
        sess.delete(am)
        sess.flush()
        self.assertEquals(methods, ['before_insert', 'after_insert', 'load', 'translate_row', 'populate_instance', 'get',
            'translate_row', 'create_instance', 'populate_instance', 'before_update', 'after_update', 'before_delete', 'after_delete'])

class RequirementsTest(TestBase, AssertsExecutionResults):
    """Tests the contract for user classes."""

    def setUpAll(self):
        global metadata, t1, t2, t3, t4, t5, t6

        metadata = MetaData(testing.db)
        t1 = Table('ht1', metadata,
                   Column('id', Integer, primary_key=True),
                   Column('value', String(10)))
        t2 = Table('ht2', metadata,
                   Column('id', Integer, primary_key=True),
                   Column('ht1_id', Integer, ForeignKey('ht1.id')),
                   Column('value', String(10)))
        t3 = Table('ht3', metadata,
                   Column('id', Integer, primary_key=True),
                   Column('value', String(10)))
        t4 = Table('ht4', metadata,
                   Column('ht1_id', Integer, ForeignKey('ht1.id'),
                          primary_key=True),
                   Column('ht3_id', Integer, ForeignKey('ht3.id'),
                          primary_key=True))
        t5 = Table('ht5', metadata,
                   Column('ht1_id', Integer, ForeignKey('ht1.id'),
                          primary_key=True),
                    )
        t6 = Table('ht6', metadata,
                   Column('ht1a_id', Integer, ForeignKey('ht1.id'),
                          primary_key=True),
                   Column('ht1b_id', Integer, ForeignKey('ht1.id'),
                          primary_key=True),
                   Column('value', String(10)))
        metadata.create_all()

    def setUp(self):
        clear_mappers()

    def tearDownAll(self):
        metadata.drop_all()

    def test_baseclass(self):
        class OldStyle:
            pass

        self.assertRaises(exceptions.ArgumentError, mapper, OldStyle, t1)

        class NoWeakrefSupport(str):
            pass

        # TODO: is weakref support detectable without an instance?
        #self.assertRaises(exceptions.ArgumentError, mapper, NoWeakrefSupport, t2)

    def test_comparison_overrides(self):
        """Simple tests to ensure users can supply comparison __methods__.

        The suite-level test --options are better suited to detect
        problems- they add selected __methods__ across the board on all
        ORM tests.  This test simply shoves a variety of operations
        through the ORM to catch basic regressions early in a standard
        test run.
        """

        # adding these methods directly to each class to avoid decoration
        # by the testlib decorators.
        class H1(object):
            def __init__(self, value='abc'):
                self.value = value
            def __nonzero__(self):
                return False
            def __hash__(self):
                return hash(self.value)
            def __eq__(self, other):
                if isinstance(other, type(self)):
                    return self.value == other.value
                return False
        class H2(object):
            def __init__(self, value='abc'):
                self.value = value
            def __nonzero__(self):
                return False
            def __hash__(self):
                return hash(self.value)
            def __eq__(self, other):
                if isinstance(other, type(self)):
                    return self.value == other.value
                return False
        class H3(object):
            def __init__(self, value='abc'):
                self.value = value
            def __nonzero__(self):
                return False
            def __hash__(self):
                return hash(self.value)
            def __eq__(self, other):
                if isinstance(other, type(self)):
                    return self.value == other.value
                return False
        class H6(object):
            def __init__(self, value='abc'):
                self.value = value
            def __nonzero__(self):
                return False
            def __hash__(self):
                return hash(self.value)
            def __eq__(self, other):
                if isinstance(other, type(self)):
                    return self.value == other.value
                return False

        mapper(H1, t1, properties={
            'h2s': relation(H2, backref='h1'),
            'h3s': relation(H3, secondary=t4, backref='h1s'),
            'h1s': relation(H1, secondary=t5, backref='parent_h1'),
            't6a': relation(H6, backref='h1a',
                            primaryjoin=t1.c.id==t6.c.ht1a_id),
            't6b': relation(H6, backref='h1b',
                            primaryjoin=t1.c.id==t6.c.ht1b_id),
            })
        mapper(H2, t2)
        mapper(H3, t3)
        mapper(H6, t6)

        s = create_session()
        for i in range(3):
            h1 = H1()
            s.save(h1)

        h1.h2s.append(H2())
        h1.h3s.extend([H3(), H3()])
        h1.h1s.append(H1())

        s.flush()
        self.assertEquals(t1.count().scalar(), 4)

        h6 = H6()
        h6.h1a = h1
        h6.h1b = h1

        h6 = H6()
        h6.h1a = h1
        h6.h1b = x = H1()
        assert x in s

        h6.h1b.h2s.append(H2())

        s.flush()

        h1.h2s.extend([H2(), H2()])
        s.flush()

        h1s = s.query(H1).options(eagerload('h2s')).all()
        self.assertEqual(len(h1s), 5)

        self.assert_unordered_result(h1s, H1,
                                     {'h2s': []},
                                     {'h2s': []},
                                     {'h2s': (H2, [{'value': 'abc'},
                                                   {'value': 'abc'},
                                                   {'value': 'abc'}])},
                                     {'h2s': []},
                                     {'h2s': (H2, [{'value': 'abc'}])})

        h1s = s.query(H1).options(eagerload('h3s')).all()

        self.assertEqual(len(h1s), 5)
        h1s = s.query(H1).options(eagerload_all('t6a.h1b'),
                                  eagerload('h2s'),
                                  eagerload_all('h3s.h1s')).all()
        self.assertEqual(len(h1s), 5)


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