From: Mike Bayer <mike_mp@zzzcomputing.com>
Date: Sat, 15 Oct 2011 16:10:31 +0000 (-0400)
Subject: - Added an exception catch + warning for the
X-Git-Tag: rel_0_7_3~7
X-Git-Url: http://git.ipfire.org/?a=commitdiff_plain;h=f1eea63468d8e5d84edceb2b0028984e5917dde0;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git

- Added an exception catch + warning for the
"return unicode detection" step within connect,
allows databases that crash on NVARCHAR to
continue initializing, assuming no NVARCHAR
type implemented.  [ticket:2299]
---

diff --git a/CHANGES b/CHANGES
index d072eba6ec..bcb433d786 100644
--- a/CHANGES
+++ b/CHANGES
@@ -218,6 +218,12 @@ CHANGES
     a '.' before being resolved as an entry
     point.  [ticket:2286]
 
+  - Added an exception catch + warning for the
+    "return unicode detection" step within connect,
+    allows databases that crash on NVARCHAR to
+    continue initializing, assuming no NVARCHAR
+    type implemented.  [ticket:2299]
+
 - types
   - Extra keyword arguments to the base Float 
     type beyond "precision" and "asdecimal" are ignored;
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index 9f6c5010e2..876a9f5b1e 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -209,29 +209,34 @@ class DefaultDialect(base.Dialect):
         # end Py2K
         # Py3K
         #cast_to = str
-        def check_unicode(type_):
+        def check_unicode(formatstr, type_):
             cursor = connection.connection.cursor()
             try:
-                cursor.execute(
-                    cast_to(
-                        expression.select( 
-                        [expression.cast(
-                            expression.literal_column(
-                                    "'test unicode returns'"), type_)
-                        ]).compile(dialect=self)
+                try:
+                    cursor.execute(
+                        cast_to(
+                            expression.select( 
+                            [expression.cast(
+                                expression.literal_column(
+                                        "'test %s returns'" % formatstr), type_)
+                            ]).compile(dialect=self)
+                        )
                     )
-                )
-                row = cursor.fetchone()
+                    row = cursor.fetchone()
 
-                return isinstance(row[0], unicode)
+                    return isinstance(row[0], unicode)
+                except self.dbapi.Error, de:
+                    util.warn("Exception attempting to "
+                            "detect unicode returns: %r" % de)
+                    return False
             finally:
                 cursor.close()
 
         # detect plain VARCHAR
-        unicode_for_varchar = check_unicode(sqltypes.VARCHAR(60))
+        unicode_for_varchar = check_unicode("plain", sqltypes.VARCHAR(60))
 
         # detect if there's an NVARCHAR type with different behavior available
-        unicode_for_unicode = check_unicode(sqltypes.Unicode(60))
+        unicode_for_unicode = check_unicode("unicode", sqltypes.Unicode(60))
 
         if unicode_for_unicode and not unicode_for_varchar:
             return "conditional"
diff --git a/test/engine/test_execute.py b/test/engine/test_execute.py
index 25108d7da6..3ae4204d21 100644
--- a/test/engine/test_execute.py
+++ b/test/engine/test_execute.py
@@ -239,6 +239,21 @@ class ExecuteTest(fixtures.TestBase):
         conn = eng.contextual_connect()
         eq_(conn._execution_options['foo'], 'hoho')
 
+    def test_unicode_test_fails_warning(self):
+        class MockCursor(engines.DBAPIProxyCursor):
+            def execute(self, stmt, params=None):
+                if "test unicode returns" in stmt:
+                    raise self.engine.dialect.dbapi.DatabaseError("boom")
+                else:
+                    return super(MockCursor, self).execute(stmt, params)
+        eng = engines.proxying_engine(cursor_cls=MockCursor)
+        assert_raises_message(
+            tsa.exc.SAWarning,
+            "Exception attempting to detect unicode returns",
+            eng.connect
+        )
+        assert eng.dialect.returns_unicode_strings in (True, False)
+        eng.dispose()
 
 class CompiledCacheTest(fixtures.TestBase):
     @classmethod
diff --git a/test/lib/engines.py b/test/lib/engines.py
index 1d56082673..69fdeb3370 100644
--- a/test/lib/engines.py
+++ b/test/lib/engines.py
@@ -255,6 +255,60 @@ def mock_engine(dialect_name=None):
     engine.assert_sql = assert_sql
     return engine
 
+class DBAPIProxyCursor(object):
+    """Proxy a DBAPI cursor.
+    
+    Tests can provide subclasses of this to intercept
+    DBAPI-level cursor operations.
+    
+    """
+    def __init__(self, engine, conn):
+        self.engine = engine
+        self.connection = conn
+        self.cursor = conn.cursor()
+
+    def execute(self, stmt, parameters=None):
+        if parameters:
+            return self.cursor.execute(stmt, parameters)
+        else:
+            return self.cursor.execute(stmt)
+
+    def executemany(self, stmt, params):
+        return self.cursor.executemany(stmt, params)
+
+    def __getattr__(self, key):
+        return getattr(self.cursor, key)
+
+class DBAPIProxyConnection(object):
+    """Proxy a DBAPI connection.
+    
+    Tests can provide subclasses of this to intercept
+    DBAPI-level cursor operations.
+    
+    """
+    def __init__(self, engine, cursor_cls):
+        self.conn = self._sqla_unwrap = engine.pool._creator()
+        self.engine = engine
+        self.cursor_cls = cursor_cls
+
+    def cursor(self):
+        return self.cursor_cls(self.engine, self.conn)
+
+    def close(self):
+        self.conn.close()
+
+    def __getattr__(self, key):
+        return getattr(self.conn, key)
+
+def proxying_engine(conn_cls=DBAPIProxyConnection, cursor_cls=DBAPIProxyCursor):
+    """Produce an engine that provides proxy hooks for 
+    common methods.
+    
+    """
+    def mock_conn():
+        return conn_cls(config.db, cursor_cls)
+    return testing_engine(options={'creator':mock_conn})
+
 class ReplayableSession(object):
     """A simple record/playback tool.