apply render_schema_translates to identity insert directives

Fixed bug where the "schema_translate_map" feature would fail to function
correctly in conjunction with an INSERT into a table that has an IDENTITY
column, where the value of the IDENTITY column were specified in the values
of the INSERT thus triggering SQLAlchemy's feature of setting IDENTITY
INSERT to "on"; it's in this directive where the schema translate map would
fail to be honored.

Fixes: #6658
Change-Id: I8235aa639dd465d038a2ad48e7a669f3e5c5c37c

diff --git a/doc/build/changelog/unreleased_14/6658.rst b/doc/build/changelog/unreleased_14/6658.rst
new file mode 100644 (file)
index 0000000..c0d8995
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/6658.rst
@@ -0,0 +1,11 @@
+.. change::
+    :tags: bug, mssql
+    :tickets: 6658
+
+    Fixed bug where the "schema_translate_map" feature would fail to function
+    correctly in conjunction with an INSERT into a table that has an IDENTITY
+    column, where the value of the IDENTITY column were specified in the values
+    of the INSERT thus triggering SQLAlchemy's feature of setting IDENTITY
+    INSERT to "on"; it's in this directive where the schema translate map would
+    fail to be honored.
+

diff --git a/lib/sqlalchemy/dialects/mssql/base.py b/lib/sqlalchemy/dialects/mssql/base.py
index 750f3743bbf2e47a62ceba61f4a99ba23c3c6da0..4ca83a697d2a7f4c0c7eb56d12180a888adcda77 100644 (file)
--- a/lib/sqlalchemy/dialects/mssql/base.py
+++ b/lib/sqlalchemy/dialects/mssql/base.py
@@ -1533,10 +1533,18 @@ class MSExecutionContext(default.DefaultExecutionContext):
     _result_strategy = None
 
     def _opt_encode(self, statement):
+
         if not self.dialect.supports_unicode_statements:
-            return self.dialect._encoder(statement)[0]
+            encoded = self.dialect._encoder(statement)[0]
         else:
-            return statement
+            encoded = statement
+
+        if self.compiled and self.compiled.schema_translate_map:
+
+            rst = self.compiled.preparer._render_schema_translates
+            encoded = rst(encoded, self.compiled.schema_translate_map)
+
+        return encoded
 
     def pre_exec(self):
         """Activate IDENTITY_INSERT if needed."""

diff --git a/test/dialect/mssql/test_query.py b/test/dialect/mssql/test_query.py
index d22016e8f9fa941c734851072195df3d7f130016..6952b26a9070658db9fc39b9e8248ef108a2b9f4 100644 (file)
--- a/test/dialect/mssql/test_query.py
+++ b/test/dialect/mssql/test_query.py
@@ -18,6 +18,7 @@ from sqlalchemy import testing
 from sqlalchemy import util
 from sqlalchemy.dialects.mssql import base as mssql
 from sqlalchemy.testing import AssertsCompiledSQL
+from sqlalchemy.testing import config
 from sqlalchemy.testing import engines
 from sqlalchemy.testing import eq_
 from sqlalchemy.testing import fixtures
@@ -118,6 +119,26 @@ class IdentityInsertTest(fixtures.TablesTest, AssertsCompiledSQL):
         conn.execute(cattable.insert().values({cattable.c.id: literal(5)}))
         eq_(conn.scalar(select(cattable.c.id)), 5)
 
+    @testing.requires.schemas
+    def test_insert_using_schema_translate(self, connection, metadata):
+
+        t = Table(
+            "t",
+            metadata,
+            Column("id", Integer),
+            Column("description", String(50)),
+            PrimaryKeyConstraint("id", name="PK_cattable"),
+            schema=None,
+        )
+        conn = connection.execution_options(
+            schema_translate_map={None: config.test_schema}
+        )
+        metadata.create_all(conn)
+
+        conn.execute(t.insert().values({"id": 1, "description": "descrip"}))
+
+        eq_(conn.execute(select(t)).first(), (1, "descrip"))
+
 
 class QueryUnicodeTest(fixtures.TestBase):