- Added Float and Time types to Firebird (FBFloat and FBTime). Fixed
BLOB SUB_TYPE for TEXT and Binary types.
+- Changed the API for the in_ operator. in_() now accepts a single argument
+ that is a sequence of values or a selectable. The old API of passing in
+ values as varargs still works but is deprecated.
+
0.4.0beta6
----------
{python}
{sql}>>> session.query(Address).filter(
- ... Address.email_address.in_('jack@google.com', 'j25@yahoo.com')
+ ... Address.email_address.in_(['jack@google.com', 'j25@yahoo.com'])
... ).count()
SELECT count(addresses.id)
FROM addresses
# only one address remains
{sql}>>> session.query(Address).filter(
- ... Address.email_address.in_('jack@google.com', 'j25@yahoo.com')
+ ... Address.email_address.in_(['jack@google.com', 'j25@yahoo.com'])
... ).count()
DELETE FROM addresses WHERE addresses.id = ?
[2]
{stop}0
{sql}>>> session.query(Address).filter(
- ... Address.email_address.in_('jack@google.com', 'j25@yahoo.com')
+ ... Address.email_address.in_(['jack@google.com', 'j25@yahoo.com'])
... ).count()
SELECT count(addresses.id)
FROM addresses
>>> metadata.bind.echo = False
>>> session = create_session(autoflush=True)
>>> (ed, harry, mary) = session.query(User).filter(
- ... User.c.user_name.in_('Ed', 'Harry', 'Mary')
+ ... User.c.user_name.in_(['Ed', 'Harry', 'Mary'])
... ).order_by(User.c.user_name).all() # doctest: +NORMALIZE_WHITESPACE
>>> del mary.addresses[1]
>>> harry_address = Address('harry2@gmail.com')
return self._in_impl(operators.in_op, operators.notin_op, *other)
def _in_impl(self, op, negate_op, *other):
- if len(other) == 0:
- return _Grouping(case([(self.__eq__(None), text('NULL'))], else_=text('0')).__eq__(text('1')))
- elif len(other) == 1:
- o = other[0]
- if _is_literal(o) or isinstance( o, _CompareMixin):
- return self.__eq__( o) #single item -> ==
- else:
- assert isinstance(o, Selectable)
- return self.__compare( op, o, negate=negate_op) #single selectable
-
+ # Handle old style *args argument passing
+ if len(other) != 1 or not isinstance(other[0], Selectable) and (not hasattr(other[0], '__iter__') or isinstance(other[0], basestring)):
+ util.warn_deprecated('passing in_ arguments as varargs is deprecated, in_ takes a single argument that is a sequence or a selectable')
+ seq_or_selectable = other
+ else:
+ seq_or_selectable = other[0]
+
+ if isinstance(seq_or_selectable, Selectable):
+ return self.__compare( op, seq_or_selectable, negate=negate_op)
+
+ # Handle non selectable arguments as sequences
args = []
- for o in other:
+ for o in seq_or_selectable:
if not _is_literal(o):
if not isinstance( o, _CompareMixin):
- raise exceptions.InvalidRequestError( "in() function accepts either non-selectable values, or a single selectable: "+repr(o) )
+ raise exceptions.InvalidRequestError( "in() function accepts either a list of non-selectable values, or a selectable: "+repr(o) )
else:
o = self._bind_param(o)
args.append(o)
+
+ if len(args) == 0:
+ # Special case handling for empty IN's
+ return _Grouping(case([(self.__eq__(None), text('NULL'))], else_=text('0')).__eq__(text('1')))
+
return self.__compare(op, ClauseList(*args).self_group(against=op), negate=negate_op)
def startswith(self, other):
column('col4'))
(s1, s2) = (
- select([t1.c.col3.label('col3'), t1.c.col4.label('col4')], t1.c.col2.in_("t1col2r1", "t1col2r2")),
- select([t2.c.col3.label('col3'), t2.c.col4.label('col4')], t2.c.col2.in_("t2col2r2", "t2col2r3"))
+ select([t1.c.col3.label('col3'), t1.c.col4.label('col4')], t1.c.col2.in_(["t1col2r1", "t1col2r2"])),
+ select([t2.c.col3.label('col3'), t2.c.col4.label('col4')], t2.c.col2.in_(["t2col2r2", "t2col2r3"]))
)
u = union(s1, s2, order_by=['col3', 'col4'])
self.assert_compile(u, "SELECT t1.col3 AS col3, t1.col4 AS col4 FROM t1 WHERE t1.col2 IN (:t1_col2, :t1_col2_1) UNION SELECT t2.col3 AS col3, t2.col4 AS col4 FROM t2 WHERE t2.col2 IN (:t2_col2, :t2_col2_1) ORDER BY col3, col4")
def overlap(self, ids, errors, update_style):
sel = counters.select(for_update=update_style,
- whereclause=counters.c.counter_id.in_(*ids))
+ whereclause=counters.c.counter_id.in_(ids))
con = testbase.db.connect()
trans = con.begin()
try:
sess.delete(l[2])
sess.flush()
- assert tables.orders.count(tables.orders.c.user_id.in_(*ids)).scalar() == 0
- assert tables.orderitems.count(tables.orders.c.user_id.in_(*ids) &(tables.orderitems.c.order_id==tables.orders.c.order_id)).scalar() == 0
- assert tables.addresses.count(tables.addresses.c.user_id.in_(*ids)).scalar() == 0
- assert tables.users.count(tables.users.c.user_id.in_(*ids)).scalar() == 0
+ assert tables.orders.count(tables.orders.c.user_id.in_(ids)).scalar() == 0
+ assert tables.orderitems.count(tables.orders.c.user_id.in_(ids) &(tables.orderitems.c.order_id==tables.orders.c.order_id)).scalar() == 0
+ assert tables.addresses.count(tables.addresses.c.user_id.in_(ids)).scalar() == 0
+ assert tables.users.count(tables.users.c.user_id.in_(ids)).scalar() == 0
def testdelete2(self):
"""test that unloaded collections are still included in a delete-cascade by default."""
jack.orders[:] = []
ids = [jack.user_id]
- self.assert_(tables.orders.count(tables.orders.c.user_id.in_(*ids)).scalar() == 1)
- self.assert_(tables.orderitems.count(tables.orders.c.user_id.in_(*ids) &(tables.orderitems.c.order_id==tables.orders.c.order_id)).scalar() == 2)
+ self.assert_(tables.orders.count(tables.orders.c.user_id.in_(ids)).scalar() == 1)
+ self.assert_(tables.orderitems.count(tables.orders.c.user_id.in_(ids) &(tables.orderitems.c.order_id==tables.orders.c.order_id)).scalar() == 2)
sess.flush()
- self.assert_(tables.orders.count(tables.orders.c.user_id.in_(*ids)).scalar() == 0)
- self.assert_(tables.orderitems.count(tables.orders.c.user_id.in_(*ids) &(tables.orderitems.c.order_id==tables.orders.c.order_id)).scalar() == 0)
+ self.assert_(tables.orders.count(tables.orders.c.user_id.in_(ids)).scalar() == 0)
+ self.assert_(tables.orderitems.count(tables.orders.c.user_id.in_(ids) &(tables.orderitems.c.order_id==tables.orders.c.order_id)).scalar() == 0)
class M2OCascadeTest(AssertMixin):
except TypeError, e:
assert "doesn't support item deletion" in str(e), str(e)
- for a in u.addresses.filter(Address.email_address.in_('c', 'e', 'f')):
+ for a in u.addresses.filter(Address.email_address.in_(['c', 'e', 'f'])):
u.addresses.remove(a)
assert [Address(email_address='a'), Address(email_address='b'), Address(email_address='d')] == list(u.addresses)
Order(id=4, user=User(id=9))
] == 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]))
assert [
Order(id=3, user=User(id=7)),
] == q.all()
# {
# 'table3' : table1.join(table3),
# 'table2' : table1.join(table2),
- # 'table1' : table1.select(table1.c.type.in_('table1', 'table1b')),
+ # 'table1' : table1.select(table1.c.type.in_(['table1', 'table1b'])),
# }, None, 'pjoin')
join = table1.outerjoin(table2).outerjoin(table3).alias('pjoin')
for x in range(0, 2):
r = session.query(Person).filter(people.c.name.like('%2')).join('status').filter_by(name="active")
assert str(list(r)) == "[Manager M2, category YYYYYYYYY, status Status active, Engineer E2, field X, status Status active]"
- r = session.query(Engineer).join('status').filter(people.c.name.in_('E2', 'E3', 'E4', 'M4', 'M2', 'M1') & (status.c.name=="active"))
+ r = session.query(Engineer).join('status').filter(people.c.name.in_(['E2', 'E3', 'E4', 'M4', 'M2', 'M1']) & (status.c.name=="active"))
assert str(list(r)) == "[Engineer E2, field X, status Status active, Engineer E3, field X, status Status active]"
# this test embeds the original polymorphic union (employee_join) fully
# into the WHERE criterion, using a correlated select. ticket #577 tracks
mapper(User, users)
q = create_session().query(User)
self.assert_(q.count()==3)
- self.assert_(q.count(users.c.user_id.in_(8,9))==2)
+ self.assert_(q.count(users.c.user_id.in_([8,9]))==2)
self.assert_(q.count_by(user_name='fred')==1)
def testmanytomany_count(self):
fwd_sql + "'\n or\n'" + rev_sql + "'")
def test_in(self):
- self._test(User.id.in_('a', 'b'),
+ self._test(User.id.in_(['a', 'b']),
"users.id IN (:users_id, :users_id_1)")
def test_between(self):
class AggregateTest(QueryTest):
def test_sum(self):
sess = create_session()
- orders = sess.query(Order).filter(Order.id.in_(2, 3, 4))
+ orders = sess.query(Order).filter(Order.id.in_([2, 3, 4]))
assert orders.sum(Order.user_id * Order.address_id) == 79
def test_apply(self):
sess = create_session()
- assert sess.query(Order).apply_sum(Order.user_id * Order.address_id).filter(Order.id.in_(2, 3, 4)).one() == 79
+ assert sess.query(Order).apply_sum(Order.user_id * Order.address_id).filter(Order.id.in_([2, 3, 4])).one() == 79
class CountTest(QueryTest):
def test_overlap_with_aliases(self):
oalias = orders.alias('oalias')
- result = create_session().query(User).select_from(users.join(oalias)).filter(oalias.c.description.in_("order 1", "order 2", "order 3")).join(['orders', 'items']).all()
+ result = create_session().query(User).select_from(users.join(oalias)).filter(oalias.c.description.in_(["order 1", "order 2", "order 3"])).join(['orders', 'items']).all()
assert [User(id=7, name='jack'), User(id=9, name='fred')] == result
- result = create_session().query(User).select_from(users.join(oalias)).filter(oalias.c.description.in_("order 1", "order 2", "order 3")).join(['orders', 'items']).filter_by(id=4).all()
+ result = create_session().query(User).select_from(users.join(oalias)).filter(oalias.c.description.in_(["order 1", "order 2", "order 3"])).join(['orders', 'items']).filter_by(id=4).all()
assert [User(id=7, name='jack')] == result
def test_aliased(self):
north_american_cities = sess.query(WeatherLocation).filter(WeatherLocation.continent == 'North America')
assert set([c.city for c in north_american_cities]) == set(['New York', 'Toronto'])
- asia_and_europe = sess.query(WeatherLocation).filter(WeatherLocation.continent.in_('Europe', 'Asia'))
+ asia_and_europe = sess.query(WeatherLocation).filter(WeatherLocation.continent.in_(['Europe', 'Asia']))
assert set([c.city for c in asia_and_europe]) == set(['Tokyo', 'London', 'Dublin'])
print repr(u.addresses)
Session.commit()
- usertable = users.select(users.c.user_id.in_(u.user_id)).execute().fetchall()
+ usertable = users.select(users.c.user_id.in_([u.user_id])).execute().fetchall()
self.assertEqual(usertable[0].values(), [u.user_id, 'one2manytester'])
- addresstable = addresses.select(addresses.c.address_id.in_(a.address_id, a2.address_id), order_by=[addresses.c.email_address]).execute().fetchall()
+ addresstable = addresses.select(addresses.c.address_id.in_([a.address_id, a2.address_id]), order_by=[addresses.c.email_address]).execute().fetchall()
self.assertEqual(addresstable[0].values(), [a2.address_id, u.user_id, 'lala@test.org'])
self.assertEqual(addresstable[1].values(), [a.address_id, u.user_id, 'one2many@test.org'])
# select both
#Session.close()
- userlist = User.query.filter(users.c.user_id.in_(u.user_id, u2.user_id)).order_by([users.c.user_name]).all()
+ userlist = User.query.filter(users.c.user_id.in_([u.user_id, u2.user_id])).order_by([users.c.user_name]).all()
print repr(u.user_id), repr(userlist[0].user_id), repr(userlist[0].user_name)
self.assert_(u.user_id == userlist[0].user_id and userlist[0].user_name == 'modifiedname')
self.assert_(u2.user_id == userlist[1].user_id and userlist[1].user_name == 'savetester2')
u = Session.get(User, id)
assert u.user_name == 'multitester'
- usertable = users.select(users.c.user_id.in_(u.foo_id)).execute().fetchall()
+ usertable = users.select(users.c.user_id.in_([u.foo_id])).execute().fetchall()
self.assertEqual(usertable[0].values(), [u.foo_id, 'multitester'])
- addresstable = addresses.select(addresses.c.address_id.in_(u.address_id)).execute().fetchall()
+ addresstable = addresses.select(addresses.c.address_id.in_([u.address_id])).execute().fetchall()
self.assertEqual(addresstable[0].values(), [u.address_id, u.foo_id, 'multi@test.org'])
u.email = 'lala@hey.com'
u.user_name = 'imnew'
Session.commit()
- usertable = users.select(users.c.user_id.in_(u.foo_id)).execute().fetchall()
+ usertable = users.select(users.c.user_id.in_([u.foo_id])).execute().fetchall()
self.assertEqual(usertable[0].values(), [u.foo_id, 'imnew'])
- addresstable = addresses.select(addresses.c.address_id.in_(u.address_id)).execute().fetchall()
+ addresstable = addresses.select(addresses.c.address_id.in_([u.address_id])).execute().fetchall()
self.assertEqual(addresstable[0].values(), [u.address_id, u.foo_id, 'lala@hey.com'])
Session.close()
item.item_name = elem['item_name']
item.keywords = []
if elem['keywords'][1]:
- klist = Session.query(keywordmapper).select(keywords.c.name.in_(*[e['name'] for e in elem['keywords'][1]]))
+ klist = Session.query(keywordmapper).select(keywords.c.name.in_([e['name'] for e in elem['keywords'][1]]))
else:
klist = []
khash = {}
Session.commit()
- l = Session.query(m).select(items.c.item_name.in_(*[e['item_name'] for e in data[1:]]), order_by=[items.c.item_name])
+ l = Session.query(m).select(items.c.item_name.in_([e['item_name'] for e in data[1:]]), order_by=[items.c.item_name])
self.assert_result(l, *data)
objects[4].item_name = 'item4updated'
Session.commit()
Session.close()
- l = Item.query.filter(items.c.item_name.in_(*[e['item_name'] for e in data[1:]])).order_by(items.c.item_name).all()
+ l = Item.query.filter(items.c.item_name.in_([e['item_name'] for e in data[1:]])).order_by(items.c.item_name).all()
self.assert_result(l, *data)
class SaveTest2(ORMTest):
order_by=[desc(func.count(_item_id)), _item_id],
limit=20)
ids = [r.id for r in top_20_q.execute().fetchall()]
- q2 = session.query(Item).filter(Item.id.in_(*ids))
+ q2 = session.query(Item).filter(Item.id.in_(ids))
for num, item in enumerate(q2):
report.append("number %s: %s" % (num + 1, item.name))
session.begin()
customer = session.query(Customer).get(customer_id)
- items = session.query(Item).filter(Item.id.in_(*item_ids))
+ items = session.query(Item).filter(Item.id.in_(item_ids))
purchase = Purchase()
purchase.customer = customer
# In operator (containedby)
assert len(fulltable(Animal.select(Animal.c.Species.like('%pede%')))) == 2
- assert len(fulltable(Animal.select(Animal.c.Species.in_('Lion', 'Tiger', 'Bear')))) == 3
+ assert len(fulltable(Animal.select(Animal.c.Species.in_(['Lion', 'Tiger', 'Bear'])))) == 3
# Try In with cell references
class thing(object): pass
pet, pet2 = thing(), thing()
pet.Name, pet2.Name = 'Slug', 'Ostrich'
- assert len(fulltable(Animal.select(Animal.c.Species.in_(pet.Name, pet2.Name)))) == 2
+ assert len(fulltable(Animal.select(Animal.c.Species.in_([pet.Name, pet2.Name])))) == 2
# logic and other functions
assert len(fulltable(Animal.select(Animal.c.Species.like('Slug')))) == 1
users.insert().execute(user_id = 8, user_name = 'fred')
users.insert().execute(user_id = 9, user_name = None)
- s = users.select(users.c.user_name.in_())
+ s = users.select(users.c.user_name.in_([]))
r = s.execute().fetchall()
# No username is in empty set
assert len(r) == 0
- s = users.select(not_(users.c.user_name.in_()))
+ s = users.select(not_(users.c.user_name.in_([])))
r = s.execute().fetchall()
# All usernames with a value are outside an empty set
assert len(r) == 2
- s = users.select(users.c.user_name.in_('jack','fred'))
+ s = users.select(users.c.user_name.in_(['jack','fred']))
r = s.execute().fetchall()
assert len(r) == 2
- s = users.select(not_(users.c.user_name.in_('jack','fred')))
+ s = users.select(not_(users.c.user_name.in_(['jack','fred'])))
r = s.execute().fetchall()
# Null values are not outside any set
assert len(r) == 0
u = bindparam('search_key')
- s = users.select(u.in_())
+ s = users.select(u.in_([]))
r = s.execute(search_key='john').fetchall()
assert len(r) == 0
r = s.execute(search_key=None).fetchall()
assert len(r) == 0
- s = users.select(not_(u.in_()))
+ s = users.select(not_(u.in_([])))
r = s.execute(search_key='john').fetchall()
assert len(r) == 3
r = s.execute(search_key=None).fetchall()
assert len(r) == 0
- s = users.select(users.c.user_name.in_() == True)
+ s = users.select(users.c.user_name.in_([]) == True)
r = s.execute().fetchall()
assert len(r) == 0
- s = users.select(users.c.user_name.in_() == False)
+ s = users.select(users.c.user_name.in_([]) == False)
r = s.execute().fetchall()
assert len(r) == 2
- s = users.select(users.c.user_name.in_() == None)
+ s = users.select(users.c.user_name.in_([]) == None)
r = s.execute().fetchall()
assert len(r) == 1
def test_union(self):
(s1, s2) = (
select([t1.c.col3.label('col3'), t1.c.col4.label('col4')],
- t1.c.col2.in_("t1col2r1", "t1col2r2")),
+ t1.c.col2.in_(["t1col2r1", "t1col2r2"])),
select([t2.c.col3.label('col3'), t2.c.col4.label('col4')],
- t2.c.col2.in_("t2col2r2", "t2col2r3"))
+ t2.c.col2.in_(["t2col2r2", "t2col2r3"]))
)
u = union(s1, s2)
def test_union_ordered(self):
(s1, s2) = (
select([t1.c.col3.label('col3'), t1.c.col4.label('col4')],
- t1.c.col2.in_("t1col2r1", "t1col2r2")),
+ t1.c.col2.in_(["t1col2r1", "t1col2r2"])),
select([t2.c.col3.label('col3'), t2.c.col4.label('col4')],
- t2.c.col2.in_("t2col2r2", "t2col2r3"))
+ t2.c.col2.in_(["t2col2r2", "t2col2r3"]))
)
u = union(s1, s2, order_by=['col3', 'col4'])
def test_union_ordered_alias(self):
(s1, s2) = (
select([t1.c.col3.label('col3'), t1.c.col4.label('col4')],
- t1.c.col2.in_("t1col2r1", "t1col2r2")),
+ t1.c.col2.in_(["t1col2r1", "t1col2r2"])),
select([t2.c.col3.label('col3'), t2.c.col4.label('col4')],
- t2.c.col2.in_("t2col2r2", "t2col2r3"))
+ t2.c.col2.in_(["t2col2r2", "t2col2r3"]))
)
u = union(s1, s2, order_by=['col3', 'col4'])
assert [str(c) for c in s.c] == ["id", "hoho"]
def testin(self):
- self.assert_compile(select([table1], table1.c.myid.in_('a')),
- "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid = :mytable_myid")
+ self.assert_compile(select([table1], table1.c.myid.in_(['a'])),
+ "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid IN (:mytable_myid)")
- self.assert_compile(select([table1], ~table1.c.myid.in_('a')),
- "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid != :mytable_myid")
+ self.assert_compile(select([table1], ~table1.c.myid.in_(['a'])),
+ "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid NOT IN (:mytable_myid)")
- self.assert_compile(select([table1], table1.c.myid.in_('a', 'b')),
+ self.assert_compile(select([table1], table1.c.myid.in_(['a', 'b'])),
"SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid IN (:mytable_myid, :mytable_myid_1)")
- self.assert_compile(select([table1], table1.c.myid.in_(literal('a'))),
- "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid = :literal")
+ self.assert_compile(select([table1], table1.c.myid.in_(iter(['a', 'b']))),
+ "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid IN (:mytable_myid, :mytable_myid_1)")
+
+ self.assert_compile(select([table1], table1.c.myid.in_([literal('a')])),
+ "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid IN (:literal)")
- self.assert_compile(select([table1], table1.c.myid.in_(literal('a'), 'b')),
+ self.assert_compile(select([table1], table1.c.myid.in_([literal('a'), 'b'])),
"SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid IN (:literal, :mytable_myid)")
- self.assert_compile(select([table1], table1.c.myid.in_(literal('a'), literal('b'))),
+ self.assert_compile(select([table1], table1.c.myid.in_([literal('a'), literal('b')])),
"SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid IN (:literal, :literal_1)")
- self.assert_compile(select([table1], table1.c.myid.in_('a', literal('b'))),
+ self.assert_compile(select([table1], table1.c.myid.in_(['a', literal('b')])),
"SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid IN (:mytable_myid, :literal)")
- self.assert_compile(select([table1], table1.c.myid.in_(literal(1) + 'a')),
- "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid = :literal + :literal_1")
+ self.assert_compile(select([table1], table1.c.myid.in_([literal(1) + 'a'])),
+ "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid IN (:literal + :literal_1)")
- self.assert_compile(select([table1], table1.c.myid.in_(literal('a') +'a', 'b')),
+ self.assert_compile(select([table1], table1.c.myid.in_([literal('a') +'a', 'b'])),
"SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid IN (:literal || :literal_1, :mytable_myid)")
- self.assert_compile(select([table1], table1.c.myid.in_(literal('a') + literal('a'), literal('b'))),
+ self.assert_compile(select([table1], table1.c.myid.in_([literal('a') + literal('a'), literal('b')])),
"SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid IN (:literal || :literal_1, :literal_2)")
- self.assert_compile(select([table1], table1.c.myid.in_(1, literal(3) + 4)),
+ self.assert_compile(select([table1], table1.c.myid.in_([1, literal(3) + 4])),
"SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid IN (:mytable_myid, :literal + :literal_1)")
- self.assert_compile(select([table1], table1.c.myid.in_(literal('a') < 'b')),
- "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid = (:literal < :literal_1)")
+ self.assert_compile(select([table1], table1.c.myid.in_([literal('a') < 'b'])),
+ "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid IN (:literal < :literal_1)")
- self.assert_compile(select([table1], table1.c.myid.in_(table1.c.myid)),
- "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid = mytable.myid")
+ self.assert_compile(select([table1], table1.c.myid.in_([table1.c.myid])),
+ "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid IN (mytable.myid)")
- self.assert_compile(select([table1], table1.c.myid.in_('a', table1.c.myid)),
+ self.assert_compile(select([table1], table1.c.myid.in_(['a', table1.c.myid])),
"SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid IN (:mytable_myid, mytable.myid)")
- self.assert_compile(select([table1], table1.c.myid.in_(literal('a'), table1.c.myid)),
+ self.assert_compile(select([table1], table1.c.myid.in_([literal('a'), table1.c.myid])),
"SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid IN (:literal, mytable.myid)")
- self.assert_compile(select([table1], table1.c.myid.in_(literal('a'), table1.c.myid +'a')),
+ self.assert_compile(select([table1], table1.c.myid.in_([literal('a'), table1.c.myid +'a'])),
"SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid IN (:literal, mytable.myid + :mytable_myid)")
- self.assert_compile(select([table1], table1.c.myid.in_(literal(1), 'a' + table1.c.myid)),
+ self.assert_compile(select([table1], table1.c.myid.in_([literal(1), 'a' + table1.c.myid])),
"SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid IN (:literal, :mytable_myid + mytable.myid)")
- self.assert_compile(select([table1], table1.c.myid.in_(1, 2, 3)),
+ self.assert_compile(select([table1], table1.c.myid.in_([1, 2, 3])),
"SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid IN (:mytable_myid, :mytable_myid_1, :mytable_myid_2)")
self.assert_compile(select([table1], table1.c.myid.in_(select([table2.c.otherid]))),
)
# test empty in clause
- self.assert_compile(select([table1], table1.c.myid.in_()),
+ self.assert_compile(select([table1], table1.c.myid.in_([])),
"SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE (CASE WHEN (mytable.myid IS NULL) THEN NULL ELSE 0 END = 1)")
+ def test_in_deprecated_api(self):
+ self.assert_compile(select([table1], table1.c.myid.in_('abc')),
+ "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid IN (:mytable_myid)")
+
+ self.assert_compile(select([table1], table1.c.myid.in_(1)),
+ "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid IN (:mytable_myid)")
+
+ self.assert_compile(select([table1], table1.c.myid.in_(1,2)),
+ "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid IN (:mytable_myid, :mytable_myid_1)")
+
+ self.assert_compile(select([table1], table1.c.myid.in_()),
+ "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE (CASE WHEN (mytable.myid IS NULL) THEN NULL ELSE 0 END = 1)")
def testcast(self):
tbl = table('casttest',
"SELECT op.field FROM op WHERE (op.field + :op_field) * :literal")
self.assert_compile(table.select((table.c.field * 5) + 6),
"SELECT op.field FROM op WHERE op.field * :op_field + :literal")
- self.assert_compile(table.select(5 + table.c.field.in_(5,6)),
+ self.assert_compile(table.select(5 + table.c.field.in_([5,6])),
"SELECT op.field FROM op WHERE :literal + (op.field IN (:op_field, :op_field_1))")
- self.assert_compile(table.select((5 + table.c.field).in_(5,6)),
+ self.assert_compile(table.select((5 + table.c.field).in_([5,6])),
"SELECT op.field FROM op WHERE :op_field + op.field IN (:literal, :literal_1)")
self.assert_compile(table.select(not_(and_(table.c.field == 5, table.c.field == 7))),
"SELECT op.field FROM op WHERE NOT (op.field = :op_field AND op.field = :op_field_1)")