diff options
| -rw-r--r-- | lib/sqlalchemy/dialects/oracle/cx_oracle.py | 26 | ||||
| -rw-r--r-- | lib/sqlalchemy/types.py | 20 | ||||
| -rw-r--r-- | test/dialect/test_oracle.py | 6 | ||||
| -rw-r--r-- | test/sql/test_query.py | 3 | ||||
| -rw-r--r-- | test/sql/test_types.py | 4 |
5 files changed, 37 insertions, 22 deletions
diff --git a/lib/sqlalchemy/dialects/oracle/cx_oracle.py b/lib/sqlalchemy/dialects/oracle/cx_oracle.py index 1748a866c..f4c2e295f 100644 --- a/lib/sqlalchemy/dialects/oracle/cx_oracle.py +++ b/lib/sqlalchemy/dialects/oracle/cx_oracle.py @@ -115,24 +115,33 @@ class _LOBMixin(object): return value return process -class _OracleChar(sqltypes.CHAR): +class _NativeUnicodeMixin(object): + def result_processor(self, dialect, coltype): + # if we know cx_Oracle will return unicode, + # don't process results + if self.convert_unicode != 'force' and \ + dialect._cx_oracle_native_nvarchar and \ + coltype == dialect.dbapi.UNICODE: + return None + else: + return super(_NativeUnicodeMixin, self).result_processor(dialect, coltype) + +class _OracleChar(_NativeUnicodeMixin, sqltypes.CHAR): def get_dbapi_type(self, dbapi): return dbapi.FIXED_CHAR -class _OracleNVarChar(sqltypes.NVARCHAR): +class _OracleNVarChar(_NativeUnicodeMixin, sqltypes.NVARCHAR): def get_dbapi_type(self, dbapi): return dbapi.UNICODE - def result_processor(self, dialect, coltype): - if dialect._cx_oracle_native_nvarchar: - return None - else: - return sqltypes.NVARCHAR.result_processor(self, dialect, coltype) class _OracleText(_LOBMixin, sqltypes.Text): def get_dbapi_type(self, dbapi): return dbapi.CLOB -class _OracleUnicodeText(sqltypes.UnicodeText): +class _OracleString(_NativeUnicodeMixin, sqltypes.String): + pass + +class _OracleUnicodeText(_NativeUnicodeMixin, sqltypes.UnicodeText): def get_dbapi_type(self, dbapi): return dbapi.NCLOB @@ -184,6 +193,7 @@ colspecs = { sqltypes.Interval : _OracleInterval, oracle.INTERVAL : _OracleInterval, sqltypes.Text : _OracleText, + sqltypes.String : _OracleString, sqltypes.UnicodeText : _OracleUnicodeText, sqltypes.CHAR : _OracleChar, sqltypes.Integer : _OracleInteger, # this is only needed for OUT parameters. diff --git a/lib/sqlalchemy/types.py b/lib/sqlalchemy/types.py index 3b7027e23..53f32fb2e 100644 --- a/lib/sqlalchemy/types.py +++ b/lib/sqlalchemy/types.py @@ -653,6 +653,7 @@ class String(Concatenable, TypeEngine): return impltype( length=self.length, convert_unicode=self.convert_unicode, + unicode_error=self.unicode_error, _warn_on_bytestring=True, ) @@ -690,7 +691,7 @@ class String(Concatenable, TypeEngine): needs_convert = wants_unicode and \ (dialect.returns_unicode_strings is not True or self.convert_unicode == 'force') - + if needs_convert: to_unicode = processors.to_unicode_processor_factory( dialect.encoding, self.unicode_error) @@ -1216,7 +1217,7 @@ class SchemaType(object): if bind is None: bind = _bind_or_error(self) t = self.dialect_impl(bind.dialect) - if t is not self: + if t is not self and isinstance(t, SchemaType): t.create(bind=bind, checkfirst=checkfirst) def drop(self, bind=None, checkfirst=False): @@ -1226,27 +1227,27 @@ class SchemaType(object): if bind is None: bind = _bind_or_error(self) t = self.dialect_impl(bind.dialect) - if t is not self: + if t is not self and isinstance(t, SchemaType): t.drop(bind=bind, checkfirst=checkfirst) def _on_table_create(self, event, target, bind, **kw): t = self.dialect_impl(bind.dialect) - if t is not self: + if t is not self and isinstance(t, SchemaType): t._on_table_create(event, target, bind, **kw) def _on_table_drop(self, event, target, bind, **kw): t = self.dialect_impl(bind.dialect) - if t is not self: + if t is not self and isinstance(t, SchemaType): t._on_table_drop(event, target, bind, **kw) def _on_metadata_create(self, event, target, bind, **kw): t = self.dialect_impl(bind.dialect) - if t is not self: + if t is not self and isinstance(t, SchemaType): t._on_metadata_create(event, target, bind, **kw) def _on_metadata_drop(self, event, target, bind, **kw): t = self.dialect_impl(bind.dialect) - if t is not self: + if t is not self and isinstance(t, SchemaType): t._on_metadata_drop(event, target, bind, **kw) class Enum(String, SchemaType): @@ -1341,13 +1342,16 @@ class Enum(String, SchemaType): table.append_constraint(e) def adapt(self, impltype): - return impltype(name=self.name, + if issubclass(impltype, Enum): + return impltype(name=self.name, quote=self.quote, schema=self.schema, metadata=self.metadata, convert_unicode=self.convert_unicode, *self.enums ) + else: + return super(Enum, self).adapt(impltype) class PickleType(MutableType, TypeDecorator): """Holds Python objects. diff --git a/test/dialect/test_oracle.py b/test/dialect/test_oracle.py index fc698f28a..5b64165d8 100644 --- a/test/dialect/test_oracle.py +++ b/test/dialect/test_oracle.py @@ -459,10 +459,10 @@ class TypesTest(TestBase, AssertsCompiledSQL): (Date(), cx_oracle._OracleDate), (oracle.OracleRaw(), cx_oracle._OracleRaw), (String(), String), - (VARCHAR(), VARCHAR), + (VARCHAR(), cx_oracle._OracleString), (DATE(), DATE), - (String(50), String), - (Unicode(), Unicode), + (String(50), cx_oracle._OracleString), + (Unicode(), cx_oracle._OracleNVarChar), (Text(), cx_oracle._OracleText), (UnicodeText(), cx_oracle._OracleUnicodeText), (NCHAR(), cx_oracle._OracleNVarChar), diff --git a/test/sql/test_query.py b/test/sql/test_query.py index a9915ab29..4ccc51713 100644 --- a/test/sql/test_query.py +++ b/test/sql/test_query.py @@ -770,8 +770,7 @@ class QueryTest(TestBase): @testing.emits_warning('.*empty sequence.*') @testing.fails_on('firebird', "kinterbasdb doesn't send full type information") @testing.fails_if(lambda: - (testing.db.name, testing.db.driver) == ('mssql', 'pyodbc') - and not testing.db.dialect.freetds, + testing.against('mssql+pyodbc') and not testing.db.dialect.freetds, "not supported by Windows ODBC driver") def test_bind_in(self): users.insert().execute(user_id = 7, user_name = 'jack') diff --git a/test/sql/test_types.py b/test/sql/test_types.py index af8881a4e..6a021b96f 100644 --- a/test/sql/test_types.py +++ b/test/sql/test_types.py @@ -312,7 +312,9 @@ class UnicodeTest(TestBase, AssertsExecutionResults): def test_native_unicode(self): """assert expected values for 'native unicode' mode""" - if testing.against('mssql+pyodbc') and not testing.db.dialect.freetds: + if \ + (testing.against('mssql+pyodbc') and not testing.db.dialect.freetds) or \ + testing.against('oracle+cx_oracle'): assert testing.db.dialect.returns_unicode_strings == 'conditional' return |