Method pre_exec

lib/sqlalchemy/dialects/mssql/base.py:1857–1916

Activate IDENTITY_INSERT if needed.

(self)

        return statement

    def pre_exec(self):
        """Activate IDENTITY_INSERT if needed."""

        if self.isinsert:
            if TYPE_CHECKING:
                assert is_sql_compiler(self.compiled)
                assert isinstance(self.compiled.compile_state, DMLState)
                assert isinstance(
                    self.compiled.compile_state.dml_table, TableClause
                )

            tbl = self.compiled.compile_state.dml_table
            id_column = tbl._autoincrement_column

            if id_column is not None and (
                not isinstance(id_column.default, Sequence)
            ):
                insert_has_identity = True
                compile_state = self.compiled.dml_compile_state
                self._enable_identity_insert = (
                    id_column.key in self.compiled_parameters[0]
                ) or (
                    compile_state._dict_parameters
                    and (id_column.key in compile_state._insert_col_keys)
                )

            else:
                insert_has_identity = False
                self._enable_identity_insert = False

            self._select_lastrowid = (
                not self.compiled.inline
                and insert_has_identity
                and not self.compiled.effective_returning
                and not self._enable_identity_insert
                and not self.executemany
            )

            if self._enable_identity_insert:
                self.root_connection._cursor_execute(
                    self.cursor,
                    self._opt_encode(
                        "SET IDENTITY_INSERT %s ON"
                        % self.identifier_preparer.format_table(tbl)
                    ),
                    (),
                    self,
                )

            # don't embed the scope_identity select into an
            # "INSERT .. DEFAULT VALUES"
            if (
                self._select_lastrowid
                and self.dialect.scope_identity_must_be_embedded
                and self.dialect.use_scope_identity
                and len(self.parameters[0])
            ):
                self._embedded_scope_identity = True