--- /dev/null
+.. change::
+ :tags: general, changed
+ :tickets: 7257
+
+ Migrated the codebase to remove all pre-2.0 behaviors and architectures
+ that were previously noted as deprecated for removal in 2.0, including:
+
+ * removal of all Python 2 code, minimum version is now Python 3.7
+
+ * :class:`_engine.Engine` and :class:`_engine.Connection` now use the
+ new 2.0 style of working, which includes "autobegin", library level
+ autocommit removed, subtransactions and "branched" connections
+ removed
+
+ * Result objects use 2.0-style behaviors; :class:`_result.Row` is fully
+ a named tuple without "mapping" behavior, use :class:`_result.RowMapping`
+ for "mapping" behavior
+
+ * All Unicode encoding/decoding architecture has been removed from
+ SQLAlchemy. All modern DBAPI implementations support Unicode
+ transparently thanks to Python 3, so the ``convert_unicode`` feature
+ as well as related mechanisms to look for bytestrings in
+ DBAPI ``cursor.description`` etc. have been removed.
+
+ * More are in progress as development continues
* Structs *
***********/
-typedef struct {
- PyObject_HEAD
- PyObject *encoding;
- PyObject *errors;
-} UnicodeResultProcessor;
-
typedef struct {
PyObject_HEAD
PyObject *type;
-/**************************
- * UnicodeResultProcessor *
- **************************/
-
-static int
-UnicodeResultProcessor_init(UnicodeResultProcessor *self, PyObject *args,
- PyObject *kwds)
-{
- PyObject *encoding, *errors = NULL;
- static char *kwlist[] = {"encoding", "errors", NULL};
-
-#if PY_MAJOR_VERSION >= 3
- if (!PyArg_ParseTupleAndKeywords(args, kwds, "U|U:__init__", kwlist,
- &encoding, &errors))
- return -1;
-#else
- if (!PyArg_ParseTupleAndKeywords(args, kwds, "S|S:__init__", kwlist,
- &encoding, &errors))
- return -1;
-#endif
-
-#if PY_MAJOR_VERSION >= 3
- encoding = PyUnicode_AsASCIIString(encoding);
-#else
- Py_INCREF(encoding);
-#endif
- self->encoding = encoding;
-
- if (errors) {
-#if PY_MAJOR_VERSION >= 3
- errors = PyUnicode_AsASCIIString(errors);
-#else
- Py_INCREF(errors);
-#endif
- } else {
-#if PY_MAJOR_VERSION >= 3
- errors = PyBytes_FromString("strict");
-#else
- errors = PyString_FromString("strict");
-#endif
- if (errors == NULL)
- return -1;
- }
- self->errors = errors;
-
- return 0;
-}
-
-static PyObject *
-UnicodeResultProcessor_process(UnicodeResultProcessor *self, PyObject *value)
-{
- const char *encoding, *errors;
- char *str;
- Py_ssize_t len;
-
- if (value == Py_None)
- Py_RETURN_NONE;
-
-#if PY_MAJOR_VERSION >= 3
- if (PyBytes_AsStringAndSize(value, &str, &len))
- return NULL;
-
- encoding = PyBytes_AS_STRING(self->encoding);
- errors = PyBytes_AS_STRING(self->errors);
-#else
- if (PyString_AsStringAndSize(value, &str, &len))
- return NULL;
-
- encoding = PyString_AS_STRING(self->encoding);
- errors = PyString_AS_STRING(self->errors);
-#endif
-
- return PyUnicode_Decode(str, len, encoding, errors);
-}
-
-static PyObject *
-UnicodeResultProcessor_conditional_process(UnicodeResultProcessor *self, PyObject *value)
-{
- const char *encoding, *errors;
- char *str;
- Py_ssize_t len;
-
- if (value == Py_None)
- Py_RETURN_NONE;
-
-#if PY_MAJOR_VERSION >= 3
- if (PyUnicode_Check(value) == 1) {
- Py_INCREF(value);
- return value;
- }
-
- if (PyBytes_AsStringAndSize(value, &str, &len))
- return NULL;
-
- encoding = PyBytes_AS_STRING(self->encoding);
- errors = PyBytes_AS_STRING(self->errors);
-#else
-
- if (PyUnicode_Check(value) == 1) {
- Py_INCREF(value);
- return value;
- }
-
- if (PyString_AsStringAndSize(value, &str, &len))
- return NULL;
-
-
- encoding = PyString_AS_STRING(self->encoding);
- errors = PyString_AS_STRING(self->errors);
-#endif
-
- return PyUnicode_Decode(str, len, encoding, errors);
-}
-
-static void
-UnicodeResultProcessor_dealloc(UnicodeResultProcessor *self)
-{
- Py_XDECREF(self->encoding);
- Py_XDECREF(self->errors);
-#if PY_MAJOR_VERSION >= 3
- Py_TYPE(self)->tp_free((PyObject*)self);
-#else
- self->ob_type->tp_free((PyObject*)self);
-#endif
-}
-
-static PyMethodDef UnicodeResultProcessor_methods[] = {
- {"process", (PyCFunction)UnicodeResultProcessor_process, METH_O,
- "The value processor itself."},
- {"conditional_process", (PyCFunction)UnicodeResultProcessor_conditional_process, METH_O,
- "Conditional version of the value processor."},
- {NULL} /* Sentinel */
-};
-
-static PyTypeObject UnicodeResultProcessorType = {
- PyVarObject_HEAD_INIT(NULL, 0)
- "sqlalchemy.cprocessors.UnicodeResultProcessor", /* tp_name */
- sizeof(UnicodeResultProcessor), /* tp_basicsize */
- 0, /* tp_itemsize */
- (destructor)UnicodeResultProcessor_dealloc, /* tp_dealloc */
- 0, /* tp_print */
- 0, /* tp_getattr */
- 0, /* tp_setattr */
- 0, /* tp_compare */
- 0, /* tp_repr */
- 0, /* tp_as_number */
- 0, /* tp_as_sequence */
- 0, /* tp_as_mapping */
- 0, /* tp_hash */
- 0, /* tp_call */
- 0, /* tp_str */
- 0, /* tp_getattro */
- 0, /* tp_setattro */
- 0, /* tp_as_buffer */
- Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE, /* tp_flags */
- "UnicodeResultProcessor objects", /* tp_doc */
- 0, /* tp_traverse */
- 0, /* tp_clear */
- 0, /* tp_richcompare */
- 0, /* tp_weaklistoffset */
- 0, /* tp_iter */
- 0, /* tp_iternext */
- UnicodeResultProcessor_methods, /* tp_methods */
- 0, /* tp_members */
- 0, /* tp_getset */
- 0, /* tp_base */
- 0, /* tp_dict */
- 0, /* tp_descr_get */
- 0, /* tp_descr_set */
- 0, /* tp_dictoffset */
- (initproc)UnicodeResultProcessor_init, /* tp_init */
- 0, /* tp_alloc */
- 0, /* tp_new */
-};
/**************************
* DecimalResultProcessor *
{
PyObject *m;
- UnicodeResultProcessorType.tp_new = PyType_GenericNew;
- if (PyType_Ready(&UnicodeResultProcessorType) < 0)
- INITERROR;
-
DecimalResultProcessorType.tp_new = PyType_GenericNew;
if (PyType_Ready(&DecimalResultProcessorType) < 0)
INITERROR;
PyDateTime_IMPORT;
- Py_INCREF(&UnicodeResultProcessorType);
- PyModule_AddObject(m, "UnicodeResultProcessor",
- (PyObject *)&UnicodeResultProcessorType);
-
Py_INCREF(&DecimalResultProcessorType);
PyModule_AddObject(m, "DecimalResultProcessor",
(PyObject *)&DecimalResultProcessorType);
supports_sane_rowcount_returning = True
supports_sane_multi_rowcount = False
- supports_unicode_statements = True
- supports_unicode_binds = True
-
supports_native_decimal = True
default_paramstyle = "named"
# hold the desired driver name
pyodbc_driver_name = None
- def __init__(
- self, supports_unicode_binds=None, use_setinputsizes=False, **kw
- ):
+ def __init__(self, use_setinputsizes=False, **kw):
super(PyODBCConnector, self).__init__(**kw)
- if supports_unicode_binds is not None:
- self.supports_unicode_binds = supports_unicode_binds
self.use_setinputsizes = use_setinputsizes
@classmethod
def _opt_encode(self, statement):
- if not self.dialect.supports_unicode_statements:
- encoded = self.dialect._encoder(statement)[0]
- else:
- encoded = statement
-
if self.compiled and self.compiled.schema_translate_map:
rst = self.compiled.preparer._render_schema_translates
- encoded = rst(encoded, self.compiled.schema_translate_map)
+ statement = rst(statement, self.compiled.schema_translate_map)
- return encoded
+ return statement
def pre_exec(self):
"""Activate IDENTITY_INSERT if needed."""
from .base import MySQLCompiler
from .base import MySQLDialect
from .base import MySQLIdentifierPreparer
-from ... import processors
from ... import util
driver = "mysqlconnector"
supports_statement_cache = True
- supports_unicode_binds = True
-
supports_sane_rowcount = True
supports_sane_multi_rowcount = True
colspecs = util.update_copy(MySQLDialect.colspecs, {BIT: _myconnpyBIT})
- def __init__(self, *arg, **kw):
- super(MySQLDialect_mysqlconnector, self).__init__(*arg, **kw)
-
- # hack description encoding since mysqlconnector randomly
- # returns bytes or not
- self._description_decoder = (
- processors.to_conditional_unicode_processor_factory
- )(self.description_encoding)
-
- def _check_unicode_description(self, connection):
- # hack description encoding since mysqlconnector randomly
- # returns bytes or not
- return False
-
- @property
- def description_encoding(self):
- # total guess
- return "latin-1"
-
- @util.memoized_property
- def supports_unicode_statements(self):
- return util.py3k or self._mysqlconnector_version_info > (2, 0)
-
@classmethod
def dbapi(cls):
from mysql import connector
description_encoding = None
- # generally, these two values should be both True
- # or both False. PyMySQL unicode tests pass all the way back
- # to 0.4 either way. See [ticket:3337]
- supports_unicode_statements = True
- supports_unicode_binds = True
-
@langhelpers.memoized_property
def supports_server_side_cursors(self):
try:
:name: Oracle
:full_support: 11.2, 18c
:normal_support: 11+
- :best_effort: 8+
+ :best_effort: 9+
Auto Increment Behavior
Oracle 8 Compatibility
----------------------
+.. warning:: The status of Oracle 8 compatibility is not known for SQLAlchemy
+ 2.0.
+
When Oracle 8 is detected, the dialect internally configures itself to the
following behaviors:
makes use of Oracle's (+) operator.
* the NVARCHAR2 and NCLOB datatypes are no longer generated as DDL when
- the :class:`~sqlalchemy.types.Unicode` is used - VARCHAR2 and CLOB are
- issued instead. This because these types don't seem to work correctly on
- Oracle 8 even though they are available. The
- :class:`~sqlalchemy.types.NVARCHAR` and
+ the :class:`~sqlalchemy.types.Unicode` is used - VARCHAR2 and CLOB are issued
+ instead. This because these types don't seem to work correctly on Oracle 8
+ even though they are available. The :class:`~sqlalchemy.types.NVARCHAR` and
:class:`~sqlalchemy.dialects.oracle.NCLOB` types will always generate
NVARCHAR2 and NCLOB.
-* the "native unicode" mode is disabled when using cx_oracle, i.e. SQLAlchemy
- encodes all Python unicode objects to "string" before passing in as bind
- parameters.
Synonym/DBLINK Reflection
-------------------------
name = "oracle"
supports_statement_cache = True
supports_alter = True
- supports_unicode_statements = False
- supports_unicode_binds = False
max_identifier_length = 128
supports_simple_order_by_label = False
# use the default
return None
- def _check_unicode_returns(self, connection):
- additional_tests = [
- expression.cast(
- expression.literal_column("'test nvarchar2 returns'"),
- sqltypes.NVARCHAR(60),
- )
- ]
- return super(OracleDialect, self)._check_unicode_returns(
- connection, additional_tests
- )
-
_isolation_lookup = ["READ COMMITTED", "SERIALIZABLE"]
def get_isolation_level(self, connection):
, such as::
e = create_engine(
- "oracle+cx_oracle://user:pass@dsn", coerce_to_unicode=False)
+ "oracle+cx_oracle://user:pass@dsn", coerce_to_decimal=False)
The parameters accepted by the cx_oracle dialect are as follows:
* ``auto_convert_lobs`` - defaults to True; See :ref:`cx_oracle_lob`.
-* ``coerce_to_unicode`` - see :ref:`cx_oracle_unicode` for detail.
-
* ``coerce_to_decimal`` - see :ref:`cx_oracle_numeric` for detail.
* ``encoding_errors`` - see :ref:`cx_oracle_unicode_encoding_errors` for detail.
-------
As is the case for all DBAPIs under Python 3, all strings are inherently
-Unicode strings. Under Python 2, cx_Oracle also supports Python Unicode
-objects directly. In all cases however, the driver requires an explicit
+Unicode strings. In all cases however, the driver requires an explicit
encoding configuration.
Ensuring the Correct Client Encoding
unless the ``use_nchar_for_unicode=True`` is passed to the dialect
when :func:`_sa.create_engine` is called.
-Unicode Coercion of result rows under Python 2
-^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
-
-When result sets are fetched that include strings, under Python 3 the cx_Oracle
-DBAPI returns all strings as Python Unicode objects, since Python 3 only has a
-Unicode string type. This occurs for data fetched from datatypes such as
-VARCHAR2, CHAR, CLOB, NCHAR, NCLOB, etc. In order to provide cross-
-compatibility under Python 2, the SQLAlchemy cx_Oracle dialect will add
-Unicode-conversion to string data under Python 2 as well. Historically, this
-made use of converters that were supplied by cx_Oracle but were found to be
-non-performant; SQLAlchemy's own converters are used for the string to Unicode
-conversion under Python 2. To disable the Python 2 Unicode conversion for
-VARCHAR2, CHAR, and CLOB, the flag ``coerce_to_unicode=False`` can be passed to
-:func:`_sa.create_engine`.
-
-.. versionchanged:: 1.3 Unicode conversion is applied to all string values
- by default under python 2. The ``coerce_to_unicode`` now defaults to True
- and can be set to False to disable the Unicode coercion of strings that are
- delivered as VARCHAR2/CHAR/CLOB data.
.. _cx_oracle_unicode_encoding_errors:
supports_sane_rowcount = True
supports_sane_multi_rowcount = True
- supports_unicode_statements = True
- supports_unicode_binds = True
-
use_setinputsizes = True
driver = "cx_oracle"
_cx_oracle_threaded = None
+ _cursor_var_unicode_kwargs = util.immutabledict()
+
@util.deprecated_params(
threaded=(
"1.3",
def __init__(
self,
auto_convert_lobs=True,
- coerce_to_unicode=True,
coerce_to_decimal=True,
arraysize=50,
encoding_errors=None,
OracleDialect.__init__(self, **kwargs)
self.arraysize = arraysize
self.encoding_errors = encoding_errors
+ if encoding_errors:
+ self._cursor_var_unicode_kwargs = {
+ "encodingErrors": encoding_errors
+ }
if threaded is not None:
self._cx_oracle_threaded = threaded
self.auto_convert_lobs = auto_convert_lobs
- self.coerce_to_unicode = coerce_to_unicode
self.coerce_to_decimal = coerce_to_decimal
if self._use_nchar_for_unicode:
self.colspecs = self.colspecs.copy()
"cx_Oracle version 5.2 and above are supported"
)
+ if encoding_errors and self.cx_oracle_ver < (6, 4):
+ util.warn(
+ "cx_oracle version %r does not support encodingErrors"
+ % (self.cx_oracle_ver,)
+ )
+ self._cursor_var_unicode_kwargs = util.immutabledict()
+
self._include_setinputsizes = {
cx_Oracle.DATETIME,
cx_Oracle.NCLOB,
self._is_cx_oracle_6 = self.cx_oracle_ver >= (6,)
- @property
- def _cursor_var_unicode_kwargs(self):
- if self.encoding_errors:
- if self.cx_oracle_ver >= (6, 4):
- return {"encodingErrors": self.encoding_errors}
- else:
- util.warn(
- "cx_oracle version %r does not support encodingErrors"
- % (self.cx_oracle_ver,)
- )
-
- return {}
-
def _parse_cx_oracle_ver(self, version):
m = re.match(r"(\d+)\.(\d+)(?:\.(\d+))?", version)
if m:
def initialize(self, connection):
super(OracleDialect_cx_oracle, self).initialize(connection)
- if self._is_oracle_8:
- self.supports_unicode_binds = False
-
self._detect_decimal_char(connection)
def get_isolation_level(self, connection):
cursor, name, default_type, size, precision, scale
)
- # allow all strings to come back natively as Unicode
+ # if unicode options were specified, add a decoder, otherwise
+ # cx_Oracle should return Unicode
elif (
- dialect.coerce_to_unicode
+ dialect._cursor_var_unicode_kwargs
and default_type
in (
cx_Oracle.STRING,
if dbtype
)
- if not self.supports_unicode_binds:
- # oracle 8 only
- collection = (
- (self.dialect._encoder(key)[0], dbtype)
- for key, dbtype in collection
- )
-
cursor.setinputsizes(**{key: dbtype for key, dbtype in collection})
def do_recover_twophase(self, connection):
driver = "asyncpg"
supports_statement_cache = True
- supports_unicode_statements = True
supports_server_side_cursors = True
- supports_unicode_binds = True
-
default_paramstyle = "format"
supports_sane_multi_rowcount = False
execution_ctx_cls = PGExecutionContext_asyncpg
:ref:`psycopg2_unicode`
-* ``use_native_unicode``: Under Python 2 only, this can be set to False to
- disable the use of psycopg2's native Unicode support.
-
- .. seealso::
-
- :ref:`psycopg2_disable_native_unicode`
-
* ``executemany_mode``, ``executemany_batch_page_size``,
``executemany_values_page_size``: Allows use of psycopg2
Unicode with Psycopg2
----------------------
-The psycopg2 DBAPI driver supports Unicode data transparently. Under Python 2
-only, the SQLAlchemy psycopg2 dialect will enable the
-``psycopg2.extensions.UNICODE`` extension by default to ensure Unicode is
-handled properly; under Python 3, this is psycopg2's default behavior.
+The psycopg2 DBAPI driver supports Unicode data transparently.
The client character encoding can be controlled for the psycopg2 dialect
in the following ways:
# encoding
client_encoding = utf8
-.. _psycopg2_disable_native_unicode:
-
-Disabling Native Unicode
-^^^^^^^^^^^^^^^^^^^^^^^^
-
-Under Python 2 only, SQLAlchemy can also be instructed to skip the usage of the
-psycopg2 ``UNICODE`` extension and to instead utilize its own unicode
-encode/decode services, which are normally reserved only for those DBAPIs that
-don't fully support unicode directly. Passing ``use_native_unicode=False`` to
-:func:`_sa.create_engine` will disable usage of ``psycopg2.extensions.
-UNICODE``. SQLAlchemy will instead encode data itself into Python bytestrings
-on the way in and coerce from bytes on the way back, using the value of the
-:func:`_sa.create_engine` ``encoding`` parameter, which defaults to ``utf-8``.
-SQLAlchemy's own unicode encode/decode functionality is steadily becoming
-obsolete as most DBAPIs now support unicode fully.
Transactions
_has_native_hstore = True
- engine_config_types = PGDialect.engine_config_types.union(
- {"use_native_unicode": util.asbool}
- )
-
colspecs = util.update_copy(
PGDialect.colspecs,
{
def __init__(
self,
- use_native_unicode=True,
client_encoding=None,
use_native_hstore=True,
use_native_uuid=True,
**kwargs
):
PGDialect.__init__(self, **kwargs)
- self.use_native_unicode = use_native_unicode
- if not use_native_unicode:
- raise exc.ArgumentError(
- "psycopg2 native_unicode mode is required under Python 3"
- )
if not use_native_hstore:
self._has_native_hstore = False
self.use_native_hstore = use_native_hstore
self.use_native_uuid = use_native_uuid
- self.supports_unicode_binds = use_native_unicode
self.client_encoding = client_encoding
# Parse executemany_mode argument, allowing it to be only one of the
executemany_values = (
"(%s)" % context.compiled.insert_single_values_expr
)
- if not self.supports_unicode_statements:
- executemany_values = executemany_values.encode(self.encoding)
# guard for statement that was altered via event hook or similar
if executemany_values not in statement:
class SQLiteDialect(default.DefaultDialect):
name = "sqlite"
supports_alter = False
- supports_unicode_statements = True
- supports_unicode_binds = True
# SQlite supports "DEFAULT VALUES" but *does not* support
# "VALUES (DEFAULT)"
additional keyword arguments. See the example
at :ref:`custom_dbapi_args`.
- :param convert_unicode=False: if set to True, causes
- all :class:`.String` datatypes to act as though the
- :paramref:`.String.convert_unicode` flag has been set to ``True``,
- regardless of a setting of ``False`` on an individual :class:`.String`
- type. This has the effect of causing all :class:`.String` -based
- columns to accommodate Python Unicode objects directly as though the
- datatype were the :class:`.Unicode` type.
-
- .. deprecated:: 1.3
-
- The :paramref:`_sa.create_engine.convert_unicode` parameter
- is deprecated and will be removed in a future release.
- All modern DBAPIs now support Python Unicode directly and this
- parameter is unnecessary.
-
:param creator: a callable which returns a DBAPI connection.
This creation function will be passed to the underlying
connection pool and will be used to create all new database
:ref:`change_4737`
- :param encoding: **legacy Python 2 value only, where it only applies to
- specific DBAPIs, not used in Python 3 for any modern DBAPI driver.
- Please refer to individual dialect documentation for client encoding
- behaviors.** Defaults to the string value ``utf-8``. This value
- refers **only** to the character encoding that is used when SQLAlchemy
- sends or receives data from a :term:`DBAPI` that does not support
- Python Unicode and **is only used under Python 2**, only for certain
- DBAPI drivers, and only in certain circumstances. **Python 3 users
- please DISREGARD this parameter and refer to the documentation for the
- specific dialect in use in order to configure character encoding
- behavior.**
-
- .. note:: The ``encoding`` parameter deals only with in-Python
- encoding issues that were prevalent with **some DBAPIS only**
- under **Python 2 only**. Under Python 3 it is not used by
- any modern dialect. For DBAPIs that require
- client encoding configurations, which are most of those outside
- of SQLite, please consult specific :ref:`dialect documentation
- <dialect_toplevel>` for details.
-
- All modern DBAPIs that work in Python 3 necessarily feature direct
- support for Python unicode strings. Under Python 2, this was not
- always the case. For those scenarios where the DBAPI is detected as
- not supporting a Python ``unicode`` object under Python 2, this
- encoding is used to determine the source/destination encoding. It is
- **not used** for those cases where the DBAPI handles unicode directly.
-
- To properly configure a system to accommodate Python ``unicode``
- objects, the DBAPI should be configured to handle unicode to the
- greatest degree as is appropriate - see the notes on unicode pertaining
- to the specific target database in use at :ref:`dialect_toplevel`.
-
- Areas where string encoding may need to be accommodated
- outside of the DBAPI, nearly always under **Python 2 only**,
- include zero or more of:
-
- * the values passed to bound parameters, corresponding to
- the :class:`.Unicode` type or the :class:`.String` type
- when ``convert_unicode`` is ``True``;
- * the values returned in result set columns corresponding
- to the :class:`.Unicode` type or the :class:`.String`
- type when ``convert_unicode`` is ``True``;
- * the string SQL statement passed to the DBAPI's
- ``cursor.execute()`` method;
- * the string names of the keys in the bound parameter
- dictionary passed to the DBAPI's ``cursor.execute()``
- as well as ``cursor.setinputsizes()`` methods;
- * the string column names retrieved from the DBAPI's
- ``cursor.description`` attribute.
-
- When using Python 3, the DBAPI is required to support all of the above
- values as Python ``unicode`` objects, which in Python 3 are just known
- as ``str``. In Python 2, the DBAPI does not specify unicode behavior
- at all, so SQLAlchemy must make decisions for each of the above values
- on a per-DBAPI basis - implementations are completely inconsistent in
- their behavior.
-
:param execution_options: Dictionary execution options which will
be applied to all connections. See
:meth:`~sqlalchemy.engine.Connection.execution_options`
"""
-import codecs
import functools
import random
import re
from .. import event
from .. import exc
from .. import pool
-from .. import processors
from .. import types as sqltypes
from .. import util
from ..sql import compiler
engine_config_types = util.immutabledict(
[
- ("convert_unicode", util.bool_or_str("force")),
("pool_timeout", util.asint),
("echo", util.bool_or_str("debug")),
("echo_pool", util.bool_or_str("debug")),
# *not* the FLOAT type however.
supports_native_decimal = False
- supports_unicode_statements = True
- supports_unicode_binds = True
- returns_unicode_strings = sqltypes.String.RETURNS_UNICODE
description_encoding = None
name = "default"
NO_DIALECT_SUPPORT = NO_DIALECT_SUPPORT
@util.deprecated_params(
- convert_unicode=(
- "1.3",
- "The :paramref:`_sa.create_engine.convert_unicode` parameter "
- "and corresponding dialect-level parameters are deprecated, "
- "and will be removed in a future release. Modern DBAPIs support "
- "Python Unicode natively and this parameter is unnecessary.",
- ),
empty_in_strategy=(
"1.4",
"The :paramref:`_sa.create_engine.empty_in_strategy` keyword is "
)
def __init__(
self,
- convert_unicode=False,
encoding="utf-8",
paramstyle=None,
dbapi=None,
else:
self.server_side_cursors = True
- self.convert_unicode = convert_unicode
self.encoding = encoding
self.positional = False
self._ischema = None
)
self.label_length = label_length
self.compiler_linting = compiler_linting
- if self.description_encoding == "use_encoding":
- self._description_decoder = (
- processors.to_unicode_processor_factory
- )(encoding)
- elif self.description_encoding is not None:
- self._description_decoder = (
- processors.to_unicode_processor_factory
- )(self.description_encoding)
- self._encoder = codecs.getencoder(self.encoding)
- self._decoder = processors.to_unicode_processor_factory(self.encoding)
def _ensure_has_table_connection(self, arg):
except NotImplementedError:
self.default_isolation_level = None
- if (
- self.description_encoding is not None
- and self._check_unicode_description(connection)
- ):
- self._description_decoder = self.description_encoding = None
-
if not self._user_defined_max_identifier_length:
max_ident_length = self._check_max_identifier_length(connection)
if max_ident_length:
"""
return self.get_isolation_level(dbapi_conn)
- def _check_unicode_description(self, connection):
- cast_to = util.text_type
-
- cursor = connection.connection.cursor()
- try:
- cursor.execute(
- cast_to(
- expression.select(
- expression.literal_column("'x'").label("some_label")
- ).compile(dialect=self)
- )
- )
- return isinstance(cursor.description[0][0], util.text_type)
- finally:
- cursor.close()
-
def type_descriptor(self, typeobj):
"""Provide a database-specific :class:`.TypeEngine` object, given
the generic object which comes from the types module.
self.unicode_statement, schema_translate_map
)
- if not dialect.supports_unicode_statements:
- self.statement = dialect._encoder(self.unicode_statement)[0]
- else:
- self.statement = self.unicode_statement
+ self.statement = self.unicode_statement
self.cursor = self.create_cursor()
self.compiled_parameters = []
# final self.unicode_statement is now assigned, encode if needed
# by dialect
- if not dialect.supports_unicode_statements:
- self.statement = self.unicode_statement.encode(
- self.dialect.encoding
- )
- else:
- self.statement = self.unicode_statement
+ self.statement = self.unicode_statement
# Convert the dictionary of bind parameter values
# into a dict or list to be sent to the DBAPI's
]
parameters.append(dialect.execute_sequence_format(param))
else:
- encode = not dialect.supports_unicode_statements
- if encode:
- encoder = dialect._encoder
for compiled_params in self.compiled_parameters:
- if encode:
- param = {
- encoder(key)[0]: processors[key](compiled_params[key])
- if key in processors
- else compiled_params[key]
- for key in compiled_params
- }
- else:
- param = {
- key: processors[key](compiled_params[key])
- if key in processors
- else compiled_params[key]
- for key in compiled_params
- }
+ param = {
+ key: processors[key](compiled_params[key])
+ if key in processors
+ else compiled_params[key]
+ for key in compiled_params
+ }
parameters.append(param)
elif isinstance(parameters[0], dialect.execute_sequence_format):
self.parameters = parameters
elif isinstance(parameters[0], dict):
- if dialect.supports_unicode_statements:
- self.parameters = parameters
- else:
- self.parameters = [
- {dialect._encoder(k)[0]: d[k] for k in d}
- for d in parameters
- ] or [{}]
+ self.parameters = parameters
else:
self.parameters = [
dialect.execute_sequence_format(p) for p in parameters
self.executemany = len(parameters) > 1
- if not dialect.supports_unicode_statements and isinstance(
- statement, util.text_type
- ):
- self.unicode_statement = statement
- self.statement = dialect._encoder(statement)[0]
- else:
- self.statement = self.unicode_statement = statement
+ self.statement = self.unicode_statement = statement
self.cursor = self.create_cursor()
return self
"""
conn = self.root_connection
- if (
- isinstance(stmt, util.text_type)
- and not self.dialect.supports_unicode_statements
- ):
- stmt = self.dialect._encoder(stmt)[0]
if "schema_translate_map" in self.execution_options:
schema_translate_map = self.execution_options.get(
"""
-import codecs
import datetime
import re
def py_fallback():
- def to_unicode_processor_factory(encoding, errors=None):
- decoder = codecs.getdecoder(encoding)
-
- def process(value):
- if value is None:
- return None
- else:
- # decoder returns a tuple: (value, len). Simply dropping the
- # len part is safe: it is done that way in the normal
- # 'xx'.decode(encoding) code path.
- return decoder(value, errors)[0]
-
- return process
-
- def to_conditional_unicode_processor_factory(encoding, errors=None):
- decoder = codecs.getdecoder(encoding)
-
- def process(value):
- if value is None:
- return None
- elif isinstance(value, util.text_type):
- return value
- else:
- # decoder returns a tuple: (value, len). Simply dropping the
- # len part is safe: it is done that way in the normal
- # 'xx'.decode(encoding) code path.
- return decoder(value, errors)[0]
-
- return process
-
def to_decimal_processor_factory(target_class, scale):
fstring = "%%.%df" % scale
from sqlalchemy.cprocessors import str_to_time # noqa
from sqlalchemy.cprocessors import to_float # noqa
from sqlalchemy.cprocessors import to_str # noqa
- from sqlalchemy.cprocessors import UnicodeResultProcessor # noqa
-
- def to_unicode_processor_factory(encoding, errors=None):
- if errors is not None:
- return UnicodeResultProcessor(encoding, errors).process
- else:
- return UnicodeResultProcessor(encoding).process
-
- def to_conditional_unicode_processor_factory(encoding, errors=None):
- if errors is not None:
- return UnicodeResultProcessor(encoding, errors).conditional_process
- else:
- return UnicodeResultProcessor(encoding).conditional_process
def to_decimal_processor_factory(target_class, scale):
# Note that the scale argument is not taken into account for integer
if isinstance(self.default, (ColumnDefault, Sequence)):
args.append(self.default)
else:
- if getattr(self.type, "_warn_on_bytestring", False):
- if isinstance(self.default, util.binary_type):
- util.warn(
- "Unicode column '%s' has non-unicode "
- "default value %r specified."
- % (self.key, self.default)
- )
args.append(ColumnDefault(self.default))
if self.server_default is not None:
"""
-import codecs
import datetime as dt
import decimal
import json
"""The base for all string and character types.
- In SQL, corresponds to VARCHAR. Can also take Python unicode objects
- and encode to the database's encoding in bind params (and the reverse for
- result sets.)
+ In SQL, corresponds to VARCHAR.
The `length` field is usually required when the `String` type is
used within a CREATE TABLE statement, as VARCHAR requires a length
__visit_name__ = "string"
- RETURNS_UNICODE = util.symbol(
- "RETURNS_UNICODE",
- """Indicates that the DBAPI returns Python Unicode for VARCHAR,
- NVARCHAR, and other character-based datatypes in all cases.
-
- This is the default value for
- :attr:`.DefaultDialect.returns_unicode_strings` under Python 3.
-
- .. versionadded:: 1.4
-
- """,
- )
-
- RETURNS_BYTES = util.symbol(
- "RETURNS_BYTES",
- """Indicates that the DBAPI returns byte objects under Python 3
- or non-Unicode string objects under Python 2 for VARCHAR, NVARCHAR,
- and other character-based datatypes in all cases.
-
- This may be applied to the
- :attr:`.DefaultDialect.returns_unicode_strings` attribute.
-
- .. versionadded:: 1.4
-
- """,
- )
-
- RETURNS_CONDITIONAL = util.symbol(
- "RETURNS_CONDITIONAL",
- """Indicates that the DBAPI may return Unicode or bytestrings for
- VARCHAR, NVARCHAR, and other character-based datatypes, and that
- SQLAlchemy's default String datatype will need to test on a per-row
- basis for Unicode or bytes.
-
- This may be applied to the
- :attr:`.DefaultDialect.returns_unicode_strings` attribute.
-
- .. versionadded:: 1.4
-
- """,
- )
-
- RETURNS_UNKNOWN = util.symbol(
- "RETURNS_UNKNOWN",
- """Indicates that the dialect should test on first connect what the
- string-returning behavior of character-based datatypes is.
-
- This is the default value for DefaultDialect.unicode_returns under
- Python 2.
-
- This may be applied to the
- :attr:`.DefaultDialect.returns_unicode_strings` attribute under
- Python 2 only. The value is disallowed under Python 3.
-
- .. versionadded:: 1.4
-
- .. deprecated:: 1.4 This value will be removed in SQLAlchemy 2.0.
-
- """,
- )
-
- @util.deprecated_params(
- convert_unicode=(
- "1.3",
- "The :paramref:`.String.convert_unicode` parameter is deprecated "
- "and will be removed in a future release. All modern DBAPIs "
- "now support Python Unicode directly and this parameter is "
- "unnecessary.",
- ),
- unicode_error=(
- "1.3",
- "The :paramref:`.String.unicode_errors` parameter is deprecated "
- "and will be removed in a future release. This parameter is "
- "unnecessary for modern Python DBAPIs and degrades performance "
- "significantly.",
- ),
- )
def __init__(
self,
length=None,
collation=None,
- convert_unicode=False,
- unicode_error=None,
- _warn_on_bytestring=False,
- _expect_unicode=False,
):
"""
Create a string-holding type.
>>> print(select(cast('some string', String(collation='utf8'))))
SELECT CAST(:param_1 AS VARCHAR COLLATE utf8) AS anon_1
- :param convert_unicode: When set to ``True``, the
- :class:`.String` type will assume that
- input is to be passed as Python Unicode objects under Python 2,
- and results returned as Python Unicode objects.
- In the rare circumstance that the DBAPI does not support
- Python unicode under Python 2, SQLAlchemy will use its own
- encoder/decoder functionality on strings, referring to the
- value of the :paramref:`_sa.create_engine.encoding` parameter
- parameter passed to :func:`_sa.create_engine` as the encoding.
-
- For the extremely rare case that Python Unicode
- is to be encoded/decoded by SQLAlchemy on a backend
- that *does* natively support Python Unicode,
- the string value ``"force"`` can be passed here which will
- cause SQLAlchemy's encode/decode services to be
- used unconditionally.
-
- .. note::
-
- SQLAlchemy's unicode-conversion flags and features only apply
- to Python 2; in Python 3, all string objects are Unicode objects.
- For this reason, as well as the fact that virtually all modern
- DBAPIs now support Unicode natively even under Python 2,
- the :paramref:`.String.convert_unicode` flag is inherently a
- legacy feature.
-
.. note::
- In the vast majority of cases, the :class:`.Unicode` or
- :class:`.UnicodeText` datatypes should be used for a
- :class:`_schema.Column` that expects to store non-ascii data.
- These
- datatypes will ensure that the correct types are used on the
- database side as well as set up the correct Unicode behaviors
- under Python 2.
-
- .. seealso::
-
- :paramref:`_sa.create_engine.convert_unicode` -
- :class:`_engine.Engine`-wide parameter
-
- :param unicode_error: Optional, a method to use to handle Unicode
- conversion errors. Behaves like the ``errors`` keyword argument to
- the standard library's ``string.decode()`` functions, requires
- that :paramref:`.String.convert_unicode` is set to
- ``"force"``
+ In most cases, the :class:`.Unicode` or :class:`.UnicodeText`
+ datatypes should be used for a :class:`_schema.Column` that expects
+ to store non-ascii data. These datatypes will ensure that the
+ correct types are used on the database.
"""
- if unicode_error is not None and convert_unicode != "force":
- raise exc.ArgumentError(
- "convert_unicode must be 'force' " "when unicode_error is set."
- )
self.length = length
self.collation = collation
- self._expect_unicode = convert_unicode or _expect_unicode
- self._expect_unicode_error = unicode_error
-
- self._warn_on_bytestring = _warn_on_bytestring
def literal_processor(self, dialect):
def process(value):
return process
def bind_processor(self, dialect):
- if self._expect_unicode or dialect.convert_unicode:
- if (
- dialect.supports_unicode_binds
- and self._expect_unicode != "force"
- ):
- if self._warn_on_bytestring:
-
- def process(value):
- if isinstance(value, util.binary_type):
- util.warn_limited(
- "Unicode type received non-unicode "
- "bind param value %r.",
- (util.ellipses_string(value),),
- )
- return value
-
- return process
- else:
- return None
- else:
- encoder = codecs.getencoder(dialect.encoding)
- warn_on_bytestring = self._warn_on_bytestring
-
- def process(value):
- if isinstance(value, util.text_type):
- return encoder(value, self._expect_unicode_error)[0]
- elif warn_on_bytestring and value is not None:
- util.warn_limited(
- "Unicode type received non-unicode bind "
- "param value %r.",
- (util.ellipses_string(value),),
- )
- return value
-
- return process
- else:
- return None
+ return None
def result_processor(self, dialect, coltype):
- wants_unicode = self._expect_unicode or dialect.convert_unicode
- needs_convert = wants_unicode and (
- dialect.returns_unicode_strings is not String.RETURNS_UNICODE
- or self._expect_unicode in ("force", "force_nocheck")
- )
- needs_isinstance = (
- needs_convert
- and dialect.returns_unicode_strings
- in (
- String.RETURNS_CONDITIONAL,
- String.RETURNS_UNICODE,
- )
- and self._expect_unicode != "force_nocheck"
- )
- if needs_convert:
- if needs_isinstance:
- return processors.to_conditional_unicode_processor_factory(
- dialect.encoding, self._expect_unicode_error
- )
- else:
- return processors.to_unicode_processor_factory(
- dialect.encoding, self._expect_unicode_error
- )
- else:
- return None
+ return None
@property
def python_type(self):
- if self._expect_unicode:
- return util.text_type
- else:
- return str
+ return util.text_type
def get_dbapi_type(self, dbapi):
return dbapi.STRING
- @classmethod
- def _warn_deprecated_unicode(cls):
- util.warn_deprecated(
- "The convert_unicode on Engine and String as well as the "
- "unicode_error flag on String are deprecated. All modern "
- "DBAPIs now support Python Unicode natively under Python 2, and "
- "under Python 3 all strings are inherently Unicode. These flags "
- "will be removed in a future release.",
- version="1.3",
- )
-
class Text(String):
"""A variably sized string type.
- In SQL, usually corresponds to CLOB or TEXT. Can also take Python
- unicode objects and encode to the database's encoding in bind
- params (and the reverse for result sets.) In general, TEXT objects
+ In SQL, usually corresponds to CLOB or TEXT. In general, TEXT objects
do not have a length; while some databases will accept a length
argument here, it will be rejected by others.
some backends implies an underlying column type that is explicitly
supporting of non-ASCII data, such as ``NVARCHAR`` on Oracle and SQL
Server. This will impact the output of ``CREATE TABLE`` statements and
- ``CAST`` functions at the dialect level, and also in some cases will
- indicate different behavior in the DBAPI itself in how it handles bound
- parameters.
+ ``CAST`` functions at the dialect level.
The character encoding used by the :class:`.Unicode` type that is used to
transmit and receive data to the database is usually determined by the
in the :ref:`dialect_toplevel` section.
In modern SQLAlchemy, use of the :class:`.Unicode` datatype does not
- typically imply any encoding/decoding behavior within SQLAlchemy itself.
- Historically, when DBAPIs did not support Python ``unicode`` objects under
- Python 2, SQLAlchemy handled unicode encoding/decoding services itself
- which would be controlled by the flag :paramref:`.String.convert_unicode`;
- this flag is deprecated as it is no longer needed for Python 3.
-
- When using Python 2, data that is passed to columns that use the
- :class:`.Unicode` datatype must be of type ``unicode``, and not ``str``
- which in Python 2 is equivalent to ``bytes``. In Python 3, all data
- passed to columns that use the :class:`.Unicode` datatype should be
- of type ``str``. See the flag :paramref:`.String.convert_unicode` for
- more discussion of unicode encode/decode behavior under Python 2.
+ imply any encoding/decoding behavior within SQLAlchemy itself. In Python
+ 3, all string objects are inherently Unicode capable, and SQLAlchemy
+ does not produce bytestring objects nor does it accommodate a DBAPI that
+ does not return Python Unicode objects in result sets for string values.
.. warning:: Some database backends, particularly SQL Server with pyodbc,
are known to have undesirable behaviors regarding data that is noted
:class:`.UnicodeText` - unlengthed textual counterpart
to :class:`.Unicode`.
- :paramref:`.String.convert_unicode`
-
:meth:`.DialectEvents.do_setinputsizes`
"""
Create a :class:`.Unicode` object.
- Parameters are the same as that of :class:`.String`,
- with the exception that ``convert_unicode``
- defaults to ``True``.
+ Parameters are the same as that of :class:`.String`.
"""
- kwargs.setdefault("_expect_unicode", True)
- kwargs.setdefault("_warn_on_bytestring", True)
super(Unicode, self).__init__(length=length, **kwargs)
"""
Create a Unicode-converting Text type.
- Parameters are the same as that of :class:`_expression.TextClause`,
- with the exception that ``convert_unicode``
- defaults to ``True``.
+ Parameters are the same as that of :class:`_expression.TextClause`.
"""
- kwargs.setdefault("_expect_unicode", True)
- kwargs.setdefault("_warn_on_bytestring", True)
super(UnicodeText, self).__init__(length=length, **kwargs)
- def _warn_deprecated_unicode(self):
- pass
-
class Integer(_LookupExpressionAdapter, TypeEngine):
__visit_name__ = "enum"
- @util.deprecated_params(
- convert_unicode=(
- "1.3",
- "The :paramref:`.Enum.convert_unicode` parameter is deprecated "
- "and will be removed in a future release. All modern DBAPIs "
- "now support Python Unicode directly and this parameter is "
- "unnecessary.",
- )
- )
def __init__(self, *enums, **kw):
r"""Construct an enum.
.. versionadded:: 1.1 a PEP-435 style enumerated class may be
passed.
- :param convert_unicode: Enable unicode-aware bind parameter and
- result-set processing for this Enum's data under Python 2 only.
- Under Python 2, this is set automatically based on the presence of
- unicode label strings. This flag will be removed in SQLAlchemy 2.0.
-
:param create_constraint: defaults to False. When creating a
non-native enumerated type, also build a CHECK constraint on the
database against the valid values.
values, objects = self._parse_into_values(enums, kw)
self._setup_for_values(values, objects, kw)
- convert_unicode = kw.pop("convert_unicode", None)
self.validate_strings = kw.pop("validate_strings", False)
- if convert_unicode is None:
- _expect_unicode = True
- else:
- _expect_unicode = convert_unicode
-
if self.enums:
length = max(len(x) for x in self.enums)
else:
self._valid_lookup[None] = self._object_lookup[None] = None
- super(Enum, self).__init__(
- length=length, _expect_unicode=_expect_unicode
- )
+ super(Enum, self).__init__(length=length)
if self.enum_class:
kw.setdefault("name", self.enum_class.__name__.lower())
op, other_comparator
)
if op is operators.concat_op:
- typ = String(
- self.type.length, _expect_unicode=self.type._expect_unicode
- )
+ typ = String(self.type.length)
return op, typ
comparator_factory = Comparator
return util.constructor_copy(self, self._generic_type_affinity, *args)
def adapt_to_emulated(self, impltype, **kw):
- kw.setdefault("_expect_unicode", self._expect_unicode)
kw.setdefault("validate_strings", self.validate_strings)
kw.setdefault("name", self.name)
kw.setdefault("schema", self.schema)
@util.memoized_property
def _str_impl(self):
- return String(_expect_unicode=True)
+ return String()
def bind_processor(self, dialect):
string_process = self._str_impl.bind_processor(dialect)
dbapi_key,
]
- platform_tokens.append(
- "nativeunicode"
- if config.db.dialect.convert_unicode
- else "dbapiunicode"
- )
+ platform_tokens.append("dbapiunicode")
_has_cext = has_compiled_ext()
platform_tokens.append(_has_cext and "cextensions" or "nocextensions")
return "_".join(platform_tokens)
from sqlalchemy.orm import subqueryload
from sqlalchemy.orm.session import _sessions
from sqlalchemy.processors import to_decimal_processor_factory
-from sqlalchemy.processors import to_unicode_processor_factory
from sqlalchemy.sql import column
from sqlalchemy.sql import util as sql_util
from sqlalchemy.sql.visitors import cloned_traverse
go()
- @testing.requires.cextensions
- def test_UnicodeResultProcessor_init(self):
- @profile_memory()
- def go():
- to_unicode_processor_factory("utf8")
-
- go()
-
@testing.requires.cextensions
def test_cycles_in_row(self):
from sqlalchemy import util
from sqlalchemy.dialects import mssql
from sqlalchemy.dialects.mssql import base
-from sqlalchemy.dialects.mssql.information_schema import CoerceUnicode
from sqlalchemy.dialects.mssql.information_schema import tables
from sqlalchemy.schema import CreateIndex
from sqlalchemy.testing import AssertsCompiledSQL
class InfoCoerceUnicodeTest(fixtures.TestBase, AssertsCompiledSQL):
- def test_info_unicode_coercion(self):
-
- dialect = mssql.dialect()
- value = CoerceUnicode().bind_processor(dialect)("a string")
- assert isinstance(value, util.text_type)
-
def test_info_unicode_cast_no_2000(self):
dialect = mssql.dialect()
dialect.server_version_info = base.MS_2000_VERSION
def test_older_cx_oracle_warning(self, cx_Oracle, cx_oracle_type):
cx_Oracle.version = "6.3"
- ignore_dialect = cx_oracle.dialect(
- dbapi=cx_Oracle, encoding_errors="ignore"
- )
- ignore_outputhandler = (
- ignore_dialect._generate_connection_outputtype_handler()
- )
-
- cursor = mock.Mock()
-
with testing.expect_warnings(
r"cx_oracle version \(6, 3\) does not support encodingErrors"
):
- ignore_outputhandler(
- cursor, "foo", cx_oracle_type, None, None, None
+ dialect = cx_oracle.dialect(
+ dbapi=cx_Oracle, encoding_errors="ignore"
)
+ eq_(dialect._cursor_var_unicode_kwargs, {})
+
@_oracle_char_combinations
def test_encoding_errors_cx_oracle(
self,
cursor = mock.Mock()
plain_outputhandler(cursor, "foo", cx_oracle_type, None, None, None)
- eq_(
- cursor.mock_calls,
- [mock.call.var(mock.ANY, None, cursor.arraysize)],
- )
+ if cx_oracle_type in (cx_Oracle.FIXED_CHAR, cx_Oracle.STRING):
+ # no calls; without encodingErrors, use cx_Oracle's default unicode
+ # handling
+ eq_(
+ cursor.mock_calls,
+ [],
+ )
+ else:
+ eq_(
+ cursor.mock_calls,
+ [mock.call.var(mock.ANY, None, cursor.arraysize)],
+ )
class ComputedReturningTest(fixtures.TablesTest):
eq_(sqla_result, cx_oracle_result)
- def test_coerce_to_unicode(self, connection):
- engine = testing_engine(options=dict(coerce_to_unicode=False))
- with engine.connect() as conn_no_coerce:
- value = exec_sql(
- conn_no_coerce, "SELECT 'hello' FROM DUAL"
- ).scalar()
- assert not isinstance(value, util.binary_type)
- assert isinstance(value, util.text_type)
-
- value = exec_sql(connection, "SELECT 'hello' FROM DUAL").scalar()
- assert isinstance(value, util.text_type)
-
def test_reflect_dates(self, metadata, connection):
Table(
"date_types",
future_connection.dialect.server_version_info,
)
- @testing.requires.psycopg2_compatibility
- def test_pg_dialect_no_native_unicode_in(self, testing_engine):
- with testing.expect_raises_message(
- exc.ArgumentError,
- "psycopg2 native_unicode mode is required under Python 3",
- ):
- testing_engine(options=dict(use_native_unicode=False))
-
def test_psycopg2_empty_connection_string(self):
dialect = psycopg2_dialect.dialect()
u = url.make_url("postgresql+psycopg2://")
result = connection.execute(
select(data_table.c.data["k1"].astext)
).first()
- if connection.dialect.returns_unicode_strings:
- assert isinstance(result[0], util.text_type)
- else:
- assert isinstance(result[0], util.string_types)
+ assert isinstance(result[0], util.text_type)
def test_query_returned_as_int(self, connection):
self._fixture_data(connection)
from sqlalchemy import Sequence
from sqlalchemy import String
from sqlalchemy import testing
-from sqlalchemy import Unicode
from sqlalchemy.schema import CreateTable
from sqlalchemy.sql import literal_column
from sqlalchemy.sql import select
from sqlalchemy.testing.schema import Table
from sqlalchemy.types import TypeDecorator
from sqlalchemy.types import TypeEngine
-from sqlalchemy.util import b
-from sqlalchemy.util import u
class DDLTest(fixtures.TestBase, AssertsCompiledSQL):
eq_(list(connection.execute(t.select())), [(5, "data")])
-class UnicodeDefaultsTest(fixtures.TestBase):
- __backend__ = True
-
- def test_no_default(self):
- Column(Unicode(32))
-
- def test_unicode_default(self):
- default = u("foo")
- Column(Unicode(32), default=default)
-
- def test_nonunicode_default(self):
- default = b("foo")
- assert_raises_message(
- sa.exc.SAWarning,
- "Unicode column 'foobar' has non-unicode "
- "default value b?'foo' specified.",
- Column,
- "foobar",
- Unicode(32),
- default=default,
- )
-
-
class InsertFromSelectTest(fixtures.TablesTest):
__backend__ = True
from sqlalchemy import case
from sqlalchemy import CHAR
from sqlalchemy import column
-from sqlalchemy import create_engine
from sqlalchemy import exc
from sqlalchemy import exists
from sqlalchemy import ForeignKey
from sqlalchemy import table
from sqlalchemy import testing
from sqlalchemy import text
-from sqlalchemy import util
from sqlalchemy.engine import default
from sqlalchemy.sql import coercions
from sqlalchemy.sql import LABEL_STYLE_TABLENAME_PLUS_COL
):
preparer.quote_schema("hi", True)
- def test_string_convert_unicode(self):
- with testing.expect_deprecated(
- "The String.convert_unicode parameter is deprecated and "
- "will be removed in a future release."
- ):
- String(convert_unicode=True)
-
- def test_string_convert_unicode_force(self):
- with testing.expect_deprecated(
- "The String.convert_unicode parameter is deprecated and "
- "will be removed in a future release."
- ):
- String(convert_unicode="force")
-
- def test_engine_convert_unicode(self):
- with testing.expect_deprecated(
- "The create_engine.convert_unicode parameter and "
- "corresponding dialect-level"
- ):
- create_engine(
- "mysql+mysqldb://", convert_unicode=True, module=mock.Mock()
- )
-
def test_empty_and_or(self):
with testing.expect_deprecated(
r"Invoking and_\(\) without arguments is deprecated, and "
)
-class ConvertUnicodeDeprecationTest(fixtures.TestBase):
-
- __backend__ = True
-
- data = util.u(
- "Alors vous imaginez ma surprise, au lever du jour, quand "
- "une drôle de petite voix m’a réveillé. "
- "Elle disait: « S’il vous plaît… dessine-moi un mouton! »"
- )
-
- def test_unicode_warnings_dialectlevel(self):
-
- unicodedata = self.data
-
- with testing.expect_deprecated(
- "The create_engine.convert_unicode parameter and "
- "corresponding dialect-level"
- ):
- dialect = default.DefaultDialect(convert_unicode=True)
- dialect.supports_unicode_binds = False
-
- s = String()
- uni = s.dialect_impl(dialect).bind_processor(dialect)
-
- uni(util.b("x"))
- assert isinstance(uni(unicodedata), util.binary_type)
-
- eq_(uni(unicodedata), unicodedata.encode("utf-8"))
-
- def test_ignoring_unicode_error(self):
- """checks String(unicode_error='ignore') is passed to
- underlying codec."""
-
- unicodedata = self.data
-
- with testing.expect_deprecated(
- "The String.convert_unicode parameter is deprecated and "
- "will be removed in a future release.",
- "The String.unicode_errors parameter is deprecated and "
- "will be removed in a future release.",
- ):
- type_ = String(
- 248, convert_unicode="force", unicode_error="ignore"
- )
- dialect = default.DefaultDialect(encoding="ascii")
- proc = type_.result_processor(dialect, 10)
-
- utfdata = unicodedata.encode("utf8")
- eq_(proc(utfdata), unicodedata.encode("ascii", "ignore").decode())
-
-
class SubqueryCoercionsTest(fixtures.TestBase, AssertsCompiledSQL):
__dialect__ = "default"
from sqlalchemy.testing import eq_
from sqlalchemy.testing import expect_deprecated_20
from sqlalchemy.testing import expect_raises
-from sqlalchemy.testing import expect_warnings
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
from sqlalchemy.testing import is_not
from sqlalchemy.testing.schema import Table
from sqlalchemy.testing.util import picklers
from sqlalchemy.testing.util import round_decimal
-from sqlalchemy.util import u
def _all_dialect_modules():
t2 = t1.adapt(Text)
eq_(t2.length, 50)
- def test_convert_unicode_text_type(self):
- with testing.expect_deprecated(
- "The String.convert_unicode parameter is deprecated"
- ):
- eq_(types.String(convert_unicode=True).python_type, util.text_type)
-
class TypeAffinityTest(fixtures.TestBase):
@testing.combinations(
eq_(a.dialect_specific_args["bar"], "bar")
-class StringConvertUnicodeTest(fixtures.TestBase):
- @testing.combinations((Unicode,), (String,), argnames="datatype")
- @testing.combinations((True,), (False,), argnames="convert_unicode")
- @testing.combinations(
- (String.RETURNS_CONDITIONAL,),
- (String.RETURNS_BYTES,),
- (String.RETURNS_UNICODE),
- argnames="returns_unicode_strings",
- )
- def test_convert_unicode(
- self, datatype, convert_unicode, returns_unicode_strings
- ):
- s1 = datatype()
- dialect = mock.Mock(
- returns_unicode_strings=returns_unicode_strings,
- encoding="utf-8",
- convert_unicode=convert_unicode,
- )
-
- proc = s1.result_processor(dialect, None)
-
- string = u("méil")
- bytestring = string.encode("utf-8")
-
- if (
- datatype is Unicode or convert_unicode
- ) and returns_unicode_strings in (
- String.RETURNS_CONDITIONAL,
- String.RETURNS_BYTES,
- ):
- eq_(proc(bytestring), string)
-
- if returns_unicode_strings is String.RETURNS_CONDITIONAL:
- eq_(proc(string), string)
- else:
- if util.py3k:
- # trying to decode a unicode
- assert_raises(TypeError, proc, string)
- else:
- assert_raises(UnicodeEncodeError, proc, string)
- else:
- is_(proc, None)
-
-
class TypeCoerceCastTest(fixtures.TablesTest):
__backend__ = True
)
-class UnicodeTest(fixtures.TestBase):
-
- """Exercise the Unicode and related types.
-
- Note: unicode round trip tests are now in
- sqlalchemy/testing/suite/test_types.py.
-
- """
-
- __backend__ = True
-
- data = util.u(
- "Alors vous imaginez ma surprise, au lever du jour, quand "
- "une drôle de petite voix m’a réveillé. "
- "Elle disait: « S’il vous plaît… dessine-moi un mouton! »"
- )
-
- def test_unicode_warnings_typelevel_native_unicode(self):
-
- unicodedata = self.data
- u = Unicode()
- dialect = default.DefaultDialect()
- dialect.supports_unicode_binds = True
- uni = u.dialect_impl(dialect).bind_processor(dialect)
- if util.py3k:
- assert_raises(exc.SAWarning, uni, b"x")
- assert isinstance(uni(unicodedata), str)
- else:
- assert_raises(exc.SAWarning, uni, "x")
- assert isinstance(uni(unicodedata), unicode) # noqa
-
- def test_unicode_warnings_typelevel_sqla_unicode(self):
- unicodedata = self.data
- u = Unicode()
- dialect = default.DefaultDialect()
- dialect.supports_unicode_binds = False
- uni = u.dialect_impl(dialect).bind_processor(dialect)
- assert_raises(exc.SAWarning, uni, util.b("x"))
- assert isinstance(uni(unicodedata), util.binary_type)
-
- eq_(uni(unicodedata), unicodedata.encode("utf-8"))
-
- def test_unicode_warnings_totally_wrong_type(self):
- u = Unicode()
- dialect = default.DefaultDialect()
- dialect.supports_unicode_binds = False
- uni = u.dialect_impl(dialect).bind_processor(dialect)
- with expect_warnings(
- "Unicode type received non-unicode bind param value 5."
- ):
- eq_(uni(5), 5)
-
-
class EnumTest(AssertsCompiledSQL, fixtures.TablesTest):
__backend__ = True
# depending on backend.
assert "('x'," in e.print_sql()
- @testing.uses_deprecated(".*convert_unicode")
def test_repr(self):
e = Enum(
"x",
"y",
name="somename",
- convert_unicode=True,
quote=True,
inherit_schema=True,
native_enum=False,
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
