Set IDENTITY_INSERT for insert.values({column: expr})

Fixed bug where the SQL Server "IDENTITY_INSERT" logic that allows an INSERT
to proceed with an explicit value on an IDENTITY column was not detecting
the case where :meth:`.Insert.values` were used with a dictionary that
contained a :class:`.Column` as key and a SQL expression as a value.

Fixes: #4499
Change-Id: Ia61cd6524b030b40a665db9c20771f0c5aa5fcd7

diff --git a/doc/build/changelog/unreleased_12/4499.rst b/doc/build/changelog/unreleased_12/4499.rst
new file mode 100644 (file)
index 0000000..ca6be04
--- /dev/null
+++ b/doc/build/changelog/unreleased_12/4499.rst
@@ -0,0 +1,8 @@
+.. change::
+   :tags: bug, mssql
+   :tickets: 4499
+
+   Fixed bug where the SQL Server "IDENTITY_INSERT" logic that allows an INSERT
+   to proceed with an explicit value on an IDENTITY column was not detecting
+   the case where :meth:`.Insert.values` were used with a dictionary that
+   contained a :class:`.Column` as key and a SQL expression as a value.

diff --git a/lib/sqlalchemy/dialects/mssql/base.py b/lib/sqlalchemy/dialects/mssql/base.py
index 93dc9d88a0eb3d2c20ee68fd6a10b89315244f1d..49410114029382ddc29dde240dd95b945a6dd660 100644 (file)
--- a/lib/sqlalchemy/dialects/mssql/base.py
+++ b/lib/sqlalchemy/dialects/mssql/base.py
@@ -1380,13 +1380,21 @@ class MSExecutionContext(default.DefaultExecutionContext):
                     and (
                         (
                             self.compiled.statement._has_multi_parameters
-                            and seq_column.key
-                            in self.compiled.statement.parameters[0]
+                            and (
+                                seq_column.key
+                                in self.compiled.statement.parameters[0]
+                                or seq_column
+                                in self.compiled.statement.parameters[0]
+                            )
                         )
                         or (
                             not self.compiled.statement._has_multi_parameters
-                            and seq_column.key
-                            in self.compiled.statement.parameters
+                            and (
+                                seq_column.key
+                                in self.compiled.statement.parameters
+                                or seq_column
+                                in self.compiled.statement.parameters
+                            )
                         )
                     )
                 )

diff --git a/test/dialect/mssql/test_query.py b/test/dialect/mssql/test_query.py
index 46af4658d3433edac8a5ed935e882282f0828a65..bf836fc14741d05029759bd912f15843195b662b 100644 (file)
--- a/test/dialect/mssql/test_query.py
+++ b/test/dialect/mssql/test_query.py
@@ -7,6 +7,7 @@ from sqlalchemy import event
 from sqlalchemy import ForeignKey
 from sqlalchemy import func
 from sqlalchemy import Integer
+from sqlalchemy import literal
 from sqlalchemy import MetaData
 from sqlalchemy import or_
 from sqlalchemy import PrimaryKeyConstraint
@@ -172,36 +173,68 @@ class IdentityInsertTest(fixtures.TestBase, AssertsCompiledSQL):
         )
 
     def test_execute(self):
-        cattable.insert().values(id=9, description="Python").execute()
+        with testing.db.connect() as conn:
+            conn.execute(cattable.insert().values(id=9, description="Python"))
 
-        cats = cattable.select().order_by(cattable.c.id).execute()
-        eq_([(9, "Python")], list(cats))
+            cats = conn.execute(cattable.select().order_by(cattable.c.id))
+            eq_([(9, "Python")], list(cats))
 
-        result = cattable.insert().values(description="PHP").execute()
-        eq_([10], result.inserted_primary_key)
-        lastcat = cattable.select().order_by(desc(cattable.c.id)).execute()
-        eq_((10, "PHP"), lastcat.first())
+            result = conn.execute(cattable.insert().values(description="PHP"))
+            eq_([10], result.inserted_primary_key)
+            lastcat = conn.execute(
+                cattable.select().order_by(desc(cattable.c.id))
+            )
+            eq_((10, "PHP"), lastcat.first())
 
     def test_executemany(self):
-        cattable.insert().execute(
-            [
-                {"id": 89, "description": "Python"},
-                {"id": 8, "description": "Ruby"},
-                {"id": 3, "description": "Perl"},
-                {"id": 1, "description": "Java"},
-            ]
-        )
-        cats = cattable.select().order_by(cattable.c.id).execute()
-        eq_(
-            [(1, "Java"), (3, "Perl"), (8, "Ruby"), (89, "Python")], list(cats)
-        )
-        cattable.insert().execute(
-            [{"description": "PHP"}, {"description": "Smalltalk"}]
-        )
-        lastcats = (
-            cattable.select().order_by(desc(cattable.c.id)).limit(2).execute()
-        )
-        eq_([(91, "Smalltalk"), (90, "PHP")], list(lastcats))
+        with testing.db.connect() as conn:
+            conn.execute(
+                cattable.insert(),
+                [
+                    {"id": 89, "description": "Python"},
+                    {"id": 8, "description": "Ruby"},
+                    {"id": 3, "description": "Perl"},
+                    {"id": 1, "description": "Java"},
+                ],
+            )
+            cats = conn.execute(cattable.select().order_by(cattable.c.id))
+            eq_(
+                [(1, "Java"), (3, "Perl"), (8, "Ruby"), (89, "Python")],
+                list(cats),
+            )
+            conn.execute(
+                cattable.insert(),
+                [{"description": "PHP"}, {"description": "Smalltalk"}],
+            )
+            lastcats = conn.execute(
+                cattable.select().order_by(desc(cattable.c.id)).limit(2)
+            )
+            eq_([(91, "Smalltalk"), (90, "PHP")], list(lastcats))
+
+    def test_insert_plain_param(self):
+        with testing.db.connect() as conn:
+            conn.execute(cattable.insert(), id=5)
+            eq_(conn.scalar(select([cattable.c.id])), 5)
+
+    def test_insert_values_key_plain(self):
+        with testing.db.connect() as conn:
+            conn.execute(cattable.insert().values(id=5))
+            eq_(conn.scalar(select([cattable.c.id])), 5)
+
+    def test_insert_values_key_expression(self):
+        with testing.db.connect() as conn:
+            conn.execute(cattable.insert().values(id=literal(5)))
+            eq_(conn.scalar(select([cattable.c.id])), 5)
+
+    def test_insert_values_col_plain(self):
+        with testing.db.connect() as conn:
+            conn.execute(cattable.insert().values({cattable.c.id: 5}))
+            eq_(conn.scalar(select([cattable.c.id])), 5)
+
+    def test_insert_values_col_expression(self):
+        with testing.db.connect() as conn:
+            conn.execute(cattable.insert().values({cattable.c.id: literal(5)}))
+            eq_(conn.scalar(select([cattable.c.id])), 5)
 
 
 class QueryUnicodeTest(fixtures.TestBase):

diff --git a/test/orm/test_unitofwork.py b/test/orm/test_unitofwork.py
index dc8a818d244799604bb6b75132f4f6aad3237001..d6ca166004c927feb112f85638ebf037aaaa9c77 100644 (file)
--- a/test/orm/test_unitofwork.py
+++ b/test/orm/test_unitofwork.py
@@ -479,6 +479,8 @@ class ForeignPKTest(fixtures.MappedTest):
 
 
 class ClauseAttributesTest(fixtures.MappedTest):
+    __backend__ = True
+
     @classmethod
     def define_tables(cls, metadata):
         Table(