From b9b0aca7575e347dfd62221c9d515decee4c75f6 Mon Sep 17 00:00:00 2001 From: Mike Bayer Date: Wed, 19 Dec 2007 19:51:46 +0000 Subject: [PATCH] - auto-reconnect support improved; a Connection can now automatically reconnect after its underlying connection is invalidated, without needing to connect() again from the engine. This allows an ORM session bound to a single Connection to not need a reconnect. Open transactions on the Connection must be rolled back after an invalidation of the underlying connection else an error is raised. Also fixed bug where disconnect detect was not being called for cursor(), rollback(), or commit(). --- CHANGES | 9 ++ lib/sqlalchemy/databases/mysql.py | 8 +- lib/sqlalchemy/databases/sqlite.py | 3 + lib/sqlalchemy/engine/base.py | 105 +++++++++----- lib/sqlalchemy/exceptions.py | 13 +- lib/sqlalchemy/pool.py | 7 +- test/engine/reconnect.py | 215 ++++++++++++++++++++++++++--- test/sql/testtypes.py | 2 +- test/testlib/engines.py | 28 +++- 9 files changed, 331 insertions(+), 59 deletions(-) diff --git a/CHANGES b/CHANGES index 1077b4c30d..4178cb33d6 100644 --- a/CHANGES +++ b/CHANGES @@ -13,6 +13,15 @@ CHANGES knows that its return type is a Date. We only have a few functions represented so far but will continue to add to the system [ticket:615] + - auto-reconnect support improved; a Connection can now automatically + reconnect after its underlying connection is invalidated, without + needing to connect() again from the engine. This allows an ORM session + bound to a single Connection to not need a reconnect. + Open transactions on the Connection must be rolled back after an invalidation + of the underlying connection else an error is raised. Also fixed + bug where disconnect detect was not being called for cursor(), rollback(), + or commit(). + - added new flag to String and create_engine(), assert_unicode=(True|False|'warn'|None). Defaults to `False` or `None` on create_engine() and String, `'warn'` on the Unicode type. When `True`, diff --git a/lib/sqlalchemy/databases/mysql.py b/lib/sqlalchemy/databases/mysql.py index 122c24bff5..a738887f47 100644 --- a/lib/sqlalchemy/databases/mysql.py +++ b/lib/sqlalchemy/databases/mysql.py @@ -1527,8 +1527,12 @@ class MySQLDialect(default.DefaultDialect): connection.ping() def is_disconnect(self, e): - return isinstance(e, self.dbapi.OperationalError) and \ - e.args[0] in (2006, 2013, 2014, 2045, 2055) + if isinstance(e, self.dbapi.OperationalError): + return e.args[0] in (2006, 2013, 2014, 2045, 2055) + elif isinstance(e, self.dbapi.InterfaceError): # if underlying connection is closed, this is the error you get + return "(0, '')" in str(e) + else: + return False def get_default_schema_name(self, connection): try: diff --git a/lib/sqlalchemy/databases/sqlite.py b/lib/sqlalchemy/databases/sqlite.py index 16dd9427c0..e028b1c539 100644 --- a/lib/sqlalchemy/databases/sqlite.py +++ b/lib/sqlalchemy/databases/sqlite.py @@ -237,6 +237,9 @@ class SQLiteDialect(default.DefaultDialect): def oid_column_name(self, column): return "oid" + def is_disconnect(self, e): + return isinstance(e, self.dbapi.ProgrammingError) and "Cannot operate on a closed database." in str(e) + def table_names(self, connection, schema): s = "SELECT name FROM sqlite_master WHERE type='table'" return [row[0] for row in connection.execute(s)] diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py index ff2245f39e..57c76e7ed3 100644 --- a/lib/sqlalchemy/engine/base.py +++ b/lib/sqlalchemy/engine/base.py @@ -542,6 +542,7 @@ class Connection(Connectable): self.__close_with_result = close_with_result self.__savepoint_seq = 0 self.__branch = _branch + self.__invalid = False def _branch(self): """Return a new Connection which references this Connection's @@ -559,12 +560,30 @@ class Connection(Connectable): return self.engine.dialect dialect = property(dialect) + def closed(self): + """return True if this connection is closed.""" + + return not self.__invalid and '_Connection__connection' not in self.__dict__ + closed = property(closed) + + def invalidated(self): + """return True if this connection was invalidated.""" + + return self.__invalid + invalidated = property(invalidated) + def connection(self): "The underlying DB-API connection managed by this Connection." try: return self.__connection except AttributeError: + if self.__invalid: + if self.__transaction is not None: + raise exceptions.InvalidRequestError("Can't reconnect until invalid transaction is rolled back") + self.__connection = self.engine.raw_connection() + self.__invalid = False + return self.__connection raise exceptions.InvalidRequestError("This Connection is closed") connection = property(connection) @@ -603,16 +622,28 @@ class Connection(Connectable): return self - def invalidate(self): - """Invalidate and close the Connection. + def invalidate(self, exception=None): + """Invalidate the underlying DBAPI connection associated with this Connection. The underlying DB-API connection is literally closed (if possible), and is discarded. Its source connection pool will typically lazilly create a new connection to replace it. + + Upon the next usage, this Connection will attempt to reconnect + to the pool with a new connection. + + Transactions in progress remain in an "opened" state (even though + the actual transaction is gone); these must be explicitly + rolled back before a reconnect on this Connection can proceed. This + is to prevent applications from accidentally continuing their transactional + operations in a non-transactional state. + """ - self.__connection.invalidate() - self.__connection = None + if self.__connection.is_valid: + self.__connection.invalidate(exception) + del self.__connection + self.__invalid = True def detach(self): """Detach the underlying DB-API connection from its connection pool. @@ -699,29 +730,31 @@ class Connection(Connectable): if self.engine._should_log_info: self.engine.logger.info("BEGIN") try: - self.engine.dialect.do_begin(self.__connection) + self.engine.dialect.do_begin(self.connection) except Exception, e: - raise exceptions.DBAPIError.instance(None, None, e) + raise self.__handle_dbapi_exception(e, None, None, None) def _rollback_impl(self): - if self.__connection.is_valid: + if not self.closed and not self.invalidated and self.__connection.is_valid: if self.engine._should_log_info: self.engine.logger.info("ROLLBACK") try: - self.engine.dialect.do_rollback(self.__connection) + self.engine.dialect.do_rollback(self.connection) + self.__transaction = None except Exception, e: - raise exceptions.DBAPIError.instance(None, None, e) - self.__transaction = None + raise self.__handle_dbapi_exception(e, None, None, None) + else: + self.__transaction = None def _commit_impl(self): if self.engine._should_log_info: self.engine.logger.info("COMMIT") try: - self.engine.dialect.do_commit(self.__connection) + self.engine.dialect.do_commit(self.connection) + self.__transaction = None except Exception, e: - raise exceptions.DBAPIError.instance(None, None, e) - self.__transaction = None - + raise self.__handle_dbapi_exception(e, None, None, None) + def _savepoint_impl(self, name=None): if name is None: self.__savepoint_seq += 1 @@ -789,6 +822,7 @@ class Connection(Connectable): if not self.__branch: self.__connection.close() self.__connection = None + self.__invalid = False del self.__connection def scalar(self, object, *multiparams, **params): @@ -872,15 +906,32 @@ class Connection(Connectable): self._autocommit(context) return context.result() - def __create_execution_context(self, **kwargs): - return self.engine.dialect.create_execution_context(connection=self, **kwargs) - def __execute_raw(self, context): if context.executemany: self._cursor_executemany(context.cursor, context.statement, context.parameters, context=context) else: self._cursor_execute(context.cursor, context.statement, context.parameters[0], context=context) + + def __handle_dbapi_exception(self, e, statement, parameters, cursor): + if not isinstance(e, self.dialect.dbapi.Error): + return e + is_disconnect = self.dialect.is_disconnect(e) + if is_disconnect: + self.invalidate(e) + self.engine.dispose() + if cursor: + cursor.close() + self._autorollback() + if self.__close_with_result: + self.close() + return exceptions.DBAPIError.instance(statement, parameters, e, connection_invalidated=is_disconnect) + def __create_execution_context(self, **kwargs): + try: + return self.engine.dialect.create_execution_context(connection=self, **kwargs) + except Exception, e: + raise self.__handle_dbapi_exception(e, kwargs.get('statement', None), kwargs.get('parameters', None), None) + def _cursor_execute(self, cursor, statement, parameters, context=None): if self.engine._should_log_info: self.engine.logger.info(statement) @@ -888,14 +939,7 @@ class Connection(Connectable): try: self.dialect.do_execute(cursor, statement, parameters, context=context) except Exception, e: - if self.dialect.is_disconnect(e): - self.__connection.invalidate(e=e) - self.engine.dispose() - cursor.close() - self._autorollback() - if self.__close_with_result: - self.close() - raise exceptions.DBAPIError.instance(statement, parameters, e) + raise self.__handle_dbapi_exception(e, statement, parameters, cursor) def _cursor_executemany(self, cursor, statement, parameters, context=None): if self.engine._should_log_info: @@ -904,14 +948,7 @@ class Connection(Connectable): try: self.dialect.do_executemany(cursor, statement, parameters, context=context) except Exception, e: - if self.dialect.is_disconnect(e): - self.__connection.invalidate(e=e) - self.engine.dispose() - cursor.close() - self._autorollback() - if self.__close_with_result: - self.close() - raise exceptions.DBAPIError.instance(statement, parameters, e) + raise self.__handle_dbapi_exception(e, statement, parameters, cursor) # poor man's multimethod/generic function thingy executors = { @@ -990,8 +1027,8 @@ class Transaction(object): def commit(self): if not self._parent._is_active: raise exceptions.InvalidRequestError("This transaction is inactive") - self._is_active = False self._do_commit() + self._is_active = False def _do_commit(self): pass diff --git a/lib/sqlalchemy/exceptions.py b/lib/sqlalchemy/exceptions.py index 8338bc554f..530ce3e3a2 100644 --- a/lib/sqlalchemy/exceptions.py +++ b/lib/sqlalchemy/exceptions.py @@ -62,7 +62,11 @@ class NoSuchColumnError(KeyError, SQLAlchemyError): class DisconnectionError(SQLAlchemyError): - """Raised within ``Pool`` when a disconnect is detected on a raw DB-API connection.""" + """Raised within ``Pool`` when a disconnect is detected on a raw DB-API connection. + + This error is consumed internally by a connection pool. It can be raised by + a ``PoolListener`` so that the host pool forces a disconnect. + """ class DBAPIError(SQLAlchemyError): @@ -84,7 +88,7 @@ class DBAPIError(SQLAlchemyError): Its type and properties are DB-API implementation specific. """ - def instance(cls, statement, params, orig): + def instance(cls, statement, params, orig, connection_invalidated=False): # Don't ever wrap these, just return them directly as if # DBAPIError didn't exist. if isinstance(orig, (KeyboardInterrupt, SystemExit)): @@ -95,10 +99,10 @@ class DBAPIError(SQLAlchemyError): if name in glob and issubclass(glob[name], DBAPIError): cls = glob[name] - return cls(statement, params, orig) + return cls(statement, params, orig, connection_invalidated) instance = classmethod(instance) - def __init__(self, statement, params, orig): + def __init__(self, statement, params, orig, connection_invalidated=False): try: text = str(orig) except (KeyboardInterrupt, SystemExit): @@ -110,6 +114,7 @@ class DBAPIError(SQLAlchemyError): self.statement = statement self.params = params self.orig = orig + self.connection_invalidated = connection_invalidated def __str__(self): return ' '.join([SQLAlchemyError.__str__(self), diff --git a/lib/sqlalchemy/pool.py b/lib/sqlalchemy/pool.py index ff38f21b87..7a5c2ef0ed 100644 --- a/lib/sqlalchemy/pool.py +++ b/lib/sqlalchemy/pool.py @@ -208,7 +208,12 @@ class _ConnectionRecord(object): if self.connection is not None: if self.__pool._should_log_info: self.__pool.log("Closing connection %s" % repr(self.connection)) - self.connection.close() + try: + self.connection.close() + except: + if self.__pool._should_log_info: + self.__pool.log("Exception closing connection %s" % repr(self.connection)) + def invalidate(self, e=None): if self.__pool._should_log_info: diff --git a/test/engine/reconnect.py b/test/engine/reconnect.py index 7c213695f2..f9d692b3d7 100644 --- a/test/engine/reconnect.py +++ b/test/engine/reconnect.py @@ -13,37 +13,39 @@ class MockDBAPI(object): self.connections = weakref.WeakKeyDictionary() def connect(self, *args, **kwargs): return MockConnection(self) - + def shutdown(self): + for c in self.connections: + c.explode[0] = True + Error = MockDisconnect + class MockConnection(object): def __init__(self, dbapi): - self.explode = False dbapi.connections[self] = True + self.explode = [False] def rollback(self): pass def commit(self): pass def cursor(self): - return MockCursor(explode=self.explode) + return MockCursor(self) def close(self): pass class MockCursor(object): - def __init__(self, explode): - self.explode = explode + def __init__(self, parent): + self.explode = parent.explode self.description = None def execute(self, *args, **kwargs): - if self.explode: + if self.explode[0]: raise MockDisconnect("Lost the DB connection") else: return def close(self): pass -class ReconnectTest(PersistTest): - def test_reconnect(self): - """test that an 'is_disconnect' condition will invalidate the connection, and additionally - dispose the previous connection pool and recreate.""" - +class MockReconnectTest(PersistTest): + def setUp(self): + global db, dbapi dbapi = MockDBAPI() # create engine using our current dburi @@ -52,6 +54,11 @@ class ReconnectTest(PersistTest): # monkeypatch disconnect checker db.dialect.is_disconnect = lambda e: isinstance(e, MockDisconnect) + def test_reconnect(self): + """test that an 'is_disconnect' condition will invalidate the connection, and additionally + dispose the previous connection pool and recreate.""" + + pid = id(db.pool) # make a connection @@ -68,17 +75,17 @@ class ReconnectTest(PersistTest): assert len(dbapi.connections) == 2 # set it to fail - conn.connection.connection.explode = True - + dbapi.shutdown() + try: - # execute should fail conn.execute("SELECT 1") assert False - except exceptions.SQLAlchemyError, e: + except exceptions.DBAPIError: pass # assert was invalidated - assert conn.connection.connection is None + assert not conn.closed + assert conn.invalidated # close shouldnt break conn.close() @@ -92,6 +99,182 @@ class ReconnectTest(PersistTest): conn.execute("SELECT 1") conn.close() assert len(dbapi.connections) == 1 + + def test_invalidate_trans(self): + conn = db.connect() + trans = conn.begin() + dbapi.shutdown() + + try: + conn.execute("SELECT 1") + assert False + except exceptions.DBAPIError: + pass + + # assert was invalidated + assert len(dbapi.connections) == 0 + assert not conn.closed + assert conn.invalidated + assert trans.is_active + + try: + conn.execute("SELECT 1") + assert False + except exceptions.InvalidRequestError, e: + assert str(e) == "Can't reconnect until invalid transaction is rolled back" + + assert trans.is_active + + try: + trans.commit() + assert False + except exceptions.InvalidRequestError, e: + assert str(e) == "Can't reconnect until invalid transaction is rolled back" + + assert trans.is_active + + trans.rollback() + assert not trans.is_active + + conn.execute("SELECT 1") + assert not conn.invalidated + + assert len(dbapi.connections) == 1 + + def test_conn_reusable(self): + conn = db.connect() + + conn.execute("SELECT 1") + + assert len(dbapi.connections) == 1 + + dbapi.shutdown() + + # raises error + try: + conn.execute("SELECT 1") + assert False + except exceptions.DBAPIError: + pass + + assert not conn.closed + assert conn.invalidated + + # ensure all connections closed (pool was recycled) + assert len(dbapi.connections) == 0 + + # test reconnects + conn.execute("SELECT 1") + assert not conn.invalidated + assert len(dbapi.connections) == 1 + + +class RealReconnectTest(PersistTest): + def setUp(self): + global engine + engine = engines.reconnecting_engine() + + def tearDown(self): + engine.dispose() + + def test_reconnect(self): + conn = engine.connect() + + self.assertEquals(conn.execute("SELECT 1").scalar(), 1) + assert not conn.closed + + engine.test_shutdown() + + try: + conn.execute("SELECT 1") + assert False + except exceptions.DBAPIError, e: + if not e.connection_invalidated: + raise + + assert not conn.closed + assert conn.invalidated + + assert conn.invalidated + self.assertEquals(conn.execute("SELECT 1").scalar(), 1) + assert not conn.invalidated + + # one more time + engine.test_shutdown() + try: + conn.execute("SELECT 1") + assert False + except exceptions.DBAPIError, e: + if not e.connection_invalidated: + raise + assert conn.invalidated + self.assertEquals(conn.execute("SELECT 1").scalar(), 1) + assert not conn.invalidated + + conn.close() + + def test_close(self): + conn = engine.connect() + self.assertEquals(conn.execute("SELECT 1").scalar(), 1) + assert not conn.closed + + engine.test_shutdown() + + try: + conn.execute("SELECT 1") + assert False + except exceptions.DBAPIError, e: + if not e.connection_invalidated: + raise + + conn.close() + conn = engine.connect() + self.assertEquals(conn.execute("SELECT 1").scalar(), 1) + + def test_with_transaction(self): + conn = engine.connect() + + trans = conn.begin() + + self.assertEquals(conn.execute("SELECT 1").scalar(), 1) + assert not conn.closed + + engine.test_shutdown() + + try: + conn.execute("SELECT 1") + assert False + except exceptions.DBAPIError, e: + if not e.connection_invalidated: + raise + + assert not conn.closed + assert conn.invalidated + assert trans.is_active + + try: + conn.execute("SELECT 1") + assert False + except exceptions.InvalidRequestError, e: + assert str(e) == "Can't reconnect until invalid transaction is rolled back" + + assert trans.is_active + + try: + trans.commit() + assert False + except exceptions.InvalidRequestError, e: + assert str(e) == "Can't reconnect until invalid transaction is rolled back" + + assert trans.is_active + + trans.rollback() + assert not trans.is_active + + assert conn.invalidated + self.assertEquals(conn.execute("SELECT 1").scalar(), 1) + assert not conn.invalidated + if __name__ == '__main__': testbase.main() diff --git a/test/sql/testtypes.py b/test/sql/testtypes.py index 69696ec642..59acaff2ff 100644 --- a/test/sql/testtypes.py +++ b/test/sql/testtypes.py @@ -342,7 +342,7 @@ class UnicodeTest(AssertMixin): assert str(e) == "Unicode type received non-unicode bind param value 'im not unicode'" finally: unicode_engine.dispose() - + @testing.unsupported('oracle') def testblanks(self): unicode_table.insert().execute(unicode_varchar=u'') diff --git a/test/testlib/engines.py b/test/testlib/engines.py index 56507618c2..b576a15364 100644 --- a/test/testlib/engines.py +++ b/test/testlib/engines.py @@ -68,7 +68,31 @@ def close_open_connections(fn): decorated.__name__ = fn.__name__ return decorated - +class ReconnectFixture(object): + def __init__(self, dbapi): + self.dbapi = dbapi + self.connections = [] + + def __getattr__(self, key): + return getattr(self.dbapi, key) + + def connect(self, *args, **kwargs): + conn = self.dbapi.connect(*args, **kwargs) + self.connections.append(conn) + return conn + + def shutdown(self): + for c in list(self.connections): + c.close() + self.connections = [] + +def reconnecting_engine(url=None, options=None): + url = url or config.db_url + dbapi = config.db.dialect.dbapi + engine = testing_engine(url, {'module':ReconnectFixture(dbapi)}) + engine.test_shutdown = engine.dialect.dbapi.shutdown + return engine + def testing_engine(url=None, options=None): """Produce an engine configured by --options with optional overrides.""" @@ -109,3 +133,5 @@ def utf8_engine(url=None, options=None): url = str(url) return testing_engine(url, options) + + -- 2.47.3