Method _handle_dbapi_exception

lib/sqlalchemy/engine/base.py:2201–2345

(
        self,
        e: BaseException,
        statement: Optional[str],
        parameters: Optional[_AnyExecuteParams],
        cursor: Optional[DBAPICursor],
        context: Optional[ExecutionContext],
        is_sub_exec: bool = False,
    )

    _is_disconnect = False

    def _handle_dbapi_exception(
        self,
        e: BaseException,
        statement: Optional[str],
        parameters: Optional[_AnyExecuteParams],
        cursor: Optional[DBAPICursor],
        context: Optional[ExecutionContext],
        is_sub_exec: bool = False,
    ) -> NoReturn:
        exc_info = sys.exc_info()

        is_exit_exception = util.is_exit_exception(e)

        if not self._is_disconnect:
            self._is_disconnect = (
                isinstance(e, self.dialect.loaded_dbapi.Error)
                and not self.closed
                and self.dialect.is_disconnect(
                    e,
                    self._dbapi_connection if not self.invalidated else None,
                    cursor,
                )
            ) or (is_exit_exception and not self.closed)

        invalidate_pool_on_disconnect = not is_exit_exception

        ismulti: bool = (
            not is_sub_exec and context.executemany
            if context is not None
            else False
        )
        if self._reentrant_error:
            raise exc.DBAPIError.instance(
                statement,
                parameters,
                e,
                self.dialect.loaded_dbapi.Error,
                hide_parameters=self.engine.hide_parameters,
                dialect=self.dialect,
                ismulti=ismulti,
            ).with_traceback(exc_info[2]) from e
        self._reentrant_error = True
        try:
            # non-DBAPI error - if we already got a context,
            # or there's no string statement, don't wrap it
            should_wrap = isinstance(e, self.dialect.loaded_dbapi.Error) or (
                not isinstance(e, exc.StatementError)
                and statement is not None
                and context is None
                and not is_exit_exception
            )

            if should_wrap:
                sqlalchemy_exception = exc.DBAPIError.instance(
                    statement,
                    parameters,
                    cast(Exception, e),
                    self.dialect.loaded_dbapi.Error,