:ticket:`4242`
+.. _change_4369:
+
+cx_Oracle connect arguments modernized, deprecated parameters removed
+---------------------------------------------------------------------
+
+A series of modernizations to the parameters accepted by the cx_oracle
+dialect as well as the URL string:
+
+* The deprecated paramters ``auto_setinputsizes``, ``allow_twophase``,
+ ``exclude_setinputsizes`` are removed.
+
+* The value of the ``threaded`` parameter, which has always been defaulted
+ to True for the SQLAlchemy dialect, is no longer generated by default.
+ The SQLAlchemy :class:`.Connection` object is not considered to be thread-safe
+ itself so there's no need for this flag to be passed.
+
+* It's deprecated to pass ``threaded`` to :func:`.create_engine` itself.
+ To set the value of ``threaded`` to ``True``, pass it to either the
+ :paramref:`.create_engine.connect_args` dictionary or use the query
+ string e.g. ``oracle+cx_oracle://...?threaded=true``.
+
+* All parameters passed on the URL query string that are not otherwise
+ specially consumed are now passed to the cx_Oracle.connect() function.
+ A selection of these are also coerced either into cx_Oracle constants
+ or booleans including ``mode``, ``purity``, ``events``, and ``threaded``.
+
+* As was the case earlier, all cx_Oracle ``.connect()`` arguments are accepted
+ via the :paramref:`.create_engine.connect_args` dictionary, the documentation
+ was inaccurate regarding this.
+
+:ticket:`4369`
+
Dialect Improvements and Changes - SQL Server
=============================================
--- /dev/null
+.. change::
+ :tags: bug, oracle
+ :tickets: 4369
+
+ Updated the parameters that can be sent to the cx_Oracle DBAPI to both allow
+ for all current parameters as well as for future parameters not added yet.
+ In addition, removed unused parameters that were deprecated in version 1.2,
+ and additionally we are now defaulting "threaded" to False.
+
+ .. seealso::
+
+ :ref:`change_4369`
}
)
+Alternatively, most cx_Oracle DBAPI arguments can also be encoded as strings
+within the URL, which includes parameters such as ``mode``, ``purity``,
+``events``, ``threaded``, and others::
+
+ e = create_engine("oracle+cx_oracle://user:pass@dsn?mode=SYSDBA&events=true")
+
+.. versionchanged:: 1.3 the cx_oracle dialect now accepts all argument names
+ within the URL string itself, to be passed to the cx_Oracle DBAPI. As
+ was the case earlier but not correctly documented, the
+ :paramref:`.create_engine.connect_args` parameter also accepts all
+ cx_Oracle DBAPI connect arguments.
+
There are also options that are consumed by the SQLAlchemy cx_oracle dialect
itself. These options are always passed directly to :func:`.create_engine`,
such as::
* ``coerce_to_decimal`` - see :ref:`cx_oracle_numeric` for detail.
-* ``threaded`` - this parameter is passed as the value of "threaded" to
- ``cx_Oracle.connect()`` and defaults to True, which is the opposite of
- cx_Oracle's default. This parameter is deprecated and will default to
- ``False`` in version 1.3 of SQLAlchemy.
-
.. _cx_oracle_unicode:
Unicode
execute_sequence_format = list
+ _cx_oracle_threaded = None
+
def __init__(self,
auto_convert_lobs=True,
- threaded=True,
coerce_to_unicode=True,
coerce_to_decimal=True,
arraysize=50,
+ threaded=None,
**kwargs):
- self._pop_deprecated_kwargs(kwargs)
-
OracleDialect.__init__(self, **kwargs)
- self.threaded = threaded
self.arraysize = arraysize
+ if threaded is not None:
+ util.warn_deprecated(
+ "The 'threaded' parameter to the cx_oracle dialect "
+ "itself is deprecated. The value now defaults to False in "
+ "any case. To pass an explicit True value, use the "
+ "create_engine connect_args dictionary or add ?threaded=true "
+ "to the URL string."
+ )
+ 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
return on_connect
def create_connect_args(self, url):
- dialect_opts = dict(url.query)
+ opts = dict(url.query)
- for opt in ('use_ansi', 'auto_setinputsizes', 'auto_convert_lobs',
- 'threaded', 'allow_twophase'):
- if opt in dialect_opts:
- util.coerce_kw_type(dialect_opts, opt, bool)
- setattr(self, opt, dialect_opts[opt])
+ for opt in ('use_ansi', 'auto_convert_lobs'):
+ if opt in opts:
+ util.warn_deprecated(
+ "cx_oracle dialect option %r should only be passed to "
+ "create_engine directly, not within the URL string" % opt)
+ util.coerce_kw_type(opts, opt, bool)
+ setattr(self, opt, opts.pop(opt))
database = url.database
- service_name = dialect_opts.get('service_name', None)
+ service_name = opts.pop('service_name', None)
if database or service_name:
# if we have a database, then we have a remote host
port = url.port
# we have a local tnsname
dsn = url.host
- opts = dict(
- threaded=self.threaded,
- )
-
if dsn is not None:
opts['dsn'] = dsn
if url.password is not None:
if url.username is not None:
opts['user'] = url.username
- if 'mode' in url.query:
- opts['mode'] = url.query['mode']
- if isinstance(opts['mode'], util.string_types):
- mode = opts['mode'].upper()
- if mode == 'SYSDBA':
- opts['mode'] = self.dbapi.SYSDBA
- elif mode == 'SYSOPER':
- opts['mode'] = self.dbapi.SYSOPER
+ if self._cx_oracle_threaded is not None:
+ opts.setdefault("threaded", self._cx_oracle_threaded)
+
+ def convert_cx_oracle_constant(value):
+ if isinstance(value, util.string_types):
+ try:
+ int_val = int(value)
+ except ValueError:
+ value = value.upper()
+ return getattr(self.dbapi, value)
else:
- util.coerce_kw_type(opts, 'mode', int)
+ return int_val
+ else:
+ return value
+
+ util.coerce_kw_type(opts, 'mode', convert_cx_oracle_constant)
+ util.coerce_kw_type(opts, 'threaded', bool)
+ util.coerce_kw_type(opts, 'events', bool)
+ util.coerce_kw_type(opts, 'purity', convert_cx_oracle_constant)
+
return ([], opts)
def _get_server_version_info(self, connection):
when coercing to boolean.
"""
- if key in kw and not isinstance(kw[key], type_) and kw[key] is not None:
+ if key in kw and (
+ not isinstance(type_, type) or not isinstance(kw[key], type_)
+ ) and kw[key] is not None:
if type_ is bool and flexi_bool:
kw[key] = asbool(kw[key])
else:
from sqlalchemy.testing import eq_
-from sqlalchemy import types as sqltypes, exc, schema
-from sqlalchemy.sql import table, column
+from sqlalchemy import exc
from sqlalchemy.testing import (fixtures,
AssertsExecutionResults,
AssertsCompiledSQL)
from sqlalchemy import testing
-from sqlalchemy import Integer, Text, LargeBinary, Unicode, UniqueConstraint,\
- Index, MetaData, select, inspect, ForeignKey, String, func, \
- TypeDecorator, bindparam, Numeric, TIMESTAMP, CHAR, text, \
- literal_column, VARCHAR, create_engine, Date, NVARCHAR, \
- ForeignKeyConstraint, Sequence, Float, DateTime, cast, UnicodeText, \
- union, except_, type_coerce, or_, outerjoin, DATE, NCHAR, outparam, \
- PrimaryKeyConstraint, FLOAT
-from sqlalchemy.util import u, b
-from sqlalchemy import util
+from sqlalchemy import create_engine
+from sqlalchemy import bindparam, outparam
+from sqlalchemy import text, Float, Integer, String, select, literal_column,\
+ Unicode, UnicodeText, Sequence
+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
-from sqlalchemy.engine import default
-import decimal
from sqlalchemy.engine import url
from sqlalchemy.testing.schema import Table, Column
-import datetime
-import os
-from sqlalchemy import sql
from sqlalchemy.testing.mock import Mock
from sqlalchemy.testing import mock
-from sqlalchemy import exc
+
class DialectTest(fixtures.TestBase):
def test_cx_oracle_version_parse(self):
eq_(result, u('’é'))
-class ServiceNameTest(fixtures.TestBase):
+class CXOracleConnectArgsTest(fixtures.TestBase):
__only_on__ = 'oracle+cx_oracle'
__backend__ = True
_initialize=False
)
+ def _test_db_opt(self, url_string, key, value):
+ import cx_Oracle
+ url_obj = url.make_url(url_string)
+ dialect = cx_oracle.dialect(dbapi=cx_Oracle)
+ arg, kw = dialect.create_connect_args(url_obj)
+ eq_(kw[key], value)
+
+ def _test_db_opt_unpresent(self, url_string, key):
+ import cx_Oracle
+ url_obj = url.make_url(url_string)
+ dialect = cx_oracle.dialect(dbapi=cx_Oracle)
+ arg, kw = dialect.create_connect_args(url_obj)
+ assert key not in kw
+
+ def _test_dialect_param_from_url(self, url_string, key, value):
+ import cx_Oracle
+ url_obj = url.make_url(url_string)
+ dialect = cx_oracle.dialect(dbapi=cx_Oracle)
+ with testing.expect_deprecated(
+ "cx_oracle dialect option %r should" % key):
+ arg, kw = dialect.create_connect_args(url_obj)
+ eq_(getattr(dialect, key), value)
+
+ # test setting it on the dialect normally
+ dialect = cx_oracle.dialect(dbapi=cx_Oracle, **{key: value})
+ eq_(getattr(dialect, key), value)
+
+ def test_mode(self):
+ import cx_Oracle
+ self._test_db_opt(
+ 'oracle+cx_oracle://scott:tiger@host/?mode=sYsDBA',
+ "mode",
+ cx_Oracle.SYSDBA
+ )
+
+ self._test_db_opt(
+ 'oracle+cx_oracle://scott:tiger@host/?mode=SYSOPER',
+ "mode",
+ cx_Oracle.SYSOPER
+ )
+
+ def test_int_mode(self):
+ self._test_db_opt(
+ 'oracle+cx_oracle://scott:tiger@host/?mode=32767',
+ "mode",
+ 32767
+ )
+
+ @testing.requires.cxoracle6_or_greater
+ def test_purity(self):
+ import cx_Oracle
+ self._test_db_opt(
+ 'oracle+cx_oracle://scott:tiger@host/?purity=attr_purity_new',
+ "purity",
+ cx_Oracle.ATTR_PURITY_NEW
+ )
+
+ def test_encoding(self):
+ self._test_db_opt(
+ "oracle+cx_oracle://scott:tiger@host/"
+ "?encoding=AMERICAN_AMERICA.UTF8",
+ "encoding",
+ "AMERICAN_AMERICA.UTF8"
+ )
+
+ def test_threaded(self):
+ self._test_db_opt(
+ 'oracle+cx_oracle://scott:tiger@host/?threaded=true',
+ "threaded",
+ True
+ )
+
+ self._test_db_opt_unpresent(
+ 'oracle+cx_oracle://scott:tiger@host/',
+ "threaded"
+ )
+
+ def test_events(self):
+ self._test_db_opt(
+ 'oracle+cx_oracle://scott:tiger@host/?events=true',
+ "events",
+ True
+ )
+
+ def test_threaded_deprecated_at_dialect_level(self):
+ with testing.expect_deprecated(
+ "The 'threaded' parameter to the cx_oracle dialect"):
+ dialect = cx_oracle.dialect(threaded=False)
+ arg, kw = dialect.create_connect_args(
+ url.make_url("oracle+cx_oracle://scott:tiger@dsn"))
+ eq_(kw['threaded'], False)
+
+ def test_deprecated_use_ansi(self):
+ self._test_dialect_param_from_url(
+ 'oracle+cx_oracle://scott:tiger@host/?use_ansi=False',
+ 'use_ansi', False
+ )
+
+ def test_deprecated_auto_convert_lobs(self):
+ self._test_dialect_param_from_url(
+ 'oracle+cx_oracle://scott:tiger@host/?auto_convert_lobs=False',
+ 'auto_convert_lobs', False
+ )
\ No newline at end of file
config.db.scalar("show server_encoding").lower() == "utf8"
)
+ @property
+ def cxoracle6_or_greater(self):
+ return only_if(
+ lambda config: against(config, "oracle+cx_oracle") and
+ config.db.dialect.cx_oracle_ver >= (6, )
+ )
+
@property
def oracle5x(self):
return only_if(
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
