--- /dev/null
+.. change::
+ :tags: usecase, sql
+ :tickets: 5576
+
+ Add support to ``FETCH {FIRST | NEXT} [ count ]
+ {ROW | ROWS} {ONLY | WITH TIES}`` in the select for the supported
+ backends, currently PostgreSQL, Oracle and MSSQL.
s = super(MSSQLCompiler, self).get_select_precolumns(select, **kw)
- if select._simple_int_limit and (
- select._offset_clause is None
- or (select._simple_int_offset and select._offset == 0)
- ):
+ if select._has_row_limiting_clause and self._use_top(select):
# ODBC drivers and possibly others
# don't support bind params in the SELECT clause on SQL Server.
# so have to use literal here.
kw["literal_execute"] = True
- s += "TOP %s " % self.process(select._limit_clause, **kw)
+ s += "TOP %s " % self.process(
+ self._get_limit_or_fetch(select), **kw
+ )
+ if select._fetch_clause is not None:
+ if select._fetch_clause_options["percent"]:
+ s += "PERCENT "
+ if select._fetch_clause_options["with_ties"]:
+ s += "WITH TIES "
return s
def get_crud_hint_text(self, table, text):
return text
- def limit_clause(self, select, **kw):
+ def _get_limit_or_fetch(self, select):
+ if select._fetch_clause is None:
+ return select._limit_clause
+ else:
+ return select._fetch_clause
+
+ def _use_top(self, select):
+ return (
+ select._offset_clause is None
+ or (
+ select._simple_int_clause(select._offset_clause)
+ and select._offset == 0
+ )
+ ) and (
+ select._simple_int_clause(select._limit_clause)
+ or (
+ # limit can use TOP with is by itself. fetch only uses TOP
+ # when it needs to because of PERCENT and/or WITH TIES
+ select._simple_int_clause(select._fetch_clause)
+ and (
+ select._fetch_clause_options["percent"]
+ or select._fetch_clause_options["with_ties"]
+ )
+ )
+ )
+
+ def fetch_clause(self, cs, **kwargs):
+ return ""
+
+ def limit_clause(self, cs, **kwargs):
+ return ""
+
+ def _check_can_use_fetch_like(self, select):
+ # to use ROW_NUMBER(), an ORDER BY is required.
+ # OFFSET are FETCH are options of the ORDER BY clause
+ if not select._order_by_clause.clauses:
+ raise exc.CompileError(
+ "MSSQL requires an order_by when "
+ "using an OFFSET or a non-simple "
+ "LIMIT clause"
+ )
+
+ if select._fetch_clause_options is not None and (
+ select._fetch_clause_options["percent"]
+ or select._fetch_clause_options["with_ties"]
+ ):
+ raise exc.CompileError(
+ "MSSQL needs TOP to use PERCENT and/or WITH TIES. "
+ "Only simple fetch without offset can be used."
+ )
+
+ def _row_limit_clause(self, select, **kw):
"""MSSQL 2012 supports OFFSET/FETCH operators
Use it instead subquery with row_number
"""
- if self.dialect._supports_offset_fetch and (
- (not select._simple_int_limit and select._limit_clause is not None)
- or (
- select._offset_clause is not None
- and not select._simple_int_offset
- or select._offset
- )
- ):
- # OFFSET are FETCH are options of the ORDER BY clause
- if not select._order_by_clause.clauses:
- raise exc.CompileError(
- "MSSQL requires an order_by when "
- "using an OFFSET or a non-simple "
- "LIMIT clause"
- )
+ if self.dialect._supports_offset_fetch and not self._use_top(select):
+ self._check_can_use_fetch_like(select)
text = ""
offset_str = "0"
text += "\n OFFSET %s ROWS" % offset_str
- if select._limit_clause is not None:
- text += "\n FETCH NEXT %s ROWS ONLY " % self.process(
- select._limit_clause, **kw
+ limit = self._get_limit_or_fetch(select)
+
+ if limit is not None:
+ text += "\n FETCH FIRST %s ROWS ONLY" % self.process(
+ limit, **kw
)
return text
else:
select = select_stmt
if (
- not self.dialect._supports_offset_fetch
- and (
- (
- not select._simple_int_limit
- and select._limit_clause is not None
- )
- or (
- select._offset_clause is not None
- and not select._simple_int_offset
- or select._offset
- )
- )
+ select._has_row_limiting_clause
+ and not self.dialect._supports_offset_fetch
+ and not self._use_top(select)
and not getattr(select, "_mssql_visit", None)
):
-
- # to use ROW_NUMBER(), an ORDER BY is required.
- if not select._order_by_clause.clauses:
- raise exc.CompileError(
- "MSSQL requires an order_by when "
- "using an OFFSET or a non-simple "
- "LIMIT clause"
- )
+ self._check_can_use_fetch_like(select)
_order_by_clauses = [
sql_util.unwrap_label_reference(elem)
for elem in select._order_by_clause.clauses
]
- limit_clause = select._limit_clause
+ limit_clause = self._get_limit_or_fetch(select)
offset_clause = select._offset_clause
select = select._generate()
select = select.where(whereclause)
select._oracle_visit = True
- limit_clause = select._limit_clause
- offset_clause = select._offset_clause
- if limit_clause is not None or offset_clause is not None:
+ # if fetch is used this is not needed
+ if (
+ select._has_row_limiting_clause
+ and select._fetch_clause is None
+ ):
+ limit_clause = select._limit_clause
+ offset_clause = select._offset_clause
# currently using form at:
# https://blogs.oracle.com/oraclemagazine/\
# on-rownum-and-limiting-results
if (
limit_clause is not None
and self.dialect.optimize_limits
- and select._simple_int_limit
+ and select._simple_int_clause(limit_clause)
):
param = sql.bindparam(
"_ora_frow",
# If needed, add the limiting clause
if limit_clause is not None:
- if select._simple_int_limit and (
- offset_clause is None or select._simple_int_offset
+ if select._simple_int_clause(limit_clause) and (
+ offset_clause is None
+ or select._simple_int_clause(offset_clause)
):
max_row = select._limit
adapter.traverse(elem) for elem in for_update.of
]
- if select._simple_int_offset:
+ if select._simple_int_clause(offset_clause):
offset_clause = sql.bindparam(
None,
select._offset,
params = self._parameters()
operation = re.sub(r"\?", lambda m: next(params), operation)
-
try:
prepared_stmt = await self._connection.prepare(operation)
text += " \n LIMIT " + self.process(select._limit_clause, **kw)
if select._offset_clause is not None:
if select._limit_clause is None:
- text += " \n LIMIT ALL"
+ text += "\n LIMIT ALL"
text += " OFFSET " + self.process(select._offset_clause, **kw)
return text
for t in extra_froms
)
+ def fetch_clause(self, select, **kw):
+ # pg requires parens for non literal clauses. It's also required for
+ # bind parameters if a ::type casts is used by the driver (asyncpg),
+ # so it's easies to just always add it
+ text = ""
+ if select._offset_clause is not None:
+ text += "\n OFFSET (%s) ROWS" % self.process(
+ select._offset_clause, **kw
+ )
+ if select._fetch_clause is not None:
+ text += "\n FETCH FIRST (%s)%s ROWS %s" % (
+ self.process(select._fetch_clause, **kw),
+ " PERCENT" if select._fetch_clause_options["percent"] else "",
+ "WITH TIES"
+ if select._fetch_clause_options["with_ties"]
+ else "ONLY",
+ )
+ return text
+
class PGDDLCompiler(compiler.DDLCompiler):
def get_column_specification(self, column, **kwargs):
"are applied." % (meth, meth)
)
+ @property
+ def _has_row_limiting_clause(self):
+ return (
+ self._limit_clause is not None or self._offset_clause is not None
+ )
+
def _get_options(
self,
populate_existing=None,
break
# don't need ORDER BY if no limit/offset
- if q._limit_clause is None and q._offset_clause is None:
+ if not q._has_row_limiting_clause:
q._order_by_clauses = ()
if q._distinct is True and q._order_by_clauses:
kwargs["include_table"] = False
text += self.group_by_clause(cs, **dict(asfrom=asfrom, **kwargs))
text += self.order_by_clause(cs, **kwargs)
- text += (
- (cs._limit_clause is not None or cs._offset_clause is not None)
- and self.limit_clause(cs, **kwargs)
- or ""
- )
+ if cs._has_row_limiting_clause:
+ text += self._row_limit_clause(cs, **kwargs)
if self.ctes and toplevel:
text = self._render_cte_clause() + text
self.stack.pop(-1)
return text
+ def _row_limit_clause(self, cs, **kwargs):
+ if cs._fetch_clause is not None:
+ return self.fetch_clause(cs, **kwargs)
+ else:
+ return self.limit_clause(cs, **kwargs)
+
def _get_operator_dispatch(self, operator_, qualifier1, qualifier2):
attrname = "visit_%s_%s%s" % (
operator_.__name__,
if select._order_by_clauses:
text += self.order_by_clause(select, **kwargs)
- if (
- select._limit_clause is not None
- or select._offset_clause is not None
- ):
- text += self.limit_clause(select, **kwargs)
+ if select._has_row_limiting_clause:
+ text += self._row_limit_clause(select, **kwargs)
if select._for_update_arg is not None:
text += self.for_update_clause(select, **kwargs)
text += " OFFSET " + self.process(select._offset_clause, **kw)
return text
+ def fetch_clause(self, select, **kw):
+ text = ""
+ if select._offset_clause is not None:
+ text += "\n OFFSET %s ROWS" % self.process(
+ select._offset_clause, **kw
+ )
+ if select._fetch_clause is not None:
+ text += "\n FETCH FIRST %s%s ROWS %s" % (
+ self.process(select._fetch_clause, **kw),
+ " PERCENT" if select._fetch_clause_options["percent"] else "",
+ "WITH TIES"
+ if select._fetch_clause_options["with_ties"]
+ else "ONLY",
+ )
+ return text
+
def visit_table(
self,
table,
_group_by_clauses = ()
_limit_clause = None
_offset_clause = None
+ _fetch_clause = None
+ _fetch_clause_options = None
_for_update_arg = None
def __init__(
"""
return self._offset_or_limit_clause_asint(self._limit_clause, "limit")
- @property
- def _simple_int_limit(self):
- """True if the LIMIT clause is a simple integer, False
+ def _simple_int_clause(self, clause):
+ """True if the clause is a simple integer, False
if it is not present or is a SQL expression.
"""
- return isinstance(self._limit_clause, _OffsetLimitParam)
-
- @property
- def _simple_int_offset(self):
- """True if the OFFSET clause is a simple integer, False
- if it is not present or is a SQL expression.
- """
- return isinstance(self._offset_clause, _OffsetLimitParam)
+ return isinstance(clause, _OffsetLimitParam)
@property
def _offset(self):
self._offset_clause, "offset"
)
+ @property
+ def _has_row_limiting_clause(self):
+ return (
+ self._limit_clause is not None
+ or self._offset_clause is not None
+ or self._fetch_clause is not None
+ )
+
@_generative
def limit(self, limit):
"""Return a new selectable with the given LIMIT criterion
support ``LIMIT`` will attempt to provide similar
functionality.
+ .. note::
+
+ The :meth:`_sql.GenerativeSelect.limit` method will replace
+ any clause applied with :meth:`_sql.GenerativeSelect.fetch`.
+
.. versionchanged:: 1.0.0 - :meth:`_expression.Select.limit` can now
accept arbitrary SQL expressions as well as integer values.
:param limit: an integer LIMIT parameter, or a SQL expression
- that provides an integer result.
+ that provides an integer result. Pass ``None`` to reset it.
+
+ .. seealso::
+
+ :meth:`_sql.GenerativeSelect.fetch`
+
+ :meth:`_sql.GenerativeSelect.offset`
"""
+ self._fetch_clause = self._fetch_clause_options = None
self._limit_clause = self._offset_or_limit_clause(limit)
+ @_generative
+ def fetch(self, count, with_ties=False, percent=False):
+ """Return a new selectable with the given FETCH FIRST criterion
+ applied.
+
+ This is a numeric value which usually renders as
+ ``FETCH {FIRST | NEXT} [ count ] {ROW | ROWS} {ONLY | WITH TIES}``
+ expression in the resulting select. This functionality is
+ is currently implemented for Oracle, PostgreSQL, MSSSQL.
+
+ Use :meth:`_sql.GenerativeSelect.offset` to specify the offset.
+
+ .. note::
+
+ The :meth:`_sql.GenerativeSelect.fetch` method will replace
+ any clause applied with :meth:`_sql.GenerativeSelect.limit`.
+
+ .. versionadded:: 1.4
+
+ :param count: an integer COUNT parameter, or a SQL expression
+ that provides an integer result. When ``percent=True`` this will
+ represent the percentage of rows to return, not the absolute value.
+ Pass ``None`` to reset it.
+
+ :param with_ties: When ``True``, the WITH TIES option is used
+ to return any additional rows that tie for the last place in the
+ result set according to the ``ORDER BY`` clause. The
+ ``ORDER BY`` may be mandatory in this case. Defaults to ``False``
+
+ :param percent: When ``True``, ``count`` represents the percentage
+ of the total number of selected rows to return. Defaults to ``False``
+
+ .. seealso::
+
+ :meth:`_sql.GenerativeSelect.limit`
+
+ :meth:`_sql.GenerativeSelect.offset`
+
+ """
+
+ self._limit_clause = None
+ if count is None:
+ self._fetch_clause = self._fetch_clause_options = None
+ else:
+ self._fetch_clause = self._offset_or_limit_clause(count)
+ self._fetch_clause_options = {
+ "with_ties": with_ties,
+ "percent": percent,
+ }
+
@_generative
def offset(self, offset):
"""Return a new selectable with the given OFFSET criterion
accept arbitrary SQL expressions as well as integer values.
:param offset: an integer OFFSET parameter, or a SQL expression
- that provides an integer result.
+ that provides an integer result. Pass ``None`` to reset it.
+
+ .. seealso::
+
+ :meth:`_sql.GenerativeSelect.limit`
+
+ :meth:`_sql.GenerativeSelect.fetch`
"""
LIMIT ? OFFSET ?
(2, 1)
+ .. note::
+
+ The :meth:`_sql.GenerativeSelect.slice` method will replace
+ any clause applied with :meth:`_sql.GenerativeSelect.fetch`.
+
.. versionadded:: 1.4 Added the :meth:`_sql.GenerativeSelect.slice`
method generalized from the ORM.
:meth:`_sql.GenerativeSelect.offset`
+ :meth:`_sql.GenerativeSelect.fetch`
+
"""
sql_util = util.preloaded.sql_util
+ self._fetch_clause = self._fetch_clause_options = None
self._limit_clause, self._offset_clause = sql_util._make_slice(
self._limit_clause, self._offset_clause, start, stop
)
("selects", InternalTraversal.dp_clauseelement_list),
("_limit_clause", InternalTraversal.dp_clauseelement),
("_offset_clause", InternalTraversal.dp_clauseelement),
+ ("_fetch_clause", InternalTraversal.dp_clauseelement),
+ ("_fetch_clause_options", InternalTraversal.dp_plain_dict),
("_order_by_clauses", InternalTraversal.dp_clauseelement_list),
("_group_by_clauses", InternalTraversal.dp_clauseelement_list),
("_for_update_arg", InternalTraversal.dp_clauseelement),
("_correlate_except", InternalTraversal.dp_clauseelement_tuple),
("_limit_clause", InternalTraversal.dp_clauseelement),
("_offset_clause", InternalTraversal.dp_clauseelement),
+ ("_fetch_clause", InternalTraversal.dp_clauseelement),
+ ("_fetch_clause_options", InternalTraversal.dp_plain_dict),
("_for_update_arg", InternalTraversal.dp_clauseelement),
("_distinct", InternalTraversal.dp_boolean),
("_distinct_on", InternalTraversal.dp_clauseelement_tuple),
subquery._columns._populate_separate_keys(prox)
def _needs_parens_for_grouping(self):
- return (
- self._limit_clause is not None
- or self._offset_clause is not None
- or bool(self._order_by_clause.clauses)
+ return self._has_row_limiting_clause or bool(
+ self._order_by_clause.clauses
)
def self_group(self, against=None):
@property
def regexp_match(self):
- """backend supports the regexp_match operator.
+ """backend supports the regexp_match operator."""
+ return exclusions.closed()
- .. versionadded:: 1.4
+ @property
+ def regexp_replace(self):
+ """backend supports the regexp_replace operator."""
+ return exclusions.closed()
- """
+ @property
+ def fetch_first(self):
+ """backend supports the fetch first clause."""
return exclusions.closed()
@property
- def regexp_replace(self):
- """backend supports the regexp_replace operator.
+ def fetch_percent(self):
+ """backend supports the fetch first clause with percent."""
+ return exclusions.closed()
- .. versionadded:: 1.4
+ @property
+ def fetch_ties(self):
+ """backend supports the fetch first clause with ties."""
+ return exclusions.closed()
+ @property
+ def fetch_no_order_by(self):
+ """backend supports the fetch first without order by"""
+ return exclusions.closed()
+ @property
+ def fetch_offset_with_options(self):
+ """backend supports the offset when using fetch first with percent
+ or ties. basically this is "not mssql"
"""
return exclusions.closed()
self._assert_result(stmt, [(1, 3), (1, 5), (1, 7)])
-class LimitOffsetTest(fixtures.TablesTest):
+class FetchLimitOffsetTest(fixtures.TablesTest):
__backend__ = True
@classmethod
{"id": 2, "x": 2, "y": 3},
{"id": 3, "x": 3, "y": 4},
{"id": 4, "x": 4, "y": 5},
+ {"id": 5, "x": 4, "y": 6},
],
)
[(1, 1, 2), (2, 2, 3)],
)
+ @testing.requires.fetch_first
+ def test_simple_fetch(self):
+ table = self.tables.some_table
+ self._assert_result(
+ select(table).order_by(table.c.id).fetch(2),
+ [(1, 1, 2), (2, 2, 3)],
+ )
+
@testing.requires.offset
def test_simple_offset(self):
table = self.tables.some_table
self._assert_result(
select(table).order_by(table.c.id).offset(2),
- [(3, 3, 4), (4, 4, 5)],
+ [(3, 3, 4), (4, 4, 5), (5, 4, 6)],
)
@testing.requires.offset
[(2, 2, 3), (3, 3, 4)],
)
+ @testing.requires.fetch_first
+ def test_simple_fetch_offset(self):
+ table = self.tables.some_table
+ self._assert_result(
+ select(table).order_by(table.c.id).fetch(2).offset(1),
+ [(2, 2, 3), (3, 3, 4)],
+ )
+
+ @testing.requires.fetch_no_order_by
+ def test_fetch_offset_no_order(self):
+ table = self.tables.some_table
+ with config.db.connect() as conn:
+ eq_(
+ set(conn.execute(select(table).fetch(10))),
+ set([(1, 1, 2), (2, 2, 3), (3, 3, 4), (4, 4, 5), (5, 4, 6)]),
+ )
+
@testing.requires.offset
def test_limit_offset_nobinds(self):
"""test that 'literal binds' mode works - no bound params."""
self._assert_result_str(sql, [(2, 2, 3), (3, 3, 4)])
+ @testing.requires.fetch_first
+ def test_fetch_offset_nobinds(self):
+ """test that 'literal binds' mode works - no bound params."""
+
+ table = self.tables.some_table
+ stmt = select(table).order_by(table.c.id).fetch(2).offset(1)
+ sql = stmt.compile(
+ dialect=config.db.dialect, compile_kwargs={"literal_binds": True}
+ )
+ sql = str(sql)
+
+ self._assert_result_str(sql, [(2, 2, 3), (3, 3, 4)])
+
@testing.requires.bound_limit_offset
def test_bound_limit(self):
table = self.tables.some_table
table = self.tables.some_table
self._assert_result(
select(table).order_by(table.c.id).offset(bindparam("o")),
- [(3, 3, 4), (4, 4, 5)],
+ [(3, 3, 4), (4, 4, 5), (5, 4, 6)],
params={"o": 2},
)
params={"l": 2, "o": 1},
)
+ @testing.requires.fetch_first
+ def test_bound_fetch_offset(self):
+ table = self.tables.some_table
+ self._assert_result(
+ select(table)
+ .order_by(table.c.id)
+ .fetch(bindparam("f"))
+ .offset(bindparam("o")),
+ [(2, 2, 3), (3, 3, 4)],
+ params={"f": 2, "o": 1},
+ )
+
@testing.requires.sql_expression_limit_offset
def test_expr_offset(self):
table = self.tables.some_table
select(table)
.order_by(table.c.id)
.offset(literal_column("1") + literal_column("2")),
- [(4, 4, 5)],
+ [(4, 4, 5), (5, 4, 6)],
)
@testing.requires.sql_expression_limit_offset
[(3, 3, 4), (4, 4, 5)],
)
+ @testing.requires.fetch_first
+ def test_expr_fetch_offset(self):
+ table = self.tables.some_table
+ self._assert_result(
+ select(table)
+ .order_by(table.c.id)
+ .fetch(literal_column("1") + literal_column("1"))
+ .offset(literal_column("1") + literal_column("1")),
+ [(3, 3, 4), (4, 4, 5)],
+ )
+
@testing.requires.sql_expression_limit_offset
def test_simple_limit_expr_offset(self):
table = self.tables.some_table
[(3, 3, 4), (4, 4, 5)],
)
+ @testing.requires.fetch_ties
+ def test_simple_fetch_ties(self):
+ table = self.tables.some_table
+ with config.db.connect() as conn:
+ eq_(
+ set(
+ conn.execute(
+ select(table)
+ .order_by(table.c.x.desc())
+ .fetch(1, with_ties=True)
+ )
+ ),
+ set([(4, 4, 5), (5, 4, 6)]),
+ )
+
+ @testing.requires.fetch_ties
+ @testing.requires.fetch_offset_with_options
+ def test_fetch_offset_ties(self):
+ table = self.tables.some_table
+ with config.db.connect() as conn:
+ fa = conn.execute(
+ select(table)
+ .order_by(table.c.x)
+ .fetch(2, with_ties=True)
+ .offset(2)
+ ).fetchall()
+ eq_(fa[0], (3, 3, 4))
+ eq_(set(fa), set([(3, 3, 4), (4, 4, 5), (5, 4, 6)]))
+
+ @testing.requires.fetch_ties
+ @testing.requires.fetch_offset_with_options
+ def test_fetch_offset_ties_exact_number(self):
+ table = self.tables.some_table
+ self._assert_result(
+ select(table)
+ .order_by(table.c.x)
+ .fetch(2, with_ties=True)
+ .offset(1),
+ [(2, 2, 3), (3, 3, 4)],
+ )
+
+ @testing.requires.fetch_percent
+ def test_simple_fetch_percent(self):
+ table = self.tables.some_table
+ self._assert_result(
+ select(table).order_by(table.c.id).fetch(20, percent=True),
+ [(1, 1, 2)],
+ )
+
+ @testing.requires.fetch_percent
+ @testing.requires.fetch_offset_with_options
+ def test_fetch_offset_percent(self):
+ table = self.tables.some_table
+ self._assert_result(
+ select(table)
+ .order_by(table.c.id)
+ .fetch(40, percent=True)
+ .offset(1),
+ [(2, 2, 3), (3, 3, 4)],
+ )
+
+ @testing.requires.fetch_ties
+ @testing.requires.fetch_percent
+ def test_simple_fetch_percent_ties(self):
+ table = self.tables.some_table
+ with config.db.connect() as conn:
+ fa = conn.execute(
+ select(table)
+ .order_by(table.c.x.desc())
+ .fetch(20, percent=True, with_ties=True)
+ ).fetchall()
+
+ eq_(len(fa), 2)
+ eq_(set(fa), set([(4, 4, 5), (5, 4, 6)]))
+
+ @testing.requires.fetch_ties
+ @testing.requires.fetch_percent
+ @testing.requires.fetch_offset_with_options
+ def test_fetch_offset_percent_ties(self):
+ table = self.tables.some_table
+ with config.db.connect() as conn:
+ fa = conn.execute(
+ select(table)
+ .order_by(table.c.x)
+ .fetch(40, percent=True, with_ties=True)
+ .offset(2)
+ ).fetchall()
+ eq_(fa[0], (3, 3, 4))
+ eq_(set(fa), set([(3, 3, 4), (4, 4, 5), (5, 4, 6)]))
+
class JoinTest(fixtures.TablesTest):
__backend__ = True
# -*- encoding: utf-8
+from sqlalchemy import bindparam
from sqlalchemy import Column
from sqlalchemy import Computed
from sqlalchemy import delete
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
+tbl = table("t", column("a"))
+
class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
__dialect__ = mssql.dialect()
+ @testing.fixture
+ def dialect_2012(self):
+ dialect = mssql.dialect()
+ dialect._supports_offset_fetch = True
+ return dialect
+
def test_true_false(self):
self.assert_compile(sql.false(), "0")
self.assert_compile(sql.true(), "1")
"foo.myid = mytable.myid",
)
- def test_noorderby_insubquery_offset_newstyle(self):
+ def test_noorderby_insubquery_offset_newstyle(self, dialect_2012):
"""test "no ORDER BY in subqueries unless TOP / LIMIT / OFFSET"
present"""
.alias("foo")
)
crit = q.c.myid == table1.c.myid
- dialect = mssql.dialect()
- dialect._supports_offset_fetch = True
self.assert_compile(
select("*").where(crit),
"SELECT * FROM (SELECT mytable.myid AS myid FROM mytable "
"ORDER BY mytable.myid OFFSET :param_1 ROWS) AS foo, "
"mytable WHERE foo.myid = mytable.myid",
- dialect=dialect,
+ dialect=dialect_2012,
)
- def test_noorderby_insubquery_limit_offset_newstyle(self):
+ def test_noorderby_insubquery_limit_offset_newstyle(self, dialect_2012):
"""test "no ORDER BY in subqueries unless TOP / LIMIT / OFFSET"
present"""
.alias("foo")
)
crit = q.c.myid == table1.c.myid
- dialect = mssql.dialect()
- dialect._supports_offset_fetch = True
self.assert_compile(
select("*").where(crit),
"SELECT * FROM (SELECT mytable.myid AS myid FROM mytable "
"ORDER BY mytable.myid OFFSET :param_1 ROWS "
- "FETCH NEXT :param_2 ROWS ONLY ) AS foo, "
+ "FETCH FIRST :param_2 ROWS ONLY) AS foo, "
"mytable WHERE foo.myid = mytable.myid",
- dialect=dialect,
+ dialect=dialect_2012,
)
def test_noorderby_parameters_insubquery(self):
assert t.c.x in set(c._create_result_map()["x"][1])
assert t.c.y in set(c._create_result_map()["y"][1])
- def test_limit_offset_using_offset_fetch(self):
+ def test_limit_offset_using_offset_fetch(self, dialect_2012):
t = table("t", column("x", Integer), column("y", Integer))
- dialect_2012 = mssql.base.MSDialect()
- dialect_2012._supports_offset_fetch = True
-
s = select(t).where(t.c.x == 5).order_by(t.c.y).limit(10).offset(20)
self.assert_compile(
"FROM t "
"WHERE t.x = :x_1 ORDER BY t.y "
"OFFSET :param_1 ROWS "
- "FETCH NEXT :param_2 ROWS ONLY ",
+ "FETCH FIRST :param_2 ROWS ONLY",
checkparams={"param_1": 20, "param_2": 10, "x_1": 5},
dialect=dialect_2012,
)
"CREATE TABLE t (x INTEGER NULL, y AS (x + 2)%s)" % text,
)
+ @testing.combinations(
+ (
+ 5,
+ 10,
+ {},
+ "OFFSET :param_1 ROWS FETCH FIRST :param_2 ROWS ONLY",
+ {"param_1": 10, "param_2": 5},
+ ),
+ (None, 10, {}, "OFFSET :param_1 ROWS", {"param_1": 10}),
+ (
+ 5,
+ None,
+ {},
+ "OFFSET 0 ROWS FETCH FIRST :param_1 ROWS ONLY",
+ {"param_1": 5},
+ ),
+ (
+ 0,
+ 0,
+ {},
+ "OFFSET :param_1 ROWS FETCH FIRST :param_2 ROWS ONLY",
+ {"param_1": 0, "param_2": 0},
+ ),
+ (
+ 5,
+ 0,
+ {"percent": True},
+ "TOP [POSTCOMPILE_param_1] PERCENT",
+ {"param_1": 5},
+ ),
+ (
+ 5,
+ None,
+ {"percent": True, "with_ties": True},
+ "TOP [POSTCOMPILE_param_1] PERCENT WITH TIES",
+ {"param_1": 5},
+ ),
+ (
+ 5,
+ 0,
+ {"with_ties": True},
+ "TOP [POSTCOMPILE_param_1] WITH TIES",
+ {"param_1": 5},
+ ),
+ (
+ literal_column("Q"),
+ literal_column("Y"),
+ {},
+ "OFFSET Y ROWS FETCH FIRST Q ROWS ONLY",
+ {},
+ ),
+ (
+ column("Q"),
+ column("Y"),
+ {},
+ "OFFSET [Y] ROWS FETCH FIRST [Q] ROWS ONLY",
+ {},
+ ),
+ (
+ bindparam("Q", 3),
+ bindparam("Y", 7),
+ {},
+ "OFFSET :Y ROWS FETCH FIRST :Q ROWS ONLY",
+ {"Q": 3, "Y": 7},
+ ),
+ (
+ literal_column("Q") + literal_column("Z"),
+ literal_column("Y") + literal_column("W"),
+ {},
+ "OFFSET Y + W ROWS FETCH FIRST Q + Z ROWS ONLY",
+ {},
+ ),
+ argnames="fetch, offset, fetch_kw, exp, params",
+ )
+ def test_fetch(self, dialect_2012, fetch, offset, fetch_kw, exp, params):
+ t = table("t", column("a"))
+ if "TOP" in exp:
+ sel = "SELECT %s t.a FROM t ORDER BY t.a" % exp
+ else:
+ sel = "SELECT t.a FROM t ORDER BY t.a " + exp
+
+ self.assert_compile(
+ select(t).order_by(t.c.a).fetch(fetch, **fetch_kw).offset(offset),
+ sel,
+ checkparams=params,
+ dialect=dialect_2012,
+ )
+
+ @testing.combinations(
+ (
+ 5,
+ 10,
+ {},
+ "mssql_rn > :param_1 AND mssql_rn <= :param_2 + :param_1",
+ {"param_1": 10, "param_2": 5},
+ ),
+ (None, 10, {}, "mssql_rn > :param_1", {"param_1": 10}),
+ (
+ 5,
+ None,
+ {},
+ "mssql_rn <= :param_1",
+ {"param_1": 5},
+ ),
+ (
+ 0,
+ 0,
+ {},
+ "mssql_rn > :param_1 AND mssql_rn <= :param_2 + :param_1",
+ {"param_1": 0, "param_2": 0},
+ ),
+ (
+ 5,
+ 0,
+ {"percent": True},
+ "TOP [POSTCOMPILE_param_1] PERCENT",
+ {"param_1": 5},
+ ),
+ (
+ 5,
+ None,
+ {"percent": True, "with_ties": True},
+ "TOP [POSTCOMPILE_param_1] PERCENT WITH TIES",
+ {"param_1": 5},
+ ),
+ (
+ 5,
+ 0,
+ {"with_ties": True},
+ "TOP [POSTCOMPILE_param_1] WITH TIES",
+ {"param_1": 5},
+ ),
+ (
+ literal_column("Q"),
+ literal_column("Y"),
+ {},
+ "mssql_rn > Y AND mssql_rn <= Q + Y",
+ {},
+ ),
+ (
+ column("Q"),
+ column("Y"),
+ {},
+ "mssql_rn > [Y] AND mssql_rn <= [Q] + [Y]",
+ {},
+ ),
+ (
+ bindparam("Q", 3),
+ bindparam("Y", 7),
+ {},
+ "mssql_rn > :Y AND mssql_rn <= :Q + :Y",
+ {"Q": 3, "Y": 7},
+ ),
+ (
+ literal_column("Q") + literal_column("Z"),
+ literal_column("Y") + literal_column("W"),
+ {},
+ "mssql_rn > Y + W AND mssql_rn <= Q + Z + Y + W",
+ {},
+ ),
+ argnames="fetch, offset, fetch_kw, exp, params",
+ )
+ def test_fetch_old_version(self, fetch, offset, fetch_kw, exp, params):
+ t = table("t", column("a"))
+ if "TOP" in exp:
+ sel = "SELECT %s t.a FROM t ORDER BY t.a" % exp
+ else:
+ sel = (
+ "SELECT anon_1.a FROM (SELECT t.a AS a, ROW_NUMBER() "
+ "OVER (ORDER BY t.a) AS mssql_rn FROM t) AS anon_1 WHERE "
+ + exp
+ )
+
+ self.assert_compile(
+ select(t).order_by(t.c.a).fetch(fetch, **fetch_kw).offset(offset),
+ sel,
+ checkparams=params,
+ )
+
+ _no_offset = (
+ "MSSQL needs TOP to use PERCENT and/or WITH TIES. "
+ "Only simple fetch without offset can be used."
+ )
+
+ _order_by = (
+ "MSSQL requires an order_by when using an OFFSET "
+ "or a non-simple LIMIT clause"
+ )
+
+ @testing.combinations(
+ (
+ select(tbl).order_by(tbl.c.a).fetch(5, percent=True).offset(3),
+ _no_offset,
+ ),
+ (
+ select(tbl).order_by(tbl.c.a).fetch(5, with_ties=True).offset(3),
+ _no_offset,
+ ),
+ (
+ select(tbl)
+ .order_by(tbl.c.a)
+ .fetch(5, percent=True, with_ties=True)
+ .offset(3),
+ _no_offset,
+ ),
+ (
+ select(tbl)
+ .order_by(tbl.c.a)
+ .fetch(bindparam("x"), with_ties=True),
+ _no_offset,
+ ),
+ (select(tbl).fetch(5).offset(3), _order_by),
+ (select(tbl).fetch(5), _order_by),
+ (select(tbl).offset(5), _order_by),
+ argnames="stmt, error",
+ )
+ def test_row_limit_compile_error(self, dialect_2012, stmt, error):
+ with testing.expect_raises_message(exc.CompileError, error):
+ print(stmt.compile(dialect=dialect_2012))
+ with testing.expect_raises_message(exc.CompileError, error):
+ print(stmt.compile(dialect=self.__dialect__))
+
class CompileIdentityTest(fixtures.TestBase, AssertsCompiledSQL):
__dialect__ = mssql.dialect()
from sqlalchemy import and_
from sqlalchemy import BigInteger
+from sqlalchemy import bindparam
from sqlalchemy import cast
from sqlalchemy import Column
from sqlalchemy import Computed
" %(myid_2)s, mytable.name)",
checkparams={"myid_1": "pattern", "myid_2": "replacement"},
)
+
+ @testing.combinations(
+ (
+ 5,
+ 10,
+ {},
+ "OFFSET (%(param_1)s) ROWS FETCH FIRST (%(param_2)s) ROWS ONLY",
+ {"param_1": 10, "param_2": 5},
+ ),
+ (None, 10, {}, "LIMIT ALL OFFSET %(param_1)s", {"param_1": 10}),
+ (
+ 5,
+ None,
+ {},
+ "FETCH FIRST (%(param_1)s) ROWS ONLY",
+ {"param_1": 5},
+ ),
+ (
+ 0,
+ 0,
+ {},
+ "OFFSET (%(param_1)s) ROWS FETCH FIRST (%(param_2)s) ROWS ONLY",
+ {"param_1": 0, "param_2": 0},
+ ),
+ (
+ 5,
+ 10,
+ {"percent": True},
+ "OFFSET (%(param_1)s) ROWS FETCH FIRST "
+ "(%(param_2)s) PERCENT ROWS ONLY",
+ {"param_1": 10, "param_2": 5},
+ ),
+ (
+ 5,
+ 10,
+ {"percent": True, "with_ties": True},
+ "OFFSET (%(param_1)s) ROWS FETCH FIRST (%(param_2)s)"
+ " PERCENT ROWS WITH TIES",
+ {"param_1": 10, "param_2": 5},
+ ),
+ (
+ 5,
+ 10,
+ {"with_ties": True},
+ "OFFSET (%(param_1)s) ROWS FETCH FIRST "
+ "(%(param_2)s) ROWS WITH TIES",
+ {"param_1": 10, "param_2": 5},
+ ),
+ (
+ literal_column("Q"),
+ literal_column("Y"),
+ {},
+ "OFFSET (Y) ROWS FETCH FIRST (Q) ROWS ONLY",
+ {},
+ ),
+ (
+ column("Q"),
+ column("Y"),
+ {},
+ 'OFFSET ("Y") ROWS FETCH FIRST ("Q") ROWS ONLY',
+ {},
+ ),
+ (
+ bindparam("Q", 3),
+ bindparam("Y", 7),
+ {},
+ "OFFSET (%(Y)s) ROWS FETCH FIRST (%(Q)s) ROWS ONLY",
+ {"Q": 3, "Y": 7},
+ ),
+ (
+ literal_column("Q") + literal_column("Z"),
+ literal_column("Y") + literal_column("W"),
+ {},
+ "OFFSET (Y + W) ROWS FETCH FIRST (Q + Z) ROWS ONLY",
+ {},
+ ),
+ )
+ def test_fetch(self, fetch, offset, fetch_kw, exp, params):
+ self.assert_compile(
+ select(1).fetch(fetch, **fetch_kw).offset(offset),
+ "SELECT 1 " + exp,
+ checkparams=params,
+ )
@property
def index_reflects_included_columns(self):
return only_on(["postgresql >= 11", "mssql"])
+
+ # mssql>= 11 -> >= MS_2012_VERSION
+
+ @property
+ def fetch_first(self):
+ return only_on(["postgresql", "mssql >= 11", "oracle >= 12"])
+
+ @property
+ def fetch_percent(self):
+ return only_on(["mssql >= 11", "oracle >= 12"])
+
+ @property
+ def fetch_ties(self):
+ return only_on(["postgresql >= 13", "mssql >= 11", "oracle >= 12"])
+
+ @property
+ def fetch_no_order_by(self):
+ return only_on(["postgresql", "oracle >= 12"])
+
+ @property
+ def fetch_offset_with_options(self):
+ return skip_if("mssql")
.where(table_a.c.b == 5)
.correlate_except(table_b),
),
+ lambda: (
+ select(table_a.c.a),
+ select(table_a.c.a).limit(2),
+ select(table_a.c.a).limit(3),
+ select(table_a.c.a).fetch(3),
+ select(table_a.c.a).fetch(2),
+ select(table_a.c.a).fetch(2, percent=True),
+ select(table_a.c.a).fetch(2, with_ties=True),
+ select(table_a.c.a).fetch(2, with_ties=True, percent=True),
+ select(table_a.c.a).fetch(2).offset(3),
+ select(table_a.c.a).fetch(2).offset(5),
+ select(table_a.c.a).limit(2).offset(5),
+ select(table_a.c.a).limit(2).offset(3),
+ select(table_a.c.a).union(select(table_a.c.a)).limit(2).offset(3),
+ union(select(table_a.c.a), select(table_a.c.b)).limit(2).offset(3),
+ union(select(table_a.c.a), select(table_a.c.b)).limit(6).offset(3),
+ union(select(table_a.c.a), select(table_a.c.b)).limit(6).offset(8),
+ union(select(table_a.c.a), select(table_a.c.b)).fetch(2).offset(8),
+ union(select(table_a.c.a), select(table_a.c.b)).fetch(6).offset(8),
+ union(select(table_a.c.a), select(table_a.c.b)).fetch(6).offset(3),
+ union(select(table_a.c.a), select(table_a.c.b))
+ .fetch(6, percent=True)
+ .offset(3),
+ union(select(table_a.c.a), select(table_a.c.b))
+ .fetch(6, with_ties=True)
+ .offset(3),
+ union(select(table_a.c.a), select(table_a.c.b))
+ .fetch(6, with_ties=True, percent=True)
+ .offset(3),
+ union(select(table_a.c.a), select(table_a.c.b)).limit(6),
+ union(select(table_a.c.a), select(table_a.c.b)).offset(6),
+ ),
lambda: (
select(table_a.c.a),
select(table_a.c.a).join(table_b, table_a.c.a == table_b.c.a),
dont_compare_values_fixtures.append(_lambda_fixtures)
+ # like fixture but returns at least two objects that compare equally
+ equal_fixtures = [
+ lambda: (
+ select(table_a.c.a).fetch(3),
+ select(table_a.c.a).fetch(2).fetch(3),
+ select(table_a.c.a).fetch(3, percent=False, with_ties=False),
+ select(table_a.c.a).limit(2).fetch(3),
+ select(table_a.c.a).slice(2, 4).fetch(3).offset(None),
+ ),
+ lambda: (
+ select(table_a.c.a).limit(3),
+ select(table_a.c.a).fetch(2).limit(3),
+ select(table_a.c.a).fetch(2).slice(0, 3).offset(None),
+ ),
+ ]
+
class CacheKeyFixture(object):
+ def _compare_equal(self, a, b, compare_values):
+ a_key = a._generate_cache_key()
+ b_key = b._generate_cache_key()
+
+ if a_key is None:
+ assert a._annotations.get("nocache")
+
+ assert b_key is None
+ else:
+
+ eq_(a_key.key, b_key.key)
+ eq_(hash(a_key.key), hash(b_key.key))
+
+ for a_param, b_param in zip(a_key.bindparams, b_key.bindparams):
+ assert a_param.compare(b_param, compare_values=compare_values)
+ return a_key, b_key
+
def _run_cache_key_fixture(self, fixture, compare_values):
case_a = fixture()
case_b = fixture()
range(len(case_a)), 2
):
if a == b:
- a_key = case_a[a]._generate_cache_key()
- b_key = case_b[b]._generate_cache_key()
-
+ a_key, b_key = self._compare_equal(
+ case_a[a], case_b[b], compare_values
+ )
if a_key is None:
- assert case_a[a]._annotations.get("nocache")
-
- assert b_key is None
continue
-
- eq_(a_key.key, b_key.key)
- eq_(hash(a_key.key), hash(b_key.key))
-
- for a_param, b_param in zip(
- a_key.bindparams, b_key.bindparams
- ):
- assert a_param.compare(
- b_param, compare_values=compare_values
- )
else:
a_key = case_a[a]._generate_cache_key()
b_key = case_b[b]._generate_cache_key()
),
)
+ def _run_cache_key_equal_fixture(self, fixture, compare_values):
+ case_a = fixture()
+ case_b = fixture()
+
+ for a, b in itertools.combinations_with_replacement(
+ range(len(case_a)), 2
+ ):
+ self._compare_equal(case_a[a], case_b[b], compare_values)
+
class CacheKeyTest(CacheKeyFixture, CoreFixtures, fixtures.TestBase):
# we are slightly breaking the policy of not having external dialect
for fixture in fixtures_:
self._run_cache_key_fixture(fixture, compare_values)
+ def test_cache_key_equal(self):
+ for fixture in self.equal_fixtures:
+ self._run_cache_key_equal_fixture(fixture, True)
+
def test_literal_binds(self):
def fixture():
return (
"_offset",
)
- def test_limit_offset(self):
- for lim, offset, exp, params in [
- (
- 5,
- 10,
- "LIMIT :param_1 OFFSET :param_2",
- {"param_1": 5, "param_2": 10},
- ),
- (None, 10, "LIMIT -1 OFFSET :param_1", {"param_1": 10}),
- (5, None, "LIMIT :param_1", {"param_1": 5}),
- (
- 0,
- 0,
- "LIMIT :param_1 OFFSET :param_2",
- {"param_1": 0, "param_2": 0},
- ),
- ]:
- self.assert_compile(
- select(1).limit(lim).offset(offset),
- "SELECT 1 " + exp,
- checkparams=params,
- )
+ @testing.combinations(
+ (
+ 5,
+ 10,
+ "LIMIT :param_1 OFFSET :param_2",
+ {"param_1": 5, "param_2": 10},
+ ),
+ (None, 10, "LIMIT -1 OFFSET :param_1", {"param_1": 10}),
+ (5, None, "LIMIT :param_1", {"param_1": 5}),
+ (
+ 0,
+ 0,
+ "LIMIT :param_1 OFFSET :param_2",
+ {"param_1": 0, "param_2": 0},
+ ),
+ (
+ literal_column("Q"),
+ literal_column("Y"),
+ "LIMIT Q OFFSET Y",
+ {},
+ ),
+ (
+ column("Q"),
+ column("Y"),
+ 'LIMIT "Q" OFFSET "Y"',
+ {},
+ ),
+ )
+ def test_limit_offset(self, lim, offset, exp, params):
+ self.assert_compile(
+ select(1).limit(lim).offset(offset),
+ "SELECT 1 " + exp,
+ checkparams=params,
+ )
+
+ @testing.combinations(
+ (
+ 5,
+ 10,
+ {},
+ "OFFSET :param_1 ROWS FETCH FIRST :param_2 ROWS ONLY",
+ {"param_1": 10, "param_2": 5},
+ ),
+ (None, 10, {}, "LIMIT -1 OFFSET :param_1", {"param_1": 10}),
+ (
+ 5,
+ None,
+ {},
+ "FETCH FIRST :param_1 ROWS ONLY",
+ {"param_1": 5},
+ ),
+ (
+ 0,
+ 0,
+ {},
+ "OFFSET :param_1 ROWS FETCH FIRST :param_2 ROWS ONLY",
+ {"param_1": 0, "param_2": 0},
+ ),
+ (
+ 5,
+ 10,
+ {"percent": True},
+ "OFFSET :param_1 ROWS FETCH FIRST :param_2 PERCENT ROWS ONLY",
+ {"param_1": 10, "param_2": 5},
+ ),
+ (
+ 5,
+ 10,
+ {"percent": True, "with_ties": True},
+ "OFFSET :param_1 ROWS FETCH FIRST :param_2 PERCENT ROWS WITH TIES",
+ {"param_1": 10, "param_2": 5},
+ ),
+ (
+ 5,
+ 10,
+ {"with_ties": True},
+ "OFFSET :param_1 ROWS FETCH FIRST :param_2 ROWS WITH TIES",
+ {"param_1": 10, "param_2": 5},
+ ),
+ (
+ literal_column("Q"),
+ literal_column("Y"),
+ {},
+ "OFFSET Y ROWS FETCH FIRST Q ROWS ONLY",
+ {},
+ ),
+ (
+ column("Q"),
+ column("Y"),
+ {},
+ 'OFFSET "Y" ROWS FETCH FIRST "Q" ROWS ONLY',
+ {},
+ ),
+ (
+ bindparam("Q", 3),
+ bindparam("Y", 7),
+ {},
+ "OFFSET :Y ROWS FETCH FIRST :Q ROWS ONLY",
+ {"Q": 3, "Y": 7},
+ ),
+ (
+ literal_column("Q") + literal_column("Z"),
+ literal_column("Y") + literal_column("W"),
+ {},
+ "OFFSET Y + W ROWS FETCH FIRST Q + Z ROWS ONLY",
+ {},
+ ),
+ )
+ def test_fetch(self, fetch, offset, fetch_kw, exp, params):
+ self.assert_compile(
+ select(1).fetch(fetch, **fetch_kw).offset(offset),
+ "SELECT 1 " + exp,
+ checkparams=params,
+ )
+
+ def test_fetch_limit_offset_self_group(self):
+ self.assert_compile(
+ select(1).limit(1).self_group(),
+ "(SELECT 1 LIMIT :param_1)",
+ checkparams={"param_1": 1},
+ )
+ self.assert_compile(
+ select(1).offset(1).self_group(),
+ "(SELECT 1 LIMIT -1 OFFSET :param_1)",
+ checkparams={"param_1": 1},
+ )
+ self.assert_compile(
+ select(1).fetch(1).self_group(),
+ "(SELECT 1 FETCH FIRST :param_1 ROWS ONLY)",
+ checkparams={"param_1": 1},
+ )
+
+ def test_limit_fetch_interaction(self):
+ self.assert_compile(
+ select(1).limit(42).fetch(1),
+ "SELECT 1 FETCH FIRST :param_1 ROWS ONLY",
+ checkparams={"param_1": 1},
+ )
+ self.assert_compile(
+ select(1).fetch(42).limit(1),
+ "SELECT 1 LIMIT :param_1",
+ checkparams={"param_1": 1},
+ )
+ self.assert_compile(
+ select(1).limit(42).offset(7).fetch(1),
+ "SELECT 1 OFFSET :param_1 ROWS FETCH FIRST :param_2 ROWS ONLY",
+ checkparams={"param_1": 7, "param_2": 1},
+ )
+ self.assert_compile(
+ select(1).fetch(1).slice(2, 5),
+ "SELECT 1 LIMIT :param_1 OFFSET :param_2",
+ checkparams={"param_1": 3, "param_2": 2},
+ )
+ self.assert_compile(
+ select(1).slice(2, 5).fetch(1),
+ "SELECT 1 OFFSET :param_1 ROWS FETCH FIRST :param_2 ROWS ONLY",
+ checkparams={"param_1": 2, "param_2": 1},
+ )
def test_select_precol_compile_ordering(self):
s1 = (
literal_binds=True,
)
+ def test_fetch_offset_select_literal_binds(self):
+ stmt = select(1).fetch(5).offset(6)
+ self.assert_compile(
+ stmt,
+ "SELECT 1 OFFSET 6 ROWS FETCH FIRST 5 ROWS ONLY",
+ literal_binds=True,
+ )
+
+ def test_fetch_offset_compound_select_literal_binds(self):
+ stmt = select(1).union(select(2)).fetch(5).offset(6)
+ self.assert_compile(
+ stmt,
+ "SELECT 1 UNION SELECT 2 OFFSET 6 ROWS FETCH FIRST 5 ROWS ONLY",
+ literal_binds=True,
+ )
+
def test_multiple_col_binds(self):
self.assert_compile(
select(literal_column("*")).where(
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
