From: Mike Bayer Date: Mon, 13 Jan 2014 15:37:15 +0000 (-0500) Subject: revert r2775c95b1ee30831216cc5 which was mostly an inadvertent commit, except for... X-Git-Tag: rel_0_9_2~65 X-Git-Url: http://git.ipfire.org/gitweb.cgi?a=commitdiff_plain;h=232e3762b7cbf425dd911ae2421512382d6024af;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git revert r2775c95b1ee30831216cc5 which was mostly an inadvertent commit, except for the changelog part --- diff --git a/doc/build/requirements.txt b/doc/build/requirements.txt index 6e2354f07d..8b98c8e457 100644 --- a/doc/build/requirements.txt +++ b/doc/build/requirements.txt @@ -1,3 +1,3 @@ mako -changelog>=0.3.4 +changelog>=0.3.2 sphinx-paramlinks>=0.2.0 diff --git a/lib/sqlalchemy/connectors/mysqldb.py b/lib/sqlalchemy/connectors/mysqldb.py index 0a752b87ea..0f250dfdbd 100644 --- a/lib/sqlalchemy/connectors/mysqldb.py +++ b/lib/sqlalchemy/connectors/mysqldb.py @@ -62,18 +62,6 @@ class MySQLDBConnector(Connector): # is overridden when pymysql is used return __import__('MySQLdb') - def _check_unicode_returns(self, connection): - # work around issue fixed in - # https://github.com/farcepest/MySQLdb1/commit/cd44524fef63bd3fcb71947392326e9742d520e8 - # unicode charset fails for a table with - additional_tests = [ - sql.collate(sql.cast( - sql.literal_column( - "'test collated returns'"), - sqltypes.TEXT), 'utf8_bin') - ] - return super(MySQLDBConnector, self)._check_unicode_returns(connection, additional_tests) - def do_executemany(self, cursor, statement, parameters, context=None): rowcount = cursor.executemany(statement, parameters) if context is not None: diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py index bcb9960b12..509d772aab 100644 --- a/lib/sqlalchemy/engine/default.py +++ b/lib/sqlalchemy/engine/default.py @@ -228,55 +228,46 @@ class DefaultDialect(interfaces.Dialect): """ return None - def _check_unicode_returns(self, connection, additional_tests=None): + def _check_unicode_returns(self, connection): if util.py2k and not self.supports_unicode_statements: cast_to = util.binary_type else: cast_to = util.text_type - if self.positional: - parameters = self.execute_sequence_format() - else: - parameters = {} - - def check_unicode(test): + def check_unicode(formatstr, type_): cursor = connection.connection.cursor() try: try: - statement = cast_to(expression.select([test]).compile(dialect=self)) - connection._cursor_execute(cursor, statement, parameters) + cursor.execute( + cast_to( + expression.select( + [expression.cast( + expression.literal_column( + "'test %s returns'" % formatstr), + type_) + ]).compile(dialect=self) + ) + ) row = cursor.fetchone() return isinstance(row[0], util.text_type) - except exc.DBAPIError as de: + except self.dbapi.Error as de: util.warn("Exception attempting to " "detect unicode returns: %r" % de) return False finally: cursor.close() - tests = [ - # detect plain VARCHAR - expression.cast( - expression.literal_column("'test plain returns'"), - sqltypes.VARCHAR(60) - ), - # detect if there's an NVARCHAR type with different behavior available - expression.cast( - expression.literal_column("'test unicode returns'"), - sqltypes.Unicode(60) - ), - ] - - if additional_tests: - tests += additional_tests - - results = set([check_unicode(test) for test in tests]) - - if results.issuperset([True, False]): + # detect plain VARCHAR + 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("unicode", sqltypes.Unicode(60)) + + if unicode_for_unicode and not unicode_for_varchar: return "conditional" else: - return results == set([True]) + return unicode_for_varchar def _check_unicode_description(self, connection): # all DBAPIs on Py2K return cursor.description as encoded, diff --git a/test/engine/test_execute.py b/test/engine/test_execute.py index d3bd3c2cda..c2479eff7a 100644 --- a/test/engine/test_execute.py +++ b/test/engine/test_execute.py @@ -1050,7 +1050,7 @@ class ResultProxyTest(fixtures.TestBase): class ExecutionOptionsTest(fixtures.TestBase): def test_dialect_conn_options(self): - engine = testing_engine("sqlite://", options=dict(_initialize=False)) + engine = testing_engine("sqlite://") engine.dialect = Mock() conn = engine.connect() c2 = conn.execution_options(foo="bar")