Disable IDENTITY_INSERT if enabled.
(self)
| 1927 | ) |
| 1928 | |
| 1929 | def post_exec(self): |
| 1930 | """Disable IDENTITY_INSERT if enabled.""" |
| 1931 | |
| 1932 | conn = self.root_connection |
| 1933 | |
| 1934 | if self.isinsert or self.isupdate or self.isdelete: |
| 1935 | self._rowcount = self.cursor.rowcount |
| 1936 | |
| 1937 | if self._select_lastrowid: |
| 1938 | if self.dialect.use_scope_identity: |
| 1939 | conn._cursor_execute( |
| 1940 | self.cursor, |
| 1941 | "SELECT scope_identity() AS lastrowid", |
| 1942 | (), |
| 1943 | self, |
| 1944 | ) |
| 1945 | else: |
| 1946 | conn._cursor_execute( |
| 1947 | self.cursor, "SELECT @@identity AS lastrowid", (), self |
| 1948 | ) |
| 1949 | # fetchall() ensures the cursor is consumed without closing it |
| 1950 | row = self.cursor.fetchall()[0] |
| 1951 | self._lastrowid = int(row[0]) |
| 1952 | |
| 1953 | self.cursor_fetch_strategy = _cursor._NO_CURSOR_DML |
| 1954 | elif ( |
| 1955 | self.compiled is not None |
| 1956 | and is_sql_compiler(self.compiled) |
| 1957 | and self.compiled.effective_returning |
| 1958 | ): |
| 1959 | self.cursor_fetch_strategy = ( |
| 1960 | _cursor.FullyBufferedCursorFetchStrategy( |
| 1961 | self.cursor, |
| 1962 | self.cursor.description, |
| 1963 | self.cursor.fetchall(), |
| 1964 | ) |
| 1965 | ) |
| 1966 | |
| 1967 | if self._enable_identity_insert: |
| 1968 | if TYPE_CHECKING: |
| 1969 | assert is_sql_compiler(self.compiled) |
| 1970 | assert isinstance(self.compiled.compile_state, DMLState) |
| 1971 | assert isinstance( |
| 1972 | self.compiled.compile_state.dml_table, TableClause |
| 1973 | ) |
| 1974 | conn._cursor_execute( |
| 1975 | self.cursor, |
| 1976 | self._opt_encode( |
| 1977 | "SET IDENTITY_INSERT %s OFF" |
| 1978 | % self.identifier_preparer.format_table( |
| 1979 | self.compiled.compile_state.dml_table |
| 1980 | ) |
| 1981 | ), |
| 1982 | (), |
| 1983 | self, |
| 1984 | ) |
| 1985 | |
| 1986 | def get_lastrowid(self): |
nothing calls this directly
no test coverage detected