sqlalchemy.databases.mssql.MSExecutionContext

Undocumented

Undocumented

Undocumented

return self.cursor.lastrowid, or equivalent, after an INSERT.

This may involve calling special cursor functions, issuing a new SELECT on the cursor (or a new one), or returning a stored value that was calculated within post_exec().

This function will only be called for dialects which support "implicit" primary key generation, keep preexecute_autoincrement_sequences set to False, and when no explicit id value was bound to the statement.

The function is called once for an INSERT statement that would need to return the last inserted primary key for those dialects that make use of the lastrowid concept. In these cases, it is called directly after .ExecutionContext.post_exec.

Return a result cursor strategy for a given result object.

This method is implemented by the .DefaultDialect and is only needed by implementing dialects in the case where some special steps regarding the cursor must be taken, such as manufacturing fake results from some other element of the cursor, or pre-buffering the cursor's results.

A simplified version of the default implementation is:

from sqlalchemy.engine.result import DefaultCursorFetchStrategy

class MyExecutionContext(DefaultExecutionContext):
    def get_result_cursor_strategy(self, result):
        return DefaultCursorFetchStrategy.create(result)

Above, the .DefaultCursorFetchStrategy will be applied to the result object. For results that are pre-buffered from a cursor that might be closed, an implementation might be:

from sqlalchemy.engine.result import (
    FullyBufferedCursorFetchStrategy
)

class MyExecutionContext(DefaultExecutionContext):
    _pre_buffered_result = None

    def pre_exec(self):
        if self.special_condition_prebuffer_cursor():
            self._pre_buffered_result = (
                self.cursor.description,
                self.cursor.fetchall()
            )

    def get_result_cursor_strategy(self, result):
        if self._pre_buffered_result:
            description, cursor_buffer = self._pre_buffered_result
            return (
                FullyBufferedCursorFetchStrategy.
                    create_from_buffer(
                        result, description, cursor_buffer
                )
            )
        else:
            return DefaultCursorFetchStrategy.create(result)

This method replaces the previous not-quite-documented get_result_proxy() method.