replace Variant with direct feature inside of TypeEngine

The :meth:`_sqltypes.TypeEngine.with_variant` method now returns a copy of
the original :class:`_sqltypes.TypeEngine` object, rather than wrapping it
inside the ``Variant`` class, which is effectively removed (the import
symbol remains for backwards compatibility with code that may be testing
for this symbol). While the previous approach maintained in-Python
behaviors, maintaining the original type allows for clearer type checking
and debugging.

Fixes: #6980
Change-Id: I158c7e56306b886b5b82b040205c428a5c4a242c

4 files changed, 103 insertions, 109 deletions

diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index a9dd6a23a..90f32a4f7 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -512,6 +512,11 @@ class TypeCompiler(metaclass=util.EnsureKWArgType):
         self.dialect = dialect
 
     def process(self, type_, **kw):
+        if (
+            type_._variant_mapping
+            and self.dialect.name in type_._variant_mapping
+        ):
+            type_ = type_._variant_mapping[self.dialect.name]
         return type_._compiler_dispatch(self, **kw)
 
     def visit_unsupported_compilation(self, element, err, **kw):
diff --git a/lib/sqlalchemy/sql/schema.py b/lib/sqlalchemy/sql/schema.py
index 9b5005b5d..dd8238450 100644
--- a/lib/sqlalchemy/sql/schema.py
+++ b/lib/sqlalchemy/sql/schema.py
@@ -1586,9 +1586,13 @@ class Column(DialectKWArgs, SchemaItem, ColumnClause):
         # check if this Column is proxying another column
         if "_proxies" in kwargs:
             self._proxies = kwargs.pop("_proxies")
-        # otherwise, add DDL-related events
-        elif isinstance(self.type, SchemaEventTarget):
-            self.type._set_parent_with_dispatch(self)
+        else:
+            # otherwise, add DDL-related events
+            if isinstance(self.type, SchemaEventTarget):
+                self.type._set_parent_with_dispatch(self)
+            for impl in self.type._variant_mapping.values():
+                if isinstance(impl, SchemaEventTarget):
+                    impl._set_parent_with_dispatch(self)
 
         if self.default is not None:
             if isinstance(self.default, (ColumnDefault, Sequence)):
diff --git a/lib/sqlalchemy/sql/sqltypes.py b/lib/sqlalchemy/sql/sqltypes.py
index 574692fed..cda7b35cd 100644
--- a/lib/sqlalchemy/sql/sqltypes.py
+++ b/lib/sqlalchemy/sql/sqltypes.py
@@ -33,7 +33,7 @@ from .type_api import NativeForEmulated  # noqa
 from .type_api import to_instance
 from .type_api import TypeDecorator
 from .type_api import TypeEngine
-from .type_api import Variant
+from .type_api import Variant  # noqa
 from .. import event
 from .. import exc
 from .. import inspection
@@ -844,12 +844,25 @@ class SchemaType(SchemaEventTarget):
             )
 
     def _set_parent(self, column, **kw):
+        # set parent hook is when this type is associated with a column.
+        # Column calls it for all SchemaEventTarget instances, either the
+        # base type and/or variants in _variant_mapping.
+
+        # we want to register a second hook to trigger when that column is
+        # associated with a table.  in that event, we and all of our variants
+        # may want to set up some state on the table such as a CheckConstraint
+        # that will conditionally render at DDL render time.
+
+        # the base SchemaType also sets up events for
+        # on_table/metadata_create/drop in this method, which is used by
+        # "native" types with a separate CREATE/DROP e.g. Postgresql.ENUM
+
         column._on_table_attach(util.portable_instancemethod(self._set_table))
 
     def _variant_mapping_for_set_table(self, column):
-        if isinstance(column.type, Variant):
-            variant_mapping = column.type.mapping.copy()
-            variant_mapping["_default"] = column.type.impl
+        if column.type._variant_mapping:
+            variant_mapping = dict(column.type._variant_mapping)
+            variant_mapping["_default"] = column.type
         else:
             variant_mapping = None
         return variant_mapping
@@ -880,8 +893,9 @@ class SchemaType(SchemaEventTarget):
             ),
         )
         if self.metadata is None:
-            # TODO: what's the difference between self.metadata
-            # and table.metadata here ?
+            # if SchemaType were created w/ a metadata argument, these
+            # events would already have been associated with that metadata
+            # and would preclude an association with table.metadata
             event.listen(
                 table.metadata,
                 "before_create",
@@ -963,9 +977,19 @@ class SchemaType(SchemaEventTarget):
 
     def _is_impl_for_variant(self, dialect, kw):
         variant_mapping = kw.pop("variant_mapping", None)
-        if variant_mapping is None:
+
+        if not variant_mapping:
             return True
 
+        # for types that have _variant_mapping, all the impls in the map
+        # that are SchemaEventTarget subclasses get set up as event holders.
+        # this is so that constructs that need
+        # to be associated with the Table at dialect-agnostic time etc. like
+        # CheckConstraints can be set up with that table.  they then add
+        # to these constraints a DDL check_rule that among other things
+        # will check this _is_impl_for_variant() method to determine when
+        # the dialect is known that we are part of the table's DDL sequence.
+
         # since PostgreSQL is the only DB that has ARRAY this can only
         # be integration tested by PG-specific tests
         def _we_are_the_impl(typ):
diff --git a/lib/sqlalchemy/sql/type_api.py b/lib/sqlalchemy/sql/type_api.py
index cc226d7e3..07cd4d95f 100644
--- a/lib/sqlalchemy/sql/type_api.py
+++ b/lib/sqlalchemy/sql/type_api.py
@@ -9,6 +9,7 @@
 
 """
 
+import typing
 
 from . import operators
 from .base import SchemaEventTarget
@@ -29,6 +30,10 @@ TABLEVALUE = None
 _resolve_value_to_type = None
 
 
+# replace with pep-673 when applicable
+SelfTypeEngine = typing.TypeVar("SelfTypeEngine", bound="TypeEngine")
+
+
 class TypeEngine(Traversible):
     """The ultimate base class for all SQL datatypes.
 
@@ -192,6 +197,8 @@ class TypeEngine(Traversible):
 
     """
 
+    _variant_mapping = util.EMPTY_DICT
+
     def evaluates_none(self):
         """Return a copy of this type which has the :attr:`.should_evaluate_none`
         flag set to True.
@@ -532,8 +539,10 @@ class TypeEngine(Traversible):
         """
         raise NotImplementedError()
 
-    def with_variant(self, type_, dialect_name):
-        r"""Produce a new type object that will utilize the given
+    def with_variant(
+        self: SelfTypeEngine, type_: "TypeEngine", dialect_name: str
+    ) -> SelfTypeEngine:
+        r"""Produce a copy of this type object that will utilize the given
         type when applied to the dialect of the given name.
 
         e.g.::
@@ -541,15 +550,21 @@ class TypeEngine(Traversible):
             from sqlalchemy.types import String
             from sqlalchemy.dialects import mysql
 
-            s = String()
+            string_type = String()
+
+            string_type = string_type.with_variant(
+                mysql.VARCHAR(collation='foo'), 'mysql'
+            )
 
-            s = s.with_variant(mysql.VARCHAR(collation='foo'), 'mysql')
+        The variant mapping indicates that when this type is
+        interpreted by a specific dialect, it will instead be
+        transmuted into the given type, rather than using the
+        primary type.
 
-        The construction of :meth:`.TypeEngine.with_variant` is always
-        from the "fallback" type to that which is dialect specific.
-        The returned type is an instance of :class:`.Variant`, which
-        itself provides a :meth:`.Variant.with_variant`
-        that can be called repeatedly.
+        .. versionchanged:: 2.0 the :meth:`_types.TypeEngine.with_variant`
+           method now works with a :class:`_types.TypeEngine` object "in
+           place", returning a copy of the original type rather than returning
+           a wrapping object; the ``Variant`` class is no longer used.
 
         :param type\_: a :class:`.TypeEngine` that will be selected
          as a variant from the originating type, when a dialect
@@ -558,7 +573,24 @@ class TypeEngine(Traversible):
          this type. (i.e. ``'postgresql'``, ``'mysql'``, etc.)
 
         """
-        return Variant(self, {dialect_name: to_instance(type_)})
+
+        if dialect_name in self._variant_mapping:
+            raise exc.ArgumentError(
+                "Dialect '%s' is already present in "
+                "the mapping for this %r" % (dialect_name, self)
+            )
+        new_type = self.copy()
+        if isinstance(type_, type):
+            type_ = type_()
+        elif type_._variant_mapping:
+            raise exc.ArgumentError(
+                "can't pass a type that already has variants as a "
+                "dialect-level type to with_variant()"
+            )
+        new_type._variant_mapping = self._variant_mapping.union(
+            {dialect_name: type_}
+        )
+        return new_type
 
     @util.memoized_property
     def _type_affinity(self):
@@ -735,7 +767,12 @@ class TypeEngine(Traversible):
             return d
 
     def _gen_dialect_impl(self, dialect):
-        return dialect.type_descriptor(self)
+        if dialect.name in self._variant_mapping:
+            return self._variant_mapping[dialect.name]._gen_dialect_impl(
+                dialect
+            )
+        else:
+            return dialect.type_descriptor(self)
 
     @util.memoized_property
     def _static_cache_key(self):
@@ -1361,7 +1398,12 @@ class TypeDecorator(ExternalType, SchemaEventTarget, TypeEngine):
         """
         #todo
         """
-        adapted = dialect.type_descriptor(self)
+        if dialect.name in self._variant_mapping:
+            adapted = dialect.type_descriptor(
+                self._variant_mapping[dialect.name]
+            )
+        else:
+            adapted = dialect.type_descriptor(self)
         if adapted is not self:
             return adapted
 
@@ -1818,98 +1860,17 @@ class TypeDecorator(ExternalType, SchemaEventTarget, TypeEngine):
 
 
 class Variant(TypeDecorator):
-    """A wrapping type that selects among a variety of
-    implementations based on dialect in use.
-
-    The :class:`.Variant` type is typically constructed
-    using the :meth:`.TypeEngine.with_variant` method.
-
-    .. seealso:: :meth:`.TypeEngine.with_variant` for an example of use.
+    """deprecated.  symbol is present for backwards-compatibility with
+    workaround recipes, however this actual type should not be used.
 
     """
 
-    cache_ok = True
-
-    def __init__(self, base, mapping):
-        """Construct a new :class:`.Variant`.
-
-        :param base: the base 'fallback' type
-        :param mapping: dictionary of string dialect names to
-          :class:`.TypeEngine` instances.
-
-        """
-        self.impl = base
-        self.mapping = mapping
-
-    @util.memoized_property
-    def _static_cache_key(self):
-        # TODO: needs tests in test/sql/test_compare.py
-        return (self.__class__,) + (
-            self.impl._static_cache_key,
-            tuple(
-                (key, self.mapping[key]._static_cache_key)
-                for key in sorted(self.mapping)
-            ),
+    def __init__(self, *arg, **kw):
+        raise NotImplementedError(
+            "Variant is no longer used in SQLAlchemy; this is a "
+            "placeholder symbol for backwards compatibility."
         )
 
-    def coerce_compared_value(self, operator, value):
-        result = self.impl.coerce_compared_value(operator, value)
-        if result is self.impl:
-            return self
-        else:
-            return result
-
-    def load_dialect_impl(self, dialect):
-        if dialect.name in self.mapping:
-            return self.mapping[dialect.name]
-        else:
-            return self.impl
-
-    def _set_parent(self, column, outer=False, **kw):
-        """Support SchemaEventTarget"""
-
-        if isinstance(self.impl, SchemaEventTarget):
-            self.impl._set_parent(column, **kw)
-        for impl in self.mapping.values():
-            if isinstance(impl, SchemaEventTarget):
-                impl._set_parent(column, **kw)
-
-    def _set_parent_with_dispatch(self, parent):
-        """Support SchemaEventTarget"""
-
-        if isinstance(self.impl, SchemaEventTarget):
-            self.impl._set_parent_with_dispatch(parent)
-        for impl in self.mapping.values():
-            if isinstance(impl, SchemaEventTarget):
-                impl._set_parent_with_dispatch(parent)
-
-    def with_variant(self, type_, dialect_name):
-        r"""Return a new :class:`.Variant` which adds the given
-        type + dialect name to the mapping, in addition to the
-        mapping present in this :class:`.Variant`.
-
-        :param type\_: a :class:`.TypeEngine` that will be selected
-         as a variant from the originating type, when a dialect
-         of the given name is in use.
-        :param dialect_name: base name of the dialect which uses
-         this type. (i.e. ``'postgresql'``, ``'mysql'``, etc.)
-
-        """
-
-        if dialect_name in self.mapping:
-            raise exc.ArgumentError(
-                "Dialect '%s' is already present in "
-                "the mapping for this Variant" % dialect_name
-            )
-        mapping = self.mapping.copy()
-        mapping[dialect_name] = type_
-        return Variant(self.impl, mapping)
-
-    @property
-    def comparator_factory(self):
-        """express comparison behavior in terms of the base type"""
-        return self.impl.comparator_factory
-
 
 def _reconstitute_comparator(expression):
     return expression.comparator