:tickets: 7259
Render Non-linear CTE with :meth:`_sql.CTE.union` and :meth:`_sql.CTE.union_all`
- that accept multiple clauses to render multiple UNIONs. To make it possible,
+ that accept multiple selectables to render multiple UNIONs. To make it possible,
:meth:`_sql.Select.union` and :meth:`_sql.Select.union_all` accept also multiple
- clauses.
+ selectables. It has also been extended to other similar methods like
+ :meth:`_sql.Select.except_`, :meth:`_sql.Select.except_all`,
+ :meth:`_sql.Select.intersect` and :meth:`_sql.Select.intersect_all` that accept
+ multiple selectables.
def union(self, *other, **kwargs):
"""Return a SQL ``UNION`` of this select() construct against
- the given selectable.
+ the given selectables provided as positional arguments.
+
+ The keyword arguments are forwarded to the method
+ :meth:`_expression.CompoundSelect._create_union`.
"""
return CompoundSelect._create_union(self, *other, **kwargs)
def union_all(self, *other, **kwargs):
"""Return a SQL ``UNION ALL`` of this select() construct against
- the given selectable.
+ the given selectables provided as positional arguments.
+
+ The keyword arguments are forwarded to the method
+ :meth:`_expression.CompoundSelect._create_union_all`.
"""
return CompoundSelect._create_union_all(self, *other, **kwargs)
- def except_(self, other, **kwargs):
+ def except_(self, *other, **kwargs):
"""Return a SQL ``EXCEPT`` of this select() construct against
- the given selectable.
+ the given selectable provided as positional arguments.
+
+ The keyword arguments are forwarded to the method
+ :meth:`_expression.CompoundSelect._create_except`.
"""
- return CompoundSelect._create_except(self, other, **kwargs)
+ return CompoundSelect._create_except(self, *other, **kwargs)
- def except_all(self, other, **kwargs):
+ def except_all(self, *other, **kwargs):
"""Return a SQL ``EXCEPT ALL`` of this select() construct against
- the given selectable.
+ the given selectables provided as positional arguments.
+
+ The keyword arguments are forwarded to the method
+ :meth:`_expression.CompoundSelect._create_except_all`.
"""
- return CompoundSelect._create_except_all(self, other, **kwargs)
+ return CompoundSelect._create_except_all(self, *other, **kwargs)
- def intersect(self, other, **kwargs):
+ def intersect(self, *other, **kwargs):
"""Return a SQL ``INTERSECT`` of this select() construct against
- the given selectable.
+ the given selectables provided as positional arguments.
+
+ The keyword arguments are forwarded to the method
+ :meth:`_expression.CompoundSelect._create_intersect`.
"""
- return CompoundSelect._create_intersect(self, other, **kwargs)
+ return CompoundSelect._create_intersect(self, *other, **kwargs)
- def intersect_all(self, other, **kwargs):
+ def intersect_all(self, *other, **kwargs):
"""Return a SQL ``INTERSECT ALL`` of this select() construct
- against the given selectable.
+ against the given selectables provided as positional arguments.
+
+ The keyword arguments are forwarded to the method
+ :meth:`_expression.CompoundSelect._create_intersect_all`.
"""
- return CompoundSelect._create_intersect_all(self, other, **kwargs)
+ return CompoundSelect._create_intersect_all(self, *other, **kwargs)
@property
@util.deprecated_20(
"foo",
)
- def test_multiple_recursive_unions(self):
+ def test_recursive_cte_with_multiple_union(self):
root_query = select(literal(1).label("val")).cte(
"increasing", recursive=True
)
rec_part_2 = select((root_query.c.val + 5).label("val")).where(
root_query.c.val < 15
)
- rec_query = root_query.union(rec_part_1, rec_part_2)
-
- stmt = select(rec_query)
-
+ union_rec_query = root_query.union(rec_part_1, rec_part_2)
+ union_stmt = select(union_rec_query)
self.assert_compile(
- stmt,
+ union_stmt,
"WITH RECURSIVE increasing(val) AS "
"(SELECT :param_1 AS val "
"UNION SELECT increasing.val + :val_1 AS val FROM increasing "
"SELECT increasing.val FROM increasing",
)
+ def test_recursive_cte_with_multiple_union_all(self):
+ root_query = select(literal(1).label("val")).cte(
+ "increasing", recursive=True
+ )
+ rec_part_1 = select((root_query.c.val + 3).label("val")).where(
+ root_query.c.val < 15
+ )
+ rec_part_2 = select((root_query.c.val + 5).label("val")).where(
+ root_query.c.val < 15
+ )
+
+ union_all_rec_query = root_query.union_all(rec_part_1, rec_part_2)
+ union_all_stmt = select(union_all_rec_query)
+ self.assert_compile(
+ union_all_stmt,
+ "WITH RECURSIVE increasing(val) AS "
+ "(SELECT :param_1 AS val "
+ "UNION ALL SELECT increasing.val + :val_1 AS val FROM increasing "
+ "WHERE increasing.val < :val_2 "
+ "UNION ALL SELECT increasing.val + :val_3 AS val FROM increasing "
+ "WHERE increasing.val < :val_4) "
+ "SELECT increasing.val FROM increasing",
+ )
+
class NestingCTETest(fixtures.TestBase, AssertsCompiledSQL):
from sqlalchemy import tuple_
from sqlalchemy import union
from sqlalchemy.sql import column
+from sqlalchemy.sql import literal
from sqlalchemy.sql import table
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import AssertsCompiledSQL
"SELECT anon_1.name FROM (SELECT mytable.name AS name, "
"(mytable.myid, mytable.name) AS anon_2 FROM mytable) AS anon_1",
)
+
+ def test_select_multiple_union_all(self):
+ stmt_union = select(literal(1)).union_all(
+ select(literal(2)), select(literal(3))
+ )
+
+ self.assert_compile(
+ stmt_union,
+ "SELECT :param_1 AS anon_1"
+ " UNION ALL SELECT :param_2 AS anon_2"
+ " UNION ALL SELECT :param_3 AS anon_3",
+ )
+
+ def test_select_multiple_union(self):
+ stmt_union = select(literal(1)).union(
+ select(literal(2)), select(literal(3))
+ )
+
+ self.assert_compile(
+ stmt_union,
+ "SELECT :param_1 AS anon_1"
+ " UNION SELECT :param_2 AS anon_2"
+ " UNION SELECT :param_3 AS anon_3",
+ )
+
+ def test_select_multiple_except(self):
+ stmt_union = select(literal(1)).except_(
+ select(literal(2)), select(literal(3))
+ )
+
+ self.assert_compile(
+ stmt_union,
+ "SELECT :param_1 AS anon_1"
+ " EXCEPT SELECT :param_2 AS anon_2"
+ " EXCEPT SELECT :param_3 AS anon_3",
+ )
+
+ def test_select_multiple_except_all(self):
+ stmt_union = select(literal(1)).except_all(
+ select(literal(2)), select(literal(3))
+ )
+
+ self.assert_compile(
+ stmt_union,
+ "SELECT :param_1 AS anon_1"
+ " EXCEPT ALL SELECT :param_2 AS anon_2"
+ " EXCEPT ALL SELECT :param_3 AS anon_3",
+ )
+
+ def test_select_multiple_intersect(self):
+ stmt_union = select(literal(1)).intersect(
+ select(literal(2)), select(literal(3))
+ )
+
+ self.assert_compile(
+ stmt_union,
+ "SELECT :param_1 AS anon_1"
+ " INTERSECT SELECT :param_2 AS anon_2"
+ " INTERSECT SELECT :param_3 AS anon_3",
+ )
+
+ def test_select_multiple_intersect_all(self):
+ stmt_union = select(literal(1)).intersect_all(
+ select(literal(2)), select(literal(3))
+ )
+
+ self.assert_compile(
+ stmt_union,
+ "SELECT :param_1 AS anon_1"
+ " INTERSECT ALL SELECT :param_2 AS anon_2"
+ " INTERSECT ALL SELECT :param_3 AS anon_3",
+ )