diff options
| -rw-r--r-- | doc/build/changelog/changelog_11.rst | 19 | ||||
| -rw-r--r-- | doc/build/changelog/migration_11.rst | 63 | ||||
| -rw-r--r-- | lib/sqlalchemy/engine/base.py | 53 | ||||
| -rw-r--r-- | lib/sqlalchemy/engine/default.py | 4 | ||||
| -rw-r--r-- | lib/sqlalchemy/engine/result.py | 12 | ||||
| -rw-r--r-- | lib/sqlalchemy/events.py | 10 | ||||
| -rw-r--r-- | lib/sqlalchemy/pool.py | 1 | ||||
| -rw-r--r-- | test/engine/test_reconnect.py | 94 |
8 files changed, 223 insertions, 33 deletions
diff --git a/doc/build/changelog/changelog_11.rst b/doc/build/changelog/changelog_11.rst index cbc03c559..f9b99b348 100644 --- a/doc/build/changelog/changelog_11.rst +++ b/doc/build/changelog/changelog_11.rst @@ -22,6 +22,25 @@ :version: 1.1.0 .. change:: + :tags: bug, sql, mysql + :tickets: 3803 + + The ``BaseException`` exception class is now intercepted by the + exception-handling routines of :class:`.Connection`, and includes + handling by the :meth:`~.ConnectionEvents.handle_error` + event. The :class:`.Connection` is now **invalidated** by default in + the case of a system level exception that is not a subclass of + ``Exception``, including ``KeyboardInterrupt`` and the greenlet + ``GreenletExit`` class, to prevent further operations from occurring + upon a database connection that is in an unknown and possibly + corrupted state. The MySQL drivers are most targeted by this change + however the change is across all DBAPIs. + + .. seealso:: + + :ref:`change_3803` + + .. change:: :tags: bug, sql :tickets: 3799 diff --git a/doc/build/changelog/migration_11.rst b/doc/build/changelog/migration_11.rst index ed64565d1..c85f1bf1f 100644 --- a/doc/build/changelog/migration_11.rst +++ b/doc/build/changelog/migration_11.rst @@ -1058,6 +1058,69 @@ into unions; these cases are not supported. New Features and Improvements - Core ==================================== +.. _change_3803: + +Engines now invalidate connections, run error handlers for BaseException +------------------------------------------------------------------------ + +.. versionadded:: 1.1 this change is a late add to the 1.1 series just + prior to 1.1 final, and is not present in the 1.1 beta releases. + +The Python ``BaseException`` class is below that of ``Exception`` but is the +identifiable base for system-level exceptions such as ``KeyboardInterrupt``, +``SystemExit``, and notably the ``GreenletExit`` exception that's used by +eventlet and gevent. This exception class is now intercepted by the exception- +handling routines of :class:`.Connection`, and includes handling by the +:meth:`~.ConnectionEvents.handle_error` event. The :class:`.Connection` is now +**invalidated** by default in the case of a system level exception that is not +a subclass of ``Exception``, as it is assumed an operation was interrupted and +the connection may be in an unusable state. The MySQL drivers are most +targeted by this change however the change is across all DBAPIs. + +Note that upon invalidation, the immediate DBAPI connection used by +:class:`.Connection` is disposed, and the :class:`.Connection`, if still +being used subsequent to the exception raise, will use a new +DBAPI connection for subsequent operations upon next use; however, the state of +any transaction in progress is lost and the appropriate ``.rollback()`` method +must be called if applicable before this re-use can proceed. + +In order to identify this change, it was straightforward to demonstrate a pymysql or +mysqlclient / MySQL-Python connection moving into a corrupted state when +these exceptions occur in the middle of the connection doing its work; +the connection would then be returned to the connection pool where subsequent +uses would fail, or even before returning to the pool would cause secondary +failures in context managers that call ``.rollback()`` upon the exception +catch. The behavior here is expected to reduce +the incidence of the MySQL error "commands out of sync", as well as the +``ResourceClosedError`` which can occur when the MySQL driver fails to +report ``cursor.description`` correctly, when running under greenlet +conditions where greenlets are killed, or where ``KeyboardInterrupt`` exceptions +are handled without exiting the program entirely. + +The behavior is distinct from the usual auto-invalidation feature, in that it +does not assume that the backend database itself has been shut down or +restarted; it does not recycle the entire connection pool as is the case +for usual DBAPI disconnect exceptions. + +This change should be a net improvement for all users with the exception +of **any application that currently intercepts ``KeyboardInterrupt`` or +``GreenletExit`` and wishes to continue working within the same transaction**. +Such an operation is theoretically possible with other DBAPIs that do not appear to be +impacted by ``KeyboardInterrupt`` such as psycopg2. For these DBAPIs, +the following workaround will disable the connection from being recycled +for specific exceptions:: + + + engine = create_engine("postgresql+psycopg2://") + + @event.listens_for(engine, "handle_error") + def cancel_disconnect(ctx): + if isinstance(ctx.original_exception, KeyboardInterrupt): + ctx.is_disconnect = False + +:ticket:`3803` + + .. _change_2551: CTE Support for INSERT, UPDATE, DELETE diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py index b8acf298f..83facbf1f 100644 --- a/lib/sqlalchemy/engine/base.py +++ b/lib/sqlalchemy/engine/base.py @@ -347,7 +347,7 @@ class Connection(Connectable): except AttributeError: try: return self._revalidate_connection() - except Exception as e: + except BaseException as e: self._handle_dbapi_exception(e, None, None, None, None) def get_isolation_level(self): @@ -383,7 +383,7 @@ class Connection(Connectable): """ try: return self.dialect.get_isolation_level(self.connection) - except Exception as e: + except BaseException as e: self._handle_dbapi_exception(e, None, None, None, None) @property @@ -685,7 +685,7 @@ class Connection(Connectable): self.engine.dialect.do_begin(self.connection) if self.connection._reset_agent is None: self.connection._reset_agent = transaction - except Exception as e: + except BaseException as e: self._handle_dbapi_exception(e, None, None, None, None) def _rollback_impl(self): @@ -699,7 +699,7 @@ class Connection(Connectable): self.engine.logger.info("ROLLBACK") try: self.engine.dialect.do_rollback(self.connection) - except Exception as e: + except BaseException as e: self._handle_dbapi_exception(e, None, None, None, None) finally: if not self.__invalid and \ @@ -719,7 +719,7 @@ class Connection(Connectable): self.engine.logger.info("COMMIT") try: self.engine.dialect.do_commit(self.connection) - except Exception as e: + except BaseException as e: self._handle_dbapi_exception(e, None, None, None, None) finally: if not self.__invalid and \ @@ -967,7 +967,7 @@ class Connection(Connectable): dialect = self.dialect ctx = dialect.execution_ctx_cls._init_default( dialect, self, conn) - except Exception as e: + except BaseException as e: self._handle_dbapi_exception(e, None, None, None, None) ret = ctx._exec_default(default, None) @@ -1114,7 +1114,7 @@ class Connection(Connectable): conn = self._revalidate_connection() context = constructor(dialect, self, conn, *args) - except Exception as e: + except BaseException as e: self._handle_dbapi_exception( e, util.text_type(statement), parameters, @@ -1180,7 +1180,7 @@ class Connection(Connectable): statement, parameters, context) - except Exception as e: + except BaseException as e: self._handle_dbapi_exception( e, statement, @@ -1245,7 +1245,7 @@ class Connection(Connectable): statement, parameters, context) - except Exception as e: + except BaseException as e: self._handle_dbapi_exception( e, statement, @@ -1286,18 +1286,24 @@ class Connection(Connectable): if context and context.exception is None: context.exception = e + is_exit_exception = not isinstance(e, Exception) + if not self._is_disconnect: - self._is_disconnect = \ - isinstance(e, self.dialect.dbapi.Error) and \ - not self.closed and \ + self._is_disconnect = ( + isinstance(e, self.dialect.dbapi.Error) and + not self.closed and self.dialect.is_disconnect( e, self.__connection if not self.invalidated else None, cursor) + ) or ( + is_exit_exception and not self.closed + ) + if context: context.is_disconnect = self._is_disconnect - invalidate_pool_on_disconnect = True + invalidate_pool_on_disconnect = not is_exit_exception if self._reentrant_error: util.raise_from_cause( @@ -1313,7 +1319,8 @@ class Connection(Connectable): # 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.dbapi.Error) or \ - (statement is not None and context is None) + (statement is not None + and context is None and not is_exit_exception) if should_wrap: sqlalchemy_exception = exc.DBAPIError.instance( @@ -1344,7 +1351,8 @@ class Connection(Connectable): ctx = ExceptionContextImpl( e, sqlalchemy_exception, self.engine, self, cursor, statement, - parameters, context, self._is_disconnect) + parameters, context, self._is_disconnect, + invalidate_pool_on_disconnect) for fn in self.dispatch.handle_error: try: @@ -1358,10 +1366,11 @@ class Connection(Connectable): newraise = _raised break - if sqlalchemy_exception and \ - self._is_disconnect != ctx.is_disconnect: - sqlalchemy_exception.connection_invalidated = \ - self._is_disconnect = ctx.is_disconnect + if self._is_disconnect != ctx.is_disconnect: + self._is_disconnect = ctx.is_disconnect + if sqlalchemy_exception: + sqlalchemy_exception.connection_invalidated = \ + ctx.is_disconnect # set up potentially user-defined value for # invalidate pool. @@ -1400,7 +1409,6 @@ class Connection(Connectable): @classmethod def _handle_dbapi_exception_noconnection(cls, e, dialect, engine): - exc_info = sys.exc_info() is_disconnect = dialect.is_disconnect(e, None, None) @@ -1422,7 +1430,7 @@ class Connection(Connectable): if engine._has_events: ctx = ExceptionContextImpl( e, sqlalchemy_exception, engine, None, None, None, - None, None, is_disconnect) + None, None, is_disconnect, True) for fn in engine.dispatch.handle_error: try: # handler returns an exception; @@ -1529,7 +1537,7 @@ class ExceptionContextImpl(ExceptionContext): def __init__(self, exception, sqlalchemy_exception, engine, connection, cursor, statement, parameters, - context, is_disconnect): + context, is_disconnect, invalidate_pool_on_disconnect): self.engine = engine self.connection = connection self.sqlalchemy_exception = sqlalchemy_exception @@ -1538,6 +1546,7 @@ class ExceptionContextImpl(ExceptionContext): self.statement = statement self.parameters = parameters self.is_disconnect = is_disconnect + self.invalidate_pool_on_disconnect = invalidate_pool_on_disconnect class Transaction(object): diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py index bd6e37d6c..733a89076 100644 --- a/lib/sqlalchemy/engine/default.py +++ b/lib/sqlalchemy/engine/default.py @@ -961,7 +961,7 @@ class DefaultExecutionContext(interfaces.ExecutionContext): inputsizes.append(dbtype) try: self.cursor.setinputsizes(*inputsizes) - except Exception as e: + except BaseException as e: self.root_connection._handle_dbapi_exception( e, None, None, None, self) else: @@ -979,7 +979,7 @@ class DefaultExecutionContext(interfaces.ExecutionContext): inputsizes[key] = dbtype try: self.cursor.setinputsizes(**inputsizes) - except Exception as e: + except BaseException as e: self.root_connection._handle_dbapi_exception( e, None, None, None, self) diff --git a/lib/sqlalchemy/engine/result.py b/lib/sqlalchemy/engine/result.py index 7fe09b2c7..8e5d79968 100644 --- a/lib/sqlalchemy/engine/result.py +++ b/lib/sqlalchemy/engine/result.py @@ -734,7 +734,7 @@ class ResultProxy(object): """ try: return self.context.rowcount - except Exception as e: + except BaseException as e: self.connection._handle_dbapi_exception( e, None, None, self.cursor, self.context) @@ -756,7 +756,7 @@ class ResultProxy(object): """ try: return self._saved_cursor.lastrowid - except Exception as e: + except BaseException as e: self.connection._handle_dbapi_exception( e, None, None, self._saved_cursor, self.context) @@ -1120,7 +1120,7 @@ class ResultProxy(object): l = self.process_rows(self._fetchall_impl()) self._soft_close() return l - except Exception as e: + except BaseException as e: self.connection._handle_dbapi_exception( e, None, None, self.cursor, self.context) @@ -1149,7 +1149,7 @@ class ResultProxy(object): if len(l) == 0: self._soft_close() return l - except Exception as e: + except BaseException as e: self.connection._handle_dbapi_exception( e, None, None, self.cursor, self.context) @@ -1178,7 +1178,7 @@ class ResultProxy(object): else: self._soft_close() return None - except Exception as e: + except BaseException as e: self.connection._handle_dbapi_exception( e, None, None, self.cursor, self.context) @@ -1197,7 +1197,7 @@ class ResultProxy(object): try: row = self._fetchone_impl() - except Exception as e: + except BaseException as e: self.connection._handle_dbapi_exception( e, None, None, self.cursor, self.context) diff --git a/lib/sqlalchemy/events.py b/lib/sqlalchemy/events.py index e99c5b5e5..8776f6562 100644 --- a/lib/sqlalchemy/events.py +++ b/lib/sqlalchemy/events.py @@ -741,6 +741,9 @@ class ConnectionEvents(event.Events): * read-only, low-level exception handling for logging and debugging purposes * exception re-writing + * Establishing or disabling whether a connection or the owning + connection pool is invalidated or expired in response to a + specific exception. The hook is called while the cursor from the failed operation (if any) is still open and accessible. Special cleanup operations @@ -806,6 +809,13 @@ class ConnectionEvents(event.Events): .. versionadded:: 0.9.7 Added the :meth:`.ConnectionEvents.handle_error` hook. + .. versionchanged:: 1.1 The :meth:`.handle_error` event will now + receive all exceptions that inherit from ``BaseException``, including + ``SystemExit`` and ``KeyboardInterrupt``. The setting for + :attr:`.ExceptionContext.is_disconnect` is ``True`` in this case + and the default for :attr:`.ExceptionContext.invalidate_pool_on_disconnect` + is ``False``. + .. versionchanged:: 1.0.0 The :meth:`.handle_error` event is now invoked when an :class:`.Engine` fails during the initial call to :meth:`.Engine.connect`, as well as when a diff --git a/lib/sqlalchemy/pool.py b/lib/sqlalchemy/pool.py index 1bdffc28b..09ffaca2a 100644 --- a/lib/sqlalchemy/pool.py +++ b/lib/sqlalchemy/pool.py @@ -343,6 +343,7 @@ class Pool(log.Identified): Connections with a start time prior to this pool's invalidation time will be recycled upon next checkout. """ + rec = getattr(connection, "_connection_record", None) if not rec or self._invalidate_time < rec.starttime: self._invalidate_time = time.time() diff --git a/test/engine/test_reconnect.py b/test/engine/test_reconnect.py index 0183df71b..f04311790 100644 --- a/test/engine/test_reconnect.py +++ b/test/engine/test_reconnect.py @@ -1,4 +1,4 @@ -from sqlalchemy.testing import eq_, assert_raises, assert_raises_message +from sqlalchemy.testing import eq_, ne_, assert_raises, assert_raises_message import time from sqlalchemy import ( select, MetaData, Integer, String, create_engine, pool, exc, util) @@ -10,6 +10,7 @@ from sqlalchemy.testing import engines from sqlalchemy.testing import fixtures from sqlalchemy.testing.engines import testing_engine from sqlalchemy.testing.mock import Mock, call, patch +from sqlalchemy import event class MockError(Exception): @@ -20,16 +21,28 @@ class MockDisconnect(MockError): pass +class MockExitIsh(BaseException): + pass + + def mock_connection(): def mock_cursor(): def execute(*args, **kwargs): if conn.explode == 'execute': raise MockDisconnect("Lost the DB connection on execute") - elif conn.explode in ('execute_no_disconnect', ): + elif conn.explode == 'interrupt': + conn.explode = "explode_no_disconnect" + raise MockExitIsh("Keyboard / greenlet / etc interruption") + elif conn.explode == 'interrupt_dont_break': + conn.explode = None + raise MockExitIsh("Keyboard / greenlet / etc interruption") + elif conn.explode in ('execute_no_disconnect', + 'explode_no_disconnect'): raise MockError( "something broke on execute but we didn't lose the " "connection") - elif conn.explode in ('rollback', 'rollback_no_disconnect'): + elif conn.explode in ('rollback', 'rollback_no_disconnect', + 'explode_no_disconnect'): raise MockError( "something broke on execute but we didn't lose the " "connection") @@ -385,6 +398,81 @@ class MockReconnectTest(fixtures.TestBase): c2 = engine.connect() eq_(Dialect.initialize.call_count, 1) + def test_invalidate_conn_w_contextmanager_interrupt(self): + # test [ticket:3803] + pool = self.db.pool + + conn = self.db.connect() + self.dbapi.shutdown("interrupt") + + def go(): + with conn.begin(): + conn.execute(select([1])) + + assert_raises( + MockExitIsh, + go + ) + + assert conn.invalidated + + eq_(pool._invalidate_time, 0) # pool not invalidated + + conn.execute(select([1])) + assert not conn.invalidated + + def test_invalidate_conn_interrupt_nodisconnect_workaround(self): + # test [ticket:3803] workaround for no disconnect on keyboard interrupt + + @event.listens_for(self.db, "handle_error") + def cancel_disconnect(ctx): + ctx.is_disconnect = False + + pool = self.db.pool + + conn = self.db.connect() + self.dbapi.shutdown("interrupt_dont_break") + + def go(): + with conn.begin(): + conn.execute(select([1])) + + assert_raises( + MockExitIsh, + go + ) + + assert not conn.invalidated + + eq_(pool._invalidate_time, 0) # pool not invalidated + + conn.execute(select([1])) + assert not conn.invalidated + + def test_invalidate_conn_w_contextmanager_disconnect(self): + # test [ticket:3803] change maintains old behavior + + pool = self.db.pool + + conn = self.db.connect() + self.dbapi.shutdown("execute") + + def go(): + with conn.begin(): + conn.execute(select([1])) + + assert_raises( + exc.DBAPIError, # wraps a MockDisconnect + go + ) + + assert conn.invalidated + + ne_(pool._invalidate_time, 0) # pool is invalidated + + conn.execute(select([1])) + assert not conn.invalidated + class CursorErrTest(fixtures.TestBase): # this isn't really a "reconnect" test, it's more of |