--- /dev/null
+.. change::
+ :tags: feature, engine
+
+ Added :attr:`.Dialect.dbapi_version`, a standardized accessor for the
+ version of the DBAPI module in use by a dialect, in contrast to
+ :attr:`.Dialect.server_version_info` which refers to the database server.
+ The implementation on :class:`.DefaultDialect` makes use of a new
+ per-dialect method :meth:`.Dialect.retrieve_dbapi_version` in order to
+ retrieve the version from the DBAPI module and return it as a
+ :class:`.VersionInfo` object, which is a tuple subclass with additional
+ properties; third-party dialects should also implement the
+ :meth:`.Dialect.retrieve_dbapi_version` method.
+
+.. change::
+ :tags: bug, postgresql
+
+ Fixed issue in the asyncpg dialect where the version of the ``asyncpg``
+ DBAPI would always be reported as ``(99, 99, 99)``, as the version was
+ looked up on the dialect's DBAPI wrapper module rather than on the
+ ``asyncpg`` module itself.
+
+.. change::
+ :tags: bug, mysql
+
+ Fixed issue where the version of the DBAPI reported by the mysqldb and
+ pymysql dialects was incorrect. Current mysqlclient releases publish
+ ``MySQLdb.version_info`` and no version string at all, so no version was
+ reported; pymysql publishes ``__version__`` and ``version_info`` as
+ mysqlclient compatibility values, so the version reported for pymysql
+ was that of the mysqlclient release it emulates, e.g. ``(2, 2, 8)``
+ rather than ``(1, 2, 0)``.
+
+.. change::
+ :tags: bug, testing
+
+ The version specifications used by testing exclusions such as
+ ``testing.fails_if("+asyncmy<0.2.13")`` now support a driver name, in
+ which case the comparison is against the version of the DBAPI rather
+ than that of the database server. Previously this form raised
+ ``AssertionError: DBAPI version specs not supported yet``.
else:
return False
- def _dbapi_version(self) -> interfaces.VersionInfoType:
- if not self.dbapi:
- return ()
- return self._parse_dbapi_version(self.dbapi.version)
-
- def _parse_dbapi_version(self, vers: str) -> interfaces.VersionInfoType:
- m = re.match(r"(?:py.*-)?([\d\.]+)(?:-(\w+))?", vers)
- if not m:
- return ()
- vers_tuple: interfaces.VersionInfoType = tuple(
- [int(x) for x in m.group(1).split(".")]
- )
- if m.group(2):
- vers_tuple += (m.group(2),)
- return vers_tuple
+ def retrieve_dbapi_version(
+ self, dbapi: interfaces.DBAPIModule
+ ) -> util.VersionInfo:
+ return util.parse_version_string(dbapi.version)
def _get_server_version_info(
self, connection: Connection
- ) -> interfaces.VersionInfoType:
+ ) -> interfaces.ServerVersionInfoType:
# NOTE: this function is not reliable, particularly when
# freetds is in use. Implement database-specific server version
# queries.
else:
return False
- def _dbapi_version(self) -> interfaces.VersionInfoType:
- if not self.dbapi:
- return ()
- return self._parse_dbapi_version(self.dbapi.version)
-
- def _parse_dbapi_version(self, vers: str) -> interfaces.VersionInfoType:
- m = re.match(r"(?:py.*-)?([\d\.]+)(?:-(\w+))?", vers)
- if not m:
- return ()
- vers_tuple: interfaces.VersionInfoType = tuple(
- [int(x) for x in m.group(1).split(".")]
- )
- if m.group(2):
- vers_tuple += (m.group(2),)
- return vers_tuple
+ def retrieve_dbapi_version(
+ self, dbapi: interfaces.DBAPIModule
+ ) -> util.VersionInfo:
+ return util.parse_version_string(dbapi.version)
def _get_server_version_info(self, connection):
vers = connection.exec_driver_sql("select @@version").scalar()
{sqltypes.Numeric: _MSNumeric_pymssql, sqltypes.Float: sqltypes.Float},
)
+ def retrieve_dbapi_version(self, dbapi):
+ return util.parse_version_string(getattr(dbapi, "__version__", None))
+
@classmethod
def import_dbapi(cls):
module = __import__("pymssql")
# pymmsql < 2.1.1 doesn't have a Binary method. we use string
- client_ver = tuple(int(x) for x in module.__version__.split("."))
+ client_ver = util.parse_version_string(module.__version__)
if client_ver < (2, 1, 1):
# TODO: monkeypatching here is less than ideal
module.Binary = lambda x: x if hasattr(x, "decode") else str(x)
and self.dbapi
and hasattr(self.dbapi.Cursor, "nextset")
)
- self._need_decimal_fix = self.dbapi and self._dbapi_version() < (
- 2,
- 1,
- 8,
+ # a version which can't be determined is assumed to be an old one
+ version = self._dbapi_version_or_none
+ self._need_decimal_fix = bool(self.dbapi) and (
+ version is None or version < (2, 1, 8)
)
self.fast_executemany = fast_executemany
if fast_executemany:
from .pymysql import _connection_ping_reconnects_true
from .pymysql import MySQLDialect_pymysql
+from ... import util
from ...connectors.asyncio import AsyncAdapt_dbapi_connection
from ...connectors.asyncio import AsyncAdapt_dbapi_cursor
from ...connectors.asyncio import AsyncAdapt_dbapi_module
__import__("aiomysql"), __import__("pymysql")
)
+ def retrieve_dbapi_version(self, dbapi: DBAPIModule) -> util.VersionInfo:
+ # the version of aiomysql itself, rather than that of the pymysql
+ # module it makes use of
+ aiomysql = getattr(dbapi, "aiomysql", None)
+ return util.parse_version_string(
+ getattr(aiomysql, "__version__", None)
+ )
+
def do_terminate(self, dbapi_connection: DBAPIConnection) -> None:
dbapi_connection.terminate()
def import_dbapi(cls) -> DBAPIModule:
return AsyncAdapt_asyncmy_dbapi(__import__("asyncmy"))
+ def retrieve_dbapi_version(self, dbapi: DBAPIModule) -> util.VersionInfo:
+ # asyncmy publishes no version of its own within the module, so
+ # the installed distribution is consulted
+ return util.parse_version_from_metadata("asyncmy")
+
def do_terminate(self, dbapi_connection: DBAPIConnection) -> None:
dbapi_connection.terminate()
from __future__ import annotations
-import re
from typing import Any
from typing import Optional
from typing import Sequence
from ...sql.type_api import _ResultProcessorType
-mariadb_cpy_minimum_version = (1, 0, 1)
+mariadb_cpy_minimum_version = util.VersionInfo((1, 0, 1))
class _MariaDBUUID(sqltypes.UUID[sqltypes._UUID_RETURN]):
driver = "mariadbconnector"
supports_statement_cache = True
+ minimum_dbapi_version = mariadb_cpy_minimum_version
+
# set this to True at the module level to prevent the driver from running
# against a backend that server detects as MySQL. currently this appears to
# be unnecessary as MariaDB client libraries have always worked against
MySQLDialect.colspecs, {sqltypes.Uuid: _MariaDBUUID}
)
- @util.memoized_property
- def _dbapi_version(self) -> tuple[int, ...]:
- if self.dbapi and hasattr(self.dbapi, "__version__"):
- return tuple(
- [
- int(x)
- for x in re.findall(
- r"(\d+)(?:[-\.]?|$)", self.dbapi.__version__
- )
- ]
- )
- else:
- return (99, 99, 99)
+ def retrieve_dbapi_version(self, dbapi: DBAPIModule) -> util.VersionInfo:
+ return util.parse_version_string(getattr(dbapi, "__version__", None))
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.paramstyle = "qmark"
- if self.dbapi is not None:
- if self._dbapi_version < mariadb_cpy_minimum_version:
- raise NotImplementedError(
- "The minimum required version for MariaDB "
- "Connector/Python is %s"
- % ".".join(str(x) for x in mariadb_cpy_minimum_version)
- )
@classmethod
def import_dbapi(cls) -> DBAPIModule:
from __future__ import annotations
-import re
from typing import Any
from typing import cast
from typing import Optional
return [], opts
- @util.memoized_property
- def _mysqlconnector_version_info(self) -> Optional[tuple[int, ...]]:
- if self.dbapi and hasattr(self.dbapi, "__version__"):
- m = re.match(r"(\d+)\.(\d+)(?:\.(\d+))?", self.dbapi.__version__)
- if m:
- return tuple(int(x) for x in m.group(1, 2, 3) if x is not None)
- return None
+ def retrieve_dbapi_version(self, dbapi: DBAPIModule) -> util.VersionInfo:
+ return util.parse_version_string(getattr(dbapi, "__version__", None))
def _detect_charset(self, connection: Connection) -> str:
return connection.connection.charset # type: ignore[no-any-return]
from __future__ import annotations
-import re
+import itertools
from typing import Any
from typing import Callable
from typing import cast
preparer = MySQLIdentifierPreparer
server_version_info: tuple[int, ...]
- def __init__(self, **kwargs: Any):
- super().__init__(**kwargs)
- self._mysql_dbapi_version = (
- self._parse_dbapi_version(self.dbapi.__version__)
- if self.dbapi is not None and hasattr(self.dbapi, "__version__")
- else (0, 0, 0)
- )
-
- def _parse_dbapi_version(self, version: str) -> tuple[int, ...]:
- 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 retrieve_dbapi_version(self, dbapi: DBAPIModule) -> util.VersionInfo:
+ # mysqlclient publishes ``version_info``, a tuple in the style of
+ # ``sys.version_info`` such as ``(2, 2, 7, "final", 0)``, and no
+ # version string of its own; MySQL-python published
+ # ``__version__``. cymysql, which subclasses this dialect, also
+ # publishes ``__version__``. pymysql publishes both of these as
+ # mysqlclient compatibility values rather than as its own version,
+ # so that dialect overrides this method; the asyncio dialects
+ # likewise override as their DBAPI is a wrapper module.
+ version_info = getattr(dbapi, "version_info", None)
+ if version_info is not None:
+ return util.VersionInfo(
+ tuple(
+ itertools.takewhile(
+ lambda part: isinstance(part, int), version_info
+ )
+ )
+ )
+ return util.parse_version_string(getattr(dbapi, "__version__", None))
@util.langhelpers.memoized_property
def supports_server_side_cursors(self) -> bool:
from typing import Union
from .mysqldb import MySQLDialect_mysqldb
+from ... import util
from ...util import langhelpers
if TYPE_CHECKING:
description_encoding = None
+ def retrieve_dbapi_version(self, dbapi: DBAPIModule) -> util.VersionInfo:
+ # pymysql publishes its own version as ``VERSION_STRING``; the
+ # ``__version__`` and ``version_info`` attributes it also publishes
+ # are mysqlclient compatibility values, e.g. ``"2.2.8"`` for a
+ # pymysql that is itself version 1.2.0
+ return util.parse_version_string(
+ getattr(dbapi, "VERSION_STRING", None)
+ )
+
@langhelpers.memoized_property
def supports_server_side_cursors(self) -> bool:
try:
bind_typing = interfaces.BindTyping.SETINPUTSIZES
+ minimum_dbapi_version = util.VersionInfo((8,))
+
driver = "cx_oracle"
colspecs = util.update_copy(
self.colspecs[sqltypes.UnicodeText] = _OracleUnicodeTextNCLOB
dbapi_module = self.dbapi
- self._load_version(dbapi_module)
if dbapi_module is not None:
# these constants will first be seen in SQLAlchemy datatypes
self._paramval = lambda value: value.getvalue()
- 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 < (8,) and self.cx_oracle_ver > (0, 0, 0):
- raise exc.InvalidRequestError(
- "cx_Oracle version 8 and above are supported"
- )
+ def retrieve_dbapi_version(self, dbapi):
+ return util.parse_version_string(getattr(dbapi, "version", None))
+
+ @property
+ def cx_oracle_ver(self):
+ """Legacy accessor for :attr:`.Dialect.dbapi_version`.
+
+ Retained for backwards compatibility; ``(0, 0, 0)`` is returned
+ when no version can be determined.
+
+ """
+ version = self._dbapi_version_or_none
+ return version if version is not None else util.VersionInfo((0, 0, 0))
@classmethod
def import_dbapi(cls):
from __future__ import annotations
import collections
-import re
from typing import Any
from typing import TYPE_CHECKING
from . import cx_oracle as _cx_oracle
-from ... import exc
+from ... import util
from ...connectors.asyncio import AsyncAdapt_dbapi_connection
from ...connectors.asyncio import AsyncAdapt_dbapi_cursor
from ...connectors.asyncio import AsyncAdapt_dbapi_module
execution_ctx_cls = OracleExecutionContext_oracledb
driver = "oracledb"
- _min_version = (1,)
+
+ minimum_dbapi_version = util.VersionInfo((1,))
def __init__(
self,
def get_async_dialect_cls(cls, url):
return OracleDialectAsync_oracledb
- 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 > (0, 0, 0)
- and self.oracledb_ver < self._min_version
- ):
- raise exc.InvalidRequestError(
- f"oracledb version {self._min_version} and above are supported"
- )
+ @property
+ def oracledb_ver(self):
+ """Legacy accessor for :attr:`.Dialect.dbapi_version`.
+
+ Retained for backwards compatibility; ``(0, 0, 0)`` is returned
+ when no version can be determined.
+
+ """
+ version = self._dbapi_version_or_none
+ return version if version is not None else util.VersionInfo((0, 0, 0))
def do_begin_twophase(self, connection, xid):
conn_xis = connection.connection.xid(*xid)
]
def _check_max_identifier_length(self, connection):
- if self.oracledb_ver >= (2, 5):
+ if self.dbapi_version >= (2, 5):
max_len = connection.connection.max_identifier_length
if max_len is not None:
return max_len
supports_statement_cache = True
execution_ctx_cls = OracleExecutionContextAsync_oracledb
- _min_version = (2, 0, 1)
+ minimum_dbapi_version = util.VersionInfo((2, 0, 1))
# thick_mode mode is not supported by asyncio, oracledb will raise
@classmethod
},
)
+ def retrieve_dbapi_version(self, dbapi):
+ return util.parse_version_string(getattr(dbapi, "__version__", None))
+
def __init__(
self,
client_encoding=None,
def _invalidate_schema_cache(self):
self._invalidate_schema_cache_asof = time.time()
- @util.memoized_property
- def _dbapi_version(self):
- if self.dbapi and hasattr(self.dbapi, "__version__"):
- return tuple(
- [
- int(x)
- for x in re.findall(
- r"(\d+)(?:[-\.]?|$)", self.dbapi.__version__
- )
- ]
- )
- else:
- return (99, 99, 99)
+ def retrieve_dbapi_version(self, dbapi):
+ # dbapi is the AsyncAdapt_asyncpg_dbapi wrapper; the version is on
+ # the asyncpg module itself, which is ``.driver``
+ driver = getattr(dbapi, "driver", None)
+ return util.parse_version_string(getattr(driver, "__version__", None))
@classmethod
def import_dbapi(cls):
""" # noqa
import decimal
-import re
from . import ranges
from .array import ARRAY as PGARRAY
driver = "pg8000"
supports_statement_cache = True
+ minimum_dbapi_version = util.VersionInfo((1, 16, 6))
+
supports_unicode_statements = True
supports_unicode_binds = True
PGDialect.__init__(self, **kwargs)
self.client_encoding = client_encoding
- if self._dbapi_version < (1, 16, 6):
- raise NotImplementedError("pg8000 1.16.6 or greater is required")
-
if self._native_inet_types:
raise NotImplementedError(
"The pg8000 dialect does not fully implement "
"CIDR is not"
)
- @util.memoized_property
- def _dbapi_version(self):
- if self.dbapi and hasattr(self.dbapi, "__version__"):
- return tuple(
- [
- int(x)
- for x in re.findall(
- r"(\d+)(?:[-\.]?|$)", self.dbapi.__version__
- )
- ]
- )
- else:
- return (99, 99, 99)
+ def retrieve_dbapi_version(self, dbapi):
+ return util.parse_version_string(getattr(dbapi, "__version__", None))
@classmethod
def import_dbapi(cls):
import collections
import logging
-import re
from types import NoneType
from typing import cast
from typing import TYPE_CHECKING
class PGDialect_psycopg(_PGDialect_common_psycopg):
driver = "psycopg"
+ minimum_dbapi_version = util.VersionInfo((3, 0, 2))
+
supports_statement_cache = True
supports_server_side_cursors = True
default_paramstyle = "pyformat"
execution_ctx_cls = PGExecutionContext_psycopg
statement_compiler = PGCompiler_psycopg
preparer = PGIdentifierPreparer_psycopg
- psycopg_version = (0, 0)
_has_native_hstore = True
_psycopg_adapters_map = None
},
)
+ @property
+ def psycopg_version(self):
+ """Legacy accessor for :attr:`.Dialect.dbapi_version`.
+
+ Retained for backwards compatibility; ``(0, 0)`` is returned when
+ no version can be determined.
+
+ """
+ version = self._dbapi_version_or_none
+ return version if version is not None else (0, 0)
+
def __init__(self, **kwargs):
super().__init__(**kwargs)
if self.dbapi:
- m = re.match(r"(\d+)\.(\d+)(?:\.(\d+))?", self.dbapi.__version__)
- if m:
- self.psycopg_version = tuple(
- int(x) for x in m.group(1, 2, 3) if x is not None
- )
-
- if self.psycopg_version < (3, 0, 2):
- raise ImportError(
- "psycopg version 3.0.2 or higher is required."
- )
-
from psycopg.adapt import AdaptersMap
self._psycopg_adapters_map = adapters_map = AdaptersMap(
import collections.abc as collections_abc
import logging
-import re
from typing import cast
from . import ranges
class PGDialect_psycopg2(_PGDialect_common_psycopg):
driver = "psycopg2"
+ minimum_dbapi_version = util.VersionInfo((2, 7))
+
supports_statement_cache = True
supports_server_side_cursors = True
supports_sane_multi_rowcount = False
execution_ctx_cls = PGExecutionContext_psycopg2
preparer = PGIdentifierPreparer_psycopg2
- psycopg2_version = (0, 0)
use_insertmanyvalues_wo_returning = True
returns_native_bytes = False
self.executemany_batch_page_size = executemany_batch_page_size
- if self.dbapi and hasattr(self.dbapi, "__version__"):
- m = re.match(r"(\d+)\.(\d+)(?:\.(\d+))?", self.dbapi.__version__)
- if m:
- self.psycopg2_version = tuple(
- int(x) for x in m.group(1, 2, 3) if x is not None
- )
+ @property
+ def psycopg2_version(self):
+ """Legacy accessor for :attr:`.Dialect.dbapi_version`.
- if self.psycopg2_version < (2, 7):
- raise ImportError(
- "psycopg2 version 2.7 or higher is required."
- )
+ Retained for backwards compatibility; ``(0, 0)`` is returned when
+ no version can be determined.
+
+ """
+ version = self._dbapi_version_or_none
+ return version if version is not None else (0, 0)
def initialize(self, connection):
super().initialize(connection)
from .base import SQLiteExecutionContext
from .pysqlite import SQLiteDialect_pysqlite
from ... import pool
+from ... import util
from ...connectors.asyncio import AsyncAdapt_dbapi_connection
from ...connectors.asyncio import AsyncAdapt_dbapi_cursor
from ...connectors.asyncio import AsyncAdapt_dbapi_module
__import__("aiosqlite"), __import__("sqlite3")
)
+ def retrieve_dbapi_version(self, dbapi: DBAPIModule) -> util.VersionInfo:
+ # the version of aiosqlite, rather than the Python version
+ # reported by the pysqlite dialect
+ aiosqlite = getattr(dbapi, "aiosqlite", None)
+ return util.parse_version_string(
+ getattr(aiosqlite, "__version__", None)
+ )
+
@classmethod
def get_pool_class(cls, url: URL) -> type[pool.Pool]:
if cls._is_url_file_db(url):
from .pysqlite import SQLiteDialect_pysqlite
from ... import pool
+from ... import util
class SQLiteDialect_pysqlcipher(SQLiteDialect_pysqlite):
return sqlcipher
+ def retrieve_dbapi_version(self, dbapi):
+ # the version of sqlcipher3 / pysqlcipher3, rather than the Python
+ # version reported by the pysqlite dialect
+ return util.parse_version_string(getattr(dbapi, "version", None))
+
@classmethod
def get_pool_class(cls, url):
return pool.SingletonThreadPool
import math
import os
import re
+import sys
from typing import Any
from typing import Callable
from typing import cast
from ...engine.interfaces import DBAPICursor
from ...engine.interfaces import DBAPIModule
from ...engine.interfaces import IsolationLevel
- from ...engine.interfaces import VersionInfoType
+ from ...engine.interfaces import ServerVersionInfoType
from ...engine.url import URL
from ...pool.base import PoolProxiedConnection
from ...sql.type_api import _BindProcessorType
else:
return pool.SingletonThreadPool
- def _get_server_version_info(self, connection: Any) -> VersionInfoType:
+ def _get_server_version_info(
+ self, connection: Any
+ ) -> ServerVersionInfoType:
return self.dbapi.sqlite_version_info # type: ignore[no-any-return, union-attr] # noqa: E501
+ def retrieve_dbapi_version(self, dbapi: DBAPIModule) -> util.VersionInfo:
+ # the ``sqlite3`` module ships with CPython and has no version of
+ # its own (the legacy ``sqlite3.version`` attribute was frozen at
+ # 2.6.0 and removed in Python 3.14), so the Python version is used
+ # here, which is what its feature set actually tracks. The version
+ # of the SQLite library itself is available as
+ # :attr:`.Dialect.server_version_info`.
+ return util.VersionInfo(sys.version_info[:3])
+
_isolation_lookup = SQLiteDialect._isolation_lookup.union(
{
"AUTOCOMMIT": None,
if insertmanyvalues_page_size is not _NoArg.NO_ARG:
self.insertmanyvalues_page_size = insertmanyvalues_page_size
+ self._check_minimum_dbapi_version()
+
@property
@util.deprecated(
"2.0",
@util.memoized_property
def loaded_dbapi(self) -> DBAPIModule:
if self.dbapi is None:
- raise exc.InvalidRequestError(
+ raise exc.NoDBAPILoaded(
f"Dialect {self} does not have a Python DBAPI established "
"and cannot be used for actual database interaction"
)
return self.dbapi
+ @util.memoized_property
+ def dbapi_version(self) -> util.VersionInfo:
+ # memoization applies to a successfully determined version only;
+ # memoized_property does not cache when the function raises, so a
+ # DBAPI which is established after this dialect was constructed is
+ # still picked up
+ if self.dbapi is None:
+ raise exc.NoDBAPILoaded(
+ f"Dialect {self.name}+{self.driver} has no DBAPI module "
+ "loaded; no DBAPI version is available"
+ )
+
+ try:
+ version = self.retrieve_dbapi_version(self.dbapi)
+ except NotImplementedError as ne:
+ raise NotImplementedError(
+ f"Dialect {self.name}+{self.driver} does not implement "
+ "retrieve_dbapi_version(); no DBAPI version is available"
+ ) from ne
+
+ if not version:
+ # the DBAPI is loaded but publishes no version of its own; as
+ # with a DBAPI that isn't loaded, this is not an error on the
+ # part of the dialect
+ # asyncio dialects have a wrapper object here rather than a
+ # module, which has no __name__
+ dbapi_name = getattr(self.dbapi, "__name__", self.driver)
+ raise exc.NoDBAPILoaded(
+ f"Dialect {self.name}+{self.driver} could not determine a "
+ f"version for its DBAPI module {dbapi_name!r}"
+ )
+
+ return version
+
+ @property
+ def _dbapi_version_or_none(self) -> Optional[util.VersionInfo]:
+ """:attr:`.Dialect.dbapi_version`, or None if it can't be
+ determined.
+
+ For use by dialect startup checks, which run before a DBAPI is
+ necessarily present and which must not fail when no version is
+ available. Only :class:`.exc.NoDBAPILoaded` is accommodated;
+ ``NotImplementedError``, indicating a dialect which does not
+ implement :meth:`.Dialect.retrieve_dbapi_version` at all, is a bug
+ in that dialect and is allowed to propagate. Deliberately not
+ memoized itself; the underlying :attr:`.Dialect.dbapi_version`
+ memoizes the success case.
+
+ """
+ try:
+ return self.dbapi_version
+ except exc.NoDBAPILoaded:
+ return None
+
+ def _check_minimum_dbapi_version(self) -> None:
+ """Enforce :attr:`.Dialect.minimum_dbapi_version`, if present.
+
+ Takes place as the dialect is constructed. No check occurs when
+ the version of the DBAPI is not available at all.
+
+ """
+ minimum = self.minimum_dbapi_version
+ if minimum is None:
+ return
+
+ version = self._dbapi_version_or_none
+ if version is not None and version < minimum:
+ dbapi_name = getattr(self.dbapi, "__name__", self.driver)
+ raise exc.InvalidRequestError(
+ f"Dialect {self.name}+{self.driver} requires version "
+ f"{minimum} or greater of the {dbapi_name} DBAPI; "
+ f"version {version} is installed"
+ )
+
@util.memoized_property
def _bind_typing_render_casts(self):
return self.bind_typing is interfaces.BindTyping.RENDER_CASTS
"""
-VersionInfoType = Tuple[Union[int, str], ...]
+ServerVersionInfoType = Tuple[Union[int, str], ...]
+"""The type of :attr:`.Dialect.server_version_info`.
+
+.. versionadded:: 2.1 Renamed from ``VersionInfoType``, which remains
+ present as a synonym. The version of the DBAPI, as opposed to that of
+ the database server, is instead a ``sqlalchemy.util.VersionInfo``; see
+ :attr:`.Dialect.dbapi_version`.
+
+"""
+
+VersionInfoType = ServerVersionInfoType
+
TableKey = Tuple[Optional[str], str]
"""
- server_version_info: Optional[Tuple[Any, ...]]
+ server_version_info: Optional[ServerVersionInfoType]
"""a tuple containing a version number for the DB backend in use.
This value is only available for supporting dialects, and is
typically populated during the initial connection to the database.
"""
+ minimum_dbapi_version: Optional[util.VersionInfo] = None
+ """The minimum version of the DBAPI which this dialect supports.
+
+ When present, :class:`.DefaultDialect` compares this against
+ :attr:`.Dialect.dbapi_version` as the dialect is constructed, raising
+ :class:`.exc.InvalidRequestError` if the DBAPI in use is older. A
+ dialect therefore does not need to implement this check itself::
+
+ class MyDialect(DefaultDialect):
+ minimum_dbapi_version = util.VersionInfo((2, 5))
+
+ No check takes place if the version of the DBAPI is not available, as
+ described at :attr:`.Dialect.dbapi_version`.
+
+ .. versionadded:: 2.1
+
+ """
+
+ @property
+ def dbapi_version(self) -> util.VersionInfo:
+ """the version number of the DBAPI in use.
+
+ In contrast to :attr:`.Dialect.server_version_info`, which refers to
+ the database server itself, this attribute refers to the version of
+ the Python DBAPI module which the dialect makes use of, and is
+ available without any database connection being established.
+
+ The value is a ``sqlalchemy.util.VersionInfo``, a tuple of integers
+ which additionally sorts pre-release versions such as ``2.0.0rc1``
+ as preceding the final release ``(2, 0, 0)``. It may be compared
+ against a plain tuple of integers directly::
+
+ if dialect.dbapi_version >= (2, 5):
+ ...
+
+ Dialects should implement
+ :meth:`.Dialect.retrieve_dbapi_version` only in order to provide
+ this value; this method in turn is used by the
+ :class:`.DefaultDialect` implementation of
+ :attr:`.DefaultDialect.dbapi_version`.
+
+ Two distinct conditions prevent a version from being available:
+
+ * :class:`.exc.NoDBAPILoaded` is raised if the dialect has no DBAPI
+ module loaded, as is the case for a dialect used only to compile
+ statements, or if its DBAPI publishes no version of its own.
+ Neither is an error on the part of the dialect.
+
+ * ``NotImplementedError`` is raised if the dialect does not
+ implement :meth:`.Dialect.retrieve_dbapi_version` at all. This
+ indicates the dialect itself needs to be fixed.
+
+ Consuming code which tolerates a dialect that has not loaded a
+ DBAPI should accommodate the former only, so that a dialect in need
+ of fixing continues to make itself known::
+
+ try:
+ dbapi_version = dialect.dbapi_version
+ except exc.NoDBAPILoaded:
+ dbapi_version = None
+
+ Note that ``hasattr()`` may **not** be used to test for support, as
+ it does not intercept ``NotImplementedError``. Note also that
+ reading this attribute does not cause a DBAPI module to be
+ imported; it reports upon the module already in use, if any.
+
+ A version, once determined, is memoized. As no memoization takes
+ place while the version remains unavailable, a DBAPI which is
+ established after the dialect was constructed is still detected.
+
+ .. versionadded:: 2.1
+
+ .. seealso::
+
+ :meth:`.Dialect.retrieve_dbapi_version`
+
+ """
+ raise NotImplementedError()
+
+ def retrieve_dbapi_version(self, dbapi: DBAPIModule) -> util.VersionInfo:
+ """Return the version of the given DBAPI module.
+
+ This is the dialect-implemented hook behind
+ :attr:`.Dialect.dbapi_version`. A dialect is responsible only for
+ locating where its particular DBAPI publishes a version and parsing
+ it, typically using ``sqlalchemy.util.parse_version_string()``::
+
+ def retrieve_dbapi_version(self, dbapi):
+ return util.parse_version_string(dbapi.__version__)
+
+ The ``dbapi`` argument is the module returned by
+ :meth:`.Dialect.import_dbapi`, which for asyncio dialects is
+ typically a wrapper object rather than the driver module itself.
+
+ This method is only invoked with a DBAPI actually loaded, and only
+ until a version has been determined; the surrounding conditions,
+ including memoization, are handled by :class:`.DefaultDialect`. An
+ empty version may be returned to indicate that no version could be
+ located, which :attr:`.Dialect.dbapi_version` translates into
+ :class:`.exc.NoDBAPILoaded`.
+
+ .. versionadded:: 2.1
+
+ """
+ raise NotImplementedError()
+
default_schema_name: Optional[str]
"""the name of the default schema. This value is only available for
supporting dialects, and is typically populated during the
no context for inspection."""
+class NoDBAPILoaded(InvalidRequestError):
+ """A DBAPI-level attribute was requested from a dialect which has no
+ DBAPI module loaded, or whose DBAPI does not publish the attribute.
+
+ This is not an error on the part of the dialect; a dialect which is
+ used only to compile statements, rather than to interact with a
+ database, has no DBAPI established, and a DBAPI module is not obliged
+ to publish a version number of its own.
+
+ .. versionadded:: 2.1
+
+ """
+
+
class PendingRollbackError(InvalidRequestError):
"""A transaction has failed and needs to be rolled back before
continuing.
class SpecPredicate(Predicate):
+ """Predicate against a database and optional version.
+
+ The ``db`` string is of the form ``<backend>``, ``<backend>+<driver>``
+ or ``+<driver>``, optionally followed by a comparison operator and a
+ dotted version number.
+
+ When a version comparison is present, it applies to the **server**
+ version when no driver is named, and to the **DBAPI** version, i.e.
+ :attr:`.Dialect.dbapi_version`, when a driver is named::
+
+ # server is older than PostgreSQL 12
+ "postgresql<12"
+
+ # the asyncmy DBAPI is older than 0.2.13
+ "+asyncmy<0.2.13"
+
+ Versions with pre-release qualifiers sort as expected, so that
+ ``0.2.13rc1`` matches ``+asyncmy<0.2.13``.
+
+ .. versionadded:: 2.1 Version comparison against a named driver, which
+ previously raised an assertion error.
+
+ """
+
def __init__(self, db, op=None, spec=None, description=None):
self.db = db
self.op = op
return False
if self.op is not None:
- assert driver is None, "DBAPI version specs not supported yet"
-
- version = _server_version(engine)
+ if driver is not None:
+ version = engine.dialect.dbapi_version
+ else:
+ version = _server_version(engine)
oper = (
hasattr(self.op, "__call__") and self.op or self._ops[self.op]
)
from .langhelpers import (
parse_user_argument_for_enum as parse_user_argument_for_enum,
)
+from .langhelpers import (
+ parse_version_from_metadata as parse_version_from_metadata,
+)
+from .langhelpers import parse_version_string as parse_version_string
from .langhelpers import PluginLoader as PluginLoader
from .langhelpers import quoted_token_parser as quoted_token_parser
from .langhelpers import restore_annotations as restore_annotations
from .langhelpers import (
unbound_method_to_callable as unbound_method_to_callable,
)
+from .langhelpers import VersionInfo as VersionInfo
from .langhelpers import walk_subclasses as walk_subclasses
from .langhelpers import warn as warn
from .langhelpers import warn_exception as warn_exception
import collections
import enum
from functools import update_wrapper
+import importlib.metadata
import importlib.util
import inspect
import itertools
return cast(_M, py_module)
+_pre_release_normalize = {
+ "a": "a",
+ "alpha": "a",
+ "b": "b",
+ "beta": "b",
+ "c": "rc",
+ "pre": "rc",
+ "preview": "rc",
+ "rc": "rc",
+}
+
+_version_string_re = re.compile(
+ r"""
+ \s*
+ (?:[a-z][a-z0-9]*[-_])? # ignored prefix, "py3-"
+ v?
+ (?P<release>\d+(?:\.\d+)*)
+ (?: # pre-release
+ [-_.]?
+ (?P<pre_l>alpha|beta|preview|pre|rc|a|b|c)
+ [-_.]?
+ (?P<pre_n>\d+)?
+ )?
+ (?: # post-release
+ [-_.]?
+ (?P<post_l>post|rev|r)
+ [-_.]?
+ (?P<post_n>\d+)?
+ )?
+ (?: # developmental release
+ [-_.]?
+ (?P<dev_l>dev)
+ [-_.]?
+ (?P<dev_n>\d+)?
+ )?
+ """,
+ re.X | re.I,
+)
+
+_VersionSortKey = Tuple[
+ Tuple[int, ...],
+ Tuple[int, str, int],
+ Tuple[int, int],
+ Tuple[int, int],
+]
+
+
+def _version_sort_key(
+ release: Tuple[int, ...],
+ pre: Optional[Tuple[str, int]],
+ post: Optional[int],
+ dev: Optional[int],
+) -> _VersionSortKey:
+ if pre is None and post is None and dev is not None:
+ # a dev release with no other qualifiers precedes every
+ # pre-release of the same release number
+ pre_key = (-1, "", 0)
+ elif pre is None:
+ pre_key = (1, "", 0)
+ else:
+ pre_key = (0, pre[0], pre[1])
+
+ return (
+ release,
+ pre_key,
+ (0, 0) if post is None else (1, post),
+ (1, 0) if dev is None else (0, dev),
+ )
+
+
+def _version_comparison(
+ op: Callable[[Any, Any], bool],
+) -> Callable[[VersionInfo, Any], Any]:
+ """Build one of :class:`.VersionInfo`'s comparison methods.
+
+ Comparison takes place against the sort key rather than the tuple
+ itself, so that pre-release and similar qualifiers are taken into
+ account. A plain tuple is interpreted as the release segment of a
+ final release; anything else is not comparable.
+
+ """
+
+ def compare(self: VersionInfo, other: Any) -> Any:
+ if isinstance(other, VersionInfo):
+ other_key = other._sort_key
+ elif isinstance(other, tuple):
+ other_key = _version_sort_key(other, None, None, None)
+ else:
+ return NotImplemented
+ return op(self._sort_key, other_key)
+
+ return compare
+
+
+class VersionInfo(Tuple[int, ...]):
+ """A version number, as a tuple of integers.
+
+ :class:`.VersionInfo` is a ``tuple`` subclass consisting of the
+ numeric "release" segment of a version only, e.g. ``2.0.0rc1``
+ is the tuple ``(2, 0, 0)``. Ordering however takes any
+ pre-release, post-release and developmental qualifiers into account
+ as described by :pep:`440`, so that ``2.0.0rc1`` compares as less than
+ ``2.0.0``, including when compared against a plain tuple such as
+ ``(2, 0, 0)``.
+
+ Plain tuples are interpreted as final releases when compared against
+ a :class:`.VersionInfo`.
+
+ .. versionadded:: 2.1
+
+ """
+
+ string: Optional[str]
+ """the string from which this version was parsed, if any."""
+
+ pre: Optional[Tuple[str, int]]
+ """normalized pre-release qualifier, e.g. ``("rc", 1)``."""
+
+ post: Optional[int]
+ """post-release number, if any."""
+
+ dev: Optional[int]
+ """developmental release number, if any."""
+
+ _sort_key: _VersionSortKey
+
+ def __new__(
+ cls,
+ release: Sequence[int] = (),
+ *,
+ string: Optional[str] = None,
+ pre: Optional[Tuple[str, int]] = None,
+ post: Optional[int] = None,
+ dev: Optional[int] = None,
+ ) -> VersionInfo:
+ # __new__ is needed as the release segment has to be passed to
+ # tuple.__new__(); the remaining state is set up in __init__
+ return tuple.__new__(cls, release)
+
+ def __init__(
+ self,
+ release: Sequence[int] = (),
+ *,
+ string: Optional[str] = None,
+ pre: Optional[Tuple[str, int]] = None,
+ post: Optional[int] = None,
+ dev: Optional[int] = None,
+ ):
+ self.string = string
+ self.pre = pre
+ self.post = post
+ self.dev = dev
+ self._sort_key = _version_sort_key(tuple(self), pre, post, dev)
+
+ def __repr__(self) -> str:
+ if self.string is not None:
+ return f"VersionInfo({tuple(self)!r}, string={self.string!r})"
+ else:
+ return f"VersionInfo({tuple(self)!r})"
+
+ def __str__(self) -> str:
+ if self.string is not None:
+ return self.string
+ else:
+ return ".".join(str(num) for num in self)
+
+ # every comparison has to be stated explicitly; ``tuple`` implements
+ # all six of them, so ``functools.total_ordering`` fills in nothing
+ # here and the ones left out would silently compare as plain tuples
+ __eq__ = _version_comparison(operator.eq)
+ __ne__ = _version_comparison(operator.ne)
+ __lt__ = _version_comparison(operator.lt)
+ __le__ = _version_comparison(operator.le)
+ __gt__ = _version_comparison(operator.gt)
+ __ge__ = _version_comparison(operator.ge)
+
+ def __hash__(self) -> int:
+ return hash(self._sort_key)
+
+
+def parse_version_string(version: Optional[str]) -> VersionInfo:
+ """Parse a DBAPI version string into a :class:`.VersionInfo`.
+
+ Leading characters that are not part of the version itself are
+ ignored, as are trailing characters following the version, so that
+ strings such as ``"py3-4.0.19-beta4"`` and
+ ``"2.9.10 (dt dec pq3 ext lo64)"`` parse correctly.
+
+ An empty :class:`.VersionInfo` is returned if no version number can be
+ located at all.
+
+ Parsing is deliberately more tolerant than that of :pep:`440`, which
+ the version strings published by DBAPIs frequently do not conform to;
+ a strict implementation such as that of the ``packaging`` library
+ rejects each of the above outright.
+
+ .. versionadded:: 2.1
+
+ """
+
+ if not version:
+ return VersionInfo((), string=version)
+
+ m = _version_string_re.match(version)
+ if m is None:
+ return VersionInfo((), string=version)
+
+ release = tuple(int(x) for x in m.group("release").split("."))
+
+ pre_l = m.group("pre_l")
+ pre: Optional[Tuple[str, int]]
+ if pre_l is not None:
+ pre = (
+ _pre_release_normalize[pre_l.lower()],
+ int(m.group("pre_n") or 0),
+ )
+ else:
+ pre = None
+
+ return VersionInfo(
+ release,
+ string=version,
+ pre=pre,
+ post=(
+ int(m.group("post_n") or 0)
+ if m.group("post_l") is not None
+ else None
+ ),
+ dev=(
+ int(m.group("dev_n") or 0)
+ if m.group("dev_l") is not None
+ else None
+ ),
+ )
+
+
+def parse_version_from_metadata(distribution: str) -> VersionInfo:
+ """Return the version of an installed distribution as a
+ :class:`.VersionInfo`.
+
+ This is intended for use by dialects whose DBAPI module does not
+ itself publish a version number, such as ``asyncmy``. As the
+ distribution name is not necessarily the same as the module name, and
+ the installed distribution is not necessarily the module that was
+ imported, this should not be used when the DBAPI module provides a
+ version of its own.
+
+ An empty :class:`.VersionInfo` is returned if the distribution is not
+ installed.
+
+ .. versionadded:: 2.1
+
+ """
+
+ try:
+ version = importlib.metadata.version(distribution)
+ except importlib.metadata.PackageNotFoundError:
+ return VersionInfo()
+ else:
+ return parse_version_string(version)
+
+
class _Missing(enum.Enum):
Missing = enum.auto()
sa_exceptions.TimeoutError,
sa_exceptions.InvalidRequestError,
sa_exceptions.IllegalStateChangeError,
+ sa_exceptions.NoDBAPILoaded,
sa_exceptions.NoInspectionAvailable,
sa_exceptions.PendingRollbackError,
sa_exceptions.ResourceClosedError,
import copy
from decimal import Decimal
+import importlib.metadata
import inspect
+import operator
from pathlib import Path
import pickle
import sys
self._test('"na.me"', ["na.me"])
+class ParseVersionStringTest(fixtures.TestBase):
+ @combinations(
+ ("2.1.8", (2, 1, 8), None, None, None),
+ ("10.15.17", (10, 15, 17), None, None, None),
+ ("v1.2.3", (1, 2, 3), None, None, None),
+ ("1.4", (1, 4), None, None, None),
+ # pyodbc style, with a prefix and a spelled out pre-release
+ ("py3-3.0.1-beta4", (3, 0, 1), ("b", 4), None, None),
+ ("py3-4.0.19", (4, 0, 19), None, None, None),
+ # psycopg2 style, with trailing information
+ ("2.9.10 (dt dec pq3 ext lo64)", (2, 9, 10), None, None, None),
+ # pep 484 pre-release, post-release, developmental release
+ ("2.0.0a1", (2, 0, 0), ("a", 1), None, None),
+ ("2.0.0b1", (2, 0, 0), ("b", 1), None, None),
+ ("2.0.0rc1", (2, 0, 0), ("rc", 1), None, None),
+ ("2.0.0c1", (2, 0, 0), ("rc", 1), None, None),
+ ("2.0.0-alpha", (2, 0, 0), ("a", 0), None, None),
+ ("2.0.0.post2", (2, 0, 0), None, 2, None),
+ ("2.0.0.dev3", (2, 0, 0), None, None, 3),
+ ("2.0.0b1.dev3", (2, 0, 0), ("b", 1), None, 3),
+ # no version at all
+ ("crap.crap.crap", (), None, None, None),
+ ("", (), None, None, None),
+ (None, (), None, None, None),
+ argnames="version, release, pre, post, dev",
+ )
+ def test_parse(self, version, release, pre, post, dev):
+ parsed = util.parse_version_string(version)
+ eq_(
+ (
+ tuple(parsed),
+ parsed.pre,
+ parsed.post,
+ parsed.dev,
+ parsed.string,
+ ),
+ (release, pre, post, dev, version),
+ )
+
+ @combinations(
+ # a VersionInfo compares equal to the plain tuple of its release
+ # segment when it has no pre / post / dev qualifiers
+ ("2.0.0", "==", (2, 0, 0)),
+ ("2.0.0", "<", (2, 0, 1)),
+ ("2.0.0", ">", (1, 9, 9)),
+ ("0.2.12", "<", (0, 2, 13)),
+ ("0.2.13", "==", (0, 2, 13)),
+ # pre-releases precede the release they qualify, including when
+ # compared against a plain tuple
+ ("2.0.0b1", "<", (2, 0, 0)),
+ ("0.2.13rc1", "<", (0, 2, 13)),
+ ("2.0.0.dev1", "<", (2, 0, 0)),
+ # post-releases follow it
+ ("2.0.0.post1", ">", (2, 0, 0)),
+ argnames="version, op, other",
+ )
+ def test_compare_to_tuple(self, version, op, other):
+ """all six comparisons consult the sort key.
+
+ ``tuple`` implements every one of them, so an operator which
+ VersionInfo fails to state explicitly silently compares as a plain
+ tuple and gets the pre-release cases wrong.
+
+ """
+
+ parsed = util.parse_version_string(version)
+ eq_(
+ (
+ parsed < other,
+ parsed <= other,
+ parsed == other,
+ parsed != other,
+ parsed >= other,
+ parsed > other,
+ ),
+ (
+ op == "<",
+ op in ("<", "=="),
+ op == "==",
+ op != "==",
+ op in (">", "=="),
+ op == ">",
+ ),
+ )
+
+ def test_reflected_comparison_to_tuple(self):
+ """a plain tuple on the left still gets VersionInfo semantics.
+
+ VersionInfo is a tuple subclass, so Python tries its reflected
+ operation first.
+
+ """
+
+ parsed = util.parse_version_string("2.0.0rc1")
+
+ # (2, 0, 0) is greater than 2.0.0rc1
+ eq_(
+ (
+ (2, 0, 0) < parsed,
+ (2, 0, 0) <= parsed,
+ (2, 0, 0) == parsed,
+ (2, 0, 0) != parsed,
+ (2, 0, 0) >= parsed,
+ (2, 0, 0) > parsed,
+ ),
+ (False, False, False, True, True, True),
+ )
+
+ @combinations("<", "<=", ">=", ">", argnames="op")
+ def test_not_comparable(self, op):
+ oper = {
+ "<": operator.lt,
+ "<=": operator.le,
+ ">=": operator.ge,
+ ">": operator.gt,
+ }[op]
+
+ with expect_raises(TypeError):
+ oper(util.parse_version_string("2.0.0"), "2.0.0")
+
+ def test_pep440_ordering(self):
+ """dev < alpha < beta < rc < final < post"""
+
+ versions = [
+ "2.0.0.dev1",
+ "2.0.0a1",
+ "2.0.0a2",
+ "2.0.0b1",
+ "2.0.0rc1",
+ "2.0.0",
+ "2.0.0.post1",
+ "2.0.1",
+ ]
+ parsed = [util.parse_version_string(v) for v in versions]
+ eq_(sorted(reversed(parsed)), parsed)
+
+ def test_hash(self):
+ eq_(
+ {util.parse_version_string("2.0.0b1")},
+ {util.parse_version_string("2.0.0b1")},
+ )
+ ne_(
+ hash(util.parse_version_string("2.0.0b1")),
+ hash(util.parse_version_string("2.0.0")),
+ )
+
+ def test_compare_to_non_tuple(self):
+ parsed = util.parse_version_string("2.0.0")
+ is_false(parsed == "2.0.0")
+ with expect_raises(TypeError):
+ parsed < "2.0.0"
+
+ def test_from_metadata(self):
+ # pytest is necessarily installed for this test to be running at
+ # all; greenlet and the like are optional
+ eq_(
+ util.parse_version_from_metadata("pytest"),
+ util.parse_version_string(importlib.metadata.version("pytest")),
+ )
+
+ def test_from_metadata_not_installed(self):
+ eq_(util.parse_version_from_metadata("no_such_distribution"), ())
+
+
class BackslashReplaceTest(fixtures.TestBase):
def test_ascii_to_utf8(self):
eq_(
from sqlalchemy import Table
from sqlalchemy import testing
from sqlalchemy.dialects import mysql
+from sqlalchemy.dialects.mysql import mysqldb
+from sqlalchemy.dialects.mysql import pymysql
from sqlalchemy.dialects.mysql.pymysql import _connection_ping_reconnects_true
from sqlalchemy.engine.url import make_url
from sqlalchemy.testing import assert_raises_message
eq_(detected, "utf8mb4")
+class DBAPIVersionTest(fixtures.TestBase):
+ """test :meth:`.Dialect.retrieve_dbapi_version` for the mysqldb family
+ of dialects."""
+
+ def test_mysqlclient(self):
+ """mysqlclient publishes version_info only"""
+
+ dbapi = mock.Mock(spec=["version_info", "paramstyle"])
+ dbapi.version_info = (2, 2, 7, "final", 0)
+ dbapi.paramstyle = "format"
+
+ dialect = mysqldb.MySQLDialect_mysqldb(dbapi=dbapi)
+ eq_(dialect.dbapi_version, (2, 2, 7))
+
+ def test_mysql_python(self):
+ """the legacy MySQL-python published __version__"""
+
+ dbapi = mock.Mock(spec=["__version__", "paramstyle"])
+ dbapi.__version__ = "1.2.5"
+ dbapi.paramstyle = "format"
+
+ dialect = mysqldb.MySQLDialect_mysqldb(dbapi=dbapi)
+ eq_(dialect.dbapi_version, (1, 2, 5))
+
+ def test_pymysql(self):
+ """pymysql's __version__ / version_info are mysqlclient
+ compatibility values; VERSION_STRING is its own"""
+
+ dbapi = mock.Mock(
+ spec=[
+ "VERSION_STRING",
+ "__version__",
+ "version_info",
+ "paramstyle",
+ ]
+ )
+ dbapi.VERSION_STRING = "1.2.0"
+ dbapi.__version__ = "2.2.8"
+ dbapi.version_info = (2, 2, 8, "final", 1)
+ dbapi.paramstyle = "format"
+
+ dialect = pymysql.MySQLDialect_pymysql(dbapi=dbapi)
+ eq_(dialect.dbapi_version, (1, 2, 0))
+
+
class ParseVersionTest(fixtures.TestBase):
def test_mariadb_madness(self):
mysql_dialect = make_url("mysql+mysqldb://").get_dialect()()
class CxOracleDialectTest(fixtures.TestBase):
def test_cx_oracle_version_parse(self):
- dialect = cx_oracle.OracleDialect_cx_oracle()
-
def check(version):
- dbapi = Mock(version=version)
- dialect._load_version(dbapi)
+ # a new dialect per version; dbapi_version is memoized
+ dialect = cx_oracle.OracleDialect_cx_oracle(
+ dbapi=Mock(version=version)
+ )
return dialect.cx_oracle_ver
eq_(check("8.2"), (8, 2))
eq_(check("8.0.1"), (8, 0, 1))
- eq_(check("9.0b1"), (9, 0))
+
+ # a pre-release sorts before the release it qualifies
+ beta = check("9.0b1")
+ eq_(tuple(beta), (9, 0))
+ is_true(beta < (9, 0))
def test_minimum_version(self):
with expect_raises_message(
exc.InvalidRequestError,
- "cx_Oracle version 8 and above are supported",
+ r"Dialect oracle\+cx_oracle requires version 8 or greater of "
+ r"the cx_oracle DBAPI; version 5.1.5 is installed",
):
cx_oracle.OracleDialect_cx_oracle(dbapi=Mock(version="5.1.5"))
__only_on__ = "oracle+oracledb"
def test_oracledb_version_parse(self):
- dialect = oracledb.OracleDialect_oracledb()
-
def check(version):
- dbapi = Mock(version=version)
- dialect._load_version(dbapi)
+ # a new dialect per version; dbapi_version is memoized
+ dialect = oracledb.OracleDialect_oracledb(
+ dbapi=Mock(version=version)
+ )
return dialect.oracledb_ver
eq_(check("7.2"), (7, 2))
eq_(check("7.0.1"), (7, 0, 1))
- eq_(check("9.0b1"), (9, 0))
+
+ # a pre-release sorts before the release it qualifies
+ beta = check("9.0b1")
+ eq_(tuple(beta), (9, 0))
+ is_true(beta < (9, 0))
def test_minimum_version(self):
with expect_raises_message(
exc.InvalidRequestError,
- r"oracledb version \(1,\) and above are supported",
+ r"Dialect oracle\+oracledb requires version 1 or greater of "
+ r"the oracledb DBAPI; version 0.1.5 is installed",
):
oracledb.OracleDialect_oracledb(dbapi=Mock(version="0.1.5"))
def test_async_minimum_version(self):
with expect_raises_message(
exc.InvalidRequestError,
- r"oracledb version \(2, 0, 1\) and above are supported",
+ r"Dialect oracle\+oracledb requires version 2.0.1 or greater "
+ r"of the oracledb DBAPI; version 2.0.0 is installed",
):
oracledb.OracleDialectAsync_oracledb(dbapi=Mock(version="2.0.0"))
dialect = oracle.dialect(
dbapi=Mock(
- version="0.0.0",
+ version="2.4.0",
paramstyle="named",
),
**kw,
+++ /dev/null
-from sqlalchemy.connectors import pyodbc
-from sqlalchemy.testing import eq_
-from sqlalchemy.testing import fixtures
-
-
-class PyODBCTest(fixtures.TestBase):
- def test_pyodbc_version(self):
- connector = pyodbc.PyODBCConnector()
- for vers, expected in [
- ("2.1.8", (2, 1, 8)),
- ("py3-3.0.1-beta4", (3, 0, 1, "beta4")),
- ("10.15.17", (10, 15, 17)),
- ("crap.crap.crap", ()),
- ]:
- eq_(connector._parse_dbapi_version(vers), expected)
import copy
+import re
from unittest.mock import call
from unittest.mock import MagicMock
from unittest.mock import Mock
eq_(dialect.name, name)
+class DBAPIVersionTest(fixtures.TestBase):
+ """test :attr:`.Dialect.dbapi_version` and the
+ :meth:`.Dialect.retrieve_dbapi_version` hook."""
+
+ def _dialect_cls(self, **kw):
+ class MyDialect(DefaultDialect):
+ name = "mydialect"
+ driver = "mydriver"
+ retrieve_calls = 0
+
+ def retrieve_dbapi_version(self, dbapi):
+ MyDialect.retrieve_calls += 1
+ return tsa.util.parse_version_string(
+ getattr(dbapi, "__version__", None)
+ )
+
+ return MyDialect
+
+ def test_version_from_hook(self):
+ d = self._dialect_cls()(dbapi=Mock(__version__="1.2.3"))
+ eq_(d.dbapi_version, (1, 2, 3))
+
+ def test_memoized(self):
+ """the hook is invoked once for repeated reads"""
+
+ cls = self._dialect_cls()
+ d = cls(dbapi=Mock(__version__="1.2.3"))
+ eq_([d.dbapi_version, d.dbapi_version, d.dbapi_version][0], (1, 2, 3))
+ eq_(cls.retrieve_calls, 1)
+
+ def test_no_dbapi_not_memoized(self):
+ """a dialect with no DBAPI raises, and picks up a DBAPI which is
+ established afterwards"""
+
+ cls = self._dialect_cls()
+ d = cls()
+
+ with expect_raises_message(
+ exc.NoDBAPILoaded,
+ "Dialect mydialect\\+mydriver has no DBAPI module loaded",
+ ):
+ d.dbapi_version
+
+ eq_(cls.retrieve_calls, 0)
+ is_false("dbapi_version" in d.__dict__)
+
+ d.dbapi = Mock(__version__="1.2.3")
+ eq_(d.dbapi_version, (1, 2, 3))
+
+ def test_hook_not_implemented(self):
+ class NoHookDialect(DefaultDialect):
+ name = "nohook"
+ driver = "nohook"
+
+ d = NoHookDialect(dbapi=Mock(__version__="1.2.3"))
+ with expect_raises_message(
+ NotImplementedError,
+ "Dialect nohook\\+nohook does not implement "
+ "retrieve_dbapi_version\\(\\)",
+ ):
+ d.dbapi_version
+
+ def test_version_not_determinable(self):
+ # a DBAPI which publishes no version at all
+ dbapi = Mock(spec=["paramstyle"], __name__="mydbapi")
+ dbapi.paramstyle = "qmark"
+ d = self._dialect_cls()(dbapi=dbapi)
+ with expect_raises_message(
+ exc.NoDBAPILoaded,
+ "Dialect mydialect\\+mydriver could not determine a version "
+ "for its DBAPI module 'mydbapi'",
+ ):
+ d.dbapi_version
+
+ def test_version_not_determinable_wrapper_dbapi(self):
+ """asyncio dialects have a wrapper object rather than a module,
+ which has no __name__ to name in the message"""
+
+ dbapi = Mock(spec=["paramstyle"])
+ dbapi.paramstyle = "qmark"
+ d = self._dialect_cls()(dbapi=dbapi)
+ with expect_raises_message(
+ exc.NoDBAPILoaded,
+ "Dialect mydialect\\+mydriver could not determine a version "
+ "for its DBAPI module 'mydriver'",
+ ):
+ d.dbapi_version
+
+ @testing.combinations("no_dbapi", "no_version", argnames="scenario")
+ def test_or_none_helper(self, scenario):
+ """_dbapi_version_or_none returns None rather than raising"""
+
+ if scenario == "no_dbapi":
+ dbapi = None
+ else:
+ dbapi = Mock(spec=["paramstyle"], __name__="mydbapi")
+ dbapi.paramstyle = "qmark"
+
+ d = self._dialect_cls()(dbapi=dbapi)
+ is_(d._dbapi_version_or_none, None)
+
+ def test_or_none_helper_present(self):
+ d = self._dialect_cls()(dbapi=Mock(__version__="1.2.3"))
+ eq_(d._dbapi_version_or_none, (1, 2, 3))
+
+
+class MinimumDBAPIVersionTest(fixtures.TestBase):
+ """test :attr:`.Dialect.minimum_dbapi_version`."""
+
+ def _dialect_cls(self):
+ class MyDialect(DefaultDialect):
+ name = "mydialect"
+ driver = "mydriver"
+
+ minimum_dbapi_version = tsa.util.VersionInfo((2, 5))
+
+ def retrieve_dbapi_version(self, dbapi):
+ return tsa.util.parse_version_string(
+ getattr(dbapi, "__version__", None)
+ )
+
+ return MyDialect
+
+ @testing.combinations("2.5", "2.5.1", "3.0", argnames="version")
+ def test_version_ok(self, version):
+ d = self._dialect_cls()(
+ dbapi=Mock(__name__="mydbapi", __version__=version)
+ )
+ eq_(d.dbapi_version, tsa.util.parse_version_string(version))
+
+ @testing.combinations("2.4.9", "2.5rc1", "1.0", argnames="version")
+ def test_version_too_low(self, version):
+ with expect_raises_message(
+ exc.InvalidRequestError,
+ "Dialect mydialect\\+mydriver requires version 2.5 or greater "
+ f"of the mydbapi DBAPI; version {re.escape(version)} is "
+ "installed",
+ ):
+ self._dialect_cls()(
+ dbapi=Mock(__name__="mydbapi", __version__=version)
+ )
+
+ @testing.combinations("no_dbapi", "no_version", argnames="scenario")
+ def test_no_version_no_check(self, scenario):
+ """a version which can't be determined skips the check"""
+
+ if scenario == "no_dbapi":
+ dbapi = None
+ else:
+ dbapi = Mock(spec=["paramstyle"], __name__="mydbapi")
+ dbapi.paramstyle = "qmark"
+
+ d = self._dialect_cls()(dbapi=dbapi)
+ is_(d._dbapi_version_or_none, None)
+
+ def test_no_minimum_no_check(self):
+ """a dialect with no minimum doesn't consult the version at all"""
+
+ class NoHookDialect(DefaultDialect):
+ name = "nohook"
+ driver = "nohook"
+
+ # retrieve_dbapi_version() would raise NotImplementedError
+ NoHookDialect(dbapi=Mock(__name__="mydbapi", __version__="1.0"))
+
+
class CreateEngineTest(fixtures.TestBase):
"""test that create_engine arguments of different types get
propagated properly"""
config, "mssql+aioodbc"
):
return False
- if config.db.dialect._dbapi_version() < (4, 0, 19):
+ if config.db.dialect.dbapi_version < (4, 0, 19):
return False
with config.db.connect() as conn:
driver_connection = conn.connection.driver_connection
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
