if name is None or name == "default":
return default.DefaultDialect()
else:
- try:
- dialect_mod = _dialect_mods[name]
- except KeyError:
- dialect_mod = getattr(
- __import__("sqlalchemy.dialects.%s" % name).dialects, name
- )
- _dialect_mods[name] = dialect_mod
- d = dialect_mod.dialect()
+ from sqlalchemy.engine import url
+
+ d = url.URL(name).get_dialect()()
+
if name == "postgresql":
d.implicit_returning = True
elif name == "mssql":
from ..util.compat import string_types
from ..util.compat import text_type
from ..util.sqla_compat import create_mock_engine
+from ..util.sqla_compat import sqla_14
testing_config = configparser.ConfigParser()
testing_config.read(["test.cfg"])
opts["as_sql"] = as_sql
if literal_binds:
opts["literal_binds"] = literal_binds
- if dialect == "mariadb":
+ if not sqla_14 and dialect == "mariadb":
ctx_dialect = _get_dialect("mysql")
- ctx_dialect.server_version_info = (10, 0, 0, "MariaDB")
+ ctx_dialect.server_version_info = (10, 4, 0, "MariaDB")
else:
ctx_dialect = _get_dialect(dialect)
def _is_mariadb(mysql_dialect):
- return (
- mysql_dialect.server_version_info
- and "MariaDB" in mysql_dialect.server_version_info
- )
+ if sqla_14:
+ return mysql_dialect.is_mariadb
+ else:
+ return mysql_dialect.server_version_info and mysql_dialect._is_mariadb
def _mariadb_normalized_version_info(mysql_dialect):
- if len(mysql_dialect.server_version_info) > 5:
- return mysql_dialect.server_version_info[3:]
- else:
- return mysql_dialect.server_version_info
+ return mysql_dialect._mariadb_normalized_version_info
if sqla_14: