diff options
| author | Mike Bayer <mike_mp@zzzcomputing.com> | 2014-08-26 16:34:41 -0400 |
|---|---|---|
| committer | Mike Bayer <mike_mp@zzzcomputing.com> | 2014-08-26 16:36:56 -0400 |
| commit | e346ee2c5776e21d4b37f9b495db943b21b47da2 (patch) | |
| tree | 6682e30162e67e7e39823fd7545605582647abf1 | |
| parent | f44a4b6437eb3d4340c6c0fa790bc4fa885085ea (diff) | |
| download | sqlalchemy-e346ee2c5776e21d4b37f9b495db943b21b47da2.tar.gz | |
- hard-won pep8 formatting of one of the giants...
| -rw-r--r-- | test/orm/test_eager_relations.py | 2489 |
1 files changed, 1476 insertions, 1013 deletions
diff --git a/test/orm/test_eager_relations.py b/test/orm/test_eager_relations.py index 645d9e00a..ad16c83e1 100644 --- a/test/orm/test_eager_relations.py +++ b/test/orm/test_eager_relations.py @@ -4,7 +4,7 @@ from sqlalchemy.testing import eq_, is_, is_not_ import sqlalchemy as sa from sqlalchemy import testing from sqlalchemy.orm import joinedload, deferred, undefer, \ - joinedload_all, backref, eagerload, Session, immediateload,\ + joinedload_all, backref, Session,\ defaultload, Load from sqlalchemy import Integer, String, Date, ForeignKey, and_, select, \ func @@ -12,8 +12,7 @@ from sqlalchemy.testing.schema import Table, Column from sqlalchemy.orm import mapper, relationship, create_session, \ lazyload, aliased, column_property from sqlalchemy.sql import operators -from sqlalchemy.testing import eq_, assert_raises, \ - assert_raises_message +from sqlalchemy.testing import assert_raises, assert_raises_message from sqlalchemy.testing.assertsql import CompiledSQL from sqlalchemy.testing import fixtures from test.orm import _fixtures @@ -27,26 +26,30 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): __dialect__ = 'default' def test_basic(self): - users, Address, addresses, User = (self.tables.users, - self.classes.Address, - self.tables.addresses, - self.classes.User) + users, Address, addresses, User = ( + self.tables.users, + self.classes.Address, + self.tables.addresses, + self.classes.User) mapper(User, users, properties={ - 'addresses':relationship(mapper(Address, addresses), lazy='joined', order_by=Address.id) + 'addresses': relationship( + mapper(Address, addresses), lazy='joined', order_by=Address.id) }) sess = create_session() q = sess.query(User) - eq_([User(id=7, addresses=[Address(id=1, email_address='jack@bean.com')])], - q.filter(User.id==7).all()) + eq_([User(id=7, addresses=[ + Address(id=1, email_address='jack@bean.com')])], + q.filter(User.id == 7).all()) eq_(self.static.user_address_result, q.order_by(User.id).all()) def test_late_compile(self): - User, Address, addresses, users = (self.classes.User, - self.classes.Address, - self.tables.addresses, - self.tables.users) + User, Address, addresses, users = ( + self.classes.User, + self.classes.Address, + self.tables.addresses, + self.tables.users) m = mapper(User, users) sess = create_session() @@ -54,44 +57,52 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): m.add_property("addresses", relationship(mapper(Address, addresses))) sess.expunge_all() + def go(): eq_( - [User(id=7, addresses=[Address(id=1, email_address='jack@bean.com')])], - sess.query(User).options(joinedload('addresses')).filter(User.id==7).all() + [User(id=7, addresses=[ + Address(id=1, email_address='jack@bean.com')])], + sess.query(User).options( + joinedload('addresses')).filter(User.id == 7).all() ) self.assert_sql_count(testing.db, go, 1) - def test_no_orphan(self): """An eagerly loaded child object is not marked as an orphan""" - users, Address, addresses, User = (self.tables.users, - self.classes.Address, - self.tables.addresses, - self.classes.User) - + users, Address, addresses, User = ( + self.tables.users, + self.classes.Address, + self.tables.addresses, + self.classes.User) mapper(User, users, properties={ - 'addresses':relationship(Address, cascade="all,delete-orphan", lazy='joined') + 'addresses': relationship( + Address, cascade="all,delete-orphan", lazy='joined') }) mapper(Address, addresses) sess = create_session() user = sess.query(User).get(7) assert getattr(User, 'addresses').\ - hasparent(sa.orm.attributes.instance_state(user.addresses[0]), optimistic=True) + hasparent( + sa.orm.attributes.instance_state( + user.addresses[0]), optimistic=True) assert not sa.orm.class_mapper(Address).\ - _is_orphan(sa.orm.attributes.instance_state(user.addresses[0])) + _is_orphan( + sa.orm.attributes.instance_state(user.addresses[0])) def test_orderby(self): - users, Address, addresses, User = (self.tables.users, - self.classes.Address, - self.tables.addresses, - self.classes.User) - - mapper(User, users, properties = { - 'addresses':relationship(mapper(Address, addresses), - lazy='joined', order_by=addresses.c.email_address), + users, Address, addresses, User = ( + self.tables.users, + self.classes.Address, + self.tables.addresses, + self.classes.User) + + mapper(User, users, properties={ + 'addresses': relationship( + mapper(Address, addresses), + lazy='joined', order_by=addresses.c.email_address), }) q = create_session().query(User) eq_([ @@ -107,18 +118,20 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): Address(id=5) ]), User(id=10, addresses=[]) - ], q.order_by(User.id).all()) + ], q.order_by(User.id).all()) def test_orderby_multi(self): - users, Address, addresses, User = (self.tables.users, - self.classes.Address, - self.tables.addresses, - self.classes.User) - - mapper(User, users, properties = { - 'addresses':relationship(mapper(Address, addresses), - lazy='joined', - order_by=[addresses.c.email_address, addresses.c.id]), + users, Address, addresses, User = ( + self.tables.users, + self.classes.Address, + self.tables.addresses, + self.classes.User) + + mapper(User, users, properties={ + 'addresses': relationship( + mapper(Address, addresses), + lazy='joined', + order_by=[addresses.c.email_address, addresses.c.id]), }) q = create_session().query(User) eq_([ @@ -134,25 +147,26 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): Address(id=5) ]), User(id=10, addresses=[]) - ], q.order_by(User.id).all()) + ], q.order_by(User.id).all()) def test_orderby_related(self): """A regular mapper select on a single table can order by a relationship to a second table""" Address, addresses, users, User = (self.classes.Address, - self.tables.addresses, - self.tables.users, - self.classes.User) - + self.tables.addresses, + self.tables.users, + self.classes.User) mapper(Address, addresses) - mapper(User, users, properties = dict( - addresses = relationship(Address, lazy='joined', order_by=addresses.c.id), + mapper(User, users, properties=dict( + addresses=relationship( + Address, lazy='joined', order_by=addresses.c.id), )) q = create_session().query(User) - l = q.filter(User.id==Address.user_id).order_by(Address.email_address).all() + l = q.filter(User.id == Address.user_id).order_by( + Address.email_address).all() eq_([ User(id=8, addresses=[ @@ -166,18 +180,19 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): User(id=7, addresses=[ Address(id=1) ]), - ], l) + ], l) def test_orderby_desc(self): Address, addresses, users, User = (self.classes.Address, - self.tables.addresses, - self.tables.users, - self.classes.User) + self.tables.addresses, + self.tables.users, + self.classes.User) mapper(Address, addresses) - mapper(User, users, properties = dict( - addresses = relationship(Address, lazy='joined', - order_by=[sa.desc(addresses.c.email_address)]), + mapper(User, users, properties=dict( + addresses=relationship( + Address, lazy='joined', + order_by=[sa.desc(addresses.c.email_address)]), )) sess = create_session() eq_([ @@ -193,37 +208,43 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): Address(id=5) ]), User(id=10, addresses=[]) - ], sess.query(User).order_by(User.id).all()) + ], sess.query(User).order_by(User.id).all()) def test_deferred_fk_col(self): - users, Dingaling, User, dingalings, Address, addresses = (self.tables.users, - self.classes.Dingaling, - self.classes.User, - self.tables.dingalings, - self.classes.Address, - self.tables.addresses) + users, Dingaling, User, dingalings, Address, addresses = ( + self.tables.users, + self.classes.Dingaling, + self.classes.User, + self.tables.dingalings, + self.classes.Address, + self.tables.addresses) mapper(Address, addresses, properties={ - 'user_id':deferred(addresses.c.user_id), - 'user':relationship(User, lazy='joined') + 'user_id': deferred(addresses.c.user_id), + 'user': relationship(User, lazy='joined') }) mapper(User, users) sess = create_session() for q in [ - sess.query(Address).filter(Address.id.in_([1, 4, 5])).order_by(Address.id), - sess.query(Address).filter(Address.id.in_([1, 4, 5])).order_by(Address.id).limit(3) + sess.query(Address).filter( + Address.id.in_([1, 4, 5]) + ).order_by(Address.id), + sess.query(Address).filter( + Address.id.in_([1, 4, 5]) + ).order_by(Address.id).limit(3) ]: sess.expunge_all() eq_(q.all(), [Address(id=1, user=User(id=7)), Address(id=4, user=User(id=8)), Address(id=5, user=User(id=9))] - ) + ) sess.expunge_all() - a = sess.query(Address).filter(Address.id==1).all()[0] + a = sess.query(Address).filter(Address.id == 1).all()[0] + def go(): eq_(a.user_id, 7) # assert that the eager loader added 'user_id' to the row and deferred @@ -231,7 +252,8 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): self.assert_sql_count(testing.db, go, 0) sess.expunge_all() - a = sess.query(Address).filter(Address.id==1).first() + a = sess.query(Address).filter(Address.id == 1).first() + def go(): eq_(a.user_id, 7) # assert that the eager loader added 'user_id' to the row and deferred @@ -245,22 +267,23 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): sa.orm.clear_mappers() mapper(Address, addresses, properties={ - 'user_id':deferred(addresses.c.user_id), + 'user_id': deferred(addresses.c.user_id), }) mapper(User, users, properties={ - 'addresses':relationship(Address, lazy='joined')}) + 'addresses': relationship(Address, lazy='joined')}) for q in [ - sess.query(User).filter(User.id==7), - sess.query(User).filter(User.id==7).limit(1) + sess.query(User).filter(User.id == 7), + sess.query(User).filter(User.id == 7).limit(1) ]: sess.expunge_all() eq_(q.all(), [User(id=7, addresses=[Address(id=1)])] - ) + ) sess.expunge_all() u = sess.query(User).get(7) + def go(): eq_(u.addresses[0].user_id, 7) # assert that the eager loader didn't have to affect 'user_id' here @@ -270,14 +293,15 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): sa.orm.clear_mappers() mapper(User, users, properties={ - 'addresses':relationship(Address, lazy='joined', - order_by=addresses.c.id)}) + 'addresses': relationship(Address, lazy='joined', + order_by=addresses.c.id)}) mapper(Address, addresses, properties={ - 'user_id':deferred(addresses.c.user_id), - 'dingalings':relationship(Dingaling, lazy='joined')}) + 'user_id': deferred(addresses.c.user_id), + 'dingalings': relationship(Dingaling, lazy='joined')}) mapper(Dingaling, dingalings, properties={ - 'address_id':deferred(dingalings.c.address_id)}) + 'address_id': deferred(dingalings.c.address_id)}) sess.expunge_all() + def go(): u = sess.query(User).get(8) eq_(User(id=8, @@ -288,33 +312,34 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): self.assert_sql_count(testing.db, go, 1) def test_options_pathing(self): - users, Keyword, orders, items, order_items, Order, Item, User, keywords, item_keywords = (self.tables.users, - self.classes.Keyword, - self.tables.orders, - self.tables.items, - self.tables.order_items, - self.classes.Order, - self.classes.Item, - self.classes.User, - self.tables.keywords, - self.tables.item_keywords) + users, Keyword, orders, items, order_items, \ + Order, Item, User, keywords, item_keywords = ( + self.tables.users, + self.classes.Keyword, + self.tables.orders, + self.tables.items, + self.tables.order_items, + self.classes.Order, + self.classes.Item, + self.classes.User, + self.tables.keywords, + self.tables.item_keywords) mapper(User, users, properties={ - 'orders':relationship(Order, order_by=orders.c.id), # o2m, m2o + 'orders': relationship(Order, order_by=orders.c.id), # o2m, m2o }) mapper(Order, orders, properties={ - 'items':relationship(Item, - secondary=order_items, order_by=items.c.id), #m2m + 'items': relationship( + Item, + secondary=order_items, order_by=items.c.id), # m2m }) mapper(Item, items, properties={ - 'keywords':relationship(Keyword, - secondary=item_keywords, - order_by=keywords.c.id) #m2m + 'keywords': relationship(Keyword, + secondary=item_keywords, + order_by=keywords.c.id) # m2m }) mapper(Keyword, keywords) - - for opt, count in [ (( joinedload(User.orders, Order.items), @@ -322,18 +347,19 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): ((joinedload("orders.items"), ), 10), (( joinedload(User.orders, ), - joinedload(User.orders, Order.items), + joinedload(User.orders, Order.items), joinedload(User.orders, Order.items, Item.keywords), - ), 1), + ), 1), (( joinedload(User.orders, Order.items, Item.keywords), ), 10), (( - joinedload(User.orders, Order.items), - joinedload(User.orders, Order.items, Item.keywords), + joinedload(User.orders, Order.items), + joinedload(User.orders, Order.items, Item.keywords), ), 5), ]: sess = create_session() + def go(): eq_( sess.query(User).options(*opt).order_by(User.id).all(), @@ -344,37 +370,39 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): def test_disable_dynamic(self): """test no joined option on a dynamic.""" - users, Address, addresses, User = (self.tables.users, - self.classes.Address, - self.tables.addresses, - self.classes.User) - + users, Address, addresses, User = ( + self.tables.users, + self.classes.Address, + self.tables.addresses, + self.classes.User) mapper(User, users, properties={ - 'addresses':relationship(Address, lazy="dynamic") + 'addresses': relationship(Address, lazy="dynamic") }) mapper(Address, addresses) sess = create_session() assert_raises_message( sa.exc.InvalidRequestError, - "User.addresses' does not support object population - eager loading cannot be applied.", + "User.addresses' does not support object " + "population - eager loading cannot be applied.", sess.query(User).options(joinedload(User.addresses)).first, ) def test_many_to_many(self): - keywords, items, item_keywords, Keyword, Item = (self.tables.keywords, - self.tables.items, - self.tables.item_keywords, - self.classes.Keyword, - self.classes.Item) - + keywords, items, item_keywords, Keyword, Item = ( + self.tables.keywords, + self.tables.items, + self.tables.item_keywords, + self.classes.Keyword, + self.classes.Item) mapper(Keyword, keywords) - mapper(Item, items, properties = dict( - keywords = relationship(Keyword, secondary=item_keywords, - lazy='joined', order_by=keywords.c.id))) + mapper(Item, items, properties=dict( + keywords=relationship(Keyword, secondary=item_keywords, + lazy='joined', order_by=keywords.c.id))) q = create_session().query(Item).order_by(Item.id) + def go(): eq_(self.static.item_keyword_result, q.all()) self.assert_sql_count(testing.db, go, 1) @@ -391,58 +419,63 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): self.assert_sql_count(testing.db, go, 1) def test_eager_option(self): - keywords, items, item_keywords, Keyword, Item = (self.tables.keywords, - self.tables.items, - self.tables.item_keywords, - self.classes.Keyword, - self.classes.Item) + keywords, items, item_keywords, Keyword, Item = ( + self.tables.keywords, + self.tables.items, + self.tables.item_keywords, + self.classes.Keyword, + self.classes.Item) mapper(Keyword, keywords) - mapper(Item, items, properties = dict( - keywords = relationship(Keyword, secondary=item_keywords, lazy='select', - order_by=keywords.c.id))) + mapper(Item, items, properties=dict( + keywords=relationship( + Keyword, secondary=item_keywords, lazy='select', + order_by=keywords.c.id))) q = create_session().query(Item) def go(): eq_(self.static.item_keyword_result[0:2], - (q.options(joinedload('keywords')). - join('keywords').filter(keywords.c.name == 'red')).order_by(Item.id).all()) + (q.options( + joinedload('keywords') + ).join('keywords'). + filter(keywords.c.name == 'red')).order_by(Item.id).all()) self.assert_sql_count(testing.db, go, 1) - def test_cyclical(self): """A circular eager relationship breaks the cycle with a lazy loader""" Address, addresses, users, User = (self.classes.Address, - self.tables.addresses, - self.tables.users, - self.classes.User) - + self.tables.addresses, + self.tables.users, + self.classes.User) mapper(Address, addresses) - mapper(User, users, properties = dict( - addresses = relationship(Address, lazy='joined', - backref=sa.orm.backref('user', lazy='joined'), - order_by=Address.id) + mapper(User, users, properties=dict( + addresses=relationship( + Address, lazy='joined', + backref=sa.orm.backref('user', lazy='joined'), + order_by=Address.id) )) eq_(sa.orm.class_mapper(User).get_property('addresses').lazy, 'joined') eq_(sa.orm.class_mapper(Address).get_property('user').lazy, 'joined') sess = create_session() - eq_(self.static.user_address_result, sess.query(User).order_by(User.id).all()) + eq_( + self.static.user_address_result, + sess.query(User).order_by(User.id).all()) def test_double(self): """Eager loading with two relationships simultaneously, from the same table, using aliases.""" - users, orders, User, Address, Order, addresses = (self.tables.users, - self.tables.orders, - self.classes.User, - self.classes.Address, - self.classes.Order, - self.tables.addresses) - + users, orders, User, Address, Order, addresses = ( + self.tables.users, + self.tables.orders, + self.classes.User, + self.classes.Address, + self.classes.Order, + self.tables.addresses) openorders = sa.alias(orders, 'openorders') closedorders = sa.alias(orders, 'closedorders') @@ -453,17 +486,18 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): open_mapper = mapper(Order, openorders, non_primary=True) closed_mapper = mapper(Order, closedorders, non_primary=True) - mapper(User, users, properties = dict( - addresses = relationship(Address, lazy='joined', order_by=addresses.c.id), - open_orders = relationship( + mapper(User, users, properties=dict( + addresses=relationship( + Address, lazy='joined', order_by=addresses.c.id), + open_orders=relationship( open_mapper, primaryjoin=sa.and_(openorders.c.isopen == 1, - users.c.id==openorders.c.user_id), + users.c.id == openorders.c.user_id), lazy='joined', order_by=openorders.c.id), - closed_orders = relationship( + closed_orders=relationship( closed_mapper, primaryjoin=sa.and_(closedorders.c.isopen == 0, - users.c.id==closedorders.c.user_id), + users.c.id == closedorders.c.user_id), lazy='joined', order_by=closedorders.c.id))) q = create_session().query(User).order_by(User.id) @@ -473,57 +507,59 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): User( id=7, addresses=[Address(id=1)], - open_orders = [Order(id=3)], - closed_orders = [Order(id=1), Order(id=5)] + 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 = [] + open_orders=[], + closed_orders=[] ), User( id=9, addresses=[Address(id=5)], - open_orders = [Order(id=4)], - closed_orders = [Order(id=2)] + open_orders=[Order(id=4)], + closed_orders=[Order(id=2)] ), User(id=10) - ], q.all()) + ], q.all()) self.assert_sql_count(testing.db, go, 1) def test_double_same_mappers(self): """Eager loading with two relationships simulatneously, from the same table, using aliases.""" - addresses, items, order_items, orders, Item, User, Address, Order, users = (self.tables.addresses, - self.tables.items, - self.tables.order_items, - self.tables.orders, - self.classes.Item, - self.classes.User, - self.classes.Address, - self.classes.Order, - self.tables.users) - + addresses, items, order_items, orders, \ + Item, User, Address, Order, users = ( + self.tables.addresses, + self.tables.items, + self.tables.order_items, + self.tables.orders, + self.classes.Item, + self.classes.User, + self.classes.Address, + self.classes.Order, + self.tables.users) mapper(Address, addresses) mapper(Order, orders, properties={ 'items': relationship(Item, secondary=order_items, lazy='joined', - order_by=items.c.id)}) + order_by=items.c.id)}) mapper(Item, items) mapper(User, users, properties=dict( - addresses=relationship(Address, lazy='joined', order_by=addresses.c.id), + addresses=relationship( + Address, lazy='joined', order_by=addresses.c.id), open_orders=relationship( Order, primaryjoin=sa.and_(orders.c.isopen == 1, - users.c.id==orders.c.user_id), + users.c.id == orders.c.user_id), lazy='joined', order_by=orders.c.id), closed_orders=relationship( Order, primaryjoin=sa.and_(orders.c.isopen == 0, - users.c.id==orders.c.user_id), + users.c.id == orders.c.user_id), lazy='joined', order_by=orders.c.id))) q = create_session().query(User).order_by(User.id) @@ -531,97 +567,100 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): eq_([ User(id=7, addresses=[ - Address(id=1)], + Address(id=1)], open_orders=[Order(id=3, items=[ - Item(id=3), - Item(id=4), - Item(id=5)])], + Item(id=3), + Item(id=4), + Item(id=5)])], closed_orders=[Order(id=1, items=[ - Item(id=1), - Item(id=2), - Item(id=3)]), + Item(id=1), + Item(id=2), + Item(id=3)]), Order(id=5, items=[ - Item(id=5)])]), + Item(id=5)])]), User(id=8, addresses=[ - Address(id=2), - Address(id=3), - Address(id=4)], - open_orders = [], - closed_orders = []), + Address(id=2), + Address(id=3), + Address(id=4)], + open_orders=[], + closed_orders=[]), User(id=9, addresses=[ - Address(id=5)], + Address(id=5)], open_orders=[ - Order(id=4, - items=[ - Item(id=1), - Item(id=5)])], + 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)])]), + Order(id=2, + items=[ + Item(id=1), + Item(id=2), + Item(id=3)])]), User(id=10) - ], q.all()) + ], q.all()) self.assert_sql_count(testing.db, go, 1) def test_no_false_hits(self): """Eager loaders don't interpret main table columns as part of their eager load.""" - addresses, orders, User, Address, Order, users = (self.tables.addresses, - self.tables.orders, - self.classes.User, - self.classes.Address, - self.classes.Order, - self.tables.users) - + addresses, orders, User, Address, Order, users = ( + self.tables.addresses, + self.tables.orders, + self.classes.User, + self.classes.Address, + self.classes.Order, + self.tables.users) mapper(User, users, properties={ - 'addresses':relationship(Address, lazy='joined'), - 'orders':relationship(Order, lazy='joined') + 'addresses': relationship(Address, lazy='joined'), + 'orders': relationship(Order, lazy='joined') }) mapper(Address, addresses) mapper(Order, orders) - allusers = create_session().query(User).all() + self.allusers = create_session().query(User).all() # using a textual select, the columns will be 'id' and 'name'. the # eager loaders have aliases which should not hit on those columns, # they should be required to locate only their aliased/fully table # qualified column name. noeagers = create_session().query(User).\ - from_statement("select * from users").all() + from_statement("select * from users").all() assert 'orders' not in noeagers[0].__dict__ assert 'addresses' not in noeagers[0].__dict__ def test_limit(self): """Limit operations combined with lazy-load relationships.""" - users, items, order_items, orders, Item, User, Address, Order, addresses = (self.tables.users, - self.tables.items, - self.tables.order_items, - self.tables.orders, - self.classes.Item, - self.classes.User, - self.classes.Address, - self.classes.Order, - self.tables.addresses) - + users, items, order_items, orders, Item, \ + User, Address, Order, addresses = ( + self.tables.users, + self.tables.items, + self.tables.order_items, + self.tables.orders, + self.classes.Item, + self.classes.User, + self.classes.Address, + self.classes.Order, + self.tables.addresses) mapper(Item, items) mapper(Order, orders, properties={ - 'items':relationship(Item, secondary=order_items, lazy='joined', - order_by=items.c.id) + 'items': relationship(Item, secondary=order_items, lazy='joined', + order_by=items.c.id) }) mapper(User, users, properties={ - 'addresses':relationship(mapper(Address, addresses), lazy='joined', order_by=addresses.c.id), - 'orders':relationship(Order, lazy='select', order_by=orders.c.id) + 'addresses': relationship( + mapper(Address, addresses), + lazy='joined', order_by=addresses.c.id), + 'orders': relationship(Order, lazy='select', order_by=orders.c.id) }) sess = create_session() @@ -632,45 +671,54 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): def test_distinct(self): Address, addresses, users, User = (self.classes.Address, - self.tables.addresses, - self.tables.users, - self.classes.User) + self.tables.addresses, + self.tables.users, + self.classes.User) # this is an involved 3x union of the users table to get a lot of rows. - # then see if the "distinct" works its way out. you actually get the same - # result with or without the distinct, just via less or more rows. + # then see if the "distinct" works its way out. you actually get + # the same result with or without the distinct, just via less or + # more rows. u2 = users.alias('u2') - s = sa.union_all(u2.select(use_labels=True), u2.select(use_labels=True), u2.select(use_labels=True)).alias('u') + s = sa.union_all( + u2.select(use_labels=True), u2.select(use_labels=True), + u2.select(use_labels=True)).alias('u') mapper(User, users, properties={ - 'addresses':relationship(mapper(Address, addresses), lazy='joined', order_by=addresses.c.id), + 'addresses': relationship( + mapper(Address, addresses), + lazy='joined', order_by=addresses.c.id), }) sess = create_session() q = sess.query(User) def go(): - l = q.filter(s.c.u2_id==User.id).distinct().order_by(User.id).all() + l = q.filter(s.c.u2_id == User.id).distinct().\ + order_by(User.id).all() eq_(self.static.user_address_result, l) self.assert_sql_count(testing.db, go, 1) def test_limit_2(self): - keywords, items, item_keywords, Keyword, Item = (self.tables.keywords, - self.tables.items, - self.tables.item_keywords, - self.classes.Keyword, - self.classes.Item) + keywords, items, item_keywords, Keyword, Item = ( + self.tables.keywords, + self.tables.items, + self.tables.item_keywords, + self.classes.Keyword, + self.classes.Item) mapper(Keyword, keywords) - mapper(Item, items, properties = dict( - keywords = relationship(Keyword, secondary=item_keywords, lazy='joined', order_by=[keywords.c.id]), - )) + mapper(Item, items, properties=dict( + keywords=relationship( + Keyword, secondary=item_keywords, + lazy='joined', order_by=[keywords.c.id]), + )) sess = create_session() q = sess.query(Item) - l = q.filter((Item.description=='item 2') | - (Item.description=='item 5') | - (Item.description=='item 3')).\ + l = q.filter((Item.description == 'item 2') | + (Item.description == 'item 5') | + (Item.description == 'item 3')).\ order_by(Item.id).limit(2).all() eq_(self.static.item_keyword_result[1:3], l) @@ -681,131 +729,149 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): 'wrapped' select statement resulting from the combination of eager loading and limit/offset clauses.""" - addresses, items, order_items, orders, Item, User, Address, Order, users = (self.tables.addresses, - self.tables.items, - self.tables.order_items, - self.tables.orders, - self.classes.Item, - self.classes.User, - self.classes.Address, - self.classes.Order, - self.tables.users) - + addresses, items, order_items, orders, \ + Item, User, Address, Order, users = ( + self.tables.addresses, + self.tables.items, + self.tables.order_items, + self.tables.orders, + self.classes.Item, + self.classes.User, + self.classes.Address, + self.classes.Order, + self.tables.users) mapper(Item, items) - mapper(Order, orders, properties = dict( - items = relationship(Item, secondary=order_items, lazy='joined') + mapper(Order, orders, properties=dict( + items=relationship(Item, secondary=order_items, lazy='joined') )) mapper(Address, addresses) - mapper(User, users, properties = dict( - addresses = relationship(Address, lazy='joined', order_by=addresses.c.id), - orders = relationship(Order, lazy='joined', order_by=orders.c.id), + mapper(User, users, properties=dict( + addresses=relationship( + Address, lazy='joined', order_by=addresses.c.id), + orders=relationship(Order, lazy='joined', order_by=orders.c.id), )) sess = create_session() q = sess.query(User) if not testing.against('mssql'): - l = q.join('orders').order_by(Order.user_id.desc()).limit(2).offset(1) + l = q.join('orders').order_by( + Order.user_id.desc()).limit(2).offset(1) eq_([ User(id=9, - orders=[Order(id=2), Order(id=4)], - addresses=[Address(id=5)] - ), + orders=[Order(id=2), Order(id=4)], + addresses=[Address(id=5)] + ), User(id=7, - orders=[Order(id=1), Order(id=3), Order(id=5)], - addresses=[Address(id=1)] - ) - ], l.all()) + orders=[Order(id=1), Order(id=3), Order(id=5)], + addresses=[Address(id=1)] + ) + ], l.all()) - l = q.join('addresses').order_by(Address.email_address.desc()).limit(1).offset(0) + l = q.join('addresses').order_by( + Address.email_address.desc()).limit(1).offset(0) eq_([ User(id=7, - orders=[Order(id=1), Order(id=3), Order(id=5)], - addresses=[Address(id=1)] - ) - ], l.all()) + orders=[Order(id=1), Order(id=3), Order(id=5)], + addresses=[Address(id=1)] + ) + ], l.all()) def test_limit_4(self): User, Order, addresses, users, orders = (self.classes.User, - self.classes.Order, - self.tables.addresses, - self.tables.users, - self.tables.orders) + self.classes.Order, + self.tables.addresses, + self.tables.users, + self.tables.orders) # tests the LIMIT/OFFSET aliasing on a mapper # against a select. original issue from ticket #904 sel = sa.select([users, addresses.c.email_address], - users.c.id==addresses.c.user_id).alias('useralias') + users.c.id == addresses.c.user_id).alias('useralias') mapper(User, sel, properties={ - 'orders':relationship(Order, primaryjoin=sel.c.id==orders.c.user_id, - lazy='joined', order_by=orders.c.id) + 'orders': relationship( + Order, primaryjoin=sel.c.id == orders.c.user_id, + lazy='joined', order_by=orders.c.id) }) mapper(Order, orders) sess = create_session() eq_(sess.query(User).first(), - User(name='jack',orders=[ - Order(address_id=1,description='order 1',isopen=0,user_id=7,id=1), - Order(address_id=1,description='order 3',isopen=1,user_id=7,id=3), - Order(address_id=None,description='order 5',isopen=0,user_id=7,id=5)], - email_address='jack@bean.com',id=7) - ) + User(name='jack', orders=[ + Order( + address_id=1, + description='order 1', + isopen=0, + user_id=7, + id=1), + Order( + address_id=1, + description='order 3', + isopen=1, + user_id=7, + id=3), + Order( + address_id=None, description='order 5', isopen=0, + user_id=7, id=5)], + email_address='jack@bean.com', id=7) + ) def test_useget_cancels_eager(self): """test that a one to many lazyload cancels the unnecessary eager many-to-one join on the other side.""" - users, Address, addresses, User = (self.tables.users, - self.classes.Address, - self.tables.addresses, - self.classes.User) - + users, Address, addresses, User = ( + self.tables.users, + self.classes.Address, + self.tables.addresses, + self.classes.User) mapper(User, users) mapper(Address, addresses, properties={ - 'user':relationship(User, lazy='joined', backref='addresses') + 'user': relationship(User, lazy='joined', backref='addresses') }) sess = create_session() - u1 = sess.query(User).filter(User.id==8).one() + u1 = sess.query(User).filter(User.id == 8).one() + def go(): eq_(u1.addresses[0].user, u1) - self.assert_sql_execution(testing.db, go, + self.assert_sql_execution( + testing.db, go, CompiledSQL( "SELECT addresses.id AS addresses_id, addresses.user_id AS " "addresses_user_id, addresses.email_address AS " "addresses_email_address FROM addresses WHERE :param_1 = " "addresses.user_id", - {'param_1': 8}) - ) - + {'param_1': 8}) + ) def test_manytoone_limit(self): """test that the subquery wrapping only occurs with limit/offset and m2m or o2m joins present.""" users, items, order_items, Order, Item, User, \ - Address, orders, addresses = (self.tables.users, - self.tables.items, - self.tables.order_items, - self.classes.Order, - self.classes.Item, - self.classes.User, - self.classes.Address, - self.tables.orders, - self.tables.addresses) - + Address, orders, addresses = ( + self.tables.users, + self.tables.items, + self.tables.order_items, + self.classes.Order, + self.classes.Item, + self.classes.User, + self.classes.Address, + self.tables.orders, + self.tables.addresses) mapper(User, users, properties=odict( orders=relationship(Order, backref='user') )) mapper(Order, orders, properties=odict([ ('items', relationship(Item, secondary=order_items, - backref='orders')), + backref='orders')), ('address', relationship(Address)) - ])) + ])) mapper(Address, addresses) mapper(Item, items) @@ -815,15 +881,17 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): sess.query(User).options(joinedload(User.orders)).limit(10), "SELECT anon_1.users_id AS anon_1_users_id, " "anon_1.users_name AS anon_1_users_name, " - "orders_1.id AS orders_1_id, orders_1.user_id AS orders_1_user_id, " + "orders_1.id AS orders_1_id, orders_1.user_id " + "AS orders_1_user_id, " "orders_1.address_id AS " - "orders_1_address_id, orders_1.description AS orders_1_description, " + "orders_1_address_id, orders_1.description " + "AS orders_1_description, " "orders_1.isopen AS orders_1_isopen " "FROM (SELECT users.id AS users_id, users.name AS users_name " "FROM users " "LIMIT :param_1) AS anon_1 LEFT OUTER JOIN orders AS orders_1 " "ON anon_1.users_id = orders_1.user_id", - {'param_1':10} + {'param_1': 10} ) self.assert_compile( @@ -835,12 +903,12 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): "users_1.id AS users_1_id, users_1.name AS users_1_name " "FROM orders LEFT OUTER JOIN users AS " "users_1 ON users_1.id = orders.user_id LIMIT :param_1", - {'param_1':10} + {'param_1': 10} ) self.assert_compile( sess.query(Order).options( - joinedload(Order.user, innerjoin=True)).limit(10), + joinedload(Order.user, innerjoin=True)).limit(10), "SELECT orders.id AS orders_id, orders.user_id AS orders_user_id, " "orders.address_id AS " "orders_address_id, orders.description AS orders_description, " @@ -848,12 +916,12 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): "users_1.id AS users_1_id, users_1.name AS users_1_name " "FROM orders JOIN users AS " "users_1 ON users_1.id = orders.user_id LIMIT :param_1", - {'param_1':10} + {'param_1': 10} ) self.assert_compile( sess.query(User).options( - joinedload_all("orders.address")).limit(10), + joinedload_all("orders.address")).limit(10), "SELECT anon_1.users_id AS anon_1_users_id, " "anon_1.users_name AS anon_1_users_name, " "addresses_1.id AS addresses_1_id, " @@ -869,16 +937,16 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): "LEFT OUTER JOIN orders AS orders_1 " "ON anon_1.users_id = orders_1.user_id LEFT OUTER JOIN " "addresses AS addresses_1 ON addresses_1.id = orders_1.address_id", - {'param_1':10} + {'param_1': 10} ) self.assert_compile( sess.query(User).options(joinedload_all("orders.items"), - joinedload("orders.address")), + joinedload("orders.address")), "SELECT users.id AS users_id, users.name AS users_name, " - "items_1.id AS items_1_id, " + "items_1.id AS items_1_id, " "items_1.description AS items_1_description, " - "addresses_1.id AS addresses_1_id, " + "addresses_1.id AS addresses_1_id, " "addresses_1.user_id AS addresses_1_user_id, " "addresses_1.email_address AS " "addresses_1_email_address, orders_1.id AS orders_1_id, " @@ -896,43 +964,62 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): ) self.assert_compile( - sess.query(User).options(joinedload("orders"), joinedload("orders.address", innerjoin=True)).limit(10), - "SELECT anon_1.users_id AS anon_1_users_id, anon_1.users_name AS anon_1_users_name, " - "addresses_1.id AS addresses_1_id, addresses_1.user_id AS addresses_1_user_id, " - "addresses_1.email_address AS addresses_1_email_address, orders_1.id AS orders_1_id, " - "orders_1.user_id AS orders_1_user_id, orders_1.address_id AS orders_1_address_id, " - "orders_1.description AS orders_1_description, orders_1.isopen AS orders_1_isopen " + sess.query(User).options( + joinedload("orders"), + joinedload( + "orders.address", + innerjoin=True)).limit(10), + "SELECT anon_1.users_id AS anon_1_users_id, " + "anon_1.users_name AS anon_1_users_name, " + "addresses_1.id AS addresses_1_id, " + "addresses_1.user_id AS addresses_1_user_id, " + "addresses_1.email_address AS addresses_1_email_address, " + "orders_1.id AS orders_1_id, " + "orders_1.user_id AS orders_1_user_id, " + "orders_1.address_id AS orders_1_address_id, " + "orders_1.description AS orders_1_description, " + "orders_1.isopen AS orders_1_isopen " "FROM (SELECT users.id AS users_id, users.name AS users_name " "FROM users " - "LIMIT :param_1) AS anon_1 LEFT OUTER JOIN orders AS orders_1 ON anon_1.users_id = " - "orders_1.user_id LEFT OUTER JOIN addresses AS addresses_1 ON addresses_1.id = orders_1.address_id", - {'param_1':10} + "LIMIT :param_1) AS anon_1 LEFT OUTER JOIN orders " + "AS orders_1 ON anon_1.users_id = " + "orders_1.user_id LEFT OUTER JOIN addresses AS addresses_1 " + "ON addresses_1.id = orders_1.address_id", + {'param_1': 10} ) self.assert_compile( - sess.query(User).options(joinedload("orders", innerjoin=True), - joinedload("orders.address", innerjoin=True)).limit(10), - "SELECT anon_1.users_id AS anon_1_users_id, anon_1.users_name AS anon_1_users_name, " - "addresses_1.id AS addresses_1_id, addresses_1.user_id AS addresses_1_user_id, " - "addresses_1.email_address AS addresses_1_email_address, orders_1.id AS orders_1_id, " - "orders_1.user_id AS orders_1_user_id, orders_1.address_id AS orders_1_address_id, " - "orders_1.description AS orders_1_description, orders_1.isopen AS orders_1_isopen " + sess.query(User).options( + joinedload("orders", innerjoin=True), + joinedload("orders.address", innerjoin=True)).limit(10), + "SELECT anon_1.users_id AS anon_1_users_id, " + "anon_1.users_name AS anon_1_users_name, " + "addresses_1.id AS addresses_1_id, " + "addresses_1.user_id AS addresses_1_user_id, " + "addresses_1.email_address AS addresses_1_email_address, " + "orders_1.id AS orders_1_id, " + "orders_1.user_id AS orders_1_user_id, " + "orders_1.address_id AS orders_1_address_id, " + "orders_1.description AS orders_1_description, " + "orders_1.isopen AS orders_1_isopen " "FROM (SELECT users.id AS users_id, users.name AS users_name " "FROM users " - "LIMIT :param_1) AS anon_1 JOIN orders AS orders_1 ON anon_1.users_id = " - "orders_1.user_id JOIN addresses AS addresses_1 ON addresses_1.id = orders_1.address_id", - {'param_1':10} + "LIMIT :param_1) AS anon_1 JOIN orders " + "AS orders_1 ON anon_1.users_id = " + "orders_1.user_id JOIN addresses AS addresses_1 " + "ON addresses_1.id = orders_1.address_id", + {'param_1': 10} ) def test_one_to_many_scalar(self): Address, addresses, users, User = (self.classes.Address, - self.tables.addresses, - self.tables.users, - self.classes.User) + self.tables.addresses, + self.tables.users, + self.classes.User) - mapper(User, users, properties = dict( - address = relationship(mapper(Address, addresses), - lazy='joined', uselist=False) + mapper(User, users, properties=dict( + address=relationship(mapper(Address, addresses), + lazy='joined', uselist=False) )) q = create_session().query(User) @@ -942,19 +1029,20 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): self.assert_sql_count(testing.db, go, 1) def test_many_to_one(self): - users, Address, addresses, User = (self.tables.users, - self.classes.Address, - self.tables.addresses, - self.classes.User) - - mapper(Address, addresses, properties = dict( - user = relationship(mapper(User, users), lazy='joined') + users, Address, addresses, User = ( + self.tables.users, + self.classes.Address, + self.tables.addresses, + self.classes.User) + + mapper(Address, addresses, properties=dict( + user=relationship(mapper(User, users), lazy='joined') )) sess = create_session() q = sess.query(Address) def go(): - a = q.filter(addresses.c.id==1).one() + a = q.filter(addresses.c.id == 1).one() is_not_(a.user, None) u1 = sess.query(User).get(7) is_(a.user, u1) @@ -967,32 +1055,35 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): """ Order, Address, addresses, orders = (self.classes.Order, - self.classes.Address, - self.tables.addresses, - self.tables.orders) - + self.classes.Address, + self.tables.addresses, + self.tables.orders) # use a primaryjoin intended to defeat SA's usage of # query.get() for a many-to-one lazyload - mapper(Order, orders, properties = dict( - address = relationship(mapper(Address, addresses), + mapper(Order, orders, properties=dict( + address=relationship( + mapper(Address, addresses), primaryjoin=and_( - addresses.c.id==orders.c.address_id, + addresses.c.id == orders.c.address_id, addresses.c.email_address != None ), - lazy='joined') + lazy='joined') )) sess = create_session() def go(): - o1 = sess.query(Order).options(lazyload('address')).filter(Order.id==5).one() + o1 = sess.query(Order).options( + lazyload('address')).filter( + Order.id == 5).one() eq_(o1.address, None) self.assert_sql_count(testing.db, go, 2) sess.expunge_all() + def go(): - o1 = sess.query(Order).filter(Order.id==5).one() + o1 = sess.query(Order).filter(Order.id == 5).one() eq_(o1.address, None) self.assert_sql_count(testing.db, go, 1) @@ -1000,22 +1091,26 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): """tests eager load for a parent object with a child object that contains a many-to-many relationship to a third object.""" - users, items, order_items, orders, Item, User, Order = (self.tables.users, - self.tables.items, - self.tables.order_items, - self.tables.orders, - self.classes.Item, - self.classes.User, - self.classes.Order) - + users, items, order_items, orders, Item, User, Order = ( + self.tables.users, + self.tables.items, + self.tables.order_items, + self.tables.orders, + self.classes.Item, + self.classes.User, + self.classes.Order) mapper(User, users, properties={ - 'orders':relationship(Order, lazy='joined', order_by=orders.c.id) + 'orders': relationship(Order, lazy='joined', order_by=orders.c.id) }) mapper(Item, items) - mapper(Order, orders, properties = dict( - items = relationship(Item, secondary=order_items, lazy='joined', order_by=items.c.id) - )) + mapper(Order, orders, properties=dict( + items=relationship( + Item, + secondary=order_items, + lazy='joined', + order_by=items.c.id) + )) q = create_session().query(User) @@ -1027,24 +1122,26 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): def test_double_with_aggregate(self): User, users, orders, Order = (self.classes.User, - self.tables.users, - self.tables.orders, - self.classes.Order) + self.tables.users, + self.tables.orders, + self.classes.Order) - max_orders_by_user = sa.select([sa.func.max(orders.c.id).label('order_id')], - group_by=[orders.c.user_id] - ).alias('max_orders_by_user') + max_orders_by_user = sa.select([ + sa.func.max(orders.c.id).label('order_id')], + group_by=[orders.c.user_id] + ).alias('max_orders_by_user') - max_orders = orders.select(orders.c.id==max_orders_by_user.c.order_id).\ - alias('max_orders') + max_orders = orders.select( + orders.c.id == max_orders_by_user.c.order_id).\ + alias('max_orders') mapper(Order, orders) mapper(User, users, properties={ - 'orders':relationship(Order, backref='user', lazy='joined', - order_by=orders.c.id), - 'max_order':relationship( - mapper(Order, max_orders, non_primary=True), - lazy='joined', uselist=False) + 'orders': relationship(Order, backref='user', lazy='joined', + order_by=orders.c.id), + 'max_order': relationship( + mapper(Order, max_orders, non_primary=True), + lazy='joined', uselist=False) }) q = create_session().query(User) @@ -1052,18 +1149,18 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): def go(): eq_([ User(id=7, orders=[ - Order(id=1), - Order(id=3), - Order(id=5), - ], + Order(id=1), + Order(id=3), + Order(id=5), + ], max_order=Order(id=5) ), User(id=8, orders=[]), - User(id=9, orders=[Order(id=2),Order(id=4)], - max_order=Order(id=4) - ), + User(id=9, orders=[Order(id=2), Order(id=4)], + max_order=Order(id=4) + ), User(id=10), - ], q.order_by(User.id).all()) + ], q.order_by(User.id).all()) self.assert_sql_count(testing.db, go, 1) def test_uselist_false_warning(self): @@ -1071,57 +1168,66 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): uselist=False raises a warning.""" User, users, orders, Order = (self.classes.User, - self.tables.users, - self.tables.orders, - self.classes.Order) - + self.tables.users, + self.tables.orders, + self.classes.Order) mapper(User, users, properties={ - 'order':relationship(Order, uselist=False) + 'order': relationship(Order, uselist=False) }) mapper(Order, orders) s = create_session() assert_raises(sa.exc.SAWarning, - s.query(User).options(joinedload(User.order)).all) + s.query(User).options(joinedload(User.order)).all) def test_wide(self): - users, items, order_items, Order, Item, User, Address, orders, addresses = (self.tables.users, - self.tables.items, - self.tables.order_items, - self.classes.Order, - self.classes.Item, - self.classes.User, - self.classes.Address, - self.tables.orders, - self.tables.addresses) - - mapper(Order, orders, properties={'items':relationship(Item, secondary=order_items, lazy='joined', - order_by=items.c.id)}) + users, items, order_items, Order, Item, \ + User, Address, orders, addresses = ( + self.tables.users, + self.tables.items, + self.tables.order_items, + self.classes.Order, + self.classes.Item, + self.classes.User, + self.classes.Address, + self.tables.orders, + self.tables.addresses) + + mapper( + Order, orders, properties={ + 'items': relationship( + Item, secondary=order_items, lazy='joined', + order_by=items.c.id)}) mapper(Item, items) - mapper(User, users, properties = dict( - addresses = relationship(mapper(Address, addresses), lazy = False, order_by=addresses.c.id), - orders = relationship(Order, lazy = False, order_by=orders.c.id), + mapper(User, users, properties=dict( + addresses=relationship( + mapper( + Address, + addresses), + lazy=False, + order_by=addresses.c.id), + orders=relationship(Order, lazy=False, order_by=orders.c.id), )) q = create_session().query(User) - l = q.all() + self.l = q.all() eq_(self.static.user_all_result, q.order_by(User.id).all()) def test_against_select(self): """test eager loading of a mapper which is against a select""" - users, items, order_items, orders, Item, User, Order = (self.tables.users, - self.tables.items, - self.tables.order_items, - self.tables.orders, - self.classes.Item, - self.classes.User, - self.classes.Order) - + users, items, order_items, orders, Item, User, Order = ( + self.tables.users, + self.tables.items, + self.tables.order_items, + self.tables.orders, + self.classes.Item, + self.classes.User, + self.classes.Order) - s = sa.select([orders], orders.c.isopen==1).alias('openorders') + s = sa.select([orders], orders.c.isopen == 1).alias('openorders') mapper(Order, s, properties={ - 'user':relationship(User, lazy='joined') + 'user': relationship(User, lazy='joined') }) mapper(User, users) mapper(Item, items) @@ -1130,96 +1236,110 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): eq_([ Order(id=3, user=User(id=7)), Order(id=4, user=User(id=9)) - ], q.all()) + ], q.all()) - q = q.select_from(s.join(order_items).join(items)).filter(~Item.id.in_([1, 2, 5])) + q = q.select_from(s.join(order_items).join(items)).filter( + ~Item.id.in_([1, 2, 5])) eq_([ Order(id=3, user=User(id=7)), - ], q.all()) + ], q.all()) def test_aliasing(self): """test that eager loading uses aliases to insulate the eager load from regular criterion against those tables.""" Address, addresses, users, User = (self.classes.Address, - self.tables.addresses, - self.tables.users, - self.classes.User) + self.tables.addresses, + self.tables.users, + self.classes.User) - - mapper(User, users, properties = dict( - addresses = relationship(mapper(Address, addresses), - lazy='joined', order_by=addresses.c.id) + mapper(User, users, properties=dict( + addresses=relationship(mapper(Address, addresses), + lazy='joined', order_by=addresses.c.id) )) q = create_session().query(User) l = q.filter(addresses.c.email_address == 'ed@lala.com').filter( - Address.user_id==User.id).order_by(User.id) + Address.user_id == User.id).order_by(User.id) eq_(self.static.user_address_result[1:2], l.all()) def test_inner_join(self): Address, addresses, users, User = (self.classes.Address, - self.tables.addresses, - self.tables.users, - self.classes.User) + self.tables.addresses, + self.tables.users, + self.classes.User) - mapper(User, users, properties = dict( - addresses = relationship(mapper(Address, addresses), lazy='joined', - innerjoin=True, order_by=addresses.c.id) + mapper(User, users, properties=dict( + addresses=relationship(mapper(Address, addresses), lazy='joined', + innerjoin=True, order_by=addresses.c.id) )) sess = create_session() eq_( - [User(id=7, addresses=[ Address(id=1) ]), - User(id=8, - addresses=[ Address(id=2, email_address='ed@wood.com'), - Address(id=3, email_address='ed@bettyboop.com'), - Address(id=4, email_address='ed@lala.com'), ]), - User(id=9, addresses=[ Address(id=5) ])] - ,sess.query(User).all() + [User(id=7, addresses=[Address(id=1)]), + User(id=8, + addresses=[Address(id=2, email_address='ed@wood.com'), + Address(id=3, email_address='ed@bettyboop.com'), + Address(id=4, email_address='ed@lala.com'), ]), + User(id=9, addresses=[Address(id=5)])], sess.query(User).all() ) - self.assert_compile(sess.query(User), - "SELECT users.id AS users_id, users.name AS users_name, " - "addresses_1.id AS addresses_1_id, addresses_1.user_id AS addresses_1_user_id, " - "addresses_1.email_address AS addresses_1_email_address FROM users JOIN " - "addresses AS addresses_1 ON users.id = addresses_1.user_id ORDER BY addresses_1.id") + self.assert_compile( + sess.query(User), + "SELECT users.id AS users_id, users.name AS users_name, " + "addresses_1.id AS addresses_1_id, " + "addresses_1.user_id AS addresses_1_user_id, " + "addresses_1.email_address AS addresses_1_email_address " + "FROM users JOIN " + "addresses AS addresses_1 ON users.id = addresses_1.user_id " + "ORDER BY addresses_1.id") def test_inner_join_chaining_options(self): - users, items, order_items, Order, Item, User, orders = (self.tables.users, - self.tables.items, - self.tables.order_items, - self.classes.Order, - self.classes.Item, - self.classes.User, - self.tables.orders) - - mapper(User, users, properties = dict( - orders =relationship(Order, innerjoin=True, - lazy=False) + users, items, order_items, Order, Item, User, orders = ( + self.tables.users, + self.tables.items, + self.tables.order_items, + self.classes.Order, + self.classes.Item, + self.classes.User, + self.tables.orders) + + mapper(User, users, properties=dict( + orders=relationship(Order, innerjoin=True, + lazy=False) )) mapper(Order, orders, properties=dict( items=relationship(Item, secondary=order_items, lazy=False, - innerjoin=True) + innerjoin=True) )) mapper(Item, items) sess = create_session() self.assert_compile( sess.query(User), - "SELECT users.id AS users_id, users.name AS users_name, items_1.id AS " - "items_1_id, items_1.description AS items_1_description, orders_1.id AS " - "orders_1_id, orders_1.user_id AS orders_1_user_id, orders_1.address_id AS " - "orders_1_address_id, orders_1.description AS orders_1_description, " - "orders_1.isopen AS orders_1_isopen FROM users JOIN orders AS orders_1 ON " - "users.id = orders_1.user_id JOIN order_items AS order_items_1 ON orders_1.id = " + "SELECT users.id AS users_id, users.name AS users_name, " + "items_1.id AS " + "items_1_id, items_1.description AS items_1_description, " + "orders_1.id AS " + "orders_1_id, orders_1.user_id AS orders_1_user_id, " + "orders_1.address_id AS " + "orders_1_address_id, orders_1.description " + "AS orders_1_description, " + "orders_1.isopen AS orders_1_isopen FROM users " + "JOIN orders AS orders_1 ON " + "users.id = orders_1.user_id JOIN order_items AS order_items_1 " + "ON orders_1.id = " "order_items_1.order_id JOIN items AS items_1 ON items_1.id = " "order_items_1.item_id" ) self.assert_compile( sess.query(User).options(joinedload(User.orders, innerjoin=False)), - "SELECT users.id AS users_id, users.name AS users_name, items_1.id AS " - "items_1_id, items_1.description AS items_1_description, orders_1.id AS " - "orders_1_id, orders_1.user_id AS orders_1_user_id, orders_1.address_id AS " - "orders_1_address_id, orders_1.description AS orders_1_description, " + "SELECT users.id AS users_id, users.name AS users_name, " + "items_1.id AS " + "items_1_id, items_1.description AS items_1_description, " + "orders_1.id AS " + "orders_1_id, orders_1.user_id AS orders_1_user_id, " + "orders_1.address_id AS " + "orders_1_address_id, orders_1.description " + "AS orders_1_description, " "orders_1.isopen AS orders_1_isopen " "FROM users LEFT OUTER JOIN orders AS orders_1 " "ON users.id = orders_1.user_id " @@ -1229,13 +1349,22 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): ) self.assert_compile( - sess.query(User).options(joinedload(User.orders, Order.items, innerjoin=False)), - "SELECT users.id AS users_id, users.name AS users_name, items_1.id AS " - "items_1_id, items_1.description AS items_1_description, orders_1.id AS " - "orders_1_id, orders_1.user_id AS orders_1_user_id, orders_1.address_id AS " - "orders_1_address_id, orders_1.description AS orders_1_description, " + sess.query(User).options( + joinedload( + User.orders, + Order.items, + innerjoin=False)), + "SELECT users.id AS users_id, users.name AS users_name, " + "items_1.id AS " + "items_1_id, items_1.description AS items_1_description, " + "orders_1.id AS " + "orders_1_id, orders_1.user_id AS orders_1_user_id, " + "orders_1.address_id AS " + "orders_1_address_id, " + "orders_1.description AS orders_1_description, " "orders_1.isopen AS orders_1_isopen " - "FROM users JOIN orders AS orders_1 ON users.id = orders_1.user_id " + "FROM users JOIN orders AS orders_1 ON " + "users.id = orders_1.user_id " "LEFT OUTER JOIN (order_items AS order_items_1 " "JOIN items AS items_1 ON items_1.id = order_items_1.item_id) " "ON orders_1.id = order_items_1.order_id" @@ -1243,33 +1372,40 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): ) def test_inner_join_nested_chaining_negative_options(self): - users, items, order_items, Order, Item, User, orders = (self.tables.users, - self.tables.items, - self.tables.order_items, - self.classes.Order, - self.classes.Item, - self.classes.User, - self.tables.orders) + users, items, order_items, Order, Item, User, orders = ( + self.tables.users, + self.tables.items, + self.tables.order_items, + self.classes.Order, + self.classes.Item, + self.classes.User, + self.tables.orders) mapper(User, users, properties=dict( orders=relationship(Order, innerjoin='nested', - lazy=False, order_by=orders.c.id) + lazy=False, order_by=orders.c.id) )) mapper(Order, orders, properties=dict( items=relationship(Item, secondary=order_items, lazy=False, - innerjoin='nested', order_by=items.c.id) + innerjoin='nested', order_by=items.c.id) )) mapper(Item, items) sess = create_session() self.assert_compile( sess.query(User), - "SELECT users.id AS users_id, users.name AS users_name, items_1.id AS " - "items_1_id, items_1.description AS items_1_description, orders_1.id AS " - "orders_1_id, orders_1.user_id AS orders_1_user_id, orders_1.address_id AS " - "orders_1_address_id, orders_1.description AS orders_1_description, " - "orders_1.isopen AS orders_1_isopen FROM users JOIN orders AS orders_1 ON " - "users.id = orders_1.user_id JOIN order_items AS order_items_1 ON orders_1.id = " + "SELECT users.id AS users_id, users.name AS users_name, " + "items_1.id AS " + "items_1_id, items_1.description AS items_1_description, " + "orders_1.id AS " + "orders_1_id, orders_1.user_id AS orders_1_user_id, " + "orders_1.address_id AS " + "orders_1_address_id, orders_1.description " + "AS orders_1_description, " + "orders_1.isopen AS orders_1_isopen FROM users " + "JOIN orders AS orders_1 ON " + "users.id = orders_1.user_id JOIN order_items " + "AS order_items_1 ON orders_1.id = " "order_items_1.order_id JOIN items AS items_1 ON items_1.id = " "order_items_1.item_id ORDER BY orders_1.id, items_1.id" ) @@ -1277,101 +1413,145 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): q = sess.query(User).options(joinedload(User.orders, innerjoin=False)) self.assert_compile( q, - "SELECT users.id AS users_id, users.name AS users_name, items_1.id AS " - "items_1_id, items_1.description AS items_1_description, orders_1.id AS " - "orders_1_id, orders_1.user_id AS orders_1_user_id, orders_1.address_id AS " - "orders_1_address_id, orders_1.description AS orders_1_description, " + "SELECT users.id AS users_id, users.name AS users_name, " + "items_1.id AS " + "items_1_id, items_1.description AS items_1_description, " + "orders_1.id AS " + "orders_1_id, orders_1.user_id AS orders_1_user_id, " + "orders_1.address_id AS " + "orders_1_address_id, orders_1.description " + "AS orders_1_description, " "orders_1.isopen AS orders_1_isopen " "FROM users LEFT OUTER JOIN " "(orders AS orders_1 JOIN order_items AS order_items_1 " - "ON orders_1.id = order_items_1.order_id " - "JOIN items AS items_1 ON items_1.id = order_items_1.item_id) " + "ON orders_1.id = order_items_1.order_id " + "JOIN items AS items_1 ON items_1.id = order_items_1.item_id) " "ON users.id = orders_1.user_id ORDER BY orders_1.id, items_1.id" ) eq_( - [ - User(id=7, - orders=[ - Order(id=1, items=[Item(id=1), Item(id=2), Item(id=3)]), - Order(id=3, items=[Item(id=3), Item(id=4), Item(id=5)]), - Order(id=5, items=[Item(id=5)])]), - User(id=8, orders=[]), - User(id=9, orders=[ - Order(id=2, items=[Item(id=1), Item(id=2), Item(id=3)]), - Order(id=4, items=[Item(id=1), Item(id=5)]) - ] - ), - User(id=10, orders=[]) - ], + [ + User(id=7, + orders=[ + Order( + id=1, items=[ + Item( + id=1), Item( + id=2), Item( + id=3)]), + Order( + id=3, items=[ + Item( + id=3), Item( + id=4), Item( + id=5)]), + Order(id=5, items=[Item(id=5)])]), + User(id=8, orders=[]), + User(id=9, orders=[ + Order(id=2, items=[Item(id=1), Item(id=2), Item(id=3)]), + Order(id=4, items=[Item(id=1), Item(id=5)]) + ] + ), + User(id=10, orders=[]) + ], q.order_by(User.id).all() ) self.assert_compile( - sess.query(User).options(joinedload(User.orders, Order.items, innerjoin=False)), - "SELECT users.id AS users_id, users.name AS users_name, items_1.id AS " - "items_1_id, items_1.description AS items_1_description, orders_1.id AS " - "orders_1_id, orders_1.user_id AS orders_1_user_id, orders_1.address_id AS " - "orders_1_address_id, orders_1.description AS orders_1_description, " + sess.query(User).options( + joinedload( + User.orders, + Order.items, + innerjoin=False)), + "SELECT users.id AS users_id, users.name AS users_name, " + "items_1.id AS " + "items_1_id, items_1.description AS items_1_description, " + "orders_1.id AS " + "orders_1_id, orders_1.user_id AS orders_1_user_id, " + "orders_1.address_id AS " + "orders_1_address_id, orders_1.description AS " + "orders_1_description, " "orders_1.isopen AS orders_1_isopen " - "FROM users JOIN orders AS orders_1 ON users.id = orders_1.user_id " + "FROM users JOIN orders AS orders_1 ON users.id = " + "orders_1.user_id " "LEFT OUTER JOIN (order_items AS order_items_1 " "JOIN items AS items_1 ON items_1.id = order_items_1.item_id) " - "ON orders_1.id = order_items_1.order_id ORDER BY orders_1.id, items_1.id" + "ON orders_1.id = order_items_1.order_id ORDER BY " + "orders_1.id, items_1.id" ) def test_inner_join_nested_chaining_positive_options(self): - users, items, order_items, Order, Item, User, orders = (self.tables.users, - self.tables.items, - self.tables.order_items, - self.classes.Order, - self.classes.Item, - self.classes.User, - self.tables.orders) + users, items, order_items, Order, Item, User, orders = ( + self.tables.users, + self.tables.items, + self.tables.order_items, + self.classes.Order, + self.classes.Item, + self.classes.User, + self.tables.orders) mapper(User, users, properties=dict( orders=relationship(Order, order_by=orders.c.id) )) mapper(Order, orders, properties=dict( - items=relationship(Item, secondary=order_items, order_by=items.c.id) + items=relationship( + Item, + secondary=order_items, + order_by=items.c.id) )) mapper(Item, items) sess = create_session() q = sess.query(User).options( - joinedload("orders", innerjoin=False).\ - joinedload("items", innerjoin="nested") - ) + joinedload("orders", innerjoin=False). + joinedload("items", innerjoin="nested") + ) self.assert_compile( q, "SELECT users.id AS users_id, users.name AS users_name, " - "items_1.id AS items_1_id, items_1.description AS items_1_description, " - "orders_1.id AS orders_1_id, orders_1.user_id AS orders_1_user_id, " - "orders_1.address_id AS orders_1_address_id, orders_1.description AS " + "items_1.id AS items_1_id, items_1.description " + "AS items_1_description, " + "orders_1.id AS orders_1_id, orders_1.user_id " + "AS orders_1_user_id, " + "orders_1.address_id AS orders_1_address_id, " + "orders_1.description AS " "orders_1_description, orders_1.isopen AS orders_1_isopen " - "FROM users LEFT OUTER JOIN (orders AS orders_1 JOIN order_items AS " - "order_items_1 ON orders_1.id = order_items_1.order_id JOIN items AS " - "items_1 ON items_1.id = order_items_1.item_id) ON users.id = orders_1.user_id " + "FROM users LEFT OUTER JOIN (orders AS orders_1 " + "JOIN order_items AS " + "order_items_1 ON orders_1.id = order_items_1.order_id " + "JOIN items AS " + "items_1 ON items_1.id = order_items_1.item_id) " + "ON users.id = orders_1.user_id " "ORDER BY orders_1.id, items_1.id" ) eq_( - [ - User(id=7, - orders=[ - Order(id=1, items=[Item(id=1), Item(id=2), Item(id=3)]), - Order(id=3, items=[Item(id=3), Item(id=4), Item(id=5)]), - Order(id=5, items=[Item(id=5)])]), - User(id=8, orders=[]), - User(id=9, orders=[ - Order(id=2, items=[Item(id=1), Item(id=2), Item(id=3)]), - Order(id=4, items=[Item(id=1), Item(id=5)]) - ] - ), - User(id=10, orders=[]) - ], + [ + User(id=7, + orders=[ + Order( + id=1, items=[ + Item( + id=1), Item( + id=2), Item( + id=3)]), + Order( + id=3, items=[ + Item( + id=3), Item( + id=4), Item( + id=5)]), + Order(id=5, items=[Item(id=5)])]), + User(id=8, orders=[]), + User(id=9, orders=[ + Order(id=2, items=[Item(id=1), Item(id=2), Item(id=3)]), + Order(id=4, items=[Item(id=1), Item(id=5)]) + ] + ), + User(id=10, orders=[]) + ], q.order_by(User.id).all() ) @@ -1452,33 +1632,34 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): # a new feature as of [ticket:2369] users, Keyword, orders, items, order_items, Order, Item, \ - User, keywords, item_keywords = (self.tables.users, - self.classes.Keyword, - self.tables.orders, - self.tables.items, - self.tables.order_items, - self.classes.Order, - self.classes.Item, - self.classes.User, - self.tables.keywords, - self.tables.item_keywords) + User, keywords, item_keywords = ( + self.tables.users, + self.classes.Keyword, + self.tables.orders, + self.tables.items, + self.tables.order_items, + self.classes.Order, + self.classes.Item, + self.classes.User, + self.tables.keywords, + self.tables.item_keywords) mapper(User, users, properties={ - 'orders': relationship(Order, backref='user'), # o2m, m2o + 'orders': relationship(Order, backref='user'), # o2m, m2o }) mapper(Order, orders, properties={ 'items': relationship(Item, secondary=order_items, - order_by=items.c.id), #m2m + order_by=items.c.id), # m2m }) mapper(Item, items, properties={ 'keywords': relationship(Keyword, secondary=item_keywords, - order_by=keywords.c.id) #m2m + order_by=keywords.c.id) # m2m }) mapper(Keyword, keywords) sess = create_session() q = sess.query(User).join(User.orders).join(Order.items).\ - options(joinedload_all("orders.items.keywords")) + options(joinedload_all("orders.items.keywords")) # here, the eager join for keywords can catch onto # join(Order.items) or the nested (orders LEFT OUTER JOIN items), @@ -1486,40 +1667,47 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): self.assert_compile( q, "SELECT users.id AS users_id, users.name AS users_name, " - "keywords_1.id AS keywords_1_id, keywords_1.name AS keywords_1_name, " - "items_1.id AS items_1_id, items_1.description AS items_1_description, " - "orders_1.id AS orders_1_id, orders_1.user_id AS orders_1_user_id, " + "keywords_1.id AS keywords_1_id, keywords_1.name " + "AS keywords_1_name, " + "items_1.id AS items_1_id, items_1.description AS " + "items_1_description, " + "orders_1.id AS orders_1_id, orders_1.user_id AS " + "orders_1_user_id, " "orders_1.address_id AS orders_1_address_id, " "orders_1.description AS orders_1_description, " "orders_1.isopen AS orders_1_isopen " "FROM users JOIN orders ON users.id = orders.user_id " - "JOIN order_items AS order_items_1 ON orders.id = order_items_1.order_id " + "JOIN order_items AS order_items_1 ON orders.id = " + "order_items_1.order_id " "JOIN items ON items.id = order_items_1.item_id " - "LEFT OUTER JOIN orders AS orders_1 ON users.id = orders_1.user_id " + "LEFT OUTER JOIN orders AS orders_1 ON users.id = " + "orders_1.user_id " "LEFT OUTER JOIN (order_items AS order_items_2 " "JOIN items AS items_1 ON items_1.id = order_items_2.item_id) " "ON orders_1.id = order_items_2.order_id " "LEFT OUTER JOIN (item_keywords AS item_keywords_1 " - "JOIN keywords AS keywords_1 ON keywords_1.id = item_keywords_1.keyword_id) " + "JOIN keywords AS keywords_1 ON keywords_1.id = " + "item_keywords_1.keyword_id) " "ON items_1.id = item_keywords_1.item_id " "ORDER BY items_1.id, keywords_1.id" ) def test_inner_join_chaining_fixed(self): - users, items, order_items, Order, Item, User, orders = (self.tables.users, - self.tables.items, - self.tables.order_items, - self.classes.Order, - self.classes.Item, - self.classes.User, - self.tables.orders) + users, items, order_items, Order, Item, User, orders = ( + self.tables.users, + self.tables.items, + self.tables.order_items, + self.classes.Order, + self.classes.Item, + self.classes.User, + self.tables.orders) mapper(User, users, properties=dict( - orders =relationship(Order, lazy=False) + orders=relationship(Order, lazy=False) )) mapper(Order, orders, properties=dict( items=relationship(Item, secondary=order_items, lazy=False, - innerjoin=True) + innerjoin=True) )) mapper(Item, items) @@ -1528,13 +1716,19 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): # joining from user, its all LEFT OUTER JOINs self.assert_compile( sess.query(User), - "SELECT users.id AS users_id, users.name AS users_name, items_1.id AS " - "items_1_id, items_1.description AS items_1_description, orders_1.id AS " - "orders_1_id, orders_1.user_id AS orders_1_user_id, orders_1.address_id AS " - "orders_1_address_id, orders_1.description AS orders_1_description, " - "orders_1.isopen AS orders_1_isopen FROM users LEFT OUTER JOIN orders AS orders_1 ON " + "SELECT users.id AS users_id, users.name AS users_name, " + "items_1.id AS " + "items_1_id, items_1.description AS items_1_description, " + "orders_1.id AS " + "orders_1_id, orders_1.user_id AS orders_1_user_id, " + "orders_1.address_id AS " + "orders_1_address_id, orders_1.description AS " + "orders_1_description, " + "orders_1.isopen AS orders_1_isopen FROM users LEFT OUTER JOIN " + "orders AS orders_1 ON " "users.id = orders_1.user_id LEFT OUTER JOIN " - "(order_items AS order_items_1 JOIN items AS items_1 ON items_1.id = " + "(order_items AS order_items_1 JOIN items AS items_1 ON " + "items_1.id = " "order_items_1.item_id) ON orders_1.id = " "order_items_1.order_id" ) @@ -1551,22 +1745,22 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): "order_items_1.item_id" ) - def test_inner_join_nested_chaining_fixed(self): - users, items, order_items, Order, Item, User, orders = (self.tables.users, - self.tables.items, - self.tables.order_items, - self.classes.Order, - self.classes.Item, - self.classes.User, - self.tables.orders) + users, items, order_items, Order, Item, User, orders = ( + self.tables.users, + self.tables.items, + self.tables.order_items, + self.classes.Order, + self.classes.Item, + self.classes.User, + self.tables.orders) mapper(User, users, properties=dict( - orders = relationship(Order, lazy=False) + orders=relationship(Order, lazy=False) )) mapper(Order, orders, properties=dict( items=relationship(Item, secondary=order_items, lazy=False, - innerjoin='nested') + innerjoin='nested') )) mapper(Item, items) @@ -1574,78 +1768,119 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): self.assert_compile( sess.query(User), - "SELECT users.id AS users_id, users.name AS users_name, items_1.id AS " - "items_1_id, items_1.description AS items_1_description, orders_1.id AS " - "orders_1_id, orders_1.user_id AS orders_1_user_id, orders_1.address_id AS " - "orders_1_address_id, orders_1.description AS orders_1_description, " + "SELECT users.id AS users_id, users.name AS users_name, " + "items_1.id AS " + "items_1_id, items_1.description AS items_1_description, " + "orders_1.id AS " + "orders_1_id, orders_1.user_id AS orders_1_user_id, " + "orders_1.address_id AS " + "orders_1_address_id, orders_1.description AS " + "orders_1_description, " "orders_1.isopen AS orders_1_isopen " "FROM users LEFT OUTER JOIN " "(orders AS orders_1 JOIN order_items AS order_items_1 " - "ON orders_1.id = order_items_1.order_id " - "JOIN items AS items_1 ON items_1.id = order_items_1.item_id) " - "ON users.id = orders_1.user_id" + "ON orders_1.id = order_items_1.order_id " + "JOIN items AS items_1 ON items_1.id = order_items_1.item_id) " + "ON users.id = orders_1.user_id" ) - - def test_inner_join_options(self): - users, items, order_items, Order, Item, User, orders = (self.tables.users, - self.tables.items, - self.tables.order_items, - self.classes.Order, - self.classes.Item, - self.classes.User, - self.tables.orders) - - mapper(User, users, properties = dict( - orders =relationship(Order, backref=backref('user', innerjoin=True), - order_by=orders.c.id) + users, items, order_items, Order, Item, User, orders = ( + self.tables.users, + self.tables.items, + self.tables.order_items, + self.classes.Order, + self.classes.Item, + self.classes.User, + self.tables.orders) + + mapper(User, users, properties=dict( + orders=relationship(Order, backref=backref('user', innerjoin=True), + order_by=orders.c.id) )) mapper(Order, orders, properties=dict( - items=relationship(Item, secondary=order_items, order_by=items.c.id) + items=relationship( + Item, + secondary=order_items, + order_by=items.c.id) )) mapper(Item, items) sess = create_session() - self.assert_compile(sess.query(User).options(joinedload(User.orders, innerjoin=True)), - "SELECT users.id AS users_id, users.name AS users_name, orders_1.id AS orders_1_id, " - "orders_1.user_id AS orders_1_user_id, orders_1.address_id AS orders_1_address_id, " - "orders_1.description AS orders_1_description, orders_1.isopen AS orders_1_isopen " - "FROM users JOIN orders AS orders_1 ON users.id = orders_1.user_id ORDER BY orders_1.id") - - self.assert_compile(sess.query(User).options(joinedload_all(User.orders, Order.items, innerjoin=True)), - "SELECT users.id AS users_id, users.name AS users_name, items_1.id AS items_1_id, " - "items_1.description AS items_1_description, orders_1.id AS orders_1_id, " - "orders_1.user_id AS orders_1_user_id, orders_1.address_id AS orders_1_address_id, " - "orders_1.description AS orders_1_description, orders_1.isopen AS orders_1_isopen " - "FROM users JOIN orders AS orders_1 ON users.id = orders_1.user_id JOIN order_items AS " - "order_items_1 ON orders_1.id = order_items_1.order_id JOIN items AS items_1 ON " - "items_1.id = order_items_1.item_id ORDER BY orders_1.id, items_1.id") + self.assert_compile( + sess.query(User).options(joinedload(User.orders, innerjoin=True)), + "SELECT users.id AS users_id, users.name AS users_name, " + "orders_1.id AS orders_1_id, " + "orders_1.user_id AS orders_1_user_id, orders_1.address_id AS " + "orders_1_address_id, " + "orders_1.description AS orders_1_description, orders_1.isopen " + "AS orders_1_isopen " + "FROM users JOIN orders AS orders_1 ON users.id = " + "orders_1.user_id ORDER BY orders_1.id") + + self.assert_compile( + sess.query(User).options( + joinedload_all(User.orders, Order.items, innerjoin=True)), + "SELECT users.id AS users_id, users.name AS users_name, " + "items_1.id AS items_1_id, " + "items_1.description AS items_1_description, " + "orders_1.id AS orders_1_id, " + "orders_1.user_id AS orders_1_user_id, orders_1.address_id " + "AS orders_1_address_id, " + "orders_1.description AS orders_1_description, orders_1.isopen " + "AS orders_1_isopen " + "FROM users JOIN orders AS orders_1 ON users.id = " + "orders_1.user_id JOIN order_items AS " + "order_items_1 ON orders_1.id = order_items_1.order_id " + "JOIN items AS items_1 ON " + "items_1.id = order_items_1.item_id ORDER BY orders_1.id, " + "items_1.id") def go(): eq_( sess.query(User).options( joinedload(User.orders, innerjoin=True), joinedload(User.orders, Order.items, innerjoin=True)). - order_by(User.id).all(), + order_by(User.id).all(), [User(id=7, - orders=[ - Order(id=1, items=[ Item(id=1), Item(id=2), Item(id=3)]), - Order(id=3, items=[ Item(id=3), Item(id=4), Item(id=5)]), - Order(id=5, items=[Item(id=5)])]), - User(id=9, orders=[ - Order(id=2, items=[ Item(id=1), Item(id=2), Item(id=3)]), - Order(id=4, items=[ Item(id=1), Item(id=5)])]) - ] + orders=[ + Order( + id=1, items=[ + Item( + id=1), Item( + id=2), Item( + id=3)]), + Order( + id=3, items=[ + Item( + id=3), Item( + id=4), Item( + id=5)]), + Order(id=5, items=[Item(id=5)])]), + User(id=9, orders=[ + Order( + id=2, items=[ + Item( + id=1), Item( + id=2), Item( + id=3)]), + Order(id=4, items=[Item(id=1), Item(id=5)])]) + ] ) self.assert_sql_count(testing.db, go, 1) # test that default innerjoin setting is used for options self.assert_compile( - sess.query(Order).options(joinedload(Order.user)).filter(Order.description == 'foo'), - "SELECT orders.id AS orders_id, orders.user_id AS orders_user_id, orders.address_id AS " - "orders_address_id, orders.description AS orders_description, orders.isopen AS " - "orders_isopen, users_1.id AS users_1_id, users_1.name AS users_1_name " + sess.query(Order).options( + joinedload( + Order.user)).filter( + Order.description == 'foo'), + "SELECT orders.id AS orders_id, orders.user_id AS orders_user_id, " + "orders.address_id AS " + "orders_address_id, orders.description AS orders_description, " + "orders.isopen AS " + "orders_isopen, users_1.id AS users_1_id, users_1.name " + "AS users_1_name " "FROM orders JOIN users AS users_1 ON users_1.id = orders.user_id " "WHERE orders.description = :description_1" ) @@ -1657,13 +1892,14 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): self._test_propagated_lazyload_wildcard(True) def _test_propagated_lazyload_wildcard(self, use_load): - users, items, order_items, Order, Item, User, orders = (self.tables.users, - self.tables.items, - self.tables.order_items, - self.classes.Order, - self.classes.Item, - self.classes.User, - self.tables.orders) + users, items, order_items, Order, Item, User, orders = ( + self.tables.users, + self.tables.items, + self.tables.order_items, + self.classes.Order, + self.classes.Item, + self.classes.User, + self.tables.orders) mapper(User, users, properties=dict( orders=relationship(Order, lazy="select") @@ -1689,16 +1925,17 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): self.sql_eq_(go, [ ("SELECT users.id AS users_id, users.name AS users_name " "FROM users WHERE users.id = :id_1", {"id_1": 7}), - ("SELECT orders.id AS orders_id, orders.user_id AS orders_user_id, " - "orders.address_id AS orders_address_id, " - "orders.description AS orders_description, " - "orders.isopen AS orders_isopen FROM orders " - "WHERE :param_1 = orders.user_id", {"param_1": 7}), + ("SELECT orders.id AS orders_id, " + "orders.user_id AS orders_user_id, " + "orders.address_id AS orders_address_id, " + "orders.description AS orders_description, " + "orders.isopen AS orders_isopen FROM orders " + "WHERE :param_1 = orders.user_id", {"param_1": 7}), ]) - class SubqueryAliasingTest(fixtures.MappedTest, testing.AssertsCompiledSQL): + """test #2188""" __dialect__ = 'default' @@ -1706,44 +1943,45 @@ class SubqueryAliasingTest(fixtures.MappedTest, testing.AssertsCompiledSQL): @classmethod def define_tables(cls, metadata): Table('a', metadata, - Column('id', Integer, primary_key=True) - ) + Column('id', Integer, primary_key=True) + ) Table('b', metadata, - Column('id', Integer, primary_key=True), - Column('a_id', Integer, ForeignKey('a.id')), - Column('value', Integer), - ) + Column('id', Integer, primary_key=True), + Column('a_id', Integer, ForeignKey('a.id')), + Column('value', Integer), + ) @classmethod def setup_classes(cls): class A(cls.Comparable): pass + class B(cls.Comparable): pass def _fixture(self, props): A, B = self.classes.A, self.classes.B b_table, a_table = self.tables.b, self.tables.a - mapper(A,a_table, properties=props) - mapper(B,b_table,properties = { - 'a':relationship(A, backref="bs") + mapper(A, a_table, properties=props) + mapper(B, b_table, properties={ + 'a': relationship(A, backref="bs") }) def test_column_property(self): - A, B = self.classes.A, self.classes.B + A = self.classes.A b_table, a_table = self.tables.b, self.tables.a cp = select([func.sum(b_table.c.value)]).\ - where(b_table.c.a_id==a_table.c.id) + where(b_table.c.a_id == a_table.c.id) self._fixture({ - 'summation':column_property(cp) + 'summation': column_property(cp) }) self.assert_compile( create_session().query(A).options(joinedload_all('bs')). - order_by(A.summation). - limit(50), + order_by(A.summation). + limit(50), "SELECT anon_1.anon_2 AS anon_1_anon_2, anon_1.a_id " "AS anon_1_a_id, b_1.id AS b_1_id, b_1.a_id AS " "b_1_a_id, b_1.value AS b_1_value FROM (SELECT " @@ -1754,18 +1992,18 @@ class SubqueryAliasingTest(fixtures.MappedTest, testing.AssertsCompiledSQL): ) def test_column_property_desc(self): - A, B = self.classes.A, self.classes.B + A = self.classes.A b_table, a_table = self.tables.b, self.tables.a cp = select([func.sum(b_table.c.value)]).\ - where(b_table.c.a_id==a_table.c.id) + where(b_table.c.a_id == a_table.c.id) self._fixture({ - 'summation':column_property(cp) + 'summation': column_property(cp) }) self.assert_compile( create_session().query(A).options(joinedload_all('bs')). - order_by(A.summation.desc()). - limit(50), + order_by(A.summation.desc()). + limit(50), "SELECT anon_1.anon_2 AS anon_1_anon_2, anon_1.a_id " "AS anon_1_a_id, b_1.id AS b_1_id, b_1.a_id AS " "b_1_a_id, b_1.value AS b_1_value FROM (SELECT " @@ -1776,19 +2014,19 @@ class SubqueryAliasingTest(fixtures.MappedTest, testing.AssertsCompiledSQL): ) def test_column_property_correlated(self): - A, B = self.classes.A, self.classes.B + A = self.classes.A b_table, a_table = self.tables.b, self.tables.a cp = select([func.sum(b_table.c.value)]).\ - where(b_table.c.a_id==a_table.c.id).\ - correlate(a_table) + where(b_table.c.a_id == a_table.c.id).\ + correlate(a_table) self._fixture({ - 'summation':column_property(cp) + 'summation': column_property(cp) }) self.assert_compile( create_session().query(A).options(joinedload_all('bs')). - order_by(A.summation). - limit(50), + order_by(A.summation). + limit(50), "SELECT anon_1.anon_2 AS anon_1_anon_2, anon_1.a_id " "AS anon_1_a_id, b_1.id AS b_1_id, b_1.a_id AS " "b_1_a_id, b_1.value AS b_1_value FROM (SELECT " @@ -1799,20 +2037,20 @@ class SubqueryAliasingTest(fixtures.MappedTest, testing.AssertsCompiledSQL): ) def test_standalone_subquery_unlabeled(self): - A, B = self.classes.A, self.classes.B + A = self.classes.A b_table, a_table = self.tables.b, self.tables.a self._fixture({}) cp = select([func.sum(b_table.c.value)]).\ - where(b_table.c.a_id == a_table.c.id).\ - correlate(a_table).as_scalar() + where(b_table.c.a_id == a_table.c.id).\ + correlate(a_table).as_scalar() # up until 0.8, this was ordering by a new subquery. # the removal of a separate _make_proxy() from ScalarSelect # fixed that. self.assert_compile( create_session().query(A).options(joinedload_all('bs')). - order_by(cp). - limit(50), + order_by(cp). + limit(50), "SELECT anon_1.a_id AS anon_1_a_id, anon_1.anon_2 " "AS anon_1_anon_2, b_1.id AS b_1_id, b_1.a_id AS " "b_1_a_id, b_1.value AS b_1_value FROM (SELECT a.id " @@ -1824,16 +2062,16 @@ class SubqueryAliasingTest(fixtures.MappedTest, testing.AssertsCompiledSQL): ) def test_standalone_subquery_labeled(self): - A, B = self.classes.A, self.classes.B + A = self.classes.A b_table, a_table = self.tables.b, self.tables.a self._fixture({}) cp = select([func.sum(b_table.c.value)]).\ - where(b_table.c.a_id==a_table.c.id).\ - correlate(a_table).as_scalar().label('foo') + where(b_table.c.a_id == a_table.c.id).\ + correlate(a_table).as_scalar().label('foo') self.assert_compile( create_session().query(A).options(joinedload_all('bs')). - order_by(cp). - limit(50), + order_by(cp). + limit(50), "SELECT anon_1.a_id AS anon_1_a_id, anon_1.foo " "AS anon_1_foo, b_1.id AS b_1_id, b_1.a_id AS " "b_1_a_id, b_1.value AS b_1_value FROM (SELECT a.id " @@ -1845,18 +2083,18 @@ class SubqueryAliasingTest(fixtures.MappedTest, testing.AssertsCompiledSQL): ) def test_standalone_negated(self): - A, B = self.classes.A, self.classes.B + A = self.classes.A b_table, a_table = self.tables.b, self.tables.a self._fixture({}) cp = select([func.sum(b_table.c.value)]).\ - where(b_table.c.a_id==a_table.c.id).\ - correlate(a_table).\ - as_scalar() + where(b_table.c.a_id == a_table.c.id).\ + correlate(a_table).\ + as_scalar() # test a different unary operator self.assert_compile( create_session().query(A).options(joinedload_all('bs')). - order_by(~cp). - limit(50), + order_by(~cp). + limit(50), "SELECT anon_1.a_id AS anon_1_a_id, anon_1.anon_2 " "AS anon_1_anon_2, b_1.id AS b_1_id, b_1.a_id AS " "b_1_a_id, b_1.value AS b_1_value FROM (SELECT a.id " @@ -1869,6 +2107,7 @@ class SubqueryAliasingTest(fixtures.MappedTest, testing.AssertsCompiledSQL): class LoadOnExistingTest(_fixtures.FixtureTest): + """test that loaders from a base Query fully populate.""" run_inserts = 'once' @@ -1878,10 +2117,10 @@ class LoadOnExistingTest(_fixtures.FixtureTest): User, Address, Dingaling = self.classes.User, \ self.classes.Address, self.classes.Dingaling mapper(User, self.tables.users, properties={ - 'addresses':relationship(Address), + 'addresses': relationship(Address), }) mapper(Address, self.tables.addresses, properties={ - 'dingaling':relationship(Dingaling) + 'dingaling': relationship(Dingaling) }) mapper(Dingaling, self.tables.dingalings) @@ -1892,10 +2131,10 @@ class LoadOnExistingTest(_fixtures.FixtureTest): User, Order, Item = self.classes.User, \ self.classes.Order, self.classes.Item mapper(User, self.tables.users, properties={ - 'orders':relationship(Order), + 'orders': relationship(Order), }) mapper(Order, self.tables.orders, properties={ - 'items':relationship(Item, secondary=self.tables.order_items), + 'items': relationship(Item, secondary=self.tables.order_items), }) mapper(Item, self.tables.items) @@ -1905,7 +2144,7 @@ class LoadOnExistingTest(_fixtures.FixtureTest): def _eager_config_fixture(self): User, Address = self.classes.User, self.classes.Address mapper(User, self.tables.users, properties={ - 'addresses':relationship(Address, lazy="joined"), + 'addresses': relationship(Address, lazy="joined"), }) mapper(Address, self.tables.addresses) sess = Session(autoflush=False) @@ -1917,6 +2156,7 @@ class LoadOnExistingTest(_fixtures.FixtureTest): u1 = sess.query(User).get(8) assert 'addresses' in u1.__dict__ sess.expire(u1) + def go(): eq_(u1.id, 8) self.assert_sql_count(testing.db, go, 1) @@ -1931,7 +2171,7 @@ class LoadOnExistingTest(_fixtures.FixtureTest): a2 = u1.addresses[0] a2.email_address = 'foo' sess.query(User).options(joinedload_all("addresses.dingaling")).\ - filter_by(id=8).all() + filter_by(id=8).all() assert u1.addresses[-1] is a1 for a in u1.addresses: if a is not a1: @@ -1949,7 +2189,7 @@ class LoadOnExistingTest(_fixtures.FixtureTest): o1 = Order() u1.orders.append(o1) sess.query(User).options(joinedload_all("orders.items")).\ - filter_by(id=7).all() + filter_by(id=7).all() for o in u1.orders: if o is not o1: assert 'items' in o.__dict__ @@ -1959,15 +2199,23 @@ class LoadOnExistingTest(_fixtures.FixtureTest): def test_load_two_levels_collection_to_scalar(self): User, Address, Dingaling, sess = self._collection_to_scalar_fixture() - u1 = sess.query(User).filter_by(id=8).options(joinedload("addresses")).one() - sess.query(User).filter_by(id=8).options(joinedload_all("addresses.dingaling")).first() + u1 = sess.query(User).filter_by( + id=8).options( + joinedload("addresses")).one() + sess.query(User).filter_by( + id=8).options( + joinedload_all("addresses.dingaling")).first() assert 'dingaling' in u1.addresses[0].__dict__ def test_load_two_levels_collection_to_collection(self): User, Order, Item, sess = self._collection_to_collection_fixture() - u1 = sess.query(User).filter_by(id=7).options(joinedload("orders")).one() - sess.query(User).filter_by(id=7).options(joinedload_all("orders.items")).first() + u1 = sess.query(User).filter_by( + id=7).options( + joinedload("orders")).one() + sess.query(User).filter_by( + id=7).options( + joinedload_all("orders.items")).first() assert 'items' in u1.orders[0].__dict__ @@ -1977,137 +2225,160 @@ class AddEntityTest(_fixtures.FixtureTest): def _assert_result(self): Item, Address, Order, User = (self.classes.Item, - self.classes.Address, - self.classes.Order, - self.classes.User) + self.classes.Address, + self.classes.Order, + self.classes.User) return [ ( User(id=7, - addresses=[Address(id=1)] - ), + addresses=[Address(id=1)] + ), Order(id=1, - items=[Item(id=1), Item(id=2), Item(id=3)] - ), + items=[Item(id=1), Item(id=2), Item(id=3)] + ), ), ( User(id=7, - addresses=[Address(id=1)] - ), + addresses=[Address(id=1)] + ), Order(id=3, - items=[Item(id=3), Item(id=4), Item(id=5)] - ), + items=[Item(id=3), Item(id=4), Item(id=5)] + ), ), ( User(id=7, - addresses=[Address(id=1)] - ), + addresses=[Address(id=1)] + ), Order(id=5, - items=[Item(id=5)] - ), + items=[Item(id=5)] + ), ), ( - User(id=9, - addresses=[Address(id=5)] - ), - Order(id=2, - items=[Item(id=1), Item(id=2), Item(id=3)] - ), - ), - ( - User(id=9, - addresses=[Address(id=5)] - ), - Order(id=4, - items=[Item(id=1), Item(id=5)] - ), - ) + User(id=9, + addresses=[Address(id=5)] + ), + Order(id=2, + items=[Item(id=1), Item(id=2), Item(id=3)] + ), + ), + ( + User(id=9, + addresses=[Address(id=5)] + ), + Order(id=4, + items=[Item(id=1), Item(id=5)] + ), + ) ] def test_mapper_configured(self): - users, items, order_items, Order, Item, User, Address, orders, addresses = (self.tables.users, - self.tables.items, - self.tables.order_items, - self.classes.Order, - self.classes.Item, - self.classes.User, - self.classes.Address, - self.tables.orders, - self.tables.addresses) + users, items, order_items, Order, \ + Item, User, Address, orders, addresses = ( + self.tables.users, + self.tables.items, + self.tables.order_items, + self.classes.Order, + self.classes.Item, + self.classes.User, + self.classes.Address, + self.tables.orders, + self.tables.addresses) mapper(User, users, properties={ - 'addresses':relationship(Address, lazy='joined'), - 'orders':relationship(Order) + 'addresses': relationship(Address, lazy='joined'), + 'orders': relationship(Order) }) mapper(Address, addresses) mapper(Order, orders, properties={ - 'items':relationship(Item, secondary=order_items, lazy='joined', order_by=items.c.id) + 'items': relationship( + Item, secondary=order_items, lazy='joined', + order_by=items.c.id) }) mapper(Item, items) - sess = create_session() oalias = sa.orm.aliased(Order) + def go(): ret = sess.query(User, oalias).join(oalias, 'orders').\ - order_by(User.id,oalias.id).all() + order_by(User.id, oalias.id).all() eq_(ret, self._assert_result()) self.assert_sql_count(testing.db, go, 1) def test_options(self): - users, items, order_items, Order, Item, User, Address, orders, addresses = (self.tables.users, - self.tables.items, - self.tables.order_items, - self.classes.Order, - self.classes.Item, - self.classes.User, - self.classes.Address, - self.tables.orders, - self.tables.addresses) + users, items, order_items, Order,\ + Item, User, Address, orders, addresses = ( + self.tables.users, + self.tables.items, + self.tables.order_items, + self.classes.Order, + self.classes.Item, + self.classes.User, + self.classes.Address, + self.tables.orders, + self.tables.addresses) mapper(User, users, properties={ - 'addresses':relationship(Address), - 'orders':relationship(Order) + 'addresses': relationship(Address), + 'orders': relationship(Order) }) mapper(Address, addresses) mapper(Order, orders, properties={ - 'items':relationship(Item, secondary=order_items, order_by=items.c.id) + 'items': relationship( + Item, secondary=order_items, order_by=items.c.id) }) mapper(Item, items) sess = create_session() oalias = sa.orm.aliased(Order) + def go(): ret = sess.query(User, oalias).options(joinedload('addresses')).\ - join(oalias, 'orders').\ - order_by(User.id, oalias.id).all() + join(oalias, 'orders').\ + order_by(User.id, oalias.id).all() eq_(ret, self._assert_result()) self.assert_sql_count(testing.db, go, 6) sess.expunge_all() + def go(): ret = sess.query(User, oalias).\ - options(joinedload('addresses'), - joinedload(oalias.items)).\ - join(oalias, 'orders').\ - order_by(User.id, oalias.id).all() + options(joinedload('addresses'), + joinedload(oalias.items)).\ + join(oalias, 'orders').\ + order_by(User.id, oalias.id).all() eq_(ret, self._assert_result()) self.assert_sql_count(testing.db, go, 1) + class OrderBySecondaryTest(fixtures.MappedTest): + @classmethod def define_tables(cls, metadata): Table('m2m', metadata, - Column('id', Integer, primary_key=True, test_needs_autoincrement=True), + Column( + 'id', + Integer, + primary_key=True, + test_needs_autoincrement=True), Column('aid', Integer, ForeignKey('a.id')), Column('bid', Integer, ForeignKey('b.id'))) Table('a', metadata, - Column('id', Integer, primary_key=True, test_needs_autoincrement=True), + Column( + 'id', + Integer, + primary_key=True, + test_needs_autoincrement=True), Column('data', String(50))) Table('b', metadata, - Column('id', Integer, primary_key=True, test_needs_autoincrement=True), + Column( + 'id', + Integer, + primary_key=True, + test_needs_autoincrement=True), Column('data', String(50))) @classmethod @@ -2132,44 +2403,56 @@ class OrderBySecondaryTest(fixtures.MappedTest): (5, 2, 3))) def test_ordering(self): - a, m2m, b = (self.tables.a, - self.tables.m2m, - self.tables.b) + a, m2m, b = ( + self.tables.a, + self.tables.m2m, + self.tables.b) + + class A(fixtures.ComparableEntity): + pass - class A(fixtures.ComparableEntity):pass - class B(fixtures.ComparableEntity):pass + class B(fixtures.ComparableEntity): + pass mapper(A, a, properties={ - 'bs':relationship(B, secondary=m2m, lazy='joined', order_by=m2m.c.id) + 'bs': relationship( + B, secondary=m2m, lazy='joined', order_by=m2m.c.id) }) mapper(B, b) sess = create_session() - eq_(sess.query(A).all(), [ - A(data='a1', bs=[B(data='b3'), B(data='b1'), B(data='b2')]), - A(bs=[B(data='b4'), B(data='b3'), B(data='b2')]) - ]) + eq_(sess.query(A).all(), + [ + A(data='a1', bs=[B(data='b3'), B(data='b1'), B(data='b2')]), + A(bs=[B(data='b4'), B(data='b3'), B(data='b2')]) + ]) class SelfReferentialEagerTest(fixtures.MappedTest): + @classmethod def define_tables(cls, metadata): Table('nodes', metadata, - Column('id', Integer, primary_key=True, test_needs_autoincrement=True), - Column('parent_id', Integer, ForeignKey('nodes.id')), - Column('data', String(30))) + Column( + 'id', + Integer, + primary_key=True, + test_needs_autoincrement=True), + Column('parent_id', Integer, ForeignKey('nodes.id')), + Column('data', String(30))) def test_basic(self): nodes = self.tables.nodes class Node(fixtures.ComparableEntity): + def append(self, node): self.children.append(node) mapper(Node, nodes, properties={ - 'children':relationship(Node, - lazy='joined', - join_depth=3, order_by=nodes.c.id) + 'children': relationship(Node, + lazy='joined', + join_depth=3, order_by=nodes.c.id) }) sess = create_session() n1 = Node(data='n1') @@ -2182,6 +2465,7 @@ class SelfReferentialEagerTest(fixtures.MappedTest): sess.add(n1) sess.flush() sess.expunge_all() + def go(): d = sess.query(Node).filter_by(data='n1').all()[0] eq_(Node(data='n1', children=[ @@ -2192,10 +2476,11 @@ class SelfReferentialEagerTest(fixtures.MappedTest): Node(data='n123') ]), Node(data='n13') - ]), d) + ]), d) self.assert_sql_count(testing.db, go, 1) sess.expunge_all() + def go(): d = sess.query(Node).filter_by(data='n1').first() eq_(Node(data='n1', children=[ @@ -2206,20 +2491,20 @@ class SelfReferentialEagerTest(fixtures.MappedTest): Node(data='n123') ]), Node(data='n13') - ]), d) + ]), d) self.assert_sql_count(testing.db, go, 1) - def test_lazy_fallback_doesnt_affect_eager(self): nodes = self.tables.nodes class Node(fixtures.ComparableEntity): + def append(self, node): self.children.append(node) mapper(Node, nodes, properties={ - 'children':relationship(Node, lazy='joined', join_depth=1, - order_by=nodes.c.id) + 'children': relationship(Node, lazy='joined', join_depth=1, + order_by=nodes.c.id) }) sess = create_session() n1 = Node(data='n1') @@ -2249,20 +2534,21 @@ class SelfReferentialEagerTest(fixtures.MappedTest): Node(data='n121'), Node(data='n122'), Node(data='n123') - ], list(n12.children)) + ], list(n12.children)) self.assert_sql_count(testing.db, go, 1) def test_with_deferred(self): nodes = self.tables.nodes class Node(fixtures.ComparableEntity): + def append(self, node): self.children.append(node) mapper(Node, nodes, properties={ - 'children':relationship(Node, lazy='joined', join_depth=3, - order_by=nodes.c.id), - 'data':deferred(nodes.c.data) + 'children': relationship(Node, lazy='joined', join_depth=3, + order_by=nodes.c.id), + 'data': deferred(nodes.c.data) }) sess = create_session() n1 = Node(data='n1') @@ -2276,14 +2562,15 @@ class SelfReferentialEagerTest(fixtures.MappedTest): eq_( Node(data='n1', children=[Node(data='n11'), Node(data='n12')]), sess.query(Node).order_by(Node.id).first(), - ) + ) self.assert_sql_count(testing.db, go, 4) sess.expunge_all() def go(): eq_(Node(data='n1', children=[Node(data='n11'), Node(data='n12')]), - sess.query(Node).options(undefer('data')).order_by(Node.id).first()) + sess.query(Node). + options(undefer('data')).order_by(Node.id).first()) self.assert_sql_count(testing.db, go, 3) sess.expunge_all() @@ -2291,19 +2578,19 @@ class SelfReferentialEagerTest(fixtures.MappedTest): def go(): eq_(Node(data='n1', children=[Node(data='n11'), Node(data='n12')]), sess.query(Node).options(undefer('data'), - undefer('children.data')).first()) + undefer('children.data')).first()) self.assert_sql_count(testing.db, go, 1) - def test_options(self): nodes = self.tables.nodes class Node(fixtures.ComparableEntity): + def append(self, node): self.children.append(node) mapper(Node, nodes, properties={ - 'children':relationship(Node, lazy='select', order_by=nodes.c.id) + 'children': relationship(Node, lazy='select', order_by=nodes.c.id) }, order_by=nodes.c.id) sess = create_session() n1 = Node(data='n1') @@ -2316,9 +2603,10 @@ class SelfReferentialEagerTest(fixtures.MappedTest): sess.add(n1) sess.flush() sess.expunge_all() + def go(): d = sess.query(Node).filter_by(data='n1').\ - options(joinedload('children.children')).first() + options(joinedload('children.children')).first() eq_(Node(data='n1', children=[ Node(data='n11'), Node(data='n12', children=[ @@ -2327,15 +2615,17 @@ class SelfReferentialEagerTest(fixtures.MappedTest): Node(data='n123') ]), Node(data='n13') - ]), d) + ]), d) self.assert_sql_count(testing.db, go, 2) def go(): - d = sess.query(Node).filter_by(data='n1').\ - options(joinedload('children.children')).first() + sess.query(Node).filter_by(data='n1').\ + options(joinedload('children.children')).first() - # test that the query isn't wrapping the initial query for eager loading. - self.assert_sql_execution(testing.db, go, + # test that the query isn't wrapping the initial query for eager + # loading. + self.assert_sql_execution( + testing.db, go, CompiledSQL( "SELECT nodes.id AS nodes_id, nodes.parent_id AS " "nodes_parent_id, nodes.data AS nodes_data FROM nodes " @@ -2348,11 +2638,12 @@ class SelfReferentialEagerTest(fixtures.MappedTest): nodes = self.tables.nodes class Node(fixtures.ComparableEntity): + def append(self, node): self.children.append(node) mapper(Node, nodes, properties={ - 'children':relationship(Node, lazy='joined') + 'children': relationship(Node, lazy='joined') }) sess = create_session() n1 = Node(data='n1') @@ -2365,6 +2656,7 @@ class SelfReferentialEagerTest(fixtures.MappedTest): sess.add(n1) sess.flush() sess.expunge_all() + def go(): d = sess.query(Node).filter_by(data='n1').first() eq_(Node(data='n1', children=[ @@ -2375,22 +2667,31 @@ class SelfReferentialEagerTest(fixtures.MappedTest): Node(data='n123') ]), Node(data='n13') - ]), d) + ]), d) self.assert_sql_count(testing.db, go, 3) + class MixedSelfReferentialEagerTest(fixtures.MappedTest): + @classmethod def define_tables(cls, metadata): Table('a_table', metadata, - Column('id', Integer, primary_key=True, test_needs_autoincrement=True) - ) + Column( + 'id', + Integer, + primary_key=True, + test_needs_autoincrement=True) + ) Table('b_table', metadata, - Column('id', Integer, primary_key=True, test_needs_autoincrement=True), - Column('parent_b1_id', Integer, ForeignKey('b_table.id')), - Column('parent_a_id', Integer, ForeignKey('a_table.id')), - Column('parent_b2_id', Integer, ForeignKey('b_table.id'))) - + Column( + 'id', + Integer, + primary_key=True, + test_needs_autoincrement=True), + Column('parent_b1_id', Integer, ForeignKey('b_table.id')), + Column('parent_a_id', Integer, ForeignKey('a_table.id')), + Column('parent_b2_id', Integer, ForeignKey('b_table.id'))) @classmethod def setup_mappers(cls): @@ -2398,23 +2699,26 @@ class MixedSelfReferentialEagerTest(fixtures.MappedTest): class A(cls.Comparable): pass + class B(cls.Comparable): pass - mapper(A,a_table) - mapper(B,b_table,properties = { - 'parent_b1': relationship(B, - remote_side = [b_table.c.id], - primaryjoin = (b_table.c.parent_b1_id ==b_table.c.id), - order_by = b_table.c.id - ), - 'parent_z': relationship(A,lazy = True), - 'parent_b2': relationship(B, - remote_side = [b_table.c.id], - primaryjoin = (b_table.c.parent_b2_id ==b_table.c.id), - order_by = b_table.c.id - ) - }); + mapper(A, a_table) + mapper(B, b_table, properties={ + 'parent_b1': relationship( + B, + remote_side=[b_table.c.id], + primaryjoin=(b_table.c.parent_b1_id == b_table.c.id), + order_by=b_table.c.id + ), + 'parent_z': relationship(A, lazy=True), + 'parent_b2': relationship( + B, + remote_side=[b_table.c.id], + primaryjoin=(b_table.c.parent_b2_id == b_table.c.id), + order_by = b_table.c.id + ) + }) @classmethod def insert_data(cls): @@ -2442,36 +2746,52 @@ class MixedSelfReferentialEagerTest(fixtures.MappedTest): A, B = self.classes.A, self.classes.B session = create_session() + def go(): eq_( - session.query(B).\ - options( - joinedload('parent_b1'), - joinedload('parent_b2'), - joinedload('parent_z') - ). - filter(B.id.in_([2, 8, 11])).order_by(B.id).all(), + session.query(B). + options( + joinedload('parent_b1'), + joinedload('parent_b2'), + joinedload('parent_z') + ). + filter(B.id.in_([2, 8, 11])).order_by(B.id).all(), [ - B(id=2, parent_z=A(id=1), parent_b1=B(id=1), parent_b2=None), - B(id=8, parent_z=A(id=2), parent_b1=B(id=1), parent_b2=B(id=2)), - B(id=11, parent_z=A(id=3), parent_b1=B(id=1), parent_b2=B(id=8)) + B(id=2, + parent_z=A(id=1), + parent_b1=B(id=1), + parent_b2=None), + B(id=8, + parent_z=A(id=2), + parent_b1=B(id=1), + parent_b2=B(id=2)), + B(id=11, + parent_z=A(id=3), + parent_b1=B(id=1), + parent_b2=B(id=8)) ] ) self.assert_sql_count(testing.db, go, 1) + class SelfReferentialM2MEagerTest(fixtures.MappedTest): + @classmethod def define_tables(cls, metadata): Table('widget', metadata, - Column('id', Integer, primary_key=True, test_needs_autoincrement=True), - Column('name', sa.String(40), nullable=False, unique=True), - ) + Column( + 'id', + Integer, + primary_key=True, + test_needs_autoincrement=True), + Column('name', sa.String(40), nullable=False, unique=True), + ) Table('widget_rel', metadata, - Column('parent_id', Integer, ForeignKey('widget.id')), - Column('child_id', Integer, ForeignKey('widget.id')), - sa.UniqueConstraint('parent_id', 'child_id'), - ) + Column('parent_id', Integer, ForeignKey('widget.id')), + Column('child_id', Integer, ForeignKey('widget.id')), + sa.UniqueConstraint('parent_id', 'child_id'), + ) def test_basic(self): widget, widget_rel = self.tables.widget, self.tables.widget_rel @@ -2480,9 +2800,10 @@ class SelfReferentialM2MEagerTest(fixtures.MappedTest): pass mapper(Widget, widget, properties={ - 'children': relationship(Widget, secondary=widget_rel, - primaryjoin=widget_rel.c.parent_id==widget.c.id, - secondaryjoin=widget_rel.c.child_id==widget.c.id, + 'children': relationship( + Widget, secondary=widget_rel, + primaryjoin=widget_rel.c.parent_id == widget.c.id, + secondaryjoin=widget_rel.c.child_id == widget.c.id, lazy='joined', join_depth=1, ) }) @@ -2496,7 +2817,8 @@ class SelfReferentialM2MEagerTest(fixtures.MappedTest): sess.expunge_all() eq_([Widget(name='w1', children=[Widget(name='w2')])], - sess.query(Widget).filter(Widget.name=='w1').all()) + sess.query(Widget).filter(Widget.name == 'w1').all()) + class MixedEntitiesTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): run_setup_mappers = 'once' @@ -2508,37 +2830,41 @@ class MixedEntitiesTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): @classmethod def setup_mappers(cls): - users, Keyword, items, order_items, orders, Item, User, Address, keywords, Order, item_keywords, addresses = (cls.tables.users, - cls.classes.Keyword, - cls.tables.items, - cls.tables.order_items, - cls.tables.orders, - cls.classes.Item, - cls.classes.User, - cls.classes.Address, - cls.tables.keywords, - cls.classes.Order, - cls.tables.item_keywords, - cls.tables.addresses) + users, Keyword, items, order_items, orders, \ + Item, User, Address, keywords, Order, \ + item_keywords, addresses = ( + cls.tables.users, + cls.classes.Keyword, + cls.tables.items, + cls.tables.order_items, + cls.tables.orders, + cls.classes.Item, + cls.classes.User, + cls.classes.Address, + cls.tables.keywords, + cls.classes.Order, + cls.tables.item_keywords, + cls.tables.addresses) mapper(User, users, properties={ - 'addresses':relationship(Address, backref='user'), - 'orders':relationship(Order, backref='user'), # o2m, m2o + 'addresses': relationship(Address, backref='user'), + 'orders': relationship(Order, backref='user'), # o2m, m2o }) mapper(Address, addresses) mapper(Order, orders, properties={ - 'items':relationship(Item, secondary=order_items, order_by=items.c.id), #m2m + 'items': relationship( + Item, secondary=order_items, order_by=items.c.id), # m2m }) mapper(Item, items, properties={ - 'keywords':relationship(Keyword, secondary=item_keywords) #m2m + 'keywords': relationship(Keyword, secondary=item_keywords) # m2m }) mapper(Keyword, keywords) def test_two_entities(self): Item, Order, User, Address = (self.classes.Item, - self.classes.Order, - self.classes.User, - self.classes.Address) + self.classes.Order, + self.classes.User, + self.classes.Address) sess = create_session() @@ -2546,12 +2872,17 @@ class MixedEntitiesTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): def go(): eq_( [ - (User(id=9, addresses=[Address(id=5)]), Order(id=2, items=[Item(id=1), Item(id=2), Item(id=3)])), - (User(id=9, addresses=[Address(id=5)]), Order(id=4, items=[Item(id=1), Item(id=5)])), + (User(id=9, addresses=[Address(id=5)]), + Order(id=2, items=[ + Item(id=1), Item(id=2), Item(id=3)])), + (User(id=9, addresses=[Address(id=5)]), + Order(id=4, items=[ + Item(id=1), Item(id=5)])), ], - sess.query(User, Order).filter(User.id==Order.user_id).\ - options(joinedload(User.addresses), joinedload(Order.items)).filter(User.id==9).\ - order_by(User.id, Order.id).all(), + sess.query(User, Order).filter(User.id == Order.user_id). + options(joinedload(User.addresses), joinedload(Order.items)). + filter(User.id == 9). + order_by(User.id, Order.id).all(), ) self.assert_sql_count(testing.db, go, 1) @@ -2559,20 +2890,27 @@ class MixedEntitiesTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): def go(): eq_( [ - (User(id=9, addresses=[Address(id=5)]), Order(id=2, items=[Item(id=1), Item(id=2), Item(id=3)])), - (User(id=9, addresses=[Address(id=5)]), Order(id=4, items=[Item(id=1), Item(id=5)])), + (User(id=9, addresses=[Address(id=5)]), + Order(id=2, items=[ + Item(id=1), Item(id=2), Item(id=3)])), + (User(id=9, addresses=[Address(id=5)]), + Order(id=4, items=[ + Item(id=1), Item(id=5)])), ], - sess.query(User, Order).join(User.orders).options(joinedload(User.addresses), joinedload(Order.items)).filter(User.id==9).\ - order_by(User.id, Order.id).all(), + sess.query(User, Order).join(User.orders). + options(joinedload(User.addresses), joinedload(Order.items)). + filter(User.id == 9). + order_by(User.id, Order.id).all(), ) self.assert_sql_count(testing.db, go, 1) - @testing.exclude('sqlite', '>', (0, ), "sqlite flat out blows it on the multiple JOINs") + @testing.exclude( + 'sqlite', '>', (0, ), "sqlite flat out blows it on the multiple JOINs") def test_two_entities_with_joins(self): Item, Order, User, Address = (self.classes.Item, - self.classes.Order, - self.classes.User, - self.classes.Address) + self.classes.Order, + self.classes.User, + self.classes.Address) sess = create_session() @@ -2583,41 +2921,92 @@ class MixedEntitiesTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): eq_( [ ( - User(addresses=[Address(email_address='fred@fred.com')], name='fred'), - Order(description='order 2', isopen=0, items=[Item(description='item 1'), Item(description='item 2'), Item(description='item 3')]), - User(addresses=[Address(email_address='jack@bean.com')], name='jack'), - Order(description='order 3', isopen=1, items=[Item(description='item 3'), Item(description='item 4'), Item(description='item 5')]) + User(addresses=[ + Address(email_address='fred@fred.com')], + name='fred'), + Order(description='order 2', isopen=0, + items=[ + Item(description='item 1'), + Item(description='item 2'), + Item(description='item 3')]), + User(addresses=[ + Address(email_address='jack@bean.com')], + name='jack'), + Order(description='order 3', isopen=1, + items=[ + Item(description='item 3'), + Item(description='item 4'), + Item(description='item 5')]) ), ( - User(addresses=[Address(email_address='fred@fred.com')], name='fred'), - Order(description='order 2', isopen=0, items=[Item(description='item 1'), Item(description='item 2'), Item(description='item 3')]), - User(addresses=[Address(email_address='jack@bean.com')], name='jack'), - Order(address_id=None, description='order 5', isopen=0, items=[Item(description='item 5')]) + User( + addresses=[ + Address( + email_address='fred@fred.com')], + name='fred'), + Order( + description='order 2', isopen=0, items=[ + Item( + description='item 1'), Item( + description='item 2'), Item( + description='item 3')]), + User( + addresses=[ + Address( + email_address='jack@bean.com')], + name='jack'), + Order( + address_id=None, + description='order 5', + isopen=0, + items=[ + Item( + description='item 5')]) ), ( - User(addresses=[Address(email_address='fred@fred.com')], name='fred'), - Order(description='order 4', isopen=1, items=[Item(description='item 1'), Item(description='item 5')]), - User(addresses=[Address(email_address='jack@bean.com')], name='jack'), - Order(address_id=None, description='order 5', isopen=0, items=[Item(description='item 5')]) + User( + addresses=[ + Address( + email_address='fred@fred.com')], + name='fred'), + Order( + description='order 4', isopen=1, items=[ + Item( + description='item 1'), Item( + description='item 5')]), + User( + addresses=[ + Address( + email_address='jack@bean.com')], + name='jack'), + Order( + address_id=None, + description='order 5', + isopen=0, + items=[ + Item( + description='item 5')]) ), ], - sess.query(User, Order, u1, o1).\ - join(Order, User.orders).options(joinedload(User.addresses), joinedload(Order.items)).filter(User.id==9).\ - join(o1, u1.orders).options(joinedload(u1.addresses), joinedload(o1.items)).filter(u1.id==7).\ - filter(Order.id<o1.id).\ - order_by(User.id, Order.id, u1.id, o1.id).all(), + sess.query(User, Order, u1, o1). + join(Order, User.orders). + options(joinedload(User.addresses), + joinedload(Order.items)).filter(User.id == 9). + join(o1, u1.orders). + options(joinedload(u1.addresses), + joinedload(o1.items)).filter(u1.id == 7). + filter(Order.id < o1.id). + order_by(User.id, Order.id, u1.id, o1.id).all(), ) self.assert_sql_count(testing.db, go, 1) - - def test_aliased_entity_one(self): Item, Order, User, Address = (self.classes.Item, - self.classes.Order, - self.classes.User, - self.classes.Address) + self.classes.Order, + self.classes.User, + self.classes.Address) sess = create_session() @@ -2627,87 +3016,114 @@ class MixedEntitiesTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL): def go(): eq_( [ - (User(id=9, addresses=[Address(id=5)]), Order(id=2, items=[Item(id=1), Item(id=2), Item(id=3)])), - (User(id=9, addresses=[Address(id=5)]), Order(id=4, items=[Item(id=1), Item(id=5)])), + ( + User( + id=9, addresses=[ + Address( + id=5)]), Order( + id=2, items=[ + Item( + id=1), Item( + id=2), Item( + id=3)])), + (User(id=9, addresses=[Address(id=5)]), Order( + id=4, items=[Item(id=1), Item(id=5)])), ], - sess.query(User, oalias).filter(User.id==oalias.user_id).\ - options(joinedload(User.addresses), joinedload(oalias.items)).filter(User.id==9).\ - order_by(User.id, oalias.id).all(), + sess.query(User, oalias).filter(User.id == oalias.user_id). + options( + joinedload(User.addresses), + joinedload(oalias.items)).filter(User.id == 9). + order_by(User.id, oalias.id).all(), ) self.assert_sql_count(testing.db, go, 1) def test_aliased_entity_two(self): Item, Order, User, Address = (self.classes.Item, - self.classes.Order, - self.classes.User, - self.classes.Address) + self.classes.Order, + self.classes.User, + self.classes.Address) sess = create_session() oalias = sa.orm.aliased(Order) - # one FROM clause def go(): eq_( [ - (User(id=9, addresses=[Address(id=5)]), Order(id=2, items=[Item(id=1), Item(id=2), Item(id=3)])), - (User(id=9, addresses=[Address(id=5)]), Order(id=4, items=[Item(id=1), Item(id=5)])), + ( + User( + id=9, addresses=[ + Address( + id=5)]), Order( + id=2, items=[ + Item( + id=1), Item( + id=2), Item( + id=3)])), + (User(id=9, addresses=[Address(id=5)]), Order( + id=4, items=[Item(id=1), Item(id=5)])), ], sess.query(User, oalias).join(oalias, User.orders). - options(joinedload(User.addresses), - joinedload(oalias.items)). - filter(User.id==9). - order_by(User.id, oalias.id).all(), + options(joinedload(User.addresses), + joinedload(oalias.items)). + filter(User.id == 9). + order_by(User.id, oalias.id).all(), ) self.assert_sql_count(testing.db, go, 1) - def test_aliased_entity_three(self): - Item, Order, User, Address = (self.classes.Item, - self.classes.Order, - self.classes.User, - self.classes.Address) + Order, User = ( + self.classes.Order, + self.classes.User) sess = create_session() oalias = sa.orm.aliased(Order) - # improper setup: oalias in the columns clause but join to usual # orders alias. this should create two FROM clauses even though the # query has a from_clause set up via the join self.assert_compile( - sess.query(User, oalias).join(User.orders). - options(joinedload(oalias.items)).with_labels().statement, - "SELECT users.id AS users_id, users.name AS users_name, " - "orders_1.id AS orders_1_id, " - "orders_1.user_id AS orders_1_user_id, " - "orders_1.address_id AS orders_1_address_id, " - "orders_1.description AS orders_1_description, " - "orders_1.isopen AS orders_1_isopen, items_1.id AS items_1_id, " - "items_1.description AS items_1_description FROM users " - "JOIN orders ON users.id = orders.user_id, " - "orders AS orders_1 LEFT OUTER JOIN (order_items AS order_items_1 " - "JOIN items AS items_1 ON items_1.id = order_items_1.item_id) " - "ON orders_1.id = order_items_1.order_id ORDER BY items_1.id" + sess.query(User, oalias).join(User.orders). + options(joinedload(oalias.items)).with_labels().statement, + "SELECT users.id AS users_id, users.name AS users_name, " + "orders_1.id AS orders_1_id, " + "orders_1.user_id AS orders_1_user_id, " + "orders_1.address_id AS orders_1_address_id, " + "orders_1.description AS orders_1_description, " + "orders_1.isopen AS orders_1_isopen, items_1.id AS items_1_id, " + "items_1.description AS items_1_description FROM users " + "JOIN orders ON users.id = orders.user_id, " + "orders AS orders_1 LEFT OUTER JOIN (order_items AS order_items_1 " + "JOIN items AS items_1 ON items_1.id = order_items_1.item_id) " + "ON orders_1.id = order_items_1.order_id ORDER BY items_1.id" ) class SubqueryTest(fixtures.MappedTest): + @classmethod def define_tables(cls, metadata): Table('users_table', metadata, - Column('id', Integer, primary_key=True, test_needs_autoincrement=True), - Column('name', String(16)) - ) + Column( + 'id', + Integer, + primary_key=True, + test_needs_autoincrement=True), + Column('name', String(16)) + ) Table('tags_table', metadata, - Column('id', Integer, primary_key=True, test_needs_autoincrement=True), - Column('user_id', Integer, ForeignKey("users_table.id")), - Column('score1', sa.Float), - Column('score2', sa.Float), - ) + Column( + 'id', + Integer, + primary_key=True, + test_needs_autoincrement=True), + Column('user_id', Integer, ForeignKey("users_table.id")), + Column('score1', sa.Float), + Column('score2', sa.Float), + ) def test_label_anonymizing(self): """Eager loading works with subqueries with labels, @@ -2723,19 +3139,23 @@ class SubqueryTest(fixtures.MappedTest): """ - tags_table, users_table = self.tables.tags_table, self.tables.users_table + tags_table, users_table = self.tables.tags_table, \ + self.tables.users_table class User(fixtures.ComparableEntity): + @property def prop_score(self): return sum([tag.prop_score for tag in self.tags]) class Tag(fixtures.ComparableEntity): + @property def prop_score(self): return self.score1 * self.score2 - for labeled, labelname in [(True, 'score'), (True, None), (False, None)]: + for labeled, labelname in [(True, 'score'), (True, None), + (False, None)]: sa.orm.clear_mappers() tag_score = (tags_table.c.score1 * tags_table.c.score2) @@ -2753,7 +3173,6 @@ class SubqueryTest(fixtures.MappedTest): 'query_score': sa.orm.column_property(tag_score), }) - mapper(User, users_table, properties={ 'tags': relationship(Tag, backref='user', lazy='joined'), 'query_score': sa.orm.column_property(user_score), @@ -2761,10 +3180,10 @@ class SubqueryTest(fixtures.MappedTest): session = create_session() session.add(User(name='joe', tags=[Tag(score1=5.0, score2=3.0), - Tag(score1=55.0, score2=1.0)])) + Tag(score1=55.0, score2=1.0)])) session.add(User(name='bar', tags=[Tag(score1=5.0, score2=4.0), - Tag(score1=50.0, score2=1.0), - Tag(score1=15.0, score2=2.0)])) + Tag(score1=50.0, score2=1.0), + Tag(score1=15.0, score2=2.0)])) session.flush() session.expunge_all() @@ -2779,7 +3198,9 @@ class SubqueryTest(fixtures.MappedTest): for t in (tags_table, users_table): t.delete().execute() + class CorrelatedSubqueryTest(fixtures.MappedTest): + """tests for #946, #947, #948. The "users" table is joined to "stuff", and the relationship @@ -2795,13 +3216,23 @@ class CorrelatedSubqueryTest(fixtures.MappedTest): @classmethod def define_tables(cls, metadata): - users = Table('users', metadata, - Column('id', Integer, primary_key=True, test_needs_autoincrement=True), + Table( + 'users', metadata, + Column( + 'id', + Integer, + primary_key=True, + test_needs_autoincrement=True), Column('name', String(50)) - ) + ) - stuff = Table('stuff', metadata, - Column('id', Integer, primary_key=True, test_needs_autoincrement=True), + Table( + 'stuff', metadata, + Column( + 'id', + Integer, + primary_key=True, + test_needs_autoincrement=True), Column('date', Date), Column('user_id', Integer, ForeignKey('users.id'))) @@ -2810,21 +3241,20 @@ class CorrelatedSubqueryTest(fixtures.MappedTest): stuff, users = cls.tables.stuff, cls.tables.users users.insert().execute( - {'id':1, 'name':'user1'}, - {'id':2, 'name':'user2'}, - {'id':3, 'name':'user3'}, + {'id': 1, 'name': 'user1'}, + {'id': 2, 'name': 'user2'}, + {'id': 3, 'name': 'user3'}, ) stuff.insert().execute( - {'id':1, 'user_id':1, 'date':datetime.date(2007, 10, 15)}, - {'id':2, 'user_id':1, 'date':datetime.date(2007, 12, 15)}, - {'id':3, 'user_id':1, 'date':datetime.date(2007, 11, 15)}, - {'id':4, 'user_id':2, 'date':datetime.date(2008, 1, 15)}, - {'id':5, 'user_id':3, 'date':datetime.date(2007, 6, 15)}, - {'id':6, 'user_id':3, 'date':datetime.date(2007, 3, 15)}, + {'id': 1, 'user_id': 1, 'date': datetime.date(2007, 10, 15)}, + {'id': 2, 'user_id': 1, 'date': datetime.date(2007, 12, 15)}, + {'id': 3, 'user_id': 1, 'date': datetime.date(2007, 11, 15)}, + {'id': 4, 'user_id': 2, 'date': datetime.date(2008, 1, 15)}, + {'id': 5, 'user_id': 3, 'date': datetime.date(2007, 6, 15)}, + {'id': 6, 'user_id': 3, 'date': datetime.date(2007, 3, 15)}, ) - def test_labeled_on_date_noalias(self): self._do_test('label', True, False) @@ -2875,20 +3305,24 @@ class CorrelatedSubqueryTest(fixtures.MappedTest): if aliasstuff: salias = stuff.alias() else: - # if we don't alias the 'stuff' table within the correlated subquery, - # it gets aliased in the eager load along with the "stuff" table to "stuff_1". + # if we don't alias the 'stuff' table within the correlated + # subquery, + # it gets aliased in the eager load along with the "stuff" table + # to "stuff_1". # but it's a scalar subquery, and this doesn't actually matter salias = stuff if ondate: # the more 'relational' way to do this, join on the max date stuff_view = select([func.max(salias.c.date).label('max_date')]).\ - where(salias.c.user_id==users.c.id).correlate(users) + where(salias.c.user_id == users.c.id).correlate(users) else: - # a common method with the MySQL crowd, which actually might perform better in some + # a common method with the MySQL crowd, which actually might + # perform better in some # cases - subquery does a limit with order by DESC, join on the id - stuff_view = select([salias.c.id]).where(salias.c.user_id==users.c.id).\ - correlate(users).order_by(salias.c.date.desc()).limit(1) + stuff_view = select([salias.c.id]).\ + where(salias.c.user_id == users.c.id).\ + correlate(users).order_by(salias.c.date.desc()).limit(1) # can't win on this one if testing.against("mssql"): @@ -2904,17 +3338,25 @@ class CorrelatedSubqueryTest(fixtures.MappedTest): if ondate: mapper(User, users, properties={ - 'stuff':relationship(Stuff, primaryjoin=and_(users.c.id==stuff.c.user_id, operator(stuff.c.date, stuff_view))) + 'stuff': relationship( + Stuff, + primaryjoin=and_(users.c.id == stuff.c.user_id, + operator(stuff.c.date, stuff_view))) }) else: mapper(User, users, properties={ - 'stuff':relationship(Stuff, primaryjoin=and_(users.c.id==stuff.c.user_id, operator(stuff.c.id, stuff_view))) + 'stuff': relationship( + Stuff, + primaryjoin=and_(users.c.id == stuff.c.user_id, + operator(stuff.c.id, stuff_view))) }) sess = create_session() + def go(): eq_( - sess.query(User).order_by(User.name).options(joinedload('stuff')).all(), + sess.query(User).order_by(User.name).options( + joinedload('stuff')).all(), [ User(name='user1', stuff=[Stuff(id=2)]), User(name='user2', stuff=[Stuff(id=4)]), @@ -2924,6 +3366,7 @@ class CorrelatedSubqueryTest(fixtures.MappedTest): self.assert_sql_count(testing.db, go, 1) sess = create_session() + def go(): eq_( sess.query(User).order_by(User.name).first(), @@ -2932,36 +3375,45 @@ class CorrelatedSubqueryTest(fixtures.MappedTest): self.assert_sql_count(testing.db, go, 2) sess = create_session() + def go(): eq_( - sess.query(User).order_by(User.name).options(joinedload('stuff')).first(), + sess.query(User).order_by(User.name).options( + joinedload('stuff')).first(), User(name='user1', stuff=[Stuff(id=2)]) ) self.assert_sql_count(testing.db, go, 1) sess = create_session() + def go(): eq_( - sess.query(User).filter(User.id==2).options(joinedload('stuff')).one(), + sess.query(User).filter(User.id == 2).options( + joinedload('stuff')).one(), User(name='user2', stuff=[Stuff(id=4)]) ) self.assert_sql_count(testing.db, go, 1) + class CyclicalInheritingEagerTestOne(fixtures.MappedTest): @classmethod def define_tables(cls, metadata): - Table('t1', metadata, - Column('c1', Integer, primary_key=True, test_needs_autoincrement=True), + Table( + 't1', metadata, + Column( + 'c1', Integer, primary_key=True, + test_needs_autoincrement=True), Column('c2', String(30)), Column('type', String(30)) - ) + ) Table('t2', metadata, - Column('c1', Integer, primary_key=True, test_needs_autoincrement=True), - Column('c2', String(30)), - Column('type', String(30)), - Column('t1.id', Integer, ForeignKey('t1.c1'))) + Column('c1', Integer, primary_key=True, + test_needs_autoincrement=True), + Column('c2', String(30)), + Column('type', String(30)), + Column('t1.id', Integer, ForeignKey('t1.c1'))) def test_basic(self): t2, t1 = self.tables.t2, self.tables.t1 @@ -2979,27 +3431,32 @@ class CyclicalInheritingEagerTestOne(fixtures.MappedTest): pass mapper(T, t1, polymorphic_on=t1.c.type, polymorphic_identity='t1') - mapper(SubT, None, inherits=T, polymorphic_identity='subt1', properties={ - 't2s': relationship(SubT2, lazy='joined', - backref=sa.orm.backref('subt', lazy='joined')) - }) + mapper( + SubT, None, inherits=T, polymorphic_identity='subt1', + properties={ + 't2s': relationship( + SubT2, lazy='joined', + backref=sa.orm.backref('subt', lazy='joined')) + }) 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 load setup create_session().query(SubT).all() + class CyclicalInheritingEagerTestTwo(fixtures.DeclarativeMappedTest, - testing.AssertsCompiledSQL): + testing.AssertsCompiledSQL): __dialect__ = 'default' @classmethod def setup_classes(cls): Base = cls.DeclarativeBasic + class PersistentObject(Base): __tablename__ = 'persistent' id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + test_needs_autoincrement=True) class Movie(PersistentObject): __tablename__ = 'movie' @@ -3013,14 +3470,14 @@ class CyclicalInheritingEagerTestTwo(fixtures.DeclarativeMappedTest, movies = relationship("Movie", foreign_keys=Movie.director_id) name = Column(String(50)) - def test_from_subclass(self): Director = self.classes.Director s = create_session() self.assert_compile( s.query(Director).options(joinedload('*')), - "SELECT director.id AS director_id, persistent.id AS persistent_id, " + "SELECT director.id AS director_id, " + "persistent.id AS persistent_id, " "director.name AS director_name, movie_1.id AS movie_1_id, " "persistent_1.id AS persistent_1_id, " "movie_1.director_id AS movie_1_director_id, " @@ -3028,7 +3485,7 @@ class CyclicalInheritingEagerTestTwo(fixtures.DeclarativeMappedTest, "FROM persistent JOIN director ON persistent.id = director.id " "LEFT OUTER JOIN " "(persistent AS persistent_1 JOIN movie AS movie_1 " - "ON persistent_1.id = movie_1.id) " + "ON persistent_1.id = movie_1.id) " "ON director.id = movie_1.director_id" ) @@ -3046,20 +3503,22 @@ class CyclicalInheritingEagerTestTwo(fixtures.DeclarativeMappedTest, session.commit() session.close_all() - d = session.query(Director).options(joinedload('*')).first() + self.d = session.query(Director).options(joinedload('*')).first() assert len(list(session)) == 3 + class CyclicalInheritingEagerTestThree(fixtures.DeclarativeMappedTest, - testing.AssertsCompiledSQL): + testing.AssertsCompiledSQL): __dialect__ = 'default' @classmethod def setup_classes(cls): Base = cls.DeclarativeBasic + class PersistentObject(Base): __tablename__ = 'persistent' id = Column(Integer, primary_key=True, - test_needs_autoincrement=True) + test_needs_autoincrement=True) __mapper_args__ = {'with_polymorphic': "*"} @@ -3069,16 +3528,17 @@ class CyclicalInheritingEagerTestThree(fixtures.DeclarativeMappedTest, other_id = Column(Integer, ForeignKey('persistent.id')) name = Column(String(50)) other = relationship(PersistentObject, - primaryjoin=other_id==PersistentObject.id, - lazy=False) - __mapper_args__ = {"inherit_condition": id==PersistentObject.id} + primaryjoin=other_id == PersistentObject.id, + lazy=False) + __mapper_args__ = {"inherit_condition": id == PersistentObject.id} def test_gen_query_nodepth(self): PersistentObject = self.classes.PersistentObject sess = create_session() self.assert_compile( sess.query(PersistentObject), - "SELECT persistent.id AS persistent_id, director.id AS director_id," + "SELECT persistent.id AS persistent_id, " + "director.id AS director_id," " director.other_id AS director_other_id, " "director.name AS director_name FROM persistent " "LEFT OUTER JOIN director ON director.id = persistent.id" @@ -3090,14 +3550,17 @@ class CyclicalInheritingEagerTestThree(fixtures.DeclarativeMappedTest, sess = create_session() self.assert_compile( sess.query(PersistentObject).options(joinedload(Director.other)), - "SELECT persistent.id AS persistent_id, director.id AS director_id, " + "SELECT persistent.id AS persistent_id, " + "director.id AS director_id, " "director.other_id AS director_other_id, " "director.name AS director_name, persistent_1.id AS " "persistent_1_id, director_1.id AS director_1_id, " "director_1.other_id AS director_1_other_id, " "director_1.name AS director_1_name " - "FROM persistent LEFT OUTER JOIN director ON director.id = persistent.id " - "LEFT OUTER JOIN (persistent AS persistent_1 LEFT OUTER JOIN director AS director_1 ON " - "director_1.id = persistent_1.id) " + "FROM persistent LEFT OUTER JOIN director " + "ON director.id = persistent.id " + "LEFT OUTER JOIN (persistent AS persistent_1 " + "LEFT OUTER JOIN director AS director_1 ON " + "director_1.id = persistent_1.id) " "ON director.other_id = persistent_1.id" ) |