def get_select_hint_text(self, byfroms):
return " ".join(
- "/*+ %s */" % text for table, text in list(byfroms.items())
+ "/*+ %s */" % text for table, text in byfroms.items()
)
def function_argspec(self, fn, **kw):
lc_value = value.lower()
return (lc_value in self.reserved_words
or value[0] in self.illegal_initial_characters
- or not self.legal_characters.match(str(value))
+ or not self.legal_characters.match(util.text_type(value))
)
def format_savepoint(self, savepoint):
def normalize_name(self, name):
if name is None:
return None
-# start Py2K
-# if isinstance(name, str):
-# name = name.decode(self.encoding)
-# end Py2K
+ if util.py2k:
+ if isinstance(name, str):
+ name = name.decode(self.encoding)
if name.upper() == name and \
not self.identifier_preparer._requires_quotes(name.lower()):
return name.lower()
return None
elif name.lower() == name and not self.identifier_preparer._requires_quotes(name.lower()):
name = name.upper()
-# start Py2K
-# if not self.supports_unicode_binds:
-# name = name.encode(self.encoding)
-# else:
-# name = unicode(name)
-# end Py2K
+ if util.py2k:
+ if not self.supports_unicode_binds:
+ name = name.encode(self.encoding)
+ else:
+ name = unicode(name)
return name
def _get_default_schema_name(self, connection):
class _NativeUnicodeMixin(object):
-# start Py3K
- pass
-# end Py3K
-# start Py2K
-# def bind_processor(self, dialect):
-# if dialect._cx_oracle_with_unicode:
-# def process(value):
-# if value is None:
-# return value
-# else:
-# return unicode(value)
-# return process
-# else:
-# return super(_NativeUnicodeMixin, self).bind_processor(dialect)
-# end Py2K
+ if util.py2k:
+ def bind_processor(self, dialect):
+ if dialect._cx_oracle_with_unicode:
+ def process(value):
+ if value is None:
+ return value
+ else:
+ return unicode(value)
+ return process
+ else:
+ return super(_NativeUnicodeMixin, self).bind_processor(dialect)
# we apply a connection output handler that returns
# unicode in all cases, so the "native_unicode" flag
(fromname.encode(self.dialect.encoding),
toname.encode(self.dialect.encoding))
for fromname, toname in
- list(quoted_bind_names.items())
+ quoted_bind_names.items()
)
for param in self.parameters:
- for fromname, toname in list(quoted_bind_names.items()):
+ for fromname, toname in quoted_bind_names.items():
param[toname] = param[fromname]
del param[fromname]
# if a single execute, check for outparams
if len(self.compiled_parameters) == 1:
- for bindparam in list(self.compiled.binds.values()):
+ for bindparam in self.compiled.binds.values():
if bindparam.isoutparam:
dbtype = bindparam.type.dialect_impl(self.dialect).\
get_dbapi_type(self.dialect.dbapi)
if hasattr(self, 'out_parameters') and self.compiled.returning:
returning_params = dict(
(k, v.getvalue())
- for k, v in list(self.out_parameters.items())
+ for k, v in self.out_parameters.items()
)
return ReturningResultProxy(self, returning_params)
len(self.compiled_parameters) == 1:
result.out_parameters = out_parameters = {}
- for bind, name in list(self.compiled.bind_names.items()):
+ for bind, name in self.compiled.bind_names.items():
if name in self.out_parameters:
type = bind.type
impl_type = type.dialect_impl(self.dialect)
else:
result.out_parameters = dict(
(k, v.getvalue())
- for k, v in list(self.out_parameters.items())
+ for k, v in self.out_parameters.items()
)
return result
"""
def __init__(self, *arg, **kw):
OracleExecutionContext_cx_oracle.__init__(self, *arg, **kw)
- self.statement = str(self.statement)
+ self.statement = util.text_type(self.statement)
def _execute_scalar(self, stmt):
return super(OracleExecutionContext_cx_oracle_with_unicode, self).\
- _execute_scalar(str(stmt))
+ _execute_scalar(util.text_type(stmt))
class ReturningResultProxy(_result.FullyBufferedResultProxy):
self.supports_unicode_statements = True
self.supports_unicode_binds = True
self._cx_oracle_with_unicode = True
-# start Py2K
-# # There's really no reason to run with WITH_UNICODE under Python 2.x.
-# # Give the user a hint.
-# util.warn("cx_Oracle is compiled under Python 2.xx using the "
-# "WITH_UNICODE flag. Consider recompiling cx_Oracle without "
-# "this flag, which is in no way necessary for full support of Unicode. "
-# "Otherwise, all string-holding bind parameters must "
-# "be explicitly typed using SQLAlchemy's String type or one of its subtypes,"
-# "or otherwise be passed as Python unicode. Plain Python strings "
-# "passed as bind parameters will be silently corrupted by cx_Oracle."
-# )
-# self.execution_ctx_cls = OracleExecutionContext_cx_oracle_with_unicode
-# end Py2K
+
+ if util.py2k:
+ # There's really no reason to run with WITH_UNICODE under Python 2.x.
+ # Give the user a hint.
+ util.warn(
+ "cx_Oracle is compiled under Python 2.xx using the "
+ "WITH_UNICODE flag. Consider recompiling cx_Oracle "
+ "without this flag, which is in no way necessary for full "
+ "support of Unicode. Otherwise, all string-holding bind "
+ "parameters must be explicitly typed using SQLAlchemy's "
+ "String type or one of its subtypes,"
+ "or otherwise be passed as Python unicode. "
+ "Plain Python strings passed as bind parameters will be "
+ "silently corrupted by cx_Oracle."
+ )
+ self.execution_ctx_cls = \
+ OracleExecutionContext_cx_oracle_with_unicode
else:
self._cx_oracle_with_unicode = False
arraysize=cursor.arraysize)
# allow all strings to come back natively as Unicode
elif defaultType in (cx_Oracle.STRING, cx_Oracle.FIXED_CHAR):
- return cursor.var(str, size, cursor.arraysize)
+ return cursor.var(util.text_type, size, cursor.arraysize)
def on_connect(conn):
conn.outputtypehandler = output_type_handler
twophase=self.allow_twophase,
)
-# start Py2K
-# if self._cx_oracle_with_unicode:
-# for k, v in opts.items():
-# if isinstance(v, str):
-# opts[k] = unicode(v)
-# else:
-# for k, v in opts.items():
-# if isinstance(v, unicode):
-# opts[k] = str(v)
-# end Py2K
+ if util.py2k:
+ if self._cx_oracle_with_unicode:
+ for k, v in opts.items():
+ if isinstance(v, str):
+ opts[k] = unicode(v)
+ else:
+ for k, v in opts.items():
+ if isinstance(v, unicode):
+ opts[k] = str(v)
if 'mode' in url.query:
opts['mode'] = url.query['mode']
- if isinstance(opts['mode'], str):
+ if isinstance(opts['mode'], util.string_types):
mode = opts['mode'].upper()
if mode == 'SYSDBA':
opts['mode'] = self.dbapi.SYSDBA
from sqlalchemy.sql import table, column
from sqlalchemy.testing import fixtures, AssertsExecutionResults, AssertsCompiledSQL
from sqlalchemy import testing
+from sqlalchemy.util import u
from sqlalchemy.testing import assert_raises, assert_raises_message
from sqlalchemy.testing.engines import testing_engine
from sqlalchemy.dialects.oracle import cx_oracle, base as oracle
autoload=True, oracle_resolve_synonyms=True
)
for row in types_table.select().execute().fetchall():
- [row[k] for k in list(row.keys())]
+ [row[k] for k in row.keys()]
@testing.provide_metadata
def test_raw_roundtrip(self):
t2.c.data.type.dialect_impl(testing.db.dialect),
cx_oracle._OracleNVarChar)
- data = 'm’a réveillé.'
+ data = u('m’a réveillé.')
t2.insert().execute(data=data)
res = t2.select().execute().first()['data']
eq_(res, data)
- assert isinstance(res, str)
+ assert isinstance(res, util.text_type)
def test_char_length(self):
m = MetaData(testing.db)
m.reflect()
eq_(
- set(t.name for t in list(m.tables.values())),
+ set(t.name for t in m.tables.values()),
set(['admin_docindex'])
)
{'_underscorecolumn': '’é'},
)
result = testing.db.execute(
- table.select().where(table.c._underscorecolumn=='’é')
+ table.select().where(table.c._underscorecolumn==u('’é'))
).scalar()
- eq_(result, '’é')
+ eq_(result, u('’é'))
@testing.provide_metadata
def test_quoted_column_unicode(self):
metadata = self.metadata
table=Table("atable", metadata,
- Column("méil", Unicode(255), primary_key=True),
+ Column(u("méil"), Unicode(255), primary_key=True),
)
metadata.create_all()
table.insert().execute(
- {'méil': '’é'},
+ {u('méil'): u('’é')},
)
result = testing.db.execute(
- table.select().where(table.c['méil']=='’é')
+ table.select().where(table.c[u('méil')] == u('’é'))
).scalar()
- eq_(result, '’é')
+ eq_(result, u('’é'))
class DBLinkReflectionTest(fixtures.TestBase):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
