mapper.py

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

        self.assert_sql_count(testing.db, go, 7)


    def test_lazyoptions(self):
        """tests that an eager relation can be upgraded to a lazy relation via the options method"""
        sess = create_session()
        mapper(User, users, properties = dict(
            addresses = relation(mapper(Address, addresses), lazy=False)
        ))
        l = sess.query(User).options(lazyload('addresses')).all()
        def go():
            self.assert_result(l, User, *user_address_result)
        self.assert_sql_count(testing.db, go, 3)

    def test_deepoptions(self):
        mapper(User, users,
            properties = {
                'orders': relation(mapper(Order, orders, properties = {
                    'items' : relation(mapper(Item, orderitems, properties = {
                        'keywords' : relation(mapper(Keyword, keywords), itemkeywords)
                    }))
                }))
            })

        sess = create_session()

        # eagerload nothing.
        u = sess.query(User).all()
        def go():
            print u[0].orders[1].items[0].keywords[1]
        self.assert_sql_count(testing.db, go, 3)
        sess.clear()


        print "-------MARK----------"
        # eagerload orders.items.keywords; eagerload_all() implies eager load of orders, orders.items
        q2 = sess.query(User).options(eagerload_all('orders.items.keywords'))
        u = q2.all()
        def go():
            print u[0].orders[1].items[0].keywords[1]
        print "-------MARK2----------"
        self.assert_sql_count(testing.db, go, 0)

        sess.clear()

        # same thing, with separate options calls
        q2 = sess.query(User).options(eagerload('orders')).options(eagerload('orders.items')).options(eagerload('orders.items.keywords'))
        u = q2.all()
        def go():
            print u[0].orders[1].items[0].keywords[1]
        self.assert_sql_count(testing.db, go, 0)

        sess.clear()

        try:
            sess.query(User).options(eagerload('items', Order))
            assert False
        except exceptions.ArgumentError, e:
            assert str(e) == "Can't find entity Mapper|Order|orders in Query.  Current list: ['Mapper|User|users']"

        # eagerload "keywords" on items.  it will lazy load "orders", then lazy load
        # the "items" on the order, but on "items" it will eager load the "keywords"
        print "-------MARK5----------"
        q3 = sess.query(User).options(eagerload('orders.items.keywords'))
        u = q3.all()
        self.assert_sql_count(testing.db, go, 2)


class DeferredTest(MapperSuperTest):

    def test_basic(self):
        """tests a basic "deferred" load"""

        m = mapper(Order, orders, properties={
            'description':deferred(orders.c.description)
        })

        o = Order()
        self.assert_(o.description is None)

        q = create_session().query(m)
        def go():
            l = q.all()
            o2 = l[2]
            print o2.description

        orderby = str(orders.default_order_by()[0].compile(bind=testing.db))
        self.assert_sql(testing.db, go, [
            ("SELECT orders.order_id AS orders_order_id, orders.user_id AS orders_user_id, orders.isopen AS orders_isopen FROM orders ORDER BY %s" % orderby, {}),
            ("SELECT orders.description AS orders_description FROM orders WHERE orders.order_id = :param_1", {'param_1':3})
        ])

    def test_unsaved(self):
        """test that deferred loading doesnt kick in when just PK cols are set"""
        m = mapper(Order, orders, properties={
            'description':deferred(orders.c.description)
        })

        sess = create_session()
        o = Order()
        sess.save(o)
        o.order_id = 7
        def go():
            o.description = "some description"
        self.assert_sql_count(testing.db, go, 0)

    def test_unsavedgroup(self):
        """test that deferred loading doesnt kick in when just PK cols are set"""
        m = mapper(Order, orders, properties={
            'description':deferred(orders.c.description, group='primary'),
            'opened':deferred(orders.c.isopen, group='primary')
        })

        sess = create_session()
        o = Order()
        sess.save(o)
        o.order_id = 7
        def go():
            o.description = "some description"
        self.assert_sql_count(testing.db, go, 0)

    def test_save(self):
        m = mapper(Order, orders, properties={
            'description':deferred(orders.c.description)
        })

        sess = create_session()
        q = sess.query(m)
        l = q.all()
        o2 = l[2]
        o2.isopen = 1
        sess.flush()

    def test_group(self):
        """tests deferred load with a group"""
        m = mapper(Order, orders, properties = {
            'userident':deferred(orders.c.user_id, group='primary'),
            'description':deferred(orders.c.description, group='primary'),
            'opened':deferred(orders.c.isopen, group='primary')
        })
        sess = create_session()
        q = sess.query(m)
        def go():
            l = q.all()
            o2 = l[2]
            print o2.opened, o2.description, o2.userident
            assert o2.opened == 1
            assert o2.userident == 7
            assert o2.description == 'order 3'
        orderby = str(orders.default_order_by()[0].compile(testing.db))
        self.assert_sql(testing.db, go, [
            ("SELECT orders.order_id AS orders_order_id FROM orders ORDER BY %s" % orderby, {}),
            ("SELECT orders.user_id AS orders_user_id, orders.description AS orders_description, orders.isopen AS orders_isopen FROM orders WHERE orders.order_id = :param_1", {'param_1':3})
        ])

        o2 = q.all()[2]
#        assert o2.opened == 1
        assert o2.description == 'order 3'
        assert o2 not in sess.dirty
        o2.description = 'order 3'
        def go():
            sess.flush()
        self.assert_sql_count(testing.db, go, 0)

    def test_preserve_changes(self):
        """test that the deferred load operation doesn't revert modifications on attributes"""

        mapper(Order, orders, properties = {
            'userident':deferred(orders.c.user_id, group='primary'),
            'description':deferred(orders.c.description, group='primary'),
            'opened':deferred(orders.c.isopen, group='primary')
        })
        sess = create_session()
        o = sess.query(Order).get(3)
        assert 'userident' not in o.__dict__
        o.description = 'somenewdescription'
        assert o.description == 'somenewdescription'
        def go():
            assert o.opened == 1
        self.assert_sql_count(testing.db, go, 1)
        assert o.description == 'somenewdescription'
        assert o in sess.dirty


    def test_commitsstate(self):
        """test that when deferred elements are loaded via a group, they get the proper CommittedState
        and dont result in changes being committed"""

        m = mapper(Order, orders, properties = {
            'userident':deferred(orders.c.user_id, group='primary'),
            'description':deferred(orders.c.description, group='primary'),
            'opened':deferred(orders.c.isopen, group='primary')
        })
        sess = create_session()
        q = sess.query(m)
        o2 = q.all()[2]
        # this will load the group of attributes
        assert o2.description == 'order 3'
        assert o2 not in sess.dirty
        # this will mark it as 'dirty', but nothing actually changed
        o2.description = 'order 3'
        def go():
            # therefore the flush() shouldnt actually issue any SQL
            sess.flush()
        self.assert_sql_count(testing.db, go, 0)

    def test_options(self):
        """tests using options on a mapper to create deferred and undeferred columns"""
        m = mapper(Order, orders)
        sess = create_session()
        q = sess.query(m)
        q2 = q.options(defer('user_id'))
        def go():
            l = q2.all()
            print l[2].user_id

        orderby = str(orders.default_order_by()[0].compile(testing.db))
        self.assert_sql(testing.db, go, [
            ("SELECT orders.order_id AS orders_order_id, orders.description AS orders_description, orders.isopen AS orders_isopen FROM orders ORDER BY %s" % orderby, {}),
            ("SELECT orders.user_id AS orders_user_id FROM orders WHERE orders.order_id = :param_1", {'param_1':3})
        ])
        sess.clear()
        q3 = q2.options(undefer('user_id'))
        def go():
            l = q3.all()
            print l[3].user_id
        self.assert_sql(testing.db, go, [
            ("SELECT orders.order_id AS orders_order_id, orders.user_id AS orders_user_id, orders.description AS orders_description, orders.isopen AS orders_isopen FROM orders ORDER BY %s" % orderby, {}),
        ])

    def test_undefergroup(self):
        """tests undefer_group()"""
        m = mapper(Order, orders, properties = {
            'userident':deferred(orders.c.user_id, group='primary'),
            'description':deferred(orders.c.description, group='primary'),
            'opened':deferred(orders.c.isopen, group='primary')
        })
        sess = create_session()
        q = sess.query(m)
        def go():
            l = q.options(undefer_group('primary')).all()
            o2 = l[2]
            print o2.opened, o2.description, o2.userident
            assert o2.opened == 1
            assert o2.userident == 7
            assert o2.description == 'order 3'
        orderby = str(orders.default_order_by()[0].compile(testing.db))
        self.assert_sql(testing.db, go, [
            ("SELECT orders.user_id AS orders_user_id, orders.description AS orders_description, orders.isopen AS orders_isopen, orders.order_id AS orders_order_id FROM orders ORDER BY %s" % orderby, {}),
        ])

    def test_locates_col(self):
        """test that manually adding a col to the result undefers the column"""
        mapper(Order, orders, properties={
            'description':deferred(orders.c.description)
        })

        sess = create_session()
        o1 = sess.query(Order).first()
        def go():
            assert o1.description == 'order 1'
        self.assert_sql_count(testing.db, go, 1)

        sess = create_session()
        o1 = sess.query(Order).add_column(orders.c.description).first()[0]
        def go():
            assert o1.description == 'order 1'
        self.assert_sql_count(testing.db, go, 0)

    def test_deepoptions(self):
        m = mapper(User, users, properties={
            'orders':relation(mapper(Order, orders, properties={
                'items':relation(mapper(Item, orderitems, properties={
                    'item_name':deferred(orderitems.c.item_name)
                }))
            }))
        })
        sess = create_session()
        q = sess.query(m)
        l = q.all()
        item = l[0].orders[1].items[1]
        def go():
            print item.item_name
        self.assert_sql_count(testing.db, go, 1)
        self.assert_(item.item_name == 'item 4')
        sess.clear()
        q2 = q.options(undefer('orders.items.item_name'))
        l = q2.all()
        item = l[0].orders[1].items[1]
        def go():
            print item.item_name
        self.assert_sql_count(testing.db, go, 0)
        self.assert_(item.item_name == 'item 4')

class CompositeTypesTest(ORMTest):
    def define_tables(self, metadata):
        global graphs, edges
        graphs = Table('graphs', metadata,
            Column('id', Integer, primary_key=True),
            Column('version_id', Integer, primary_key=True),
            Column('name', String(30)))

        edges = Table('edges', metadata,
            Column('id', Integer, primary_key=True),
            Column('graph_id', Integer, nullable=False),
            Column('graph_version_id', Integer, nullable=False),
            Column('x1', Integer),
            Column('y1', Integer),
            Column('x2', Integer),
            Column('y2', Integer),
            ForeignKeyConstraint(['graph_id', 'graph_version_id'], ['graphs.id', 'graphs.version_id'])
            )

    def test_basic(self):
        class Point(object):
            def __init__(self, x, y):
                self.x = x
                self.y = y
            def __composite_values__(self):
                return [self.x, self.y]
            def __eq__(self, other):
                return other.x == self.x and other.y == self.y
            def __ne__(self, other):
                return not self.__eq__(other)

        class Graph(object):
            pass
        class Edge(object):
            def __init__(self, start, end):
                self.start = start
                self.end = end

        mapper(Graph, graphs, properties={
            'edges':relation(Edge)
        })
        mapper(Edge, edges, properties={
            'start':composite(Point, edges.c.x1, edges.c.y1),
            'end':composite(Point, edges.c.x2, edges.c.y2)
        })

        sess = create_session()
        g = Graph()
        g.id = 1
        g.version_id=1
        g.edges.append(Edge(Point(3, 4), Point(5, 6)))
        g.edges.append(Edge(Point(14, 5), Point(2, 7)))
        sess.save(g)
        sess.flush()

        sess.clear()
        g2 = sess.query(Graph).get([g.id, g.version_id])
        for e1, e2 in zip(g.edges, g2.edges):
            assert e1.start == e2.start
            assert e1.end == e2.end

        g2.edges[1].end = Point(18, 4)
        sess.flush()
        sess.clear()
        e = sess.query(Edge).get(g2.edges[1].id)
        assert e.end == Point(18, 4)

        e.end.x = 19
        e.end.y = 5
        sess.flush()
        sess.clear()
        assert sess.query(Edge).get(g2.edges[1].id).end == Point(19, 5)

        g.edges[1].end = Point(19, 5)

        sess.clear()
        def go():
            g2 = sess.query(Graph).options(eagerload('edges')).get([g.id, g.version_id])
            for e1, e2 in zip(g.edges, g2.edges):
                assert e1.start == e2.start
                assert e1.end == e2.end
        self.assert_sql_count(testing.db, go, 1)

        # test comparison of CompositeProperties to their object instances
        g = sess.query(Graph).get([1, 1])
        assert sess.query(Edge).filter(Edge.start==Point(3, 4)).one() is g.edges[0]

        assert sess.query(Edge).filter(Edge.start!=Point(3, 4)).first() is g.edges[1]

        assert sess.query(Edge).filter(Edge.start==None).all() == []


    def test_pk(self):
        """test using a composite type as a primary key"""

        class Version(object):
            def __init__(self, id, version):