consider IDENTITY_INSERT requiring explicitness

this would use execution options:

Index: lib/sqlalchemy/dialects/mssql/base.py
===================================================================
--- lib/sqlalchemy/dialects/mssql/base.py   (revision 6795)
+++ lib/sqlalchemy/dialects/mssql/base.py   (working copy)
@@ -794,16 +794,13 @@
             seq_column = tbl._autoincrement_column
             insert_has_sequence = seq_column is not None

-            if insert_has_sequence:
-                self._enable_identity_insert = seq_column.key in self.compiled_parameters[0](0)
-            else:
-                self._enable_identity_insert = False
-            
             self._select_lastrowid = insert_has_sequence and \
                                         not self.compiled.returning and \
                                         not self._enable_identity_insert and \
                                         not self.executemany

+            self._enable_identity_insert = self.execution_options.get('identity_insert', False)
+            
             if self._enable_identity_insert:
                 self.cursor.execute("SET IDENTITY_INSERT %s ON" % 
                     self.dialect.identifier_preparer.format_table(tbl))

Usage would be like:

connection.execution_options(identity_insert=True).\
                 execute(table.insert(), id=1, foo='bar')

engine.execute(
        table.insert().
             execution_options(identity_insert=True), [...](...)
    )

# etc.

if you attempt to overwrite an IDENTITY column otherwise, we let it pass through where the DB will raise an error. We generally don't catch things that the DB will raise on anyway (might be some exceptions though).

A "persistent" "identity_insert" feature could be implemented as a context manager:

from sqlalchemy.dialects.mssql import identity_insert
conn = engine.connect()
with identity_insert(conn, "tablea", "tableb", ...):
    conn.execute(...)
    conn.execute(...)

you should be able to piggyback a begin() to put it in a transaction too:

from sqlalchemy.dialects.mssql import identity_insert
conn = engine.connect()
with identity_insert(conn, "tablea", "tableb", ...).begin():
    conn.execute(...)
    conn.execute(...)

Comments (2)