"""
# collect CTEs to tack on top of a SELECT
self.ctes = util.OrderedDict()
+ # Detect same CTE references
self.ctes_by_name = {}
+ self.level_by_ctes = {}
self.ctes_recursive = False
if self.positional:
self.cte_positional = {}
if cs._has_row_limiting_clause:
text += self._row_limit_clause(cs, **kwargs)
- if self.ctes and toplevel:
- text = self._render_cte_clause() + text
+ if self.ctes:
+ nesting_level = len(self.stack) if not toplevel else None
+ text = (
+ self._render_cte_clause(
+ nesting_level=nesting_level, include_following_stack=True
+ )
+ + text
+ )
self.stack.pop(-1)
return text
):
self._init_cte_state()
+ cte_level = len(self.stack) if cte.nesting else 1
+
kwargs["visiting_cte"] = cte
- if isinstance(cte.name, elements._truncated_label):
- cte_name = self._truncated_identifier("alias", cte.name)
- else:
- cte_name = cte.name
+
+ cte_name = cte.name
+
+ if isinstance(cte_name, elements._truncated_label):
+ cte_name = self._truncated_identifier("alias", cte_name)
is_new_cte = True
embedded_in_current_named_cte = False
- if cte_name in self.ctes_by_name:
- existing_cte = self.ctes_by_name[cte_name]
+ if cte in self.level_by_ctes:
+ cte_level = self.level_by_ctes[cte]
+
+ cte_level_name = (cte_level, cte_name)
+ if cte_level_name in self.ctes_by_name:
+ existing_cte = self.ctes_by_name[cte_level_name]
embedded_in_current_named_cte = visiting_cte is existing_cte
# we've generated a same-named CTE that we are enclosed in,
# enclosed in us - we take precedence, so
# discard the text for the "inner".
del self.ctes[existing_cte]
+ del self.level_by_ctes[existing_cte]
else:
raise exc.CompileError(
"Multiple, unrelated CTEs found with "
cte_pre_alias_name = None
if is_new_cte:
- self.ctes_by_name[cte_name] = cte
+ self.ctes_by_name[cte_level_name] = cte
if (
"autocommit" in cte.element._execution_options
)
self.ctes[cte] = text
+ self.level_by_ctes[cte] = cte_level
if asfrom:
if from_linter:
self,
select_stmt,
asfrom=False,
+ insert_into=False,
fromhints=None,
compound_index=None,
select_wraps_for=None,
if toplevel and not self.compile_state:
self.compile_state = compile_state
+ is_embedded_select = compound_index is not None or insert_into
+
# translate step for Oracle, SQL Server which often need to
# restructure the SELECT to allow for LIMIT/OFFSET and possibly
# other conditions
if per_dialect:
text += " " + self.get_statement_hint_text(per_dialect)
- if self.ctes and toplevel:
- text = self._render_cte_clause() + text
+ if self.ctes:
+ # In compound query, CTEs are shared at the compound level
+ if not is_embedded_select:
+ nesting_level = len(self.stack) if not toplevel else None
+ text = (
+ self._render_cte_clause(nesting_level=nesting_level) + text
+ )
if select_stmt._suffixes:
text += " " + self._generate_prefixes(
clause += " "
return clause
- def _render_cte_clause(self):
+ def _render_cte_clause(
+ self,
+ nesting_level=None,
+ include_following_stack=False,
+ ):
+ """
+ include_following_stack
+ Also render the nesting CTEs on the next stack. Useful for
+ SQL structures like UNION or INSERT that can wrap SELECT
+ statements containing nesting CTEs.
+ """
+ if not self.ctes:
+ return ""
+
+ if nesting_level and nesting_level > 1:
+ ctes = util.OrderedDict()
+ for cte in list(self.ctes.keys()):
+ cte_level = self.level_by_ctes[cte]
+ is_rendered_level = cte_level == nesting_level or (
+ include_following_stack and cte_level == nesting_level + 1
+ )
+ if not (cte.nesting and is_rendered_level):
+ continue
+
+ ctes[cte] = self.ctes[cte]
+
+ del self.ctes[cte]
+ del self.level_by_ctes[cte]
+
+ cte_name = cte.name
+ if isinstance(cte_name, elements._truncated_label):
+ cte_name = self._truncated_identifier("alias", cte_name)
+
+ del self.ctes_by_name[(cte_level, cte_name)]
+ else:
+ ctes = self.ctes
+
+ if not ctes:
+ return ""
+
+ ctes_recursive = any([cte.recursive for cte in ctes])
+
if self.positional:
self.positiontup = (
- sum([self.cte_positional[cte] for cte in self.ctes], [])
+ sum([self.cte_positional[cte] for cte in ctes], [])
+ self.positiontup
)
- cte_text = self.get_cte_preamble(self.ctes_recursive) + " "
- cte_text += ", \n".join([txt for txt in self.ctes.values()])
+ cte_text = self.get_cte_preamble(ctes_recursive) + " "
+ cte_text += ", \n".join([txt for txt in ctes.values()])
cte_text += "\n "
return cte_text
if insert_stmt.select is not None:
# placed here by crud.py
select_text = self.process(
- self.stack[-1]["insert_from_select"], **kw
+ self.stack[-1]["insert_from_select"], insert_into=True, **kw
)
- if self.ctes and toplevel and self.dialect.cte_follows_insert:
- text += " %s%s" % (self._render_cte_clause(), select_text)
+ if self.ctes and self.dialect.cte_follows_insert:
+ nesting_level = len(self.stack) if not toplevel else None
+ text += " %s%s" % (
+ self._render_cte_clause(
+ nesting_level=nesting_level,
+ include_following_stack=True,
+ ),
+ select_text,
+ )
else:
text += " %s" % select_text
elif not crud_params and supports_default_values:
if returning_clause and not self.returning_precedes_values:
text += " " + returning_clause
- if self.ctes and toplevel and not self.dialect.cte_follows_insert:
- text = self._render_cte_clause() + text
+ if self.ctes and not self.dialect.cte_follows_insert:
+ nesting_level = len(self.stack) if not toplevel else None
+ text = (
+ self._render_cte_clause(
+ nesting_level=nesting_level, include_following_stack=True
+ )
+ + text
+ )
self.stack.pop(-1)
update_stmt, self.returning or update_stmt._returning
)
- if self.ctes and toplevel:
- text = self._render_cte_clause() + text
+ if self.ctes:
+ nesting_level = len(self.stack) if not toplevel else None
+ text = self._render_cte_clause(nesting_level=nesting_level) + text
self.stack.pop(-1)
delete_stmt, delete_stmt._returning
)
- if self.ctes and toplevel:
- text = self._render_cte_clause() + text
+ if self.ctes:
+ nesting_level = len(self.stack) if not toplevel else None
+ text = self._render_cte_clause(nesting_level=nesting_level) + text
self.stack.pop(-1)
selectable,
name=None,
recursive=False,
+ nesting=False,
_cte_alias=None,
_restates=(),
_prefixes=None,
_suffixes=None,
):
self.recursive = recursive
+ self.nesting = nesting
self._cte_alias = _cte_alias
self._restates = _restates
if _prefixes:
self.element,
name=name,
recursive=self.recursive,
+ nesting=self.nesting,
_cte_alias=self,
_prefixes=self._prefixes,
_suffixes=self._suffixes,
self.element.union(other),
name=self.name,
recursive=self.recursive,
+ nesting=self.nesting,
_restates=self._restates + (self,),
_prefixes=self._prefixes,
_suffixes=self._suffixes,
self.element.union_all(other),
name=self.name,
recursive=self.recursive,
+ nesting=self.nesting,
_restates=self._restates + (self,),
_prefixes=self._prefixes,
_suffixes=self._suffixes,
cte = coercions.expect(roles.IsCTERole, cte)
self._independent_ctes += (cte,)
- def cte(self, name=None, recursive=False):
+ def cte(self, name=None, recursive=False, nesting=False):
r"""Return a new :class:`_expression.CTE`,
or Common Table Expression instance.
A recursive common table expression is intended to be used in
conjunction with UNION ALL in order to derive rows
from those already selected.
+ :param nesting: if ``True``, will render the CTE locally to the
+ actual statement.
+
+ .. versionadded:: 1.4.24
The following examples include two from PostgreSQL's documentation at
https://www.postgresql.org/docs/current/static/queries-with.html,
connection.execute(upsert)
+ Example 4, Nesting CTE::
+
+ value_a = select(
+ literal("root").label("n")
+ ).cte("value_a")
+
+ # A nested CTE with the same name as the root one
+ value_a_nested = select(
+ literal("nesting").label("n")
+ ).cte("value_a", nesting=True)
+
+ # Nesting CTEs takes ascendency locally
+ # over the CTEs at a higher level
+ value_b = select(value_a_nested.c.n).cte("value_b")
+
+ value_ab = select(value_a.c.n.label("a"), value_b.c.n.label("b"))
+
+ The above query will render the second CTE nested inside the first,
+ shown with inline parameters below as::
+
+ WITH
+ value_a AS
+ (SELECT 'root' AS n),
+ value_b AS
+ (WITH value_a AS
+ (SELECT 'nesting' AS n)
+ SELECT value_a.n AS n FROM value_a)
+ SELECT value_a.n AS a, value_b.n AS b
+ FROM value_a, value_b
+
.. seealso::
:meth:`_orm.Query.cte` - ORM version of
:meth:`_expression.HasCTE.cte`.
"""
- return CTE._construct(self, name=name, recursive=recursive)
+ return CTE._construct(
+ self, name=name, recursive=recursive, nesting=nesting
+ )
class Subquery(AliasedReturnsRows):
a_stmt,
"foo",
)
+
+
+class NestingCTETest(fixtures.TestBase, AssertsCompiledSQL):
+
+ __dialect__ = "default_enhanced"
+
+ def test_select_with_nesting_cte_in_cte(self):
+ nesting_cte = select(literal(1).label("inner_cte")).cte(
+ "nesting", nesting=True
+ )
+ stmt = select(
+ select(nesting_cte.c.inner_cte.label("outer_cte")).cte("cte")
+ )
+
+ self.assert_compile(
+ stmt,
+ "WITH cte AS (WITH nesting AS (SELECT :param_1 AS inner_cte) "
+ "SELECT nesting.inner_cte AS outer_cte FROM nesting) "
+ "SELECT cte.outer_cte FROM cte",
+ )
+
+ def test_nesting_cte_in_cte_with_same_name(self):
+ nesting_cte = select(literal(1).label("inner_cte")).cte(
+ "some_cte", nesting=True
+ )
+ stmt = select(
+ select(nesting_cte.c.inner_cte.label("outer_cte")).cte("some_cte")
+ )
+
+ self.assert_compile(
+ stmt,
+ "WITH some_cte AS (WITH some_cte AS "
+ "(SELECT :param_1 AS inner_cte) "
+ "SELECT some_cte.inner_cte AS outer_cte "
+ "FROM some_cte) "
+ "SELECT some_cte.outer_cte FROM some_cte",
+ )
+
+ def test_nesting_cte_at_top_level(self):
+ nesting_cte = select(literal(1).label("val")).cte(
+ "nesting_cte", nesting=True
+ )
+ cte = select(literal(2).label("val")).cte("cte")
+ stmt = select(nesting_cte.c.val, cte.c.val)
+
+ self.assert_compile(
+ stmt,
+ "WITH nesting_cte AS (SELECT :param_1 AS val)"
+ ", cte AS (SELECT :param_2 AS val)"
+ " SELECT nesting_cte.val, cte.val AS val_1 FROM nesting_cte, cte",
+ )
+
+ def test_double_nesting_cte_in_cte(self):
+ """
+ Validate that the SELECT in the 2nd nesting CTE does not render
+ the 1st CTE.
+
+ It implies that nesting CTE level is taken in account.
+ """
+ select_1_cte = select(literal(1).label("inner_cte")).cte(
+ "nesting_1", nesting=True
+ )
+ select_2_cte = select(literal(2).label("inner_cte")).cte(
+ "nesting_2", nesting=True
+ )
+
+ stmt = select(
+ select(
+ select_1_cte.c.inner_cte.label("outer_1"),
+ select_2_cte.c.inner_cte.label("outer_2"),
+ ).cte("cte")
+ )
+
+ self.assert_compile(
+ stmt,
+ "WITH cte AS ("
+ "WITH nesting_1 AS (SELECT :param_1 AS inner_cte)"
+ ", nesting_2 AS (SELECT :param_2 AS inner_cte)"
+ " SELECT nesting_1.inner_cte AS outer_1"
+ ", nesting_2.inner_cte AS outer_2"
+ " FROM nesting_1, nesting_2"
+ ") SELECT cte.outer_1, cte.outer_2 FROM cte",
+ )
+
+ def test_double_nesting_cte_with_cross_reference_in_cte(self):
+ select_1_cte = select(literal(1).label("inner_cte_1")).cte(
+ "nesting_1", nesting=True
+ )
+ select_2_cte = select(
+ (select_1_cte.c.inner_cte_1 + 1).label("inner_cte_2")
+ ).cte("nesting_2", nesting=True)
+
+ # 1 next 2
+
+ nesting_cte_1_2 = select(select_1_cte, select_2_cte).cte("cte")
+ stmt_1_2 = select(nesting_cte_1_2)
+ self.assert_compile(
+ stmt_1_2,
+ "WITH cte AS ("
+ "WITH nesting_1 AS (SELECT :param_1 AS inner_cte_1)"
+ ", nesting_2 AS (SELECT nesting_1.inner_cte_1 + :inner_cte_1_1"
+ " AS inner_cte_2 FROM nesting_1)"
+ " SELECT nesting_1.inner_cte_1 AS inner_cte_1"
+ ", nesting_2.inner_cte_2 AS inner_cte_2"
+ " FROM nesting_1, nesting_2"
+ ") SELECT cte.inner_cte_1, cte.inner_cte_2 FROM cte",
+ )
+
+ # 2 next 1
+
+ # Reorganize order with add_cte
+ nesting_cte_2_1 = (
+ select(select_2_cte, select_1_cte).add_cte(select_1_cte).cte("cte")
+ )
+ stmt_2_1 = select(nesting_cte_2_1)
+ self.assert_compile(
+ stmt_2_1,
+ "WITH cte AS ("
+ "WITH nesting_1 AS (SELECT :param_1 AS inner_cte_1)"
+ ", nesting_2 AS (SELECT nesting_1.inner_cte_1 + :inner_cte_1_1"
+ " AS inner_cte_2 FROM nesting_1)"
+ " SELECT nesting_2.inner_cte_2 AS inner_cte_2"
+ ", nesting_1.inner_cte_1 AS inner_cte_1"
+ " FROM nesting_2, nesting_1"
+ ") SELECT cte.inner_cte_2, cte.inner_cte_1 FROM cte",
+ )
+
+ def test_nesting_cte_in_nesting_cte_in_cte(self):
+ select_1_cte = select(literal(1).label("inner_cte")).cte(
+ "nesting_1", nesting=True
+ )
+ select_2_cte = select(select_1_cte.c.inner_cte.label("inner_2")).cte(
+ "nesting_2", nesting=True
+ )
+
+ stmt = select(
+ select(select_2_cte.c.inner_2.label("outer_cte")).cte("cte")
+ )
+
+ self.assert_compile(
+ stmt,
+ "WITH cte AS ("
+ "WITH nesting_2 AS ("
+ "WITH nesting_1 AS (SELECT :param_1 AS inner_cte)"
+ " SELECT nesting_1.inner_cte AS inner_2 FROM nesting_1"
+ ") SELECT nesting_2.inner_2 AS outer_cte FROM nesting_2"
+ ") SELECT cte.outer_cte FROM cte",
+ )
+
+ def test_compound_select_with_nesting_cte_in_cte(self):
+ select_1_cte = select(literal(1).label("inner_cte")).cte(
+ "nesting_1", nesting=True
+ )
+ select_2_cte = select(literal(2).label("inner_cte")).cte(
+ "nesting_2", nesting=True
+ )
+
+ nesting_cte = (
+ select(select_1_cte).union(select(select_2_cte)).subquery()
+ )
+
+ stmt = select(
+ select(nesting_cte.c.inner_cte.label("outer_cte")).cte("cte")
+ )
+
+ self.assert_compile(
+ stmt,
+ "WITH cte AS ("
+ "SELECT anon_1.inner_cte AS outer_cte FROM ("
+ "WITH nesting_1 AS (SELECT :param_1 AS inner_cte)"
+ ", nesting_2 AS (SELECT :param_2 AS inner_cte)"
+ " SELECT nesting_1.inner_cte AS inner_cte FROM nesting_1"
+ " UNION"
+ " SELECT nesting_2.inner_cte AS inner_cte FROM nesting_2"
+ ") AS anon_1"
+ ") SELECT cte.outer_cte FROM cte",
+ )
+
+ def test_nesting_cte_in_recursive_cte(self):
+ nesting_cte = select(literal(1).label("inner_cte")).cte(
+ "nesting", nesting=True
+ )
+ stmt = select(
+ select(nesting_cte.c.inner_cte.label("outer_cte")).cte(
+ "cte", recursive=True
+ )
+ )
+
+ self.assert_compile(
+ stmt,
+ "WITH RECURSIVE cte(outer_cte) AS (WITH nesting AS "
+ "(SELECT :param_1 AS inner_cte) "
+ "SELECT nesting.inner_cte AS outer_cte FROM nesting) "
+ "SELECT cte.outer_cte FROM cte",
+ )
+
+ def test_recursive_nesting_cte_in_cte(self):
+ nesting_cte = select(literal(1).label("inner_cte")).cte(
+ "nesting", nesting=True, recursive=True
+ )
+ stmt = select(
+ select(nesting_cte.c.inner_cte.label("outer_cte")).cte("cte")
+ )
+
+ self.assert_compile(
+ stmt,
+ "WITH cte AS (WITH RECURSIVE nesting(inner_cte) AS "
+ "(SELECT :param_1 AS inner_cte) "
+ "SELECT nesting.inner_cte AS outer_cte FROM nesting) "
+ "SELECT cte.outer_cte FROM cte",
+ )
+
+ def test_same_nested_cte_is_not_generated_twice(self):
+ # Same = name and query
+ nesting_cte_used_twice = select(literal(1).label("inner_cte_1")).cte(
+ "nesting_cte", nesting=True
+ )
+ select_add_cte = select(
+ (nesting_cte_used_twice.c.inner_cte_1 + 1).label("next_value")
+ ).cte("nesting_2", nesting=True)
+
+ union_cte = (
+ select(
+ (nesting_cte_used_twice.c.inner_cte_1 - 1).label("next_value")
+ )
+ .union(select(select_add_cte))
+ .cte("wrapper", nesting=True)
+ )
+
+ stmt = (
+ select(union_cte)
+ .add_cte(nesting_cte_used_twice)
+ .union(select(nesting_cte_used_twice))
+ )
+
+ self.assert_compile(
+ stmt,
+ "WITH nesting_cte AS "
+ "(SELECT :param_1 AS inner_cte_1)"
+ ", wrapper AS "
+ "(WITH nesting_2 AS "
+ "(SELECT nesting_cte.inner_cte_1 + :inner_cte_1_2 "
+ "AS next_value "
+ "FROM nesting_cte)"
+ " SELECT nesting_cte.inner_cte_1 - :inner_cte_1_1 "
+ "AS next_value "
+ "FROM nesting_cte UNION SELECT nesting_2.next_value AS next_value "
+ "FROM nesting_2)"
+ " SELECT wrapper.next_value "
+ "FROM wrapper UNION SELECT nesting_cte.inner_cte_1 "
+ "FROM nesting_cte",
+ )
+
+ def test_recursive_nesting_cte_in_recursive_cte(self):
+ nesting_cte = select(literal(1).label("inner_cte")).cte(
+ "nesting", nesting=True, recursive=True
+ )
+ stmt = select(
+ select(nesting_cte.c.inner_cte.label("outer_cte")).cte(
+ "cte", recursive=True
+ )
+ )
+
+ self.assert_compile(
+ stmt,
+ "WITH RECURSIVE cte(outer_cte) AS "
+ "(WITH RECURSIVE nesting(inner_cte) "
+ "AS (SELECT :param_1 AS inner_cte) "
+ "SELECT nesting.inner_cte AS outer_cte FROM nesting) "
+ "SELECT cte.outer_cte FROM cte",
+ )
+
+ def test_select_from_insert_cte_with_nesting(self):
+ products = table("products", column("id"), column("price"))
+
+ generator_cte = select(
+ literal(1).label("id"), literal(27.0).label("price")
+ ).cte("generator", nesting=True)
+
+ cte = (
+ products.insert()
+ .from_select(
+ [products.c.id, products.c.price],
+ select(generator_cte),
+ )
+ .returning(*products.c)
+ .cte("insert_cte")
+ )
+
+ stmt = select(cte)
+
+ assert "autocommit" not in stmt._execution_options
+
+ eq_(stmt.compile().execution_options["autocommit"], True)
+
+ self.assert_compile(
+ stmt,
+ "WITH insert_cte AS "
+ "(WITH generator AS "
+ "(SELECT :param_1 AS id, :param_2 AS price) "
+ "INSERT INTO products (id, price) "
+ "SELECT generator.id AS id, generator.price "
+ "AS price FROM generator "
+ "RETURNING products.id, products.price) "
+ "SELECT insert_cte.id, insert_cte.price "
+ "FROM insert_cte",
+ )
+ eq_(stmt.compile().isinsert, False)
+
+ def test_select_from_update_cte_with_nesting(self):
+ t1 = table("table_1", column("id"), column("price"))
+
+ generator_cte = select(
+ literal(1).label("id"), literal(27.0).label("price")
+ ).cte("generator", nesting=True)
+
+ cte = (
+ t1.update()
+ .values(price=generator_cte.c.price)
+ .where(t1.c.id == generator_cte.c.id)
+ .returning(t1.c.id, t1.c.price)
+ ).cte("update_cte")
+
+ qry = select(cte)
+
+ self.assert_compile(
+ qry,
+ "WITH update_cte AS "
+ "(WITH generator AS "
+ "(SELECT :param_1 AS id, :param_2 AS price) "
+ "UPDATE table_1 SET price=generator.price "
+ "FROM generator WHERE table_1.id = generator.id "
+ "RETURNING table_1.id, table_1.price) "
+ "SELECT update_cte.id, update_cte.price FROM update_cte",
+ )
+
+ def test_select_from_delete_cte_with_nesting(self):
+ t1 = table("table_1", column("id"), column("price"))
+
+ generator_cte = select(literal(1).label("id")).cte(
+ "generator", nesting=True
+ )
+
+ dlt = (
+ t1.delete()
+ .where(t1.c.id == generator_cte.c.id)
+ .returning(t1.c.id, t1.c.price)
+ )
+
+ cte = dlt.cte("delete_cte")
+
+ qry = select(cte)
+
+ self.assert_compile(
+ qry,
+ "WITH delete_cte AS "
+ "(WITH generator AS "
+ "(SELECT %(param_1)s AS id) "
+ "DELETE FROM table_1 USING generator "
+ "WHERE table_1.id = generator.id RETURNING table_1.id, "
+ "table_1.price) SELECT delete_cte.id, delete_cte.price "
+ "FROM delete_cte",
+ dialect="postgresql",
+ )
+
+ def test_compound_select_with_nesting_cte_in_custom_order(self):
+ select_1_cte = select(literal(1).label("inner_cte")).cte(
+ "nesting_1", nesting=True
+ )
+ select_2_cte = select(literal(2).label("inner_cte")).cte(
+ "nesting_2", nesting=True
+ )
+
+ nesting_cte = (
+ select(select_1_cte)
+ .union(select(select_2_cte))
+ # Generate "select_2_cte" first
+ .add_cte(select_2_cte)
+ .subquery()
+ )
+
+ stmt = select(
+ select(nesting_cte.c.inner_cte.label("outer_cte")).cte("cte")
+ )
+
+ self.assert_compile(
+ stmt,
+ "WITH cte AS ("
+ "SELECT anon_1.inner_cte AS outer_cte FROM ("
+ "WITH nesting_2 AS (SELECT :param_1 AS inner_cte)"
+ ", nesting_1 AS (SELECT :param_2 AS inner_cte)"
+ " SELECT nesting_1.inner_cte AS inner_cte FROM nesting_1"
+ " UNION"
+ " SELECT nesting_2.inner_cte AS inner_cte FROM nesting_2"
+ ") AS anon_1"
+ ") SELECT cte.outer_cte FROM cte",
+ )
+
+ def test_recursive_cte_referenced_multiple_times_with_nesting_cte(self):
+ rec_root = select(literal(1).label("the_value")).cte(
+ "recursive_cte", recursive=True
+ )
+
+ # Allow to reference the recursive CTE more than once
+ rec_root_ref = rec_root.select().cte(
+ "allow_multiple_ref", nesting=True
+ )
+ should_continue = select(
+ exists(
+ select(rec_root_ref.c.the_value)
+ .where(rec_root_ref.c.the_value < 10)
+ .limit(1)
+ ).label("val")
+ ).cte("should_continue", nesting=True)
+
+ rec_part_1 = select(rec_root_ref.c.the_value * 2).where(
+ should_continue.c.val != True
+ )
+ rec_part_2 = select(rec_root_ref.c.the_value * 3).where(
+ should_continue.c.val != True
+ )
+
+ rec_part = rec_part_1.add_cte(rec_root_ref).union_all(rec_part_2)
+
+ rec_cte = rec_root.union_all(rec_part)
+
+ stmt = rec_cte.select()
+
+ self.assert_compile(
+ stmt,
+ "WITH RECURSIVE recursive_cte(the_value) AS ("
+ "SELECT :param_1 AS the_value UNION ALL ("
+ "WITH allow_multiple_ref AS ("
+ "SELECT recursive_cte.the_value AS the_value FROM recursive_cte)"
+ ", should_continue AS (SELECT EXISTS ("
+ "SELECT allow_multiple_ref.the_value FROM allow_multiple_ref"
+ " WHERE allow_multiple_ref.the_value < :the_value_2"
+ " LIMIT :param_2) AS val) "
+ "SELECT allow_multiple_ref.the_value * :the_value_1 AS anon_1"
+ " FROM allow_multiple_ref, should_continue "
+ "WHERE should_continue.val != true"
+ " UNION ALL SELECT allow_multiple_ref.the_value * :the_value_3"
+ " AS anon_2 FROM allow_multiple_ref, should_continue"
+ " WHERE should_continue.val != true))"
+ " SELECT recursive_cte.the_value FROM recursive_cte",
+ )
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
