Ensure DML provides named_with_column for CTE(Alias)

Fixed bug in new CTE feature for update/insert/delete whereby
an anoymous (e.g. no name passed) :class:`.CTE` construct around
the statement would fail.  The Alias base class of CTE checks
for the "named_with_column" attribute in order to detect if
the underlying selectable has a name; UpdateBase now provides
this as False.

Change-Id: I4b0309db21379a4c0cb93085298c86da3cf840e4
Fixes: #3744

diff --git a/doc/build/changelog/changelog_11.rst b/doc/build/changelog/changelog_11.rst
index 75c434da8e3f227a2fcfa04127a57d10b7e34874..284c8e97cd0e6a26a041b56d26c0b9ec8ddc9b8f 100644 (file)
--- a/doc/build/changelog/changelog_11.rst
+++ b/doc/build/changelog/changelog_11.rst
@@ -30,6 +30,14 @@
         oninsert and onupdate values weren't called upon for the embedded
         statement.
 
+    .. change::
+        :tags: bug, sql
+        :tickets: 3744
+
+         Fixed bug in new CTE feature for update/insert/delete whereby
+         an anoymous (e.g. no name passed) :class:`.CTE` construct around
+         the statement would fail.
+
     .. change::
         :tags: bug, ext
 

diff --git a/lib/sqlalchemy/sql/dml.py b/lib/sqlalchemy/sql/dml.py
index b54b4792dead0eb92179a063432acf7bc983b57f..31e4d3d9c8ea02daa75e32b75b2736cc888c0258 100644 (file)
--- a/lib/sqlalchemy/sql/dml.py
+++ b/lib/sqlalchemy/sql/dml.py
@@ -32,6 +32,7 @@ class UpdateBase(
     _hints = util.immutabledict()
     _parameter_ordering = None
     _prefixes = ()
+    named_with_column = False
 
     def _process_colparams(self, parameters):
         def process_single(p):

diff --git a/test/sql/test_cte.py b/test/sql/test_cte.py
index aa674403e6228fe98aa864d20b7b8bf9775c96a8..35d4552571d7cfcf5f9f30ca27bfee3c3f116980 100644 (file)
--- a/test/sql/test_cte.py
+++ b/test/sql/test_cte.py
@@ -530,6 +530,38 @@ class CTETest(fixtures.TestBase, AssertsCompiledSQL):
             "upsert.quantity FROM upsert))"
         )
 
+    def test_anon_update_cte(self):
+        orders = table(
+            'orders',
+            column('region')
+        )
+        stmt = orders.update().where(orders.c.region == 'x').\
+            values(region='y').\
+            returning(orders.c.region).cte()
+
+        self.assert_compile(
+            stmt.select(),
+            "WITH anon_1 AS (UPDATE orders SET region=:region "
+            "WHERE orders.region = :region_1 RETURNING orders.region) "
+            "SELECT anon_1.region FROM anon_1"
+        )
+
+    def test_anon_insert_cte(self):
+        orders = table(
+            'orders',
+            column('region')
+        )
+        stmt = orders.insert().\
+            values(region='y').\
+            returning(orders.c.region).cte()
+
+        self.assert_compile(
+            stmt.select(),
+            "WITH anon_1 AS (INSERT INTO orders (region) "
+            "VALUES (:region) RETURNING orders.region) "
+            "SELECT anon_1.region FROM anon_1"
+        )
+
     def test_pg_example_one(self):
         products = table('products', column('id'), column('date'))
         products_log = table('products_log', column('id'), column('date'))