summaryrefslogtreecommitdiff
path: root/test/dialect/postgresql/test_compiler.py
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2022-05-13 16:08:34 -0400
committerMike Bayer <mike_mp@zzzcomputing.com>2022-05-15 17:27:27 -0400
commit927abc3b33f10464ed04db3d7a454faeb6e729f2 (patch)
treef7d6b727a3178c3c01ccc9e78b0f8c957ed1fefa /test/dialect/postgresql/test_compiler.py
parent257de6ebe15d3076e19f05f93c5b3c7fae25a4d3 (diff)
downloadsqlalchemy-927abc3b33f10464ed04db3d7a454faeb6e729f2.tar.gz
render col name in on conflict set clause, not given key
Fixed bug where the PostgreSQL :meth:`_postgresql.Insert.on_conflict` method and the SQLite :meth:`_sqlite.Insert.on_conflict` method would both fail to correctly accommodate a column with a separate ".key" when specifying the column using its key name in the dictionary passed to ``set_``, as well as if the :attr:`_sqlite.Insert.excluded` or :attr:`_postgresql.Insert.excluded` collection were used as the dictionary directly. Fixes: #8014 Change-Id: I67226aeedcb2c683e22405af64720cc1f990f274
Diffstat (limited to 'test/dialect/postgresql/test_compiler.py')
-rw-r--r--test/dialect/postgresql/test_compiler.py88
1 files changed, 75 insertions, 13 deletions
diff --git a/test/dialect/postgresql/test_compiler.py b/test/dialect/postgresql/test_compiler.py
index 0fe5f7066..a4c6cc087 100644
--- a/test/dialect/postgresql/test_compiler.py
+++ b/test/dialect/postgresql/test_compiler.py
@@ -2265,41 +2265,103 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
)
-class InsertOnConflictTest(fixtures.TestBase, AssertsCompiledSQL):
+class InsertOnConflictTest(fixtures.TablesTest, AssertsCompiledSQL):
__dialect__ = postgresql.dialect()
- def setup_test(self):
- self.table1 = table1 = table(
+ run_create_tables = None
+
+ @classmethod
+ def define_tables(cls, metadata):
+ cls.table1 = table1 = table(
"mytable",
column("myid", Integer),
column("name", String(128)),
column("description", String(128)),
)
- md = MetaData()
- self.table_with_metadata = Table(
+ cls.table_with_metadata = Table(
"mytable",
- md,
+ metadata,
Column("myid", Integer, primary_key=True),
Column("name", String(128)),
Column("description", String(128)),
)
- self.unique_constr = schema.UniqueConstraint(
+ cls.unique_constr = schema.UniqueConstraint(
table1.c.name, name="uq_name"
)
- self.excl_constr = ExcludeConstraint(
+ cls.excl_constr = ExcludeConstraint(
(table1.c.name, "="),
(table1.c.description, "&&"),
name="excl_thing",
)
- self.excl_constr_anon = ExcludeConstraint(
- (self.table_with_metadata.c.name, "="),
- (self.table_with_metadata.c.description, "&&"),
- where=self.table_with_metadata.c.description != "foo",
+ cls.excl_constr_anon = ExcludeConstraint(
+ (cls.table_with_metadata.c.name, "="),
+ (cls.table_with_metadata.c.description, "&&"),
+ where=cls.table_with_metadata.c.description != "foo",
)
- self.goofy_index = Index(
+ cls.goofy_index = Index(
"goofy_index", table1.c.name, postgresql_where=table1.c.name > "m"
)
+ Table(
+ "users",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("name", String(50)),
+ )
+
+ Table(
+ "users_w_key",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("name", String(50), key="name_keyed"),
+ )
+
+ @testing.combinations("control", "excluded", "dict")
+ def test_set_excluded(self, scenario):
+ """test #8014, sending all of .excluded to set"""
+
+ if scenario == "control":
+ users = self.tables.users
+
+ stmt = insert(users)
+ self.assert_compile(
+ stmt.on_conflict_do_update(
+ constraint=users.primary_key, set_=stmt.excluded
+ ),
+ "INSERT INTO users (id, name) VALUES (%(id)s, %(name)s) ON "
+ "CONFLICT (id) DO UPDATE "
+ "SET id = excluded.id, name = excluded.name",
+ )
+ else:
+ users_w_key = self.tables.users_w_key
+
+ stmt = insert(users_w_key)
+
+ if scenario == "excluded":
+ self.assert_compile(
+ stmt.on_conflict_do_update(
+ constraint=users_w_key.primary_key, set_=stmt.excluded
+ ),
+ "INSERT INTO users_w_key (id, name) "
+ "VALUES (%(id)s, %(name_keyed)s) ON "
+ "CONFLICT (id) DO UPDATE "
+ "SET id = excluded.id, name = excluded.name",
+ )
+ else:
+ self.assert_compile(
+ stmt.on_conflict_do_update(
+ constraint=users_w_key.primary_key,
+ set_={
+ "id": stmt.excluded.id,
+ "name_keyed": stmt.excluded.name_keyed,
+ },
+ ),
+ "INSERT INTO users_w_key (id, name) "
+ "VALUES (%(id)s, %(name_keyed)s) ON "
+ "CONFLICT (id) DO UPDATE "
+ "SET id = excluded.id, name = excluded.name",
+ )
+
def test_on_conflict_do_no_call_twice(self):
users = self.table1