class FromSelfTest(QueryTest, AssertsCompiledSQL):
def test_filter(self):
- assert [User(id=8), User(id=9)] == create_session().query(User).filter(User.id.in_([8,9])).from_self().all()
-
- assert [User(id=8), User(id=9)] == create_session().query(User).order_by(User.id).slice(1,3).from_self().all()
- assert [User(id=8)] == list(create_session().query(User).filter(User.id.in_([8,9])).from_self().order_by(User.id)[0:1])
+ assert [User(id=8), User(id=9)] == \
+ create_session().query(User).\
+ filter(User.id.in_([8,9])).from_self().all()
+
+ assert [User(id=8), User(id=9)] == \
+ create_session().query(User).\
+ order_by(User.id).slice(1,3).from_self().all()
+ assert [User(id=8)] == \
+ list(create_session().query(User).\
+ filter(User.id.in_([8,9])).from_self().\
+ order_by(User.id)[0:1])
def test_join(self):
assert [
(User(id=8), Address(id=3)),
(User(id=8), Address(id=4)),
(User(id=9), Address(id=5))
- ] == create_session().query(User).filter(User.id.in_([8,9])).from_self().\
- join('addresses').add_entity(Address).order_by(User.id, Address.id).all()
+ ] == create_session().query(User).\
+ filter(User.id.in_([8,9])).from_self().\
+ join('addresses').add_entity(Address).\
+ order_by(User.id, Address.id).all()
def test_group_by(self):
eq_(
- create_session().query(Address.user_id, func.count(Address.id).label('count')).\
- group_by(Address.user_id).order_by(Address.user_id).all(),
+ create_session().query(Address.user_id,
+ func.count(Address.id).label('count')).\
+ group_by(Address.user_id).
+ order_by(Address.user_id).all(),
[(7, 1), (8, 3), (9, 1)]
)
eq_(
- create_session().query(Address.user_id, Address.id).\
- from_self(Address.user_id, func.count(Address.id)).\
- group_by(Address.user_id).order_by(Address.user_id).all(),
+ create_session().query(Address.user_id,
+ Address.id).\
+ from_self(Address.user_id,
+ func.count(Address.id)).\
+ group_by(Address.user_id).
+ order_by(Address.user_id).all(),
[(7, 1), (8, 3), (9, 1)]
)
+ def test_having(self):
+ s = create_session()
+
+ self.assert_compile(
+ s.query(User.id).group_by(User.id).having(User.id>5).
+ from_self(),
+ "SELECT anon_1.users_id AS anon_1_users_id FROM "
+ "(SELECT users.id AS users_id FROM users GROUP "
+ "BY users.id HAVING users.id > :id_1) AS anon_1",
+ use_default_dialect=True
+ )
+
def test_no_joinedload(self):
- """test that joinedloads are pushed outwards and not rendered in subqueries."""
+ """test that joinedloads are pushed outwards and not
+ rendered in subqueries."""
s = create_session()
oracle_as = not testing.against('oracle') and "AS " or ""
self.assert_compile(
- s.query(User).options(joinedload(User.addresses)).from_self().statement,
- "SELECT anon_1.users_id, anon_1.users_name, addresses_1.id, addresses_1.user_id, "\
- "addresses_1.email_address FROM (SELECT users.id AS users_id, users.name AS users_name FROM users) %(oracle_as)sanon_1 "\
- "LEFT OUTER JOIN addresses %(oracle_as)saddresses_1 ON anon_1.users_id = addresses_1.user_id ORDER BY addresses_1.id" % {
+ s.query(User).
+ options(joinedload(User.addresses)).
+ from_self().statement,
+ "SELECT anon_1.users_id, anon_1.users_name, "
+ "addresses_1.id, addresses_1.user_id, "
+ "addresses_1.email_address FROM (SELECT users.id "
+ "AS users_id, users.name AS users_name FROM users) "
+ "%(oracle_as)sanon_1 "
+ "LEFT OUTER JOIN addresses %(oracle_as)saddresses_1 "
+ "ON anon_1.users_id = addresses_1.user_id ORDER "
+ "BY addresses_1.id" % {
'oracle_as':oracle_as
}
)
ualias = aliased(User)
eq_(
- s.query(User, ualias).filter(User.id > ualias.id).from_self(User.name, ualias.name).
+ s.query(User, ualias).filter(User.id > ualias.id).
+ from_self(User.name, ualias.name).
order_by(User.name, ualias.name).all(),
[
(u'chuck', u'ed'),
)
eq_(
- s.query(User, ualias).filter(User.id > ualias.id).from_self(User.name, ualias.name).filter(ualias.name=='ed')\
+ s.query(User, ualias).filter(User.id > ualias.id).
+ from_self(User.name, ualias.name).filter(ualias.name=='ed')\
.order_by(User.name, ualias.name).all(),
[(u'chuck', u'ed'), (u'fred', u'ed')]
)
eq_(
- s.query(User, ualias).filter(User.id > ualias.id).from_self(ualias.name, Address.email_address).
- join(ualias.addresses).order_by(ualias.name, Address.email_address).all(),
+ s.query(User, ualias).filter(User.id > ualias.id).
+ from_self(ualias.name, Address.email_address).
+ join(ualias.addresses).
+ order_by(ualias.name, Address.email_address).all(),
[
(u'ed', u'fred@fred.com'),
(u'jack', u'ed@bettyboop.com'),
sess = create_session()
eq_(
- sess.query(User, Address).filter(User.id==Address.user_id).filter(Address.id.in_([2, 5])).from_self().all(),
+ sess.query(User, Address).
+ filter(User.id==Address.user_id).
+ filter(Address.id.in_([2, 5])).from_self().all(),
[
(User(id=8), Address(id=2)),
(User(id=9), Address(id=5))
)
eq_(
- sess.query(User, Address).filter(User.id==Address.user_id).filter(Address.id.in_([2, 5])).from_self().options(joinedload('addresses')).first(),
+ sess.query(User, Address).
+ filter(User.id==Address.user_id).
+ filter(Address.id.in_([2, 5])).
+ from_self().options(joinedload('addresses')).first(),
# order_by(User.id, Address.id).first(),
(User(id=8, addresses=[Address(), Address(), Address()]), Address(id=2)),