aiosqlite_file sqlite+aiosqlite:///async_querytest.db
asyncmy mysql+asyncmy://scott:tiger@127.0.0.1:3306/test?charset=utf8mb4
asyncmy_fallback mysql+asyncmy://scott:tiger@127.0.0.1:3306/test?charset=utf8mb4&async_fallback=true
- asyncpg postgresql+asyncpg://scott:tiger@192.168.0.199/test
- asyncpg_async_fallback postgresql+asyncpg://scott:tiger@192.168.0.199/test?async_fallback=true
+ asyncpg postgresql+asyncpg://scott:tiger@127.0.0.1:5432/test
asyncpg_fallback postgresql+asyncpg://scott:tiger@127.0.0.1:5432/test?async_fallback=true
default sqlite:///:memory:
docker_mssql mssql+pymssql://scott:tiger^5HHH@127.0.0.1:1433/test
mariadb mariadb+mysqldb://scott:tiger@127.0.0.1:3306/test
- mssql mssql+pyodbc://scott:tiger^5HHH@localhost:1433/test?driver=ODBC+Driver+17+for+SQL+Server
- mssql_199 mssql+pyodbc://scott:tiger^5HHH@192.168.0.199:1433/test?driver=ODBC+Driver+17+for+SQL+Server
+ mariadb_connector mariadb+mariadbconnector://scott:tiger@127.0.0.1:3306/test
+ mssql mssql+pyodbc://scott:tiger^5HHH@mssql2017:1433/test?driver=ODBC+Driver+13+for+SQL+Server
mssql_pymssql mssql+pymssql://scott:tiger@ms_2008
mysql mysql+mysqldb://scott:tiger@127.0.0.1:3306/test?charset=utf8mb4
- oracle oracle+cx_oracle://scott:tiger@127.0.0.1:1521
- oracle8 oracle+cx_oracle://scott:tiger@127.0.0.1:1521/?use_ansi=0
+ oracle oracle+cx_oracle://scott:tiger@oracle18c
+ oracle_oracledb oracle+oracledb://scott:tiger@oracle18c
pg8000 postgresql+pg8000://scott:tiger@127.0.0.1:5432/test
- postgresql postgresql+psycopg2://scott:tiger@192.168.0.199/test
+ postgresql postgresql+psycopg2://scott:tiger@127.0.0.1:5432/test
postgresql_psycopg2cffi postgresql+psycopg2cffi://scott:tiger@127.0.0.1:5432/test
+ psycopg postgresql+psycopg://scott:tiger@127.0.0.1:5432/test
+ psycopg2 postgresql+psycopg2://scott:tiger@127.0.0.1:5432/test
+ psycopg_async postgresql+psycopg_async://scott:tiger@127.0.0.1:5432/test
+ psycopg_async_fallback postgresql+psycopg_async://scott:tiger@127.0.0.1:5432/test?async_fallback=true
pymysql mysql+pymysql://scott:tiger@127.0.0.1:3306/test?charset=utf8mb4
pysqlcipher_file sqlite+pysqlcipher://:test@/querytest.db.enc
sqlite sqlite:///:memory:
--- /dev/null
+.. change::
+ :tags: oracle, feature
+ :tickets: 8054
+
+ Add support for the new oracle driver ``oracledb``.
.. automodule:: sqlalchemy.dialects.oracle.cx_oracle
+python-oracledb
+---------------
+
+.. automodule:: sqlalchemy.dialects.oracle.oracledb
+
from . import base # noqa
from . import cx_oracle # noqa
+from . import oracledb # noqa
from .base import BFILE
from .base import BINARY_DOUBLE
from .base import BINARY_FLOAT
@util.deprecated_params(
threaded=(
"1.3",
- "The 'threaded' parameter to the cx_oracle dialect "
+ "The 'threaded' parameter to the cx_oracle/oracledb dialect "
"is deprecated as a dialect-level argument, and will be removed "
"in a future release. As of version 1.3, it defaults to False "
"rather than True. The 'threaded' option can be passed to "
self.colspecs[sqltypes.Unicode] = _OracleUnicodeStringNCHAR
self.colspecs[sqltypes.UnicodeText] = _OracleUnicodeTextNCLOB
- cx_Oracle = self.dbapi
-
- if cx_Oracle is None:
- self.cx_oracle_ver = (0, 0, 0)
- else:
- self.cx_oracle_ver = self._parse_cx_oracle_ver(cx_Oracle.version)
- if self.cx_oracle_ver < (7,) and self.cx_oracle_ver > (0, 0, 0):
- raise exc.InvalidRequestError(
- "cx_Oracle version 7 and above are supported"
- )
+ dbapi_module = self.dbapi
+ self._load_version(dbapi_module)
+ if dbapi_module is not None:
self.include_set_input_sizes = {
- cx_Oracle.DATETIME,
- cx_Oracle.NCLOB,
- cx_Oracle.CLOB,
- cx_Oracle.LOB,
- cx_Oracle.NCHAR,
- cx_Oracle.FIXED_NCHAR,
- cx_Oracle.BLOB,
- cx_Oracle.FIXED_CHAR,
- cx_Oracle.TIMESTAMP,
+ dbapi_module.DATETIME,
+ dbapi_module.NCLOB,
+ dbapi_module.CLOB,
+ dbapi_module.LOB,
+ dbapi_module.NCHAR,
+ dbapi_module.FIXED_NCHAR,
+ dbapi_module.BLOB,
+ dbapi_module.FIXED_CHAR,
+ dbapi_module.TIMESTAMP,
int, # _OracleInteger,
# _OracleBINARY_FLOAT, _OracleBINARY_DOUBLE,
- cx_Oracle.NATIVE_FLOAT,
+ dbapi_module.NATIVE_FLOAT,
}
self._paramval = lambda value: value.getvalue()
- def _parse_cx_oracle_ver(self, version):
- m = re.match(r"(\d+)\.(\d+)(?:\.(\d+))?", version)
- if m:
- return tuple(int(x) for x in m.group(1, 2, 3) if x is not None)
- else:
- return (0, 0, 0)
+ def _load_version(self, dbapi_module):
+ version = (0, 0, 0)
+ if dbapi_module is not None:
+ m = re.match(r"(\d+)\.(\d+)(?:\.(\d+))?", dbapi_module.version)
+ if m:
+ version = tuple(
+ int(x) for x in m.group(1, 2, 3) if x is not None
+ )
+ self.cx_oracle_ver = version
+ if self.cx_oracle_ver < (7,) and self.cx_oracle_ver > (0, 0, 0):
+ raise exc.InvalidRequestError(
+ "cx_Oracle version 7 and above are supported"
+ )
@classmethod
def import_dbapi(cls):
return cx_Oracle
def initialize(self, connection):
- super(OracleDialect_cx_oracle, self).initialize(connection)
+ super().initialize(connection)
self._detect_decimal_char(connection)
def get_isolation_level(self, dbapi_connection):
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,
+ f"{self.driver} dialect option {opt!r} should only be "
+ "passed to create_engine directly, not within the URL "
+ "string",
version="1.3",
)
util.coerce_kw_type(opts, opt, bool)
--- /dev/null
+# Copyright (C) 2005-2022 the SQLAlchemy authors and contributors
+# <see AUTHORS file>
+#
+# This module is part of SQLAlchemy and is released under
+# the MIT License: https://www.opensource.org/licenses/mit-license.php
+# mypy: ignore-errors
+
+r"""
+.. dialect:: oracle+oracledb
+ :name: python-oracledb
+ :dbapi: oracledb
+ :connectstring: oracle+oracledb://user:pass@hostname:port[/dbname][?service_name=<service>[&key=value&key=value...]]
+ :url: https://oracle.github.io/python-oracledb/
+
+python-oracledb is released by Oracle to supersede the cx_Oracle driver.
+It is fully compatible with cx_Oracle and features both a "thin" client
+mode that requires no dependencies, as well as a "thick" mode that uses
+the Oracle Client Interface in the same way as cx_Oracle.
+
+.. seealso::
+
+ :ref:`cx_oracle` - all of cx_Oracle's notes apply to the oracledb driver
+ as well.
+
+Thick mode support
+------------------
+
+By default the ``python-oracledb`` is started in thin mode, that does not
+require oracle client libraries to be installed in the system. The
+``python-oracledb`` driver also support a "thick" mode, that behaves
+similarly to ``cx_oracle`` and requires that Oracle Client Interface (OCI)
+is installed.
+
+To enable this mode, the user may call ``oracledb.init_oracle_client``
+manually, or by passing the parameter ``thick_mode=True`` to
+:func:`_sa.create_engine`. To pass custom arguments to ``init_oracle_client``,
+like the ``lib_dir`` path, a dict may be passed to this parameter, as in::
+
+ engine = sa.create_engine("oracle+oracledb://...", thick_mode={
+ "lib_dir": "/path/to/oracle/client/lib", "driver_name": "my-app"
+ })
+
+.. seealso::
+
+ https://python-oracledb.readthedocs.io/en/latest/api_manual/module.html#oracledb.init_oracle_client
+
+
+.. versionadded:: 2.0.0 added support for oracledb driver.
+
+""" # noqa
+import re
+
+from .cx_oracle import OracleDialect_cx_oracle as _OracleDialect_cx_oracle
+from ... import exc
+
+
+class OracleDialect_oracledb(_OracleDialect_cx_oracle):
+ supports_statement_cache = True
+ driver = "oracledb"
+
+ def __init__(
+ self,
+ auto_convert_lobs=True,
+ coerce_to_decimal=True,
+ arraysize=50,
+ encoding_errors=None,
+ thick_mode=None,
+ **kwargs,
+ ):
+
+ super().__init__(
+ auto_convert_lobs,
+ coerce_to_decimal,
+ arraysize,
+ encoding_errors,
+ **kwargs,
+ )
+
+ if thick_mode is not None:
+ kw = thick_mode if isinstance(thick_mode, dict) else {}
+ self.dbapi.init_oracle_client(**kw)
+
+ @classmethod
+ def import_dbapi(cls):
+ import oracledb
+
+ return oracledb
+
+ @classmethod
+ def is_thin_mode(cls, connection):
+ return connection.connection.dbapi_connection.thin
+
+ def _load_version(self, dbapi_module):
+ version = (0, 0, 0)
+ if dbapi_module is not None:
+ m = re.match(r"(\d+)\.(\d+)(?:\.(\d+))?", dbapi_module.version)
+ if m:
+ version = tuple(
+ int(x) for x in m.group(1, 2, 3) if x is not None
+ )
+ self.oracledb_ver = version
+ if self.oracledb_ver < (1,) and self.oracledb_ver > (0, 0, 0):
+ raise exc.InvalidRequestError(
+ "oracledb version 1 and above are supported"
+ )
+
+
+dialect = OracleDialect_oracledb
mariadb>=1.0.1
oracle =
cx_oracle>=7
+oracle_oracledb =
+ oracledb>=1
postgresql = psycopg2>=2.7
postgresql_pg8000 = pg8000>=1.16.6,!=1.29.0
postgresql_asyncpg =
mssql = mssql+pyodbc://scott:tiger^5HHH@mssql2017:1433/test?driver=ODBC+Driver+13+for+SQL+Server
mssql_pymssql = mssql+pymssql://scott:tiger@ms_2008
docker_mssql = mssql+pymssql://scott:tiger^5HHH@127.0.0.1:1433/test
-oracle = oracle+cx_oracle://scott:tiger@oracle18c
+oracle = oracle+cx_oracle://scott:tiger@oracle18c/xe
+cxoracle = oracle+cx_oracle://scott:tiger@oracle18c/xe
+oracle_oracledb = oracle+oracledb://scott:tiger@oracle18c/xe
+oracledb = oracle+oracledb://scott:tiger@oracle18c/xe
--- /dev/null
+# do not import sqlalchemy testing feature in this file, since it's
+# run directly, not passing through pytest
+from sqlalchemy import create_engine
+
+
+def _get_version(conn):
+ # this is the suggested way of finding the mode, from
+ # https://python-oracledb.readthedocs.io/en/latest/user_guide/tracing.html#vsessconinfo
+ sql = (
+ "SELECT UNIQUE CLIENT_DRIVER "
+ "FROM V$SESSION_CONNECT_INFO "
+ "WHERE SID = SYS_CONTEXT('USERENV', 'SID')"
+ )
+ return conn.exec_driver_sql(sql).scalar()
+
+
+def run_thin_mode(url, queue):
+ e = create_engine(url)
+ with e.connect() as conn:
+ res = _get_version(conn)
+ queue.put((res, e.dialect.is_thin_mode(conn)))
+ e.dispose()
+
+
+def run_thick_mode(url, queue):
+ e = create_engine(url, thick_mode={"driver_name": "custom-driver-name"})
+ with e.connect() as conn:
+ res = _get_version(conn)
+ queue.put((res, e.dialect.is_thin_mode(conn)))
+ e.dispose()
from sqlalchemy import union
from sqlalchemy.dialects.oracle import base as oracle
from sqlalchemy.dialects.oracle import cx_oracle
+from sqlalchemy.dialects.oracle import oracledb
from sqlalchemy.engine import default
from sqlalchemy.sql import column
from sqlalchemy.sql import ddl
"(__[POSTCOMPILE_uid])",
dialect=cx_oracle.dialect(),
)
+ self.assert_compile(
+ bindparam("uid", expanding=True),
+ "(__[POSTCOMPILE_uid])",
+ dialect=oracledb.dialect(),
+ )
def test_cte(self):
part = table(
# coding: utf-8
+from multiprocessing import get_context
import re
from unittest import mock
from unittest.mock import Mock
from sqlalchemy import UnicodeText
from sqlalchemy.dialects.oracle import base as oracle
from sqlalchemy.dialects.oracle import cx_oracle
+from sqlalchemy.dialects.oracle import oracledb
from sqlalchemy.engine import url
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import engines
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
+from sqlalchemy.testing import is_false
+from sqlalchemy.testing import is_true
from sqlalchemy.testing.assertions import expect_raises_message
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import pep435_enum
from sqlalchemy.testing.suite import test_select
-class DialectTest(fixtures.TestBase):
+class CxOracleDialectTest(fixtures.TestBase):
def test_cx_oracle_version_parse(self):
dialect = cx_oracle.OracleDialect_cx_oracle()
- eq_(dialect._parse_cx_oracle_ver("5.2"), (5, 2))
+ def check(version):
+ dbapi = Mock(version=version)
+ dialect._load_version(dbapi)
+ return dialect.cx_oracle_ver
- eq_(dialect._parse_cx_oracle_ver("5.0.1"), (5, 0, 1))
-
- eq_(dialect._parse_cx_oracle_ver("6.0b1"), (6, 0))
+ eq_(check("7.2"), (7, 2))
+ eq_(check("7.0.1"), (7, 0, 1))
+ eq_(check("9.0b1"), (9, 0))
def test_minimum_version(self):
- with mock.patch(
- "sqlalchemy.dialects.oracle.cx_oracle.OracleDialect_cx_oracle."
- "_parse_cx_oracle_ver",
- lambda self, vers: (5, 1, 5),
+ with expect_raises_message(
+ exc.InvalidRequestError,
+ "cx_Oracle version 7 and above are supported",
):
- assert_raises_message(
- exc.InvalidRequestError,
- "cx_Oracle version 7 and above are supported",
- cx_oracle.OracleDialect_cx_oracle,
- dbapi=mock.Mock(),
- )
+ cx_oracle.OracleDialect_cx_oracle(dbapi=Mock(version="5.1.5"))
+
+ dialect = cx_oracle.OracleDialect_cx_oracle(
+ dbapi=Mock(version="7.1.0")
+ )
+ eq_(dialect.cx_oracle_ver, (7, 1, 0))
+
+
+class OracleDbDialectTest(fixtures.TestBase):
+ def test_oracledb_version_parse(self):
+ dialect = oracledb.OracleDialect_oracledb()
- with mock.patch(
- "sqlalchemy.dialects.oracle.cx_oracle.OracleDialect_cx_oracle."
- "_parse_cx_oracle_ver",
- lambda self, vers: (7, 1, 0),
+ def check(version):
+ dbapi = Mock(version=version)
+ dialect._load_version(dbapi)
+ return dialect.oracledb_ver
+
+ eq_(check("7.2"), (7, 2))
+ eq_(check("7.0.1"), (7, 0, 1))
+ eq_(check("9.0b1"), (9, 0))
+
+ def test_minimum_version(self):
+ with expect_raises_message(
+ exc.InvalidRequestError,
+ "oracledb version 1 and above are supported",
):
- cx_oracle.OracleDialect_cx_oracle(dbapi=mock.Mock())
+ oracledb.OracleDialect_oracledb(dbapi=Mock(version="0.1.5"))
+
+ dialect = oracledb.OracleDialect_oracledb(dbapi=Mock(version="7.1.0"))
+ eq_(dialect.oracledb_ver, (7, 1, 0))
+
+
+class OracledbMode(fixtures.TestBase):
+ __backend__ = True
+ __only_on__ = "oracle+oracledb"
+
+ def _run_in_process(self, fn):
+ ctx = get_context("spawn")
+ queue = ctx.Queue()
+ process = ctx.Process(target=fn, args=(str(config.db_url), queue))
+ try:
+ process.start()
+ process.join(10)
+ eq_(process.exitcode, 0)
+ return queue.get_nowait()
+ finally:
+ process.kill()
+
+ def test_thin_mode(self):
+ from ._oracledb_mode import run_thin_mode
+
+ mode, is_thin = self._run_in_process(run_thin_mode)
+ is_true(is_thin)
+ is_true(mode.startswith("python-oracledb thn"))
+
+ def test_thick_mode(self):
+ from ._oracledb_mode import run_thick_mode
+
+ mode, is_thin = self._run_in_process(run_thick_mode)
+ is_false(is_thin)
+ eq_(mode.strip(), "custom-driver-name")
class DialectWBackendTest(fixtures.TestBase):
False,
),
)
- @testing.only_on("oracle+cx_oracle")
+ @testing.only_on(["oracle+cx_oracle", "oracle+oracledb"])
def test_is_disconnect(self, message, code, expected):
dialect = testing.db.dialect
argnames="cx_oracle_type",
id_="ia",
)
+ _dialect = testing.combinations(
+ cx_oracle.dialect, oracledb.dialect, argnames="dialect_cls"
+ )
def _assert_errorhandler(self, outconverter, has_errorhandler):
data = "\uee2c\u9a66" # this is u"\uee2c\u9a66"
assert_raises(UnicodeDecodeError, outconverter, utf8_w_errors)
@_oracle_char_combinations
+ @_dialect
def test_encoding_errors_cx_oracle(
- self,
- cx_Oracle,
- cx_oracle_type,
+ self, cx_Oracle, cx_oracle_type, dialect_cls
):
- ignore_dialect = cx_oracle.dialect(
- dbapi=cx_Oracle, encoding_errors="ignore"
- )
+ ignore_dialect = dialect_cls(dbapi=cx_Oracle, encoding_errors="ignore")
ignore_outputhandler = (
ignore_dialect._generate_connection_outputtype_handler()
)
@_oracle_char_combinations
+ @_dialect
def test_no_encoding_errors_cx_oracle(
- self,
- cx_Oracle,
- cx_oracle_type,
+ self, cx_Oracle, cx_oracle_type, dialect_cls
):
- plain_dialect = cx_oracle.dialect(dbapi=cx_Oracle)
+ plain_dialect = dialect_cls(dbapi=cx_Oracle)
plain_outputhandler = (
plain_dialect._generate_connection_outputtype_handler()
class OutParamTest(fixtures.TestBase, AssertsExecutionResults):
- __only_on__ = "oracle+cx_oracle"
+ __only_on__ = ("oracle+cx_oracle", "oracle+oracledb")
__backend__ = True
@classmethod
eq_(result, "’é")
-class CXOracleConnectArgsTest(fixtures.TestBase):
- __only_on__ = "oracle+cx_oracle"
- __backend__ = True
+class BaseConnectArgsTest:
+ @property
+ def name(self):
+ raise NotImplementedError
+
+ @property
+ def dbapi(self):
+ raise NotImplementedError
+
+ @property
+ def dialect_cls(self):
+ raise NotImplementedError
def test_cx_oracle_service_name(self):
- url_string = "oracle+cx_oracle://scott:tiger@host/?service_name=hr"
+ url_string = f"oracle+{self.name}://scott:tiger@host/?service_name=hr"
eng = create_engine(url_string, _initialize=False)
cargs, cparams = eng.dialect.create_connect_args(eng.url)
assert "SID=hr" not in cparams["dsn"]
def test_cx_oracle_service_name_bad(self):
- url_string = "oracle+cx_oracle://scott:tiger@host/hr1?service_name=hr2"
+ url_string = (
+ f"oracle+{self.name}://scott:tiger@host/hr1?service_name=hr2"
+ )
assert_raises(
exc.InvalidRequestError,
create_engine,
)
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)
+ dialect = self.dialect_cls(dbapi=self.dbapi)
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)
+ dialect = self.dialect_cls(dbapi=self.dbapi)
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)
+ dialect = self.dialect_cls(dbapi=self.dbapi)
with testing.expect_deprecated(
- "cx_oracle dialect option %r should" % key
+ f"{self.name} 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})
+ dialect = self.dialect_cls(dbapi=self.dbapi, **{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",
+ f"oracle+{self.name}://scott:tiger@host/?mode=sYsDBA",
"mode",
- cx_Oracle.SYSDBA,
+ self.dbapi.SYSDBA,
)
self._test_db_opt(
- "oracle+cx_oracle://scott:tiger@host/?mode=SYSOPER",
+ f"oracle+{self.name}://scott:tiger@host/?mode=SYSOPER",
"mode",
- cx_Oracle.SYSOPER,
+ self.dbapi.SYSOPER,
)
def test_int_mode(self):
self._test_db_opt(
- "oracle+cx_oracle://scott:tiger@host/?mode=32767", "mode", 32767
+ f"oracle+{self.name}://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",
+ f"oracle+{self.name}://scott:tiger@host/?purity=attr_purity_new",
"purity",
- cx_Oracle.ATTR_PURITY_NEW,
+ self.dbapi.ATTR_PURITY_NEW,
)
def test_encoding(self):
self._test_db_opt(
- "oracle+cx_oracle://scott:tiger@host/"
+ f"oracle+{self.name}://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",
+ f"oracle+{self.name}://scott:tiger@host/?threaded=true",
"threaded",
True,
)
self._test_db_opt_unpresent(
- "oracle+cx_oracle://scott:tiger@host/", "threaded"
+ f"oracle+{self.name}://scott:tiger@host/", "threaded"
)
def test_events(self):
self._test_db_opt(
- "oracle+cx_oracle://scott:tiger@host/?events=true", "events", True
+ f"oracle+{self.name}://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"
+ "The 'threaded' parameter to the cx_oracle/oracledb dialect"
):
- dialect = cx_oracle.dialect(threaded=False)
+ dialect = self.dialect_cls(threaded=False)
arg, kw = dialect.create_connect_args(
- url.make_url("oracle+cx_oracle://scott:tiger@dsn")
+ url.make_url(f"oracle+{self.name}://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",
+ f"oracle+{self.name}://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",
+ f"oracle+{self.name}://scott:tiger@host/?auto_convert_lobs=False",
"auto_convert_lobs",
False,
)
+class CXOracleConnectArgsTest(BaseConnectArgsTest, fixtures.TestBase):
+ __only_on__ = "oracle+cx_oracle"
+ __backend__ = True
+
+ @property
+ def name(self):
+ return "cx_oracle"
+
+ @property
+ def dbapi(self):
+ import cx_Oracle
+
+ return cx_Oracle
+
+ @property
+ def dialect_cls(self):
+ return cx_oracle.dialect
+
+
+class OracleDbConnectArgsTest(BaseConnectArgsTest, fixtures.TestBase):
+ __only_on__ = "oracle+oracledb"
+ __backend__ = True
+
+ @property
+ def name(self):
+ return "oracledb"
+
+ @property
+ def dbapi(self):
+ import oracledb
+
+ return oracledb
+
+ @property
+ def dialect_cls(self):
+ return oracledb.dialect
+
+
class TableValuedTest(fixtures.TestBase):
__backend__ = True
__only_on__ = "oracle"
from sqlalchemy import VARCHAR
from sqlalchemy.dialects.oracle import base as oracle
from sqlalchemy.dialects.oracle import cx_oracle
+from sqlalchemy.dialects.oracle import oracledb
from sqlalchemy.sql import column
from sqlalchemy.sql.sqltypes import NullType
from sqlalchemy.testing import assert_raises_message
(NCHAR(), cx_oracle._OracleNChar),
(NVARCHAR(), cx_oracle._OracleUnicodeStringNCHAR),
(oracle.RAW(50), cx_oracle._OracleRaw),
+ argnames="start, test",
)
- def test_type_adapt(self, start, test):
- dialect = cx_oracle.dialect()
+ @testing.combinations(cx_oracle, oracledb, argnames="module")
+ def test_type_adapt(self, start, test, module):
+ dialect = module.dialect()
assert isinstance(
start.dialect_impl(dialect), test
(UnicodeText(), cx_oracle._OracleUnicodeTextNCLOB),
(NCHAR(), cx_oracle._OracleNChar),
(NVARCHAR(), cx_oracle._OracleUnicodeStringNCHAR),
+ argnames="start, test",
)
- def test_type_adapt_nchar(self, start, test):
- dialect = cx_oracle.dialect(use_nchar_for_unicode=True)
+ @testing.combinations(cx_oracle, oracledb, argnames="module")
+ def test_type_adapt_nchar(self, start, test, module):
+ dialect = module.dialect(use_nchar_for_unicode=True)
assert isinstance(
start.dialect_impl(dialect), test
"SELECT CAST(-9999999999999999999 AS NUMBER(19,0)) FROM dual",
),
)
- @testing.only_on("oracle+cx_oracle", "cx_oracle-specific feature")
+ @testing.only_on(
+ ["oracle+cx_oracle", "oracle+oracledb"],
+ "cx_oracle/oracledb specific feature",
+ )
def test_raw_numerics(self, title, stmt):
with testing.db.connect() as conn:
# get a brand new connection that definitely is not
assert isinstance(t2.c.nv_data.type, sqltypes.NVARCHAR)
assert isinstance(t2.c.c_data.type, sqltypes.NCHAR)
- if testing.against("oracle+cx_oracle"):
+ if testing.against("oracle+cx_oracle", "oracle+oracledb"):
assert isinstance(
t2.c.nv_data.type.dialect_impl(connection.dialect),
cx_oracle._OracleUnicodeStringNCHAR,
t2 = Table("tnv", m2, autoload_with=connection)
assert isinstance(t2.c.data.type, sqltypes.VARCHAR)
- if testing.against("oracle+cx_oracle"):
+ if testing.against("oracle+cx_oracle", "oracle+oracledb"):
assert isinstance(
t2.c.data.type.dialect_impl(connection.dialect),
cx_oracle._OracleString,
test the numeric output_type_handler when using non-US locale for NLS_LANG.
"""
- __only_on__ = "oracle+cx_oracle"
+ __only_on__ = ("oracle+cx_oracle", "oracle+oracledb")
__backend__ = True
def setup_test(self):
def teardown_test(self):
self.engine.dispose()
+ # https://python-oracledb.readthedocs.io/en/latest/user_guide/appendix_b.html#globalization-in-thin-and-thick-modes
+ @testing.requires.fail_on_oracledb_thin
def test_were_getting_a_comma(self):
connection = self.engine.pool._creator()
cursor = connection.cursor()
class SetInputSizesTest(fixtures.TestBase):
- __only_on__ = "oracle+cx_oracle"
+ __only_on__ = ("oracle+cx_oracle", "oracle+oracledb")
__backend__ = True
@testing.combinations(
"oracle+cx_oracle",
"cx_oracle exception seems to be having some issue with pickling",
)
+ @testing.fails_on(
+ "oracle+oracledb",
+ "oracledb exception seems to be having some issue with pickling",
+ )
def test_stmt_exception_pickleable_plus_dbapi(self):
raw = testing.db.raw_connection()
the_orig = None
self.meta.drop_all(conn)
self.engine.dispose()
- @testing.crashes(
- "oracle",
- "cx_oracle 6 doesn't allow a close like this due to open cursors",
- )
@testing.fails_if(
[
"+mysqlconnector",
self.assert_(h2.foober == h3.foober == h4.foober == "im foober")
eq_(h5.foober, "im the new foober")
- @testing.fails_on("oracle+cx_oracle", "seems like a cx_oracle bug")
def test_eager_defaults(self):
hohoval, default_t, Hoho = (
self.other.hohoval,
@property
def named_paramstyle(self):
- return only_on(["sqlite", "oracle+cx_oracle"])
+ return only_on(["sqlite", "oracle+cx_oracle", "oracle+oracledb"])
@property
def format_paramstyle(self):
else:
return num > 0
- return skip_if(
- [
- no_support("mssql", "two-phase xact not supported by drivers"),
- no_support(
- "sqlite", "two-phase xact not supported by database"
- ),
- # in Ia3cbbf56d4882fcc7980f90519412f1711fae74d
- # we are evaluating which modern MySQL / MariaDB versions
- # can handle two-phase testing without too many problems
- # no_support(
- # "mysql",
- # "recent MySQL communiity editions have too many issues "
- # "(late 2016), disabling for now",
- # ),
- NotPredicate(
- LambdaPredicate(
- pg_prepared_transaction,
- "max_prepared_transactions not available or zero",
- )
- ),
- ]
+ return (
+ skip_if(
+ [
+ no_support(
+ "mssql", "two-phase xact not supported by drivers"
+ ),
+ no_support(
+ "sqlite", "two-phase xact not supported by database"
+ ),
+ # in Ia3cbbf56d4882fcc7980f90519412f1711fae74d
+ # we are evaluating which modern MySQL / MariaDB versions
+ # can handle two-phase testing without too many problems
+ # no_support(
+ # "mysql",
+ # "recent MySQL community editions have too many "
+ # "issues (late 2016), disabling for now",
+ # ),
+ NotPredicate(
+ LambdaPredicate(
+ pg_prepared_transaction,
+ "max_prepared_transactions not available or zero",
+ )
+ ),
+ ]
+ )
+ + self.fail_on_oracledb_thin
)
@property
@property
def cxoracle6_or_greater(self):
- return only_if(
- lambda config: against(config, "oracle+cx_oracle")
- and config.db.dialect.cx_oracle_ver >= (6,)
- )
+ def go(config):
+ return (
+ against(config, "oracle+cx_oracle")
+ and config.db.dialect.cx_oracle_ver >= (6,)
+ ) or (
+ against(config, "oracle+oracledb")
+ and config.db.dialect.oracledb_ver >= (1,)
+ )
+
+ return only_if(go)
@property
def oracle5x(self):
and config.db.dialect.cx_oracle_ver < (6,)
)
+ @property
+ def fail_on_oracledb_thin(self):
+ def go(config):
+ if against(config, "oracle+oracledb"):
+ with config.db.connect() as conn:
+ return config.db.dialect.is_thin_mode(conn)
+ return False
+
+ return fails_if(go)
+
@property
def computed_columns(self):
return skip_if(["postgresql < 12", "sqlite < 3.31", "mysql < 5.7"])
mysql: .[mariadb_connector]
oracle: .[oracle]
+ oracle: .[oracle_oracledb]
mssql: .[mssql]
oracle: WORKERS={env:TOX_WORKERS:-n2 --max-worker-restart=5}
oracle: ORACLE={env:TOX_ORACLE:--db oracle}
+ py3{,7,8,9,10,11}-oracle: EXTRA_ORACLE_DRIVERS={env:EXTRA_ORACLE_DRIVERS:--dbdriver cx_oracle --dbdriver oracledb}
sqlite: SQLITE={env:TOX_SQLITE:--db sqlite}
sqlite_file: SQLITE={env:TOX_SQLITE_FILE:--db sqlite_file}
# tox as of 2.0 blocks all environment variables from the
# outside, unless they are here (or in TOX_TESTENV_PASSENV,
# wildcards OK). Need at least these
-passenv=ORACLE_HOME NLS_LANG TOX_POSTGRESQL TOX_POSTGRESQL_PY2K TOX_MYSQL TOX_MYSQL_PY2K TOX_ORACLE TOX_MSSQL TOX_SQLITE TOX_SQLITE_FILE TOX_WORKERS EXTRA_SQLITE_DRIVERS EXTRA_PG_DRIVERS EXTRA_MYSQL_DRIVERS
+passenv=ORACLE_HOME NLS_LANG TOX_POSTGRESQL TOX_POSTGRESQL_PY2K TOX_MYSQL TOX_MYSQL_PY2K TOX_ORACLE TOX_MSSQL TOX_SQLITE TOX_SQLITE_FILE TOX_WORKERS EXTRA_SQLITE_DRIVERS EXTRA_PG_DRIVERS EXTRA_MYSQL_DRIVERS EXTRA_ORACLE_DRIVERS
commands=
# that flag for coverage mode.
nocext: sh -c "rm -f lib/sqlalchemy/*.so"
- {env:BASECOMMAND} {env:WORKERS} {env:SQLITE:} {env:EXTRA_SQLITE_DRIVERS:} {env:POSTGRESQL:} {env:EXTRA_PG_DRIVERS:} {env:MYSQL:} {env:EXTRA_MYSQL_DRIVERS:} {env:ORACLE:} {env:MSSQL:} {env:IDENTS:} {env:PYTEST_EXCLUDES:} {env:COVERAGE:} {posargs}
+ {env:BASECOMMAND} {env:WORKERS} {env:SQLITE:} {env:EXTRA_SQLITE_DRIVERS:} {env:POSTGRESQL:} {env:EXTRA_PG_DRIVERS:} {env:MYSQL:} {env:EXTRA_MYSQL_DRIVERS:} {env:ORACLE:} {env:EXTRA_ORACLE_DRIVERS:} {env:MSSQL:} {env:IDENTS:} {env:PYTEST_EXCLUDES:} {env:COVERAGE:} {posargs}
oracle,mssql,sqlite_file: python reap_dbs.py db_idents.txt
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
