eager_relations.py

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

        mapper(Item, items)


        sess = create_session()
        def go():
            ret = sess.query(User).add_entity(Order).join('orders', aliased=True).order_by(User.id).order_by(Order.id).all()
            self.assertEquals(ret, self._assert_result())
        self.assert_sql_count(testing.db, go, 1)

    def test_options(self):
        mapper(User, users, properties={
            'addresses':relation(Address),
            'orders':relation(Order)
        })
        mapper(Address, addresses)
        mapper(Order, orders, properties={
            'items':relation(Item, secondary=order_items, order_by=items.c.id)
        })
        mapper(Item, items)

        sess = create_session()

        def go():
            ret = sess.query(User).options(eagerload('addresses')).add_entity(Order).join('orders', aliased=True).order_by(User.id).order_by(Order.id).all()
            self.assertEquals(ret, self._assert_result())
        self.assert_sql_count(testing.db, go, 6)

        sess.clear()
        def go():
            ret = sess.query(User).options(eagerload('addresses')).add_entity(Order).options(eagerload('items', Order)).join('orders', aliased=True).order_by(User.id).order_by(Order.id).all()
            self.assertEquals(ret, self._assert_result())
        self.assert_sql_count(testing.db, go, 1)

class SelfReferentialEagerTest(ORMTest):
    def define_tables(self, metadata):
        global nodes
        nodes = Table('nodes', metadata,
            Column('id', Integer, Sequence('node_id_seq', optional=True), primary_key=True),
            Column('parent_id', Integer, ForeignKey('nodes.id')),
            Column('data', String(30)))

    @testing.fails_on('maxdb')
    def test_basic(self):
        class Node(Base):
            def append(self, node):
                self.children.append(node)

        mapper(Node, nodes, properties={
            'children':relation(Node, lazy=False, join_depth=3)
        })
        sess = create_session()
        n1 = Node(data='n1')
        n1.append(Node(data='n11'))
        n1.append(Node(data='n12'))
        n1.append(Node(data='n13'))
        n1.children[1].append(Node(data='n121'))
        n1.children[1].append(Node(data='n122'))
        n1.children[1].append(Node(data='n123'))
        sess.save(n1)
        sess.flush()
        sess.clear()
        def go():
            d = sess.query(Node).filter_by(data='n1').first()
            assert Node(data='n1', children=[
                Node(data='n11'),
                Node(data='n12', children=[
                    Node(data='n121'),
                    Node(data='n122'),
                    Node(data='n123')
                ]),
                Node(data='n13')
            ]) == d
        self.assert_sql_count(testing.db, go, 1)


    def test_lazy_fallback_doesnt_affect_eager(self):
        class Node(Base):
            def append(self, node):
                self.children.append(node)

        mapper(Node, nodes, properties={
            'children':relation(Node, lazy=False, join_depth=1)
        })
        sess = create_session()
        n1 = Node(data='n1')
        n1.append(Node(data='n11'))
        n1.append(Node(data='n12'))
        n1.append(Node(data='n13'))
        n1.children[1].append(Node(data='n121'))
        n1.children[1].append(Node(data='n122'))
        n1.children[1].append(Node(data='n123'))
        sess.save(n1)
        sess.flush()
        sess.clear()

        # eager load with join depth 1.  when eager load of 'n1'
        # hits the children of 'n12', no columns are present, eager loader
        # degrades to lazy loader; fine.  but then, 'n12' is *also* in the
        # first level of columns since we're loading the whole table.
        # when those rows arrive, now we *can* eager load its children and an
        # eager collection should be initialized.  essentially the 'n12' instance
        # is present in not just two different rows but two distinct sets of columns
        # in this result set.
        def go():
            allnodes = sess.query(Node).order_by(Node.data).all()
            n12 = allnodes[2]
            assert n12.data == 'n12'
            print "N12 IS", id(n12)
            print [c.data for c in n12.children]
            assert [
                Node(data='n121'),
                Node(data='n122'),
                Node(data='n123')
            ] == list(n12.children)
        self.assert_sql_count(testing.db, go, 1)

    def test_with_deferred(self):
        class Node(Base):
            def append(self, node):
                self.children.append(node)

        mapper(Node, nodes, properties={
            'children':relation(Node, lazy=False, join_depth=3),
            'data':deferred(nodes.c.data)
        })
        sess = create_session()
        n1 = Node(data='n1')
        n1.append(Node(data='n11'))
        n1.append(Node(data='n12'))
        sess.save(n1)
        sess.flush()
        sess.clear()

        def go():
            assert Node(data='n1', children=[Node(data='n11'), Node(data='n12')]) == sess.query(Node).first()
        self.assert_sql_count(testing.db, go, 4)

        sess.clear()

        def go():
            assert Node(data='n1', children=[Node(data='n11'), Node(data='n12')]) == sess.query(Node).options(undefer('data')).first()
        self.assert_sql_count(testing.db, go, 3)

        sess.clear()

        def go():
            assert Node(data='n1', children=[Node(data='n11'), Node(data='n12')]) == sess.query(Node).options(undefer('data'), undefer('children.data')).first()
        self.assert_sql_count(testing.db, go, 1)



    def test_options(self):
        class Node(Base):
            def append(self, node):
                self.children.append(node)

        mapper(Node, nodes, properties={
            'children':relation(Node, lazy=True)
        })
        sess = create_session()
        n1 = Node(data='n1')
        n1.append(Node(data='n11'))
        n1.append(Node(data='n12'))
        n1.append(Node(data='n13'))
        n1.children[1].append(Node(data='n121'))
        n1.children[1].append(Node(data='n122'))
        n1.children[1].append(Node(data='n123'))
        sess.save(n1)
        sess.flush()
        sess.clear()
        def go():
            d = sess.query(Node).filter_by(data='n1').options(eagerload('children.children')).first()
            assert Node(data='n1', children=[
                Node(data='n11'),
                Node(data='n12', children=[
                    Node(data='n121'),
                    Node(data='n122'),
                    Node(data='n123')
                ]),
                Node(data='n13')
            ]) == d
        self.assert_sql_count(testing.db, go, 2)

        def go():
            d = sess.query(Node).filter_by(data='n1').options(eagerload('children.children')).first()

        # test that the query isn't wrapping the initial query for eager loading.
        # testing only sqlite for now since the query text is slightly different on other
        # dialects
        if testing.against('sqlite'):
            self.assert_sql(testing.db, go, [
                (
                    "SELECT nodes.id AS nodes_id, nodes.parent_id AS nodes_parent_id, nodes.data AS nodes_data FROM nodes WHERE nodes.data = :nodes_data_1 ORDER BY nodes.oid  LIMIT 1 OFFSET 0",
                    {'nodes_data_1': 'n1'}
                ),
            ])

    @testing.fails_on('maxdb')
    def test_no_depth(self):
        class Node(Base):
            def append(self, node):
                self.children.append(node)

        mapper(Node, nodes, properties={
            'children':relation(Node, lazy=False)
        })
        sess = create_session()
        n1 = Node(data='n1')
        n1.append(Node(data='n11'))
        n1.append(Node(data='n12'))
        n1.append(Node(data='n13'))
        n1.children[1].append(Node(data='n121'))
        n1.children[1].append(Node(data='n122'))
        n1.children[1].append(Node(data='n123'))
        sess.save(n1)
        sess.flush()
        sess.clear()
        def go():
            d = sess.query(Node).filter_by(data='n1').first()
            assert Node(data='n1', children=[
                Node(data='n11'),
                Node(data='n12', children=[
                    Node(data='n121'),
                    Node(data='n122'),
                    Node(data='n123')
                ]),
                Node(data='n13')
            ]) == d
        self.assert_sql_count(testing.db, go, 3)

class SelfReferentialM2MEagerTest(ORMTest):
    def define_tables(self, metadata):
        global widget, widget_rel

        widget = Table('widget', metadata,
            Column('id', Integer, primary_key=True),
            Column('name', Unicode(40), nullable=False, unique=True),
        )

        widget_rel = Table('widget_rel', metadata,
            Column('parent_id', Integer, ForeignKey('widget.id')),
            Column('child_id', Integer, ForeignKey('widget.id')),
            UniqueConstraint('parent_id', 'child_id'),
        )
    def test_basic(self):
        class Widget(Base):
            pass

        mapper(Widget, widget, properties={
            'children': relation(Widget, secondary=widget_rel,
                primaryjoin=widget_rel.c.parent_id==widget.c.id,
                secondaryjoin=widget_rel.c.child_id==widget.c.id,
                lazy=False, join_depth=1,
            )
        })

        sess = create_session()
        w1 = Widget(name=u'w1')
        w2 = Widget(name=u'w2')
        w1.children.append(w2)
        sess.save(w1)
        sess.flush()
        sess.clear()

#        l = sess.query(Widget).filter(Widget.name=='w1').all()
#        print l
        assert [Widget(name='w1', children=[Widget(name='w2')])] == sess.query(Widget).filter(Widget.name==u'w1').all()

class CyclicalInheritingEagerTest(ORMTest):
    def define_tables(self, metadata):
        global t1, t2
        t1 = Table('t1', metadata,
            Column('c1', Integer, primary_key=True),
            Column('c2', String(30)),
            Column('type', String(30))
            )

        t2 = Table('t2', metadata,
            Column('c1', Integer, primary_key=True),
            Column('c2', String(30)),
            Column('type', String(30)),
            Column('t1.id', Integer, ForeignKey('t1.c1')))

    def test_basic(self):
        class T(object):
            pass

        class SubT(T):
            pass

        class T2(object):
            pass

        class SubT2(T2):
            pass

        mapper(T, t1, polymorphic_on=t1.c.type, polymorphic_identity='t1')
        mapper(SubT, None, inherits=T, polymorphic_identity='subt1', properties={
            't2s':relation(SubT2, lazy=False, backref=backref('subt', lazy=False))
        })
        mapper(T2, t2, polymorphic_on=t2.c.type, polymorphic_identity='t2')
        mapper(SubT2, None, inherits=T2, polymorphic_identity='subt2')

        # testing a particular endless loop condition in eager join setup
        create_session().query(SubT).all()

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