actual server version in order to assist with migration of Oracle databases,
and may be configured within the Oracle server itself. This compatibility
version is retrieved using the query ``SELECT value FROM v$parameter WHERE
-name = 'compatible';``. The SQLAlchemy Oracle dialect
-will use this query upon first connect in order to determine the effective
-compatibility version of the server, which determines what the maximum allowed
-identifier length is for the server.
+name = 'compatible';``. The SQLAlchemy Oracle dialect, when tasked with
+determining the default max identifier length, will attempt to use this query
+upon first connect in order to determine the effective compatibility version of
+the server, which determines what the maximum allowed identifier length is for
+the server. If the table is not available, the server version information is
+used instead.
As of SQLAlchemy 1.4, the default max identifier length for the Oracle dialect
is 128 characters. Upon first connect, the compatibility version is detected
supports_unicode_binds = False
max_identifier_length = 128
- # this should be set to
- # "SELECT value FROM v$parameter WHERE name = 'compatible'"
- # upon connect.
- _compat_server_version_info = None
-
supports_simple_order_by_label = False
cte_follows_insert = True
def initialize(self, connection):
super(OracleDialect, self).initialize(connection)
- _compat_server_version_info = self._get_compat_server_version_info(
- connection
- )
- if _compat_server_version_info is not None:
- self._compat_server_version_info = _compat_server_version_info
-
self.implicit_returning = self.__dict__.get(
"implicit_returning", self.server_version_info > (10,)
)
self.colspecs.pop(sqltypes.Interval)
self.use_ansi = False
- def _get_compat_server_version_info(self, connection):
+ def _get_effective_compat_server_version_info(self, connection):
+ # dialect does not need compat levels below 12.2, so don't query
+ # in those cases
+
+ if self.server_version_info < (12, 2):
+ return self.server_version_info
try:
- return connection.execute(
+ compat = connection.execute(
"SELECT value FROM v$parameter WHERE name = 'compatible'"
).scalar()
- except exc.DBAPIError as err:
- util.warn("Could not determine compatibility version: %s" % err)
-
- @property
- def _effective_compat_server_version_info(self):
- if self._compat_server_version_info is not None:
- return self._compat_server_version_info
+ except exc.DBAPIError:
+ compat = None
+
+ if compat:
+ try:
+ return tuple(int(x) for x in compat.split("."))
+ except:
+ return self.server_version_info
else:
return self.server_version_info
pass
def _check_max_identifier_length(self, connection):
- if self._effective_compat_server_version_info < (12, 2):
+ if self._get_effective_compat_server_version_info(connection) < (
+ 12,
+ 2,
+ ):
return 30
else:
# use the default
# coding: utf-8
+import re
from sqlalchemy import bindparam
from sqlalchemy import create_engine
self.assert_compile(Unicode(50), "NVARCHAR2(50)", dialect=dialect)
self.assert_compile(UnicodeText(), "NCLOB", dialect=dialect)
+ def test_ident_length_in_13_is_30(self):
+ from sqlalchemy import __version__
+
+ m = re.match(r"(\d+)\.(\d+)(?:\.(\d+))?", __version__)
+ version = tuple(int(x) for x in m.group(1, 2, 3) if x is not None)
+ if version >= (1, 4):
+ length = 128
+ else:
+ length = 30
+
+ eq_(oracle.OracleDialect.max_identifier_length, length)
+
+ dialect = self._dialect((12, 2, 0))
+ conn = mock.Mock(
+ execute=mock.Mock(return_value=mock.Mock(scalar=lambda: "12.2.0"))
+ )
+ dialect.initialize(conn)
+ eq_(dialect.server_version_info, (12, 2, 0))
+ eq_(
+ dialect._get_effective_compat_server_version_info(conn), (12, 2, 0)
+ )
+ eq_(dialect.max_identifier_length, length)
+
+ def test_max_ident_122(self):
+ dialect = self._dialect((12, 2, 0))
+
+ conn = mock.Mock(
+ execute=mock.Mock(return_value=mock.Mock(scalar=lambda: "12.2.0"))
+ )
+ dialect.initialize(conn)
+ eq_(dialect.server_version_info, (12, 2, 0))
+ eq_(
+ dialect._get_effective_compat_server_version_info(conn), (12, 2, 0)
+ )
+ eq_(
+ dialect.max_identifier_length,
+ oracle.OracleDialect.max_identifier_length,
+ )
+
+ def test_max_ident_112(self):
+ dialect = self._dialect((11, 2, 0))
+
+ conn = mock.Mock(
+ execute=mock.Mock(return_value=mock.Mock(scalar="11.0.0"))
+ )
+ dialect.initialize(conn)
+ eq_(dialect.server_version_info, (11, 2, 0))
+ eq_(
+ dialect._get_effective_compat_server_version_info(conn), (11, 2, 0)
+ )
+ eq_(dialect.max_identifier_length, 30)
+
+ def test_max_ident_122_11compat(self):
+ dialect = self._dialect((12, 2, 0))
+
+ conn = mock.Mock(
+ execute=mock.Mock(return_value=mock.Mock(scalar=lambda: "11.0.0"))
+ )
+ dialect.initialize(conn)
+ eq_(dialect.server_version_info, (12, 2, 0))
+ eq_(
+ dialect._get_effective_compat_server_version_info(conn), (11, 0, 0)
+ )
+ eq_(dialect.max_identifier_length, 30)
+
+ def test_max_ident_122_11compat_vparam_raises(self):
+ dialect = self._dialect((12, 2, 0))
+
+ def c122():
+ raise exc.DBAPIError(
+ "statement", None, "no such table", None, None
+ )
+
+ conn = mock.Mock(
+ execute=mock.Mock(return_value=mock.Mock(scalar=c122))
+ )
+ dialect.initialize(conn)
+ eq_(dialect.server_version_info, (12, 2, 0))
+ eq_(
+ dialect._get_effective_compat_server_version_info(conn), (12, 2, 0)
+ )
+ eq_(
+ dialect.max_identifier_length,
+ oracle.OracleDialect.max_identifier_length,
+ )
+
+ def test_max_ident_122_11compat_vparam_cant_parse(self):
+ dialect = self._dialect((12, 2, 0))
+
+ def c122():
+ return "12.thisiscrap.0"
+
+ conn = mock.Mock(
+ execute=mock.Mock(return_value=mock.Mock(scalar=c122))
+ )
+ dialect.initialize(conn)
+ eq_(dialect.server_version_info, (12, 2, 0))
+ eq_(
+ dialect._get_effective_compat_server_version_info(conn), (12, 2, 0)
+ )
+ eq_(
+ dialect.max_identifier_length,
+ oracle.OracleDialect.max_identifier_length,
+ )
+
class ExecuteTest(fixtures.TestBase):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
