use the stack to get the insert statement in on conflict

Fixed bug in Postgresql :meth:`.postgresql.dml.Insert.on_conflict_do_update`
which would prevent the insert statement from being used as a CTE,
e.g. via :meth:`.Insert.cte`, within another statement.

Change-Id: Ie20972a05e194290bc9d92819750845872949ecc
Fixes: #4074
(cherry picked from commit 70516536107a44230762206342c51239c5d85417)

diff --git a/doc/build/changelog/unreleased_11/4074.rst b/doc/build/changelog/unreleased_11/4074.rst
new file mode 100644 (file)
index 0000000..f60ab1b
--- /dev/null
+++ b/doc/build/changelog/unreleased_11/4074.rst
@@ -0,0 +1,8 @@
+.. change::
+    :tags: bug, postgresql
+    :tickets: 4074
+    :versions: 1.2.0b3
+
+    Fixed bug in Postgresql :meth:`.postgresql.dml.Insert.on_conflict_do_update`
+    which would prevent the insert statement from being used as a CTE,
+    e.g. via :meth:`.Insert.cte`, within another statement.
\ No newline at end of file

diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index 7cb0ea7b0c4a3e6da8f03e50202dc5f01913f087..0ee65651de60aca229e9eeb05655bf5285702e11 100644 (file)
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -1520,7 +1520,9 @@ class PGCompiler(compiler.SQLCompiler):
 
         set_parameters = dict(clause.update_values_to_set)
         # create a list of column assignment clauses as tuples
-        cols = self.statement.table.c
+
+        insert_statement = self.stack[-1]['selectable']
+        cols = insert_statement.table.c
         for c in cols:
             col_key = c.key
             if col_key in set_parameters:

diff --git a/test/dialect/postgresql/test_compiler.py b/test/dialect/postgresql/test_compiler.py
index 0533016cd14bc33ae123a59b9d66b2eccbebe754..18940ed5f866d903d93605823d88d5b62f499365 100644 (file)
--- a/test/dialect/postgresql/test_compiler.py
+++ b/test/dialect/postgresql/test_compiler.py
@@ -1396,6 +1396,30 @@ class InsertOnConflictTest(fixtures.TestBase, AssertsCompiledSQL):
                                     "param_1": "somename",
                                     "param_2": "unknown"})
 
+    def test_on_conflict_as_cte(self):
+        i = insert(
+            self.table1, values=dict(name='foo'))
+        i = i.on_conflict_do_update(
+            constraint=self.excl_constr_anon,
+            set_=dict(name=i.excluded.name),
+            where=(
+                (self.table1.c.name != i.excluded.name))
+        ).returning(literal_column("1")).cte("i_upsert")
+
+        stmt = select([i])
+
+        self.assert_compile(
+            stmt,
+            "WITH i_upsert AS "
+            "(INSERT INTO mytable (name) VALUES (%(name)s) "
+            "ON CONFLICT (name, description) "
+            "WHERE description != %(description_1)s "
+            "DO UPDATE SET name = excluded.name "
+            "WHERE mytable.name != excluded.name RETURNING 1) "
+            "SELECT i_upsert.1 "
+            "FROM i_upsert"
+        )
+
     def test_quote_raw_string_col(self):
         t = table('t', column("FancyName"), column("other name"))