- added is_disconnect() support for oracle

- fixed _handle_dbapi_error to detect endless loops, doesn't call rollback/cursor.close
etc. in case of disconnect

5 files changed, 66 insertions, 44 deletions

diff --git a/CHANGES b/CHANGES
index 4f9cf26f9..402ef6e6e 100644
--- a/CHANGES
+++ b/CHANGES
@@ -270,7 +270,10 @@ CHANGES
 
    - sqlite SLDate type will not erroneously render "microseconds" portion 
      of a datetime or time object.
-     
+   
+   - oracle
+      - added disconnect detection support for Oracle
+   
    - MSSQL
       - PyODBC no longer has a global "set nocount on".
       - Fix non-identity integer PKs on autload [ticket:824]
diff --git a/lib/sqlalchemy/databases/oracle.py b/lib/sqlalchemy/databases/oracle.py
index 9c9c54bf6..c57ed006c 100644
--- a/lib/sqlalchemy/databases/oracle.py
+++ b/lib/sqlalchemy/databases/oracle.py
@@ -311,6 +311,12 @@ class OracleDialect(default.DefaultDialect):
 
         return ([], opts)
 
+    def is_disconnect(self, e):
+        if isinstance(e, self.dbapi.InterfaceError):
+            return "not connected" in str(e)
+        else:
+            return "ORA-03114" in str(e) or "ORA-03113" in str(e)
+
     def type_descriptor(self, typeobj):
         return sqltypes.adapt_type(typeobj, colspecs)
 
diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py
index 57c76e7ed..83f22729c 100644
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -732,7 +732,7 @@ class Connection(Connectable):
         try:
             self.engine.dialect.do_begin(self.connection)
         except Exception, e:
-            raise self.__handle_dbapi_exception(e, None, None, None)
+            raise self._handle_dbapi_exception(e, None, None, None)
 
     def _rollback_impl(self):
         if not self.closed and not self.invalidated and self.__connection.is_valid:
@@ -742,7 +742,7 @@ class Connection(Connectable):
                 self.engine.dialect.do_rollback(self.connection)
                 self.__transaction = None
             except Exception, e:
-                raise self.__handle_dbapi_exception(e, None, None, None)
+                raise self._handle_dbapi_exception(e, None, None, None)
         else:
             self.__transaction = None
 
@@ -753,7 +753,7 @@ class Connection(Connectable):
             self.engine.dialect.do_commit(self.connection)
             self.__transaction = None
         except Exception, e:
-            raise self.__handle_dbapi_exception(e, None, None, None)
+            raise self._handle_dbapi_exception(e, None, None, None)
         
     def _savepoint_impl(self, name=None):
         if name is None:
@@ -912,25 +912,32 @@ class Connection(Connectable):
         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 _handle_dbapi_exception(self, e, statement, parameters, cursor):
+        if getattr(self, '_reentrant_error', False):
+            return exceptions.DBAPIError.instance(None, None, e)
+        self._reentrant_error = True
+        try:
+            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()
+            else:
+                if cursor:
+                    cursor.close()
+                self._autorollback()
+                if self.__close_with_result:
+                    self.close()
+            return exceptions.DBAPIError.instance(statement, parameters, e, connection_invalidated=is_disconnect)
+        finally:
+            del self._reentrant_error
         
     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)
+            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:
@@ -939,7 +946,7 @@ class Connection(Connectable):
         try:
             self.dialect.do_execute(cursor, statement, parameters, context=context)
         except Exception, e:
-            raise self.__handle_dbapi_exception(e, statement, parameters, cursor)
+            raise self._handle_dbapi_exception(e, statement, parameters, cursor)
 
     def _cursor_executemany(self, cursor, statement, parameters, context=None):
         if self.engine._should_log_info:
@@ -948,7 +955,7 @@ class Connection(Connectable):
         try:
             self.dialect.do_executemany(cursor, statement, parameters, context=context)
         except Exception, e:
-            raise self.__handle_dbapi_exception(e, statement, parameters, cursor)
+            raise self._handle_dbapi_exception(e, statement, parameters, cursor)
 
     # poor man's multimethod/generic function thingy
     executors = {
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index b06f862f7..d113bc50f 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -334,7 +334,10 @@ class DefaultExecutionContext(base.ExecutionContext):
                dbtype = typeengine.dialect_impl(self.dialect).get_dbapi_type(self.dialect.dbapi)
                if dbtype is not None:
                     inputsizes.append(dbtype)
-            self.cursor.setinputsizes(*inputsizes)
+            try:
+                self.cursor.setinputsizes(*inputsizes)
+            except Exception, e:
+                raise self._connection._handle_dbapi_exception(e, None, None, None)
         else:
             inputsizes = {}
             for key in self.compiled.bind_names.values():
@@ -342,7 +345,10 @@ class DefaultExecutionContext(base.ExecutionContext):
                 dbtype = typeengine.dialect_impl(self.dialect).get_dbapi_type(self.dialect.dbapi)
                 if dbtype is not None:
                     inputsizes[key.encode(self.dialect.encoding)] = dbtype
-            self.cursor.setinputsizes(**inputsizes)
+            try:
+                self.cursor.setinputsizes(**inputsizes)
+            except Exception, e:
+                raise self._connection._handle_dbapi_exception(e, None, None, None)
 
     def __process_defaults(self):
         """generate default values for compiled insert/update statements,
diff --git a/test/engine/reconnect.py b/test/engine/reconnect.py
index f9d692b3d..40b838e3f 100644
--- a/test/engine/reconnect.py
+++ b/test/engine/reconnect.py
@@ -1,6 +1,6 @@
 import testbase
 import sys, weakref
-from sqlalchemy import create_engine, exceptions
+from sqlalchemy import create_engine, exceptions, select
 from testlib import *
 
 
@@ -65,7 +65,7 @@ class MockReconnectTest(PersistTest):
         conn = db.connect()
         
         # connection works
-        conn.execute("SELECT 1")
+        conn.execute(select([1]))
         
         # create a second connection within the pool, which we'll ensure also goes away
         conn2 = db.connect()
@@ -78,7 +78,7 @@ class MockReconnectTest(PersistTest):
         dbapi.shutdown()
 
         try:
-            conn.execute("SELECT 1")
+            conn.execute(select([1]))
             assert False
         except exceptions.DBAPIError:
             pass
@@ -96,7 +96,7 @@ class MockReconnectTest(PersistTest):
         assert len(dbapi.connections) == 0
         
         conn =db.connect()
-        conn.execute("SELECT 1")
+        conn.execute(select([1]))
         conn.close()
         assert len(dbapi.connections) == 1
     
@@ -106,7 +106,7 @@ class MockReconnectTest(PersistTest):
         dbapi.shutdown()
 
         try:
-            conn.execute("SELECT 1")
+            conn.execute(select([1]))
             assert False
         except exceptions.DBAPIError:
             pass
@@ -118,7 +118,7 @@ class MockReconnectTest(PersistTest):
         assert trans.is_active
 
         try:
-            conn.execute("SELECT 1")
+            conn.execute(select([1]))
             assert False
         except exceptions.InvalidRequestError, e:
             assert str(e) == "Can't reconnect until invalid transaction is rolled back"
@@ -136,7 +136,7 @@ class MockReconnectTest(PersistTest):
         trans.rollback()
         assert not trans.is_active
         
-        conn.execute("SELECT 1")
+        conn.execute(select([1]))
         assert not conn.invalidated
         
         assert len(dbapi.connections) == 1
@@ -144,7 +144,7 @@ class MockReconnectTest(PersistTest):
     def test_conn_reusable(self):
         conn = db.connect()
         
-        conn.execute("SELECT 1")
+        conn.execute(select([1]))
 
         assert len(dbapi.connections) == 1
         
@@ -152,7 +152,7 @@ class MockReconnectTest(PersistTest):
 
         # raises error
         try:
-            conn.execute("SELECT 1")
+            conn.execute(select([1]))
             assert False
         except exceptions.DBAPIError:
             pass
@@ -164,7 +164,7 @@ class MockReconnectTest(PersistTest):
         assert len(dbapi.connections) == 0
             
         # test reconnects
-        conn.execute("SELECT 1")
+        conn.execute(select([1]))
         assert not conn.invalidated
         assert len(dbapi.connections) == 1
         
@@ -180,13 +180,13 @@ class RealReconnectTest(PersistTest):
     def test_reconnect(self):
         conn = engine.connect()
 
-        self.assertEquals(conn.execute("SELECT 1").scalar(), 1) 
+        self.assertEquals(conn.execute(select([1])).scalar(), 1) 
         assert not conn.closed
 
         engine.test_shutdown()
 
         try:
-            conn.execute("SELECT 1")
+            conn.execute(select([1]))
             assert False
         except exceptions.DBAPIError, e:
             if not e.connection_invalidated:
@@ -196,32 +196,32 @@ class RealReconnectTest(PersistTest):
         assert conn.invalidated
 
         assert conn.invalidated
-        self.assertEquals(conn.execute("SELECT 1").scalar(), 1) 
+        self.assertEquals(conn.execute(select([1])).scalar(), 1) 
         assert not conn.invalidated
 
         # one more time
         engine.test_shutdown()
         try:
-            conn.execute("SELECT 1")
+            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) 
+        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) 
+        self.assertEquals(conn.execute(select([1])).scalar(), 1) 
         assert not conn.closed
 
         engine.test_shutdown()
 
         try:
-            conn.execute("SELECT 1")
+            conn.execute(select([1]))
             assert False
         except exceptions.DBAPIError, e:
             if not e.connection_invalidated:
@@ -229,20 +229,20 @@ class RealReconnectTest(PersistTest):
 
         conn.close()
         conn = engine.connect()
-        self.assertEquals(conn.execute("SELECT 1").scalar(), 1) 
+        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) 
+        self.assertEquals(conn.execute(select([1])).scalar(), 1) 
         assert not conn.closed
 
         engine.test_shutdown()
 
         try:
-            conn.execute("SELECT 1")
+            conn.execute(select([1]))
             assert False
         except exceptions.DBAPIError, e:
             if not e.connection_invalidated:
@@ -253,7 +253,7 @@ class RealReconnectTest(PersistTest):
         assert trans.is_active
 
         try:
-            conn.execute("SELECT 1")
+            conn.execute(select([1]))
             assert False
         except exceptions.InvalidRequestError, e:
             assert str(e) == "Can't reconnect until invalid transaction is rolled back"
@@ -272,7 +272,7 @@ class RealReconnectTest(PersistTest):
         assert not trans.is_active
 
         assert conn.invalidated
-        self.assertEquals(conn.execute("SELECT 1").scalar(), 1) 
+        self.assertEquals(conn.execute(select([1])).scalar(), 1) 
         assert not conn.invalidated