from ..util import EMPTY_DICT
from ..util import HasMemoized as HasMemoized
from ..util import hybridmethod
+from ..util import warn_deprecated
from ..util.typing import Self
from ..util.typing import TypeVarTuple
from ..util.typing import Unpack
("dialect_options", InternalTraversal.dp_dialect_options)
]
+ def get_dialect_option(
+ self,
+ dialect: Dialect,
+ argument_name: str,
+ *,
+ else_: Any = None,
+ deprecated_fallback: Optional[str] = None,
+ ) -> Any:
+ r"""Return the value of a dialect-specific option, or *else_* if
+ this dialect does not register the given argument.
+
+ This is useful for DDL compilers that may be inherited by
+ third-party dialects whose ``construct_arguments`` do not
+ include the same set of keys as the parent dialect.
+
+ :param dialect: The dialect for which to retrieve the option.
+ :param argument_name: The name of the argument to retrieve.
+ :param else\_: The value to return if the argument is not present.
+ :param deprecated_fallback: Optional dialect name to fall back to
+ if the argument is not present for the current dialect. If the
+ argument is present for the fallback dialect but not the current
+ dialect, a deprecation warning will be emitted.
+
+ """
+
+ registry = DialectKWArgs._kw_registry[dialect.name]
+ if registry is None:
+ return else_
+
+ if argument_name in registry.get(self.__class__, {}):
+ if (
+ deprecated_fallback is None
+ or dialect.name == deprecated_fallback
+ ):
+ return self.dialect_options[dialect.name][argument_name]
+
+ # deprecated_fallback is present; need to look in two places
+
+ # Current dialect has this option registered.
+ # Check if user explicitly set it.
+ if (
+ dialect.name in self.dialect_options
+ and argument_name
+ in self.dialect_options[dialect.name]._non_defaults
+ ):
+ # User explicitly set this dialect's option - use it
+ return self.dialect_options[dialect.name][argument_name]
+
+ # User didn't set current dialect's option.
+ # Check for deprecated fallback.
+ elif (
+ deprecated_fallback in self.dialect_options
+ and argument_name
+ in self.dialect_options[deprecated_fallback]._non_defaults
+ ):
+ # User set fallback option but not current dialect's option
+ warn_deprecated(
+ f"Using '{deprecated_fallback}_{argument_name}' "
+ f"with the '{dialect.name}' dialect is deprecated; "
+ f"please additionally specify "
+ f"'{dialect.name}_{argument_name}'.",
+ version="2.1",
+ )
+ return self.dialect_options[deprecated_fallback][argument_name]
+
+ # Return default value
+ return self.dialect_options[dialect.name][argument_name]
+ else:
+ # Current dialect doesn't have the option registered at all.
+ # Don't warn - if a third-party dialect doesn't support an
+ # option, that's their choice, not a deprecation case.
+ return else_
+
@classmethod
def argument_for(
cls, dialect_name: str, argument_name: str, default: Any
+import contextlib
import random
from sqlalchemy import BLOB
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import eq_
from sqlalchemy.testing import eq_ignore_whitespace
+from sqlalchemy.testing import expect_deprecated
from sqlalchemy.testing import expect_raises
from sqlalchemy.testing import expect_warnings
from sqlalchemy.testing import fixtures
schema.CreateIndex(idx), "CREATE INDEX test_idx1 ON testtbl (data)"
)
- def test_create_index_with_prefix(self):
+ def test_create_index_delegates_to_dialect_option(
+ self, thirdparty_dialect
+ ):
+ """test the original 3rd party dialect issue in #13134, which is
+ that visit_create_index() doesn't assume kw like "length" are
+ present"""
+
+ @thirdparty_dialect
+ class ThirdPartyDialect(mysql.MySQLDialect):
+ name = "thirdparty"
+ construct_arguments = []
+
+ m = MetaData()
+ tbl = Table("testtbl", m, Column("data", String(255)))
+ idx = Index("test_idx1", tbl.c.data)
+
+ self.assert_compile(
+ schema.CreateIndex(idx),
+ "CREATE INDEX test_idx1 ON testtbl (data)",
+ dialect=ThirdPartyDialect(),
+ )
+
+ @testing.combinations("mysql", "mariadb", argnames="dialect_name")
+ def test_create_index_with_prefix(self, dialect_name):
m = MetaData()
tbl = Table("testtbl", m, Column("data", String(255)))
idx = Index(
- "test_idx1", tbl.c.data, mysql_length=10, mysql_prefix="FULLTEXT"
+ "test_idx1",
+ tbl.c.data,
+ **{
+ f"{dialect_name}_length": 10,
+ f"{dialect_name}_prefix": "FULLTEXT",
+ },
)
self.assert_compile(
schema.CreateIndex(idx),
"CREATE FULLTEXT INDEX test_idx1 ON testtbl (data(10))",
+ dialect=dialect_name,
)
def test_create_index_with_text(self):
"CREATE INDEX test_idx1 ON testtbl (created_at desc)",
)
- def test_create_index_with_parser(self):
+ @testing.combinations("mysql", "mariadb", argnames="dialect_name")
+ def test_create_index_with_parser(self, dialect_name):
m = MetaData()
tbl = Table("testtbl", m, Column("data", String(255)))
idx = Index(
"test_idx1",
tbl.c.data,
- mysql_length=10,
- mysql_prefix="FULLTEXT",
- mysql_with_parser="ngram",
+ **{
+ f"{dialect_name}_length": 10,
+ f"{dialect_name}_prefix": "FULLTEXT",
+ f"{dialect_name}_with_parser": "ngram",
+ },
)
self.assert_compile(
schema.CreateIndex(idx),
"CREATE FULLTEXT INDEX test_idx1 "
"ON testtbl (data(10)) WITH PARSER ngram",
+ dialect=dialect_name,
)
- def test_create_index_with_length(self):
+ @testing.combinations("mysql", "mariadb", argnames="dialect_name")
+ def test_create_index_with_length(self, dialect_name):
m = MetaData()
tbl = Table("testtbl", m, Column("data", String(255)))
- idx1 = Index("test_idx1", tbl.c.data, mysql_length=10)
- idx2 = Index("test_idx2", tbl.c.data, mysql_length=5)
+ idx1 = Index("test_idx1", tbl.c.data, **{f"{dialect_name}_length": 10})
+ idx2 = Index("test_idx2", tbl.c.data, **{f"{dialect_name}_length": 5})
self.assert_compile(
schema.CreateIndex(idx1),
"CREATE INDEX test_idx1 ON testtbl (data(10))",
+ dialect=dialect_name,
)
self.assert_compile(
schema.CreateIndex(idx2),
"CREATE INDEX test_idx2 ON testtbl (data(5))",
+ dialect=dialect_name,
)
def test_drop_constraint_mysql(self):
"CREATE INDEX test_idx3 ON testtbl (a(30), b(30))",
)
- def test_create_index_with_using(self):
+ @testing.combinations("mysql", "mariadb", argnames="dialect_name")
+ def test_create_index_with_using(self, dialect_name):
m = MetaData()
tbl = Table("testtbl", m, Column("data", String(255)))
- idx1 = Index("test_idx1", tbl.c.data, mysql_using="btree")
- idx2 = Index("test_idx2", tbl.c.data, mysql_using="hash")
+ idx1 = Index(
+ "test_idx1", tbl.c.data, **{f"{dialect_name}_using": "btree"}
+ )
+ idx2 = Index(
+ "test_idx2", tbl.c.data, **{f"{dialect_name}_using": "hash"}
+ )
self.assert_compile(
schema.CreateIndex(idx1),
"CREATE INDEX test_idx1 ON testtbl (data) USING btree",
+ dialect=dialect_name,
)
self.assert_compile(
schema.CreateIndex(idx2),
"CREATE INDEX test_idx2 ON testtbl (data) USING hash",
+ dialect=dialect_name,
)
def test_create_pk_plain(self):
"PRIMARY KEY (data))",
)
- def test_create_pk_with_using(self):
+ @testing.combinations("mysql", "mariadb", argnames="dialect_name")
+ def test_create_pk_with_using(self, dialect_name):
m = MetaData()
tbl = Table(
"testtbl",
m,
Column("data", String(255)),
- PrimaryKeyConstraint("data", mysql_using="btree"),
+ PrimaryKeyConstraint("data", **{f"{dialect_name}_using": "btree"}),
)
self.assert_compile(
schema.CreateTable(tbl),
"CREATE TABLE testtbl (data VARCHAR(255) NOT NULL, "
"PRIMARY KEY (data) USING btree)",
+ dialect=dialect_name,
+ )
+
+ @testing.combinations(
+ ("with_parser", "ngram", "WITH PARSER ngram"),
+ ("using", "btree", "USING btree"),
+ argnames="paramname, value, expected",
+ )
+ @testing.variation("use_deprecated", [True, False])
+ def test_create_index_mysql_option_mariadb_deprecated(
+ self, paramname, value, expected, use_deprecated
+ ):
+ """Test that mysql_with_parser emits deprecation with mariadb
+ dialect"""
+ m = MetaData()
+ tbl = Table("testtbl", m, Column("data", String(255)))
+ idx = Index(
+ "test_idx1",
+ tbl.c.data,
+ mariadb_length=10,
+ mariadb_prefix="FULLTEXT",
+ **{
+ f"{'mysql' if use_deprecated else 'mariadb'}"
+ f"_{paramname}": value
+ },
)
+ if use_deprecated:
+ expect = expect_deprecated(
+ f"Using 'mysql_{paramname}' with the 'mariadb' dialect is "
+ f"deprecated; please additionally specify "
+ f"'mariadb_{paramname}'.",
+ )
+ else:
+ expect = contextlib.nullcontext()
+
+ with expect:
+ self.assert_compile(
+ schema.CreateIndex(idx),
+ "CREATE FULLTEXT INDEX test_idx1 "
+ f"ON testtbl (data(10)) {expected}",
+ dialect="mariadb",
+ )
+
+ def test_create_pk_with_using_mysql_option_mariadb_deprecated(self):
+ """Test that mysql_using for PK emits deprecation with mariadb
+ dialect"""
+ m = MetaData()
+ tbl = Table(
+ "testtbl",
+ m,
+ Column("data", String(255)),
+ PrimaryKeyConstraint("data", mysql_using="btree"),
+ )
+
+ with expect_deprecated(
+ "Using 'mysql_using' with the 'mariadb' dialect is deprecated; "
+ "please additionally specify 'mariadb_using'."
+ ):
+ self.assert_compile(
+ schema.CreateTable(tbl),
+ "CREATE TABLE testtbl (data VARCHAR(255) NOT NULL, "
+ "PRIMARY KEY (data) USING btree)",
+ dialect="mariadb",
+ )
+
@testing.combinations(
(True, True, (10, 2, 2)),
(True, True, (10, 2, 1)),
5,
)
+ def test_get_dialect_option_participating(self):
+ with self._fixture():
+ idx = Index("a", "b", "c", participating_x=7)
+ dialect = mock.Mock()
+ dialect.name = "participating"
+ eq_(idx.get_dialect_option(dialect, "x"), 7)
+ eq_(idx.get_dialect_option(dialect, "y"), False)
+
+ def test_get_dialect_option_participating_default(self):
+ with self._fixture():
+ idx = Index("a", "b", "c")
+ dialect = mock.Mock()
+ dialect.name = "participating"
+ eq_(idx.get_dialect_option(dialect, "x"), 5)
+ eq_(idx.get_dialect_option(dialect, "z_one"), None)
+
+ def test_get_dialect_option_participating_unregistered_arg(self):
+ with self._fixture():
+ idx = Index("a", "b", "c", participating_x=7)
+ dialect = mock.Mock()
+ dialect.name = "participating"
+ eq_(idx.get_dialect_option(dialect, "nonexistent"), None)
+ eq_(
+ idx.get_dialect_option(
+ dialect, "nonexistent", else_="fallback"
+ ),
+ "fallback",
+ )
+
+ def test_get_dialect_option_nonparticipating(self):
+ with self._fixture():
+ idx = Index("a", "b", "c")
+ dialect = mock.Mock()
+ dialect.name = "nonparticipating"
+ eq_(idx.get_dialect_option(dialect, "x"), None)
+ eq_(
+ idx.get_dialect_option(dialect, "x", else_="fallback"),
+ "fallback",
+ )
+
class NamingConventionTest(fixtures.TestBase, AssertsCompiledSQL):
__dialect__ = "default"
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
