eager_relations.py

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

"""basic tests of eager loaded attributes"""

import testenv; testenv.configure_for_tests()
from sqlalchemy import *
from sqlalchemy.orm import *
from testlib import *
from testlib.fixtures import *
from query import QueryTest

class EagerTest(FixtureTest):
    keep_mappers = False
    keep_data = True

    def test_basic(self):
        mapper(User, users, properties={
            'addresses':relation(mapper(Address, addresses), lazy=False)
        })
        sess = create_session()
        q = sess.query(User)

        assert [User(id=7, addresses=[Address(id=1, email_address='[e-mail]')])] == q.filter(User.id==7).all()
        assert fixtures.user_address_result == q.all()

    def test_no_orphan(self):
        """test that an eagerly loaded child object is not marked as an orphan"""

        mapper(User, users, properties={
            'addresses':relation(Address, cascade="all,delete-orphan", lazy=False)
        })
        mapper(Address, addresses)

        sess = create_session()
        user = sess.query(User).get(7)
        assert getattr(User, 'addresses').hasparent(user.addresses[0], optimistic=True)
        assert not class_mapper(Address)._is_orphan(user.addresses[0])

    def test_orderby(self):
        mapper(User, users, properties = {
            'addresses':relation(mapper(Address, addresses), lazy=False, order_by=addresses.c.email_address),
        })
        q = create_session().query(User)
        assert [
            User(id=7, addresses=[
                Address(id=1)
            ]),
            User(id=8, addresses=[
                Address(id=3, email_address='[e-mail]'),
                Address(id=4, email_address='[e-mail]'),
                Address(id=2, email_address='[e-mail]')
            ]),
            User(id=9, addresses=[
                Address(id=5)
            ]),
            User(id=10, addresses=[])
        ] == q.all()

    def test_orderby_multi(self):
        mapper(User, users, properties = {
            'addresses':relation(mapper(Address, addresses), lazy=False, order_by=[addresses.c.email_address, addresses.c.id]),
        })
        q = create_session().query(User)
        assert [
            User(id=7, addresses=[
                Address(id=1)
            ]),
            User(id=8, addresses=[
                Address(id=3, email_address='[e-mail]'),
                Address(id=4, email_address='[e-mail]'),
                Address(id=2, email_address='[e-mail]')
            ]),
            User(id=9, addresses=[
                Address(id=5)
            ]),
            User(id=10, addresses=[])
        ] == q.all()

    def test_orderby_secondary(self):
        """tests that a regular mapper select on a single table can order by a relation to a second table"""

        mapper(Address, addresses)

        mapper(User, users, properties = dict(
            addresses = relation(Address, lazy=False),
        ))

        q = create_session().query(User)
        l = q.filter(User.id==Address.user_id).order_by(Address.email_address).all()

        assert [
            User(id=8, addresses=[
                Address(id=2, email_address='[e-mail]'),
                Address(id=3, email_address='[e-mail]'),
                Address(id=4, email_address='[e-mail]'),
            ]),
            User(id=9, addresses=[
                Address(id=5)
            ]),
            User(id=7, addresses=[
                Address(id=1)
            ]),
        ] == l

    def test_orderby_desc(self):
        mapper(Address, addresses)

        mapper(User, users, properties = dict(
            addresses = relation(Address, lazy=False, order_by=[desc(addresses.c.email_address)]),
        ))
        sess = create_session()
        assert [
            User(id=7, addresses=[
                Address(id=1)
            ]),
            User(id=8, addresses=[
                Address(id=2, email_address='[e-mail]'),
                Address(id=4, email_address='[e-mail]'),
                Address(id=3, email_address='[e-mail]'),
            ]),
            User(id=9, addresses=[
                Address(id=5)
            ]),
            User(id=10, addresses=[])
        ] == sess.query(User).all()

    def test_deferred_fk_col(self):
        mapper(Address, addresses, properties={
            'user_id':deferred(addresses.c.user_id),
            'user':relation(User, lazy=False)
        })
        mapper(User, users)

        assert [Address(id=1, user=User(id=7)), Address(id=4, user=User(id=8)), Address(id=5, user=User(id=9))] == create_session().query(Address).filter(Address.id.in_([1, 4, 5])).all()

        assert [Address(id=1, user=User(id=7)), Address(id=4, user=User(id=8)), Address(id=5, user=User(id=9))] == create_session().query(Address).filter(Address.id.in_([1, 4, 5])).limit(3).all()

        sess = create_session()
        a = sess.query(Address).get(1)
        def go():
            assert a.user_id==7
        # assert that the eager loader added 'user_id' to the row
        # and deferred loading of that col was disabled
        self.assert_sql_count(testing.db, go, 0)

        # do the mapping in reverse
        # (we would have just used an "addresses" backref but the test fixtures then require the whole
        # backref to be set up, lazy loaders trigger, etc.)
        clear_mappers()

        mapper(Address, addresses, properties={
            'user_id':deferred(addresses.c.user_id),
        })
        mapper(User, users, properties={'addresses':relation(Address, lazy=False)})

        assert [User(id=7, addresses=[Address(id=1)])] == create_session().query(User).filter(User.id==7).all()

        assert [User(id=7, addresses=[Address(id=1)])] == create_session().query(User).limit(1).filter(User.id==7).all()

        sess = create_session()
        u = sess.query(User).get(7)
        def go():
            assert u.addresses[0].user_id==7
        # assert that the eager loader didn't have to affect 'user_id' here
        # and that its still deferred
        self.assert_sql_count(testing.db, go, 1)

        clear_mappers()

        mapper(User, users, properties={'addresses':relation(Address, lazy=False)})
        mapper(Address, addresses, properties={
            'user_id':deferred(addresses.c.user_id),
            'dingalings':relation(Dingaling, lazy=False)
        })
        mapper(Dingaling, dingalings, properties={
            'address_id':deferred(dingalings.c.address_id)
        })
        sess = create_session()
        def go():
            u = sess.query(User).limit(1).get(8)
            assert User(id=8, addresses=[Address(id=2, dingalings=[Dingaling(id=1)]), Address(id=3), Address(id=4)]) == u
        self.assert_sql_count(testing.db, go, 1)

    def test_many_to_many(self):

        mapper(Keyword, keywords)
        mapper(Item, items, properties = dict(
                keywords = relation(Keyword, secondary=item_keywords, lazy=False, order_by=keywords.c.id),
        ))

        q = create_session().query(Item)
        def go():
            assert fixtures.item_keyword_result == q.all()
        self.assert_sql_count(testing.db, go, 1)

        def go():
            assert fixtures.item_keyword_result[0:2] == q.join('keywords').filter(keywords.c.name == 'red').all()
        self.assert_sql_count(testing.db, go, 1)

        def go():
            assert fixtures.item_keyword_result[0:2] == q.join('keywords', aliased=True).filter(keywords.c.name == 'red').all()
        self.assert_sql_count(testing.db, go, 1)


    def test_eager_option(self):
        mapper(Keyword, keywords)
        mapper(Item, items, properties = dict(
                keywords = relation(Keyword, secondary=item_keywords, lazy=True, order_by=keywords.c.id),
        ))

        q = create_session().query(Item)

        def go():
            assert fixtures.item_keyword_result[0:2] == q.options(eagerload('keywords')).join('keywords').filter(keywords.c.name == 'red').all()

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

    def test_cyclical(self):
        """test that a circular eager relationship breaks the cycle with a lazy loader"""

        mapper(Address, addresses)
        mapper(User, users, properties = dict(
            addresses = relation(Address, lazy=False, backref=backref('user', lazy=False))
        ))
        assert class_mapper(User).get_property('addresses').lazy is False
        assert class_mapper(Address).get_property('user').lazy is False

        sess = create_session()
        assert fixtures.user_address_result == sess.query(User).all()

    def test_double(self):
        """tests eager loading with two relations simulatneously, from the same table, using aliases.  """
        openorders = alias(orders, 'openorders')
        closedorders = alias(orders, 'closedorders')

        mapper(Address, addresses)

        mapper(User, users, properties = dict(
            addresses = relation(Address, lazy=False),
            open_orders = relation(mapper(Order, openorders, entity_name='open'), primaryjoin = and_(openorders.c.isopen == 1, users.c.id==openorders.c.user_id), lazy=False),
            closed_orders = relation(mapper(Order, closedorders,entity_name='closed'), primaryjoin = and_(closedorders.c.isopen == 0, users.c.id==closedorders.c.user_id), lazy=False)
        ))
        q = create_session().query(User)

        def go():
            assert [
                User(
                    id=7,
                    addresses=[Address(id=1)],
                    open_orders = [Order(id=3)],
                    closed_orders = [Order(id=1), Order(id=5)]
                ),
                User(
                    id=8,
                    addresses=[Address(id=2), Address(id=3), Address(id=4)],
                    open_orders = [],
                    closed_orders = []
                ),
                User(
                    id=9,
                    addresses=[Address(id=5)],
                    open_orders = [Order(id=4)],
                    closed_orders = [Order(id=2)]
                ),
                User(id=10)

            ] == q.all()
        self.assert_sql_count(testing.db, go, 1)

    def test_double_same_mappers(self):
        """tests eager loading with two relations simulatneously, from the same table, using aliases.  """

        mapper(Address, addresses)
        mapper(Order, orders, properties={
            'items':relation(Item, secondary=order_items, lazy=False, order_by=items.c.id),
        })
        mapper(Item, items)
        mapper(User, users, properties = dict(
            addresses = relation(Address, lazy=False),
            open_orders = relation(Order, primaryjoin = and_(orders.c.isopen == 1, users.c.id==orders.c.user_id), lazy=False),
            closed_orders = relation(Order, primaryjoin = and_(orders.c.isopen == 0, users.c.id==orders.c.user_id), lazy=False)
        ))
        q = create_session().query(User)

        def go():
            assert [
                User(
                    id=7,
                    addresses=[Address(id=1)],
                    open_orders = [Order(id=3, items=[Item(id=3), Item(id=4), Item(id=5)])],
                    closed_orders = [Order(id=1, items=[Item(id=1), Item(id=2), Item(id=3)]), Order(id=5, items=[Item(id=5)])]
                ),
                User(
                    id=8,
                    addresses=[Address(id=2), Address(id=3), Address(id=4)],
                    open_orders = [],
                    closed_orders = []
                ),
                User(
                    id=9,
                    addresses=[Address(id=5)],
                    open_orders = [Order(id=4, items=[Item(id=1), Item(id=5)])],
                    closed_orders = [Order(id=2, items=[Item(id=1), Item(id=2), Item(id=3)])]
                ),
                User(id=10)

            ] == q.all()
        self.assert_sql_count(testing.db, go, 1)

    def test_no_false_hits(self):
        """test that eager loaders don't interpret main table columns as part of their eager load."""