result = ""
if select._limit:
- result += "FIRST %d " % select._limit
+ result += "FIRST %s " % self.process(sql.literal(select._limit))
if select._offset:
- result +="SKIP %d " % select._offset
+ result +="SKIP %s " % self.process(sql.literal(select._offset))
if select._distinct:
result += "DISTINCT "
return result
def limit_clause(self, select):
# The docs say offsets are supported with LIMIT. But they're not.
# TODO: maybe emulate by adding a ROWNO/ROWNUM predicate?
+ # TODO: does MaxDB support bind params for LIMIT / TOP ?
if self.is_subquery():
# sub queries need TOP
return ''
if select._distinct or select._limit:
s = select._distinct and "DISTINCT " or ""
+ # ODBC drivers and possibly others
+ # don't support bind params in the SELECT clause on SQL Server.
+ # so have to use literal here.
if select._limit:
if not select._offset:
- s += "TOP %s " % (select._limit,)
+ s += "TOP %d " % select._limit
return s
return compiler.SQLCompiler.get_select_precolumns(self, select)
limitselect = sql.select([c for c in select.c if
c.key!='mssql_rn'])
- limitselect.append_whereclause("mssql_rn>%d" % _offset)
+ limitselect.append_whereclause("mssql_rn>%s" % self.process(sql.literal(_offset)))
if _limit is not None:
- limitselect.append_whereclause("mssql_rn<=%d" %
- (_limit + _offset))
+ limitselect.append_whereclause("mssql_rn<=%s" %
+ (self.process(sql.literal(_limit + _offset))))
return self.process(limitselect, iswrapper=True, **kwargs)
else:
return compiler.SQLCompiler.visit_select(self, select, **kwargs)
# artificial limit if one wasn't provided
if limit is None:
limit = 18446744073709551615
- return ' \n LIMIT %s, %s' % (offset, limit)
+ return ' \n LIMIT %s, %s' % (
+ self.process(sql.literal(offset)),
+ self.process(sql.literal(limit)))
else:
# No offset provided, so just use the limit
- return ' \n LIMIT %s' % (limit,)
+ return ' \n LIMIT %s' % (self.process(sql.literal(limit)),)
def visit_update(self, update_stmt):
self.stack.append({'from': set([update_stmt.table])})
def limit_clause(self, select):
text = ""
if select._limit is not None:
- text += " \n LIMIT " + str(select._limit)
+ text += " \n LIMIT " + self.process(sql.literal(select._limit))
if select._offset is not None:
if select._limit is None:
text += " \n LIMIT ALL"
- text += " OFFSET " + str(select._offset)
+ text += " OFFSET " + self.process(sql.literal(select._offset))
return text
def get_select_precolumns(self, select):
def limit_clause(self, select):
text = ""
if select._limit is not None:
- text += " \n LIMIT " + str(select._limit)
+ text += "\n LIMIT " + self.process(sql.literal(select._limit))
if select._offset is not None:
if select._limit is None:
- text += " \n LIMIT -1"
- text += " OFFSET " + str(select._offset)
+ text += "\n LIMIT " + self.process(sql.literal(-1))
+ text += " OFFSET " + self.process(sql.literal(select._offset))
else:
- text += " OFFSET 0"
+ text += " OFFSET " + self.process(sql.literal(0))
return text
def for_update_clause(self, select):
def get_select_precolumns(self, select):
s = select._distinct and "DISTINCT " or ""
+ # TODO: don't think Sybase supports
+ # bind params for FIRST / TOP
if select._limit:
#if select._limit == 1:
#s += "FIRST "
def limit_clause(self, select):
text = ""
if select._limit is not None:
- text += " \n LIMIT " + str(select._limit)
+ text += "\n LIMIT " + self.process(sql.literal(select._limit))
if select._offset is not None:
if select._limit is None:
- text += " \n LIMIT -1"
- text += " OFFSET " + str(select._offset)
+ text += "\n LIMIT -1"
+ text += " OFFSET " + self.process(sql.literal(select._offset))
return text
def visit_table(self, table, asfrom=False, ashint=False, fromhints=None, **kwargs):
self.assert_compile(
select([t]).limit(10).offset(20),
- "SELECT t.col1, t.col2 FROM t LIMIT 20, 10"
+ "SELECT t.col1, t.col2 FROM t LIMIT %s, %s",
+ {'param_1':20, 'param_2':10}
)
self.assert_compile(
select([t]).limit(10),
- "SELECT t.col1, t.col2 FROM t LIMIT 10")
+ "SELECT t.col1, t.col2 FROM t LIMIT %s",
+ {'param_1':10})
+
self.assert_compile(
select([t]).offset(10),
- "SELECT t.col1, t.col2 FROM t LIMIT 10, 18446744073709551615"
+ "SELECT t.col1, t.col2 FROM t LIMIT %s, %s",
+ {'param_1':10, 'param_2':18446744073709551615}
)
def test_varchar_raise(self):
"anon_1.parent_cls AS anon_1_parent_cls, anon_2.parent_id AS anon_2_parent_id, "\
"anon_2.child2_id AS anon_2_child2_id, anon_2.parent_cls AS anon_2_parent_cls FROM "\
"(SELECT parent.id AS parent_id, child1.id AS child1_id, parent.cls AS parent_cls FROM parent "\
- "JOIN child1 ON parent.id = child1.id LIMIT 1) AS anon_1 LEFT OUTER JOIN secondary AS secondary_1 "\
+ "JOIN child1 ON parent.id = child1.id LIMIT :param_1) AS anon_1 LEFT OUTER JOIN secondary AS secondary_1 "\
"ON anon_1.parent_id = secondary_1.right_id LEFT OUTER JOIN (SELECT parent.id AS parent_id, "\
"parent.cls AS parent_cls, child2.id AS child2_id FROM parent JOIN child2 ON parent.id = child2.id) "\
- "AS anon_2 ON anon_2.parent_id = secondary_1.left_id"
- , dialect=default.DefaultDialect())
+ "AS anon_2 ON anon_2.parent_id = secondary_1.left_id",
+ {'param_1':1},
+ dialect=default.DefaultDialect())
# another way to check
assert q.limit(1).with_labels().subquery().count().scalar() == 1
"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 10) AS anon_1 LEFT OUTER JOIN orders AS orders_1 ON anon_1.users_id = orders_1.user_id"
- ,use_default_dialect=True
+ "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},
+ use_default_dialect=True
)
self.assert_compile(
"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 LEFT OUTER JOIN users AS "
- "users_1 ON users_1.id = orders.user_id LIMIT 10"
- ,use_default_dialect=True
+ "users_1 ON users_1.id = orders.user_id LIMIT :param_1",
+ {'param_1':10},
+ use_default_dialect=True
)
self.assert_compile(
"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 LIMIT 10"
- ,use_default_dialect=True
+ "users_1 ON users_1.id = orders.user_id LIMIT :param_1",
+ {'param_1':10},
+ use_default_dialect=True
)
self.assert_compile(
"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 10) AS anon_1 "
+ "(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"
- ,use_default_dialect=True
+ "addresses AS addresses_1 ON addresses_1.id = orders_1.address_id",
+ {'param_1':10},
+ use_default_dialect=True
)
self.assert_compile(
"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 10) AS anon_1 LEFT OUTER 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"
- ,use_default_dialect=True
+ "LIMIT :param_1) AS anon_1 LEFT OUTER 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},
+ use_default_dialect=True
)
@testing.resolve_artifact_names
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 "
- "WHERE nodes.data = :data_1 ORDER BY nodes.id LIMIT 1 OFFSET 0",
+ "WHERE nodes.data = :data_1 ORDER BY nodes.id LIMIT :param_1 OFFSET :param_2",
{'data_1': 'n1'}
)
)
assert create_session().query(User).filter(User.id==27).first() is None
- @testing.fails_on_everything_except('sqlite')
+ @testing.only_on('sqlite', 'testing execution but db-specific syntax')
def test_limit_offset_applies(self):
"""Test that the expected LIMIT/OFFSET is applied for slices.
q = sess.query(User)
self.assert_sql(testing.db, lambda: q[10:20], [
- ("SELECT users.id AS users_id, users.name AS users_name FROM users LIMIT 10 OFFSET 10", {})
+ ("SELECT users.id AS users_id, users.name AS users_name FROM users LIMIT :param_1 OFFSET :param_2", {'param_1':10, 'param_2':10})
])
self.assert_sql(testing.db, lambda: q[:20], [
- ("SELECT users.id AS users_id, users.name AS users_name FROM users LIMIT 20 OFFSET 0", {})
+ ("SELECT users.id AS users_id, users.name AS users_name FROM users LIMIT :param_1 OFFSET :param_2", {'param_1':20, 'param_2':0})
])
self.assert_sql(testing.db, lambda: q[5:], [
- ("SELECT users.id AS users_id, users.name AS users_name FROM users LIMIT -1 OFFSET 5", {})
+ ("SELECT users.id AS users_id, users.name AS users_name FROM users LIMIT :param_1 OFFSET :param_2", {'param_1':-1, 'param_2':5})
])
self.assert_sql(testing.db, lambda: q[2:2], [])
"nodes_1.id = nodes.parent_id JOIN nodes AS nodes_2 "
"ON nodes_2.id = nodes_1.parent_id "
"WHERE nodes.data = :data_1 AND nodes_1.data = :data_2 AND "
- "nodes_2.data = :data_3) AS anon_1 LIMIT 1",
+ "nodes_2.data = :data_3) AS anon_1 LIMIT :param_1",
+ {'param_1':1},
use_default_dialect=True
)
assert u1.corresponding_column(table2.c.otherid) is u1.c.myid
- # TODO - why is there an extra space before the LIMIT ?
self.assert_compile(
union(
select([table1.c.myid, table1.c.name]),
),
"SELECT mytable.myid, mytable.name "
"FROM mytable UNION SELECT myothertable.otherid, myothertable.othername "
- "FROM myothertable ORDER BY myid LIMIT 5 OFFSET 10"
+ "FROM myothertable ORDER BY myid LIMIT :param_1 OFFSET :param_2",
+ {'param_1':5, 'param_2':10}
)
self.assert_compile(
# self_group() is honored
self.assert_compile(
union(s.order_by("foo").self_group(), s.order_by("bar").limit(10).self_group()),
- "(SELECT foo, bar ORDER BY foo) UNION (SELECT foo, bar ORDER BY bar LIMIT 10)"
+ "(SELECT foo, bar ORDER BY foo) UNION (SELECT foo, bar ORDER BY bar LIMIT :param_1)",
+ {'param_1':10}
+
)
def test_compound_grouping(self):
),
"SELECT mytable.myid, mytable.name, mytable.description, myothertable.otherid, myothertable.othername FROM mytable "\
"JOIN myothertable ON mytable.myid = myothertable.otherid WHERE myothertable.otherid IN (SELECT myothertable.otherid "\
- "FROM myothertable ORDER BY myothertable.othername LIMIT 10) ORDER BY mytable.myid"
+ "FROM myothertable ORDER BY myothertable.othername LIMIT :param_1) ORDER BY mytable.myid",
+ {'param_1':10}
)
def test_tuple(self):
s2 = select([s1]).limit(5).offset(10).alias()
self.assert_compile(sql_util.ClauseAdapter(s2).traverse(s1),
- "SELECT foo.col1, foo.col2, foo.col3 FROM (SELECT table1.col1 AS col1, table1.col2 AS col2, table1.col3 AS col3 FROM table1) AS foo LIMIT 5 OFFSET 10")
+ "SELECT foo.col1, foo.col2, foo.col3 FROM (SELECT table1.col1 AS col1, table1.col2 AS col2, table1.col3 AS col3 FROM table1) AS foo LIMIT :param_1 OFFSET :param_2",
+ {'param_1':5, 'param_2':10}
+ )
j = s1.outerjoin(t2, s1.c.col1==t2.c.col1)
self.assert_compile(sql_util.ClauseAdapter(s2).traverse(j).select(),
"SELECT anon_1.col1, anon_1.col2, anon_1.col3, table2.col1, table2.col2, table2.col3 FROM "\
"(SELECT foo.col1 AS col1, foo.col2 AS col2, foo.col3 AS col3 FROM "\
- "(SELECT table1.col1 AS col1, table1.col2 AS col2, table1.col3 AS col3 FROM table1) AS foo LIMIT 5 OFFSET 10) AS anon_1 "\
- "LEFT OUTER JOIN table2 ON anon_1.col1 = table2.col1")
+ "(SELECT table1.col1 AS col1, table1.col2 AS col2, table1.col3 AS col3 FROM table1) AS foo LIMIT :param_1 OFFSET :param_2) AS anon_1 "\
+ "LEFT OUTER JOIN table2 ON anon_1.col1 = table2.col1",
+ {'param_1':5, 'param_2':10}
+ )
talias = t1.alias('bar')
j = s1.outerjoin(talias, s1.c.col1==talias.c.col1)
self.assert_compile(sql_util.ClauseAdapter(s2).traverse(j).select(),
"SELECT anon_1.col1, anon_1.col2, anon_1.col3, bar.col1, bar.col2, bar.col3 FROM "\
"(SELECT foo.col1 AS col1, foo.col2 AS col2, foo.col3 AS col3 FROM "\
- "(SELECT table1.col1 AS col1, table1.col2 AS col2, table1.col3 AS col3 FROM table1) AS foo LIMIT 5 OFFSET 10) AS anon_1 "\
- "LEFT OUTER JOIN table1 AS bar ON anon_1.col1 = bar.col1")
+ "(SELECT table1.col1 AS col1, table1.col2 AS col2, table1.col3 AS col3 FROM table1) AS foo LIMIT :param_1 OFFSET :param_2) AS anon_1 "\
+ "LEFT OUTER JOIN table1 AS bar ON anon_1.col1 = bar.col1",
+ {'param_1':5, 'param_2':10})
def test_functions(self):
self.assert_compile(sql_util.ClauseAdapter(t1.alias()).traverse(func.count(t1.c.col1)), "count(table1_1.col1)")
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
