@testing.resolve_artifact_names
def test_refresh_collection_exception(self):
- """test graceful failure for currently unsupported immediate refresh of a collection"""
+ """test graceful failure for currently unsupported
+ immediate refresh of a collection"""
mapper(User, users, properties={
'addresses':relationship(Address, order_by=addresses.c.email_address)
mapper(Address, addresses)
s = create_session(autoflush=True, autocommit=False)
u = s.query(User).get(8)
- assert_raises_message(sa_exc.InvalidRequestError, "properties specified for refresh", s.refresh, u, ['addresses'])
+ assert_raises_message(sa_exc.InvalidRequestError,
+ "properties specified for refresh",
+ s.refresh, u, ['addresses'])
# in contrast to a regular query with no columns
- assert_raises_message(sa_exc.InvalidRequestError, "no columns with which to SELECT", s.query().all)
+ assert_raises_message(sa_exc.InvalidRequestError,
+ "no columns with which to SELECT", s.query().all)
@testing.resolve_artifact_names
def test_refresh_cancels_expire(self):
@testing.resolve_artifact_names
def test_expire_all(self):
mapper(User, users, properties={
- 'addresses':relationship(Address, backref='user', lazy='joined'),
+ 'addresses':relationship(Address, backref='user', lazy='joined',
+ order_by=addresses.c.id),
})
mapper(Address, addresses)
def test_from_alias(self):
- query = users.select(users.c.id==7).union(users.select(users.c.id>7)).alias('ulist').outerjoin(addresses).select(use_labels=True,order_by=['ulist.id', addresses.c.id])
+ query = users.select(users.c.id==7).\
+ union(users.select(users.c.id>7)).\
+ alias('ulist').\
+ outerjoin(addresses).\
+ select(use_labels=True,
+ order_by=['ulist.id', addresses.c.id])
sess =create_session()
q = sess.query(User)
def go():
- l = list(q.options(contains_alias('ulist'), contains_eager('addresses')).instances(query.execute()))
+ l = list(q.options(contains_alias('ulist'),
+ contains_eager('addresses')).\
+ instances(query.execute()))
assert self.static.user_address_result == l
self.assert_sql_count(testing.db, go, 1)
sess.expunge_all()
def go():
- l = q.options(contains_alias('ulist'), contains_eager('addresses')).from_statement(query).all()
+ l = q.options(contains_alias('ulist'),
+ contains_eager('addresses')).\
+ from_statement(query).all()
assert self.static.user_address_result == l
self.assert_sql_count(testing.db, go, 1)
# better way. use select_from()
def go():
- l = sess.query(User).select_from(query).options(contains_eager('addresses')).all()
+ l = sess.query(User).select_from(query).\
+ options(contains_eager('addresses')).all()
assert self.static.user_address_result == l
self.assert_sql_count(testing.db, go, 1)
- # same thing, but alias addresses, so that the adapter generated by select_from() is wrapped within
+ # same thing, but alias addresses, so that the adapter
+ # generated by select_from() is wrapped within
# the adapter created by contains_eager()
adalias = addresses.alias()
- query = users.select(users.c.id==7).union(users.select(users.c.id>7)).alias('ulist').outerjoin(adalias).select(use_labels=True,order_by=['ulist.id', adalias.c.id])
+ query = users.select(users.c.id==7).\
+ union(users.select(users.c.id>7)).\
+ alias('ulist').\
+ outerjoin(adalias).\
+ select(use_labels=True,
+ order_by=['ulist.id', adalias.c.id])
def go():
- l = sess.query(User).select_from(query).options(contains_eager('addresses', alias=adalias)).all()
+ l = sess.query(User).select_from(query).\
+ options(contains_eager('addresses', alias=adalias)).all()
assert self.static.user_address_result == l
self.assert_sql_count(testing.db, go, 1)
sess = create_session()
# test that contains_eager suppresses the normal outer join rendering
- q = sess.query(User).outerjoin(User.addresses).options(contains_eager(User.addresses)).order_by(User.id)
- self.assert_compile(q.with_labels().statement,
- "SELECT addresses.id AS addresses_id, addresses.user_id AS addresses_user_id, "\
- "addresses.email_address AS addresses_email_address, users.id AS users_id, "\
- "users.name AS users_name FROM users LEFT OUTER JOIN addresses "\
- "ON users.id = addresses.user_id ORDER BY users.id"
- , dialect=default.DefaultDialect())
+ q = sess.query(User).outerjoin(User.addresses).\
+ options(contains_eager(User.addresses)).\
+ order_by(User.id, addresses.c.id)
+ self.assert_compile(q.with_labels().statement,
+ 'SELECT addresses.id AS addresses_id, '
+ 'addresses.user_id AS addresses_user_id, '
+ 'addresses.email_address AS '
+ 'addresses_email_address, users.id AS '
+ 'users_id, users.name AS users_name FROM '
+ 'users LEFT OUTER JOIN addresses ON '
+ 'users.id = addresses.user_id ORDER BY '
+ 'users.id, addresses.id',
+ dialect=default.DefaultDialect())
def go():
assert self.static.user_address_result == q.all()
sess.expunge_all()
adalias = addresses.alias()
- q = sess.query(User).select_from(users.outerjoin(adalias)).options(contains_eager(User.addresses, alias=adalias))
+ q = sess.query(User).select_from(users.outerjoin(adalias)).options(contains_eager(User.addresses, alias=adalias)).order_by(User.id, adalias.c.id)
def go():
eq_(self.static.user_address_result, q.order_by(User.id).all())
self.assert_sql_count(testing.db, go, 1)