from sqlalchemy.testing import eq_, assert_raises, assert_raises_message
from sqlalchemy.testing import fixtures
-from sqlalchemy import Integer, String, ForeignKey, or_, and_, exc, \
+from sqlalchemy import Integer, String, ForeignKey, or_, exc, \
select, func, Boolean, case, text
from sqlalchemy.orm import mapper, relationship, backref, Session, \
- joinedload, aliased
+ joinedload
from sqlalchemy import testing
from sqlalchemy.testing.schema import Table, Column
):
assert_raises_message(
exc.InvalidRequestError,
- r"Can't call Query.update\(\) when %s\(\) has been called" % mname,
+ r"Can't call Query.update\(\) when "
+ "%s\(\) has been called" % mname,
q.update,
{'name': 'ed'})
assert_raises_message(
exc.InvalidRequestError,
- r"Can't call Query.delete\(\) when %s\(\) has been called" % mname,
+ r"Can't call Query.delete\(\) when "
+ "%s\(\) has been called" % mname,
q.delete)
def test_delete(self):
sess = Session()
john, jack, jill, jane = sess.query(User).order_by(User.id).all()
- sess.query(User).filter(or_(User.name == 'john', User.name == 'jill')).\
+ sess.query(User).filter(
+ or_(User.name == 'john', User.name == 'jill')).\
delete(synchronize_session='evaluate')
assert john not in sess and jill not in sess
sess.rollback()
sess = Session()
john, jack, jill, jane = sess.query(User).order_by(User.id).all()
- sess.query(User).filter(or_(User.name == 'john', User.name == 'jill')).\
+ sess.query(User).filter(
+ or_(User.name == 'john', User.name == 'jill')).\
delete(synchronize_session='fetch')
assert john not in sess and jill not in sess
sess.rollback()
sess = Session()
john, jack, jill, jane = sess.query(User).order_by(User.id).all()
- sess.query(User).filter(or_(User.name == 'john', User.name == 'jill')).\
+ sess.query(User).filter(
+ or_(User.name == 'john', User.name == 'jill')).\
delete(synchronize_session=False)
assert john in sess and jill in sess
sess = Session()
john, jack, jill, jane = sess.query(User).order_by(User.id).all()
- sess.query(User).filter(or_(User.name == 'john', User.name == 'jill')).\
+ sess.query(User).filter(
+ or_(User.name == 'john', User.name == 'jill')).\
delete(synchronize_session='fetch')
assert john not in sess and jill not in sess
sess.query(User).filter_by(name='j2').\
delete(
- synchronize_session='evaluate')
+ synchronize_session='evaluate')
assert john not in sess
def test_autoflush_before_fetch_delete(self):
sess.query(User).filter_by(name='j2').\
delete(
- synchronize_session='fetch')
+ synchronize_session='fetch')
assert john not in sess
def test_evaluate_before_update(self):
sess.query(User).filter_by(name='john').\
filter_by(age=25).\
delete(
- synchronize_session='evaluate')
+ synchronize_session='evaluate')
assert john not in sess
def test_fetch_before_delete(self):
sess.query(User).filter_by(name='john').\
filter_by(age=25).\
delete(
- synchronize_session='fetch')
+ synchronize_session='fetch')
assert john not in sess
sess = Session()
john, jack, jill, jane = sess.query(User).order_by(User.id).all()
- sess.query(User).options(joinedload(User.documents)).filter(User.age > 29).\
+ sess.query(User).options(
+ joinedload(User.documents)).filter(User.age > 29).\
update({'age': User.age - 10}, synchronize_session='fetch')
eq_([john.age, jack.age, jill.age, jane.age], [25, 37, 29, 27])
set([
(1, True), (2, None),
(3, None), (4, True),
- (5, True), (6, None),
- ])
+ (5, True), (6, None)])
)
def test_no_eval_against_multi_table_criteria(self):
set([
(1, True), (2, None),
(3, None), (4, True),
- (5, True), (6, None),
- ])
+ (5, True), (6, None)])
)
@testing.requires.update_where_target_in_subquery
set([
(1, True), (2, False),
(3, False), (4, True),
- (5, True), (6, False),
- ])
+ (5, True), (6, False)])
)
@testing.only_on('mysql', 'Multi table update')
filter(User.id == 2).update({
Document.samename: 'd_samename',
User.samename: 'u_samename'
- }, synchronize_session=False
- )
+ }, synchronize_session=False)
eq_(
s.query(User.id, Document.samename, User.samename).
filter(User.id == Document.user_id).