Merge "Initial ORM typing layout" into main

28 files changed, 1048 insertions, 567 deletions

diff --git a/lib/sqlalchemy/engine/result.py b/lib/sqlalchemy/engine/result.py
index 15c9b1d95..3f916fea0 100644
--- a/lib/sqlalchemy/engine/result.py
+++ b/lib/sqlalchemy/engine/result.py
@@ -6,7 +6,6 @@
 # the MIT License: https://www.opensource.org/licenses/mit-license.php
 
 """Define generic result set constructs."""
-
 import collections.abc as collections_abc
 import functools
 import itertools
@@ -19,11 +18,13 @@ from .. import util
 from ..sql.base import _generative
 from ..sql.base import HasMemoized
 from ..sql.base import InPlaceGenerative
+from ..util._has_cy import HAS_CYEXTENSION
 
-try:
-    from sqlalchemy.cyextension.resultproxy import tuplegetter
-except ImportError:
+
+if typing.TYPE_CHECKING or not HAS_CYEXTENSION:
     from ._py_row import tuplegetter
+else:
+    from sqlalchemy.cyextension.resultproxy import tuplegetter
 
 
 class ResultMetaData:
diff --git a/lib/sqlalchemy/engine/row.py b/lib/sqlalchemy/engine/row.py
index 39c69c2ff..16215ccc4 100644
--- a/lib/sqlalchemy/engine/row.py
+++ b/lib/sqlalchemy/engine/row.py
@@ -7,21 +7,21 @@
 
 """Define row constructs including :class:`.Row`."""
 
-
 import collections.abc as collections_abc
 import operator
+import typing
 
 from ..sql import util as sql_util
+from ..util._has_cy import HAS_CYEXTENSION
 
-
-try:
-    from sqlalchemy.cyextension.resultproxy import BaseRow
-    from sqlalchemy.cyextension.resultproxy import KEY_INTEGER_ONLY
-    from sqlalchemy.cyextension.resultproxy import KEY_OBJECTS_ONLY
-except ImportError:
+if typing.TYPE_CHECKING or not HAS_CYEXTENSION:
     from ._py_row import BaseRow
     from ._py_row import KEY_INTEGER_ONLY
     from ._py_row import KEY_OBJECTS_ONLY
+else:
+    from sqlalchemy.cyextension.resultproxy import BaseRow
+    from sqlalchemy.cyextension.resultproxy import KEY_INTEGER_ONLY
+    from sqlalchemy.cyextension.resultproxy import KEY_OBJECTS_ONLY
 
 
 class Row(BaseRow, collections_abc.Sequence):
diff --git a/lib/sqlalchemy/engine/url.py b/lib/sqlalchemy/engine/url.py
index 9157ff008..ec5ab2bec 100644
--- a/lib/sqlalchemy/engine/url.py
+++ b/lib/sqlalchemy/engine/url.py
@@ -16,6 +16,11 @@ be used directly and is also accepted directly by ``create_engine()``.
 
 import collections.abc as collections_abc
 import re
+from typing import Dict
+from typing import NamedTuple
+from typing import Optional
+from typing import Tuple
+from typing import Union
 from urllib.parse import parse_qsl
 from urllib.parse import quote_plus
 from urllib.parse import unquote
@@ -27,20 +32,7 @@ from ..dialects import plugins
 from ..dialects import registry
 
 
-class URL(
-    util.namedtuple(
-        "URL",
-        [
-            "drivername",
-            "username",
-            "password",
-            "host",
-            "port",
-            "database",
-            "query",
-        ],
-    )
-):
+class URL(NamedTuple):
     """
     Represent the components of a URL used to connect to a database.
 
@@ -86,17 +78,13 @@ class URL(
 
     """
 
-    def __new__(self, *arg, **kw):
-        if kw.pop("_new_ok", False):
-            return super(URL, self).__new__(self, *arg, **kw)
-        else:
-            util.warn_deprecated(
-                "Calling URL() directly is deprecated and will be disabled "
-                "in a future release.  The public constructor for URL is "
-                "now the URL.create() method.",
-                "1.4",
-            )
-            return URL.create(*arg, **kw)
+    drivername: str
+    username: Optional[str]
+    password: Optional[str]
+    host: Optional[str]
+    port: Optional[int]
+    database: Optional[str]
+    query: Dict[str, Union[str, Tuple[str]]]
 
     @classmethod
     def create(
@@ -153,7 +141,6 @@ class URL(
             cls._assert_port(port),
             cls._assert_none_str(database, "database"),
             cls._str_dict(query),
-            _new_ok=True,
         )
 
     @classmethod
@@ -264,10 +251,10 @@ class URL(
         if query is not None:
             kw["query"] = query
 
-        return self._replace(**kw)
+        return self._assert_replace(**kw)
 
-    def _replace(self, **kw):
-        """Override ``namedtuple._replace()`` to provide argument checking."""
+    def _assert_replace(self, **kw):
+        """argument checks before calling _replace()"""
 
         if "drivername" in kw:
             self._assert_str(kw["drivername"], "drivername")
@@ -279,7 +266,7 @@ class URL(
         if "query" in kw:
             kw["query"] = self._str_dict(kw["query"])
 
-        return super(URL, self)._replace(**kw)
+        return self._replace(**kw)
 
     def update_query_string(self, query_string, append=False):
         """Return a new :class:`_engine.URL` object with the :attr:`_engine.URL.query`
@@ -467,7 +454,6 @@ class URL(
                     for key in set(self.query).difference(names)
                 }
             ),
-            _new_ok=True,
         )
 
     @util.memoized_property
diff --git a/lib/sqlalchemy/ext/hybrid.py b/lib/sqlalchemy/ext/hybrid.py
index 52817e838..c7d9d4f88 100644
--- a/lib/sqlalchemy/ext/hybrid.py
+++ b/lib/sqlalchemy/ext/hybrid.py
@@ -802,10 +802,15 @@ advanced and/or patient developers, there's probably a whole lot of amazing
 things it can be used for.
 
 """  # noqa
+from typing import Any
+from typing import TypeVar
+
 from .. import util
 from ..orm import attributes
 from ..orm import interfaces
 
+_T = TypeVar("_T", bound=Any)
+
 HYBRID_METHOD = util.symbol("HYBRID_METHOD")
 """Symbol indicating an :class:`InspectionAttr` that's
    of type :class:`.hybrid_method`.
@@ -1147,7 +1152,7 @@ class hybrid_property(interfaces.InspectionAttrInfo):
         return expr_comparator
 
 
-class Comparator(interfaces.PropComparator):
+class Comparator(interfaces.PropComparator[_T]):
     """A helper class that allows easy construction of custom
     :class:`~.orm.interfaces.PropComparator`
     classes for usage with hybrids."""
@@ -1168,7 +1173,7 @@ class Comparator(interfaces.PropComparator):
         return self
 
 
-class ExprComparator(Comparator):
+class ExprComparator(Comparator[_T]):
     def __init__(self, cls, expression, hybrid):
         self.cls = cls
         self.expression = expression
diff --git a/lib/sqlalchemy/ext/mypy/names.py b/lib/sqlalchemy/ext/mypy/names.py
index 8ec15a6d4..b6f911979 100644
--- a/lib/sqlalchemy/ext/mypy/names.py
+++ b/lib/sqlalchemy/ext/mypy/names.py
@@ -104,7 +104,7 @@ _lookup: Dict[str, Tuple[int, Set[str]]] = {
         },
     ),
     "TypeEngine": (TYPEENGINE, {"sqlalchemy.sql.type_api.TypeEngine"}),
-    "Mapped": (MAPPED, {"sqlalchemy.orm.attributes.Mapped"}),
+    "Mapped": (MAPPED, {NAMED_TYPE_SQLA_MAPPED}),
     "declarative_base": (
         DECLARATIVE_BASE,
         {
diff --git a/lib/sqlalchemy/ext/mypy/plugin.py b/lib/sqlalchemy/ext/mypy/plugin.py
index 8687012a1..0a21feb51 100644
--- a/lib/sqlalchemy/ext/mypy/plugin.py
+++ b/lib/sqlalchemy/ext/mypy/plugin.py
@@ -112,6 +112,8 @@ class SQLAlchemyPlugin(Plugin):
         self, file: MypyFile
     ) -> List[Tuple[int, str, int]]:
         return [
+            #
+            (10, "sqlalchemy.orm", -1),
             (10, "sqlalchemy.orm.attributes", -1),
             (10, "sqlalchemy.orm.decl_api", -1),
         ]
@@ -270,7 +272,7 @@ def _add_globals(ctx: Union[ClassDefContext, DynamicClassDefContext]) -> None:
 
     """
 
-    util.add_global(ctx, "sqlalchemy.orm.attributes", "Mapped", "__sa_Mapped")
+    util.add_global(ctx, "sqlalchemy.orm", "Mapped", "__sa_Mapped")
 
 
 def _set_declarative_metaclass(
diff --git a/lib/sqlalchemy/orm/__init__.py b/lib/sqlalchemy/orm/__init__.py
index 17167a7de..55f2f3100 100644
--- a/lib/sqlalchemy/orm/__init__.py
+++ b/lib/sqlalchemy/orm/__init__.py
@@ -25,18 +25,22 @@ from ._orm_constructors import contains_alias as contains_alias
 from ._orm_constructors import create_session as create_session
 from ._orm_constructors import deferred as deferred
 from ._orm_constructors import dynamic_loader as dynamic_loader
+from ._orm_constructors import mapped_column as mapped_column
 from ._orm_constructors import query_expression as query_expression
 from ._orm_constructors import relationship as relationship
 from ._orm_constructors import synonym as synonym
 from ._orm_constructors import with_loader_criteria as with_loader_criteria
 from .attributes import AttributeEvent as AttributeEvent
 from .attributes import InstrumentedAttribute as InstrumentedAttribute
-from .attributes import Mapped as Mapped
 from .attributes import QueryableAttribute as QueryableAttribute
+from .base import Mapped as Mapped
 from .context import QueryContext as QueryContext
+from .decl_api import add_mapped_attribute as add_mapped_attribute
 from .decl_api import as_declarative as as_declarative
 from .decl_api import declarative_base as declarative_base
 from .decl_api import declarative_mixin as declarative_mixin
+from .decl_api import DeclarativeBase as DeclarativeBase
+from .decl_api import DeclarativeBaseNoMeta as DeclarativeBaseNoMeta
 from .decl_api import DeclarativeMeta as DeclarativeMeta
 from .decl_api import declared_attr as declared_attr
 from .decl_api import has_inherited_table as has_inherited_table
diff --git a/lib/sqlalchemy/orm/_orm_constructors.py b/lib/sqlalchemy/orm/_orm_constructors.py
index be0d23d00..80607670e 100644
--- a/lib/sqlalchemy/orm/_orm_constructors.py
+++ b/lib/sqlalchemy/orm/_orm_constructors.py
@@ -6,11 +6,16 @@
 # the MIT License: https://www.opensource.org/licenses/mit-license.php
 
 import typing
+from typing import Any
 from typing import Callable
+from typing import Collection
+from typing import Optional
+from typing import overload
 from typing import Type
 from typing import Union
 
 from . import mapper as mapperlib
+from .base import Mapped
 from .descriptor_props import CompositeProperty
 from .descriptor_props import SynonymProperty
 from .properties import ColumnProperty
@@ -21,6 +26,11 @@ from .util import LoaderCriteriaOption
 from .. import sql
 from .. import util
 from ..exc import InvalidRequestError
+from ..sql.schema import Column
+from ..sql.schema import SchemaEventTarget
+from ..sql.type_api import TypeEngine
+from ..util.typing import Literal
+
 
 _RC = typing.TypeVar("_RC")
 _T = typing.TypeVar("_T")
@@ -41,9 +51,138 @@ def contains_alias(alias) -> "AliasOption":
     return AliasOption(alias)
 
 
+@overload
+def mapped_column(
+    *args: SchemaEventTarget,
+    nullable: bool = ...,
+    primary_key: bool = ...,
+    **kw: Any,
+) -> "Mapped":
+    ...
+
+
+@overload
+def mapped_column(
+    __type: Union[Type["TypeEngine[_T]"], "TypeEngine[_T]"],
+    *args: SchemaEventTarget,
+    nullable: Union[Literal[None], Literal[True]] = ...,
+    primary_key: Union[Literal[None], Literal[False]] = ...,
+    **kw: Any,
+) -> "Mapped[Optional[_T]]":
+    ...
+
+
+@overload
+def mapped_column(
+    __type: Union[Type["TypeEngine[_T]"], "TypeEngine[_T]"],
+    *args: SchemaEventTarget,
+    nullable: Union[Literal[None], Literal[True]] = ...,
+    primary_key: Union[Literal[None], Literal[False]] = ...,
+    **kw: Any,
+) -> "Mapped[Optional[_T]]":
+    ...
+
+
+@overload
+def mapped_column(
+    __type: Union[Type["TypeEngine[_T]"], "TypeEngine[_T]"],
+    *args: SchemaEventTarget,
+    nullable: Union[Literal[None], Literal[False]] = ...,
+    primary_key: Literal[True] = True,
+    **kw: Any,
+) -> "Mapped[_T]":
+    ...
+
+
+@overload
+def mapped_column(
+    __type: Union[Type["TypeEngine[_T]"], "TypeEngine[_T]"],
+    *args: SchemaEventTarget,
+    nullable: Literal[False] = ...,
+    primary_key: bool = ...,
+    **kw: Any,
+) -> "Mapped[_T]":
+    ...
+
+
+@overload
+def mapped_column(
+    __name: str,
+    __type: Union[Type["TypeEngine[_T]"], "TypeEngine[_T]"],
+    *args: SchemaEventTarget,
+    nullable: Union[Literal[None], Literal[True]] = ...,
+    primary_key: Union[Literal[None], Literal[False]] = ...,
+    **kw: Any,
+) -> "Mapped[Optional[_T]]":
+    ...
+
+
+@overload
+def mapped_column(
+    __name: str,
+    __type: Union[Type["TypeEngine[_T]"], "TypeEngine[_T]"],
+    *args: SchemaEventTarget,
+    nullable: Union[Literal[None], Literal[True]] = ...,
+    primary_key: Union[Literal[None], Literal[False]] = ...,
+    **kw: Any,
+) -> "Mapped[Optional[_T]]":
+    ...
+
+
+@overload
+def mapped_column(
+    __name: str,
+    __type: Union[Type["TypeEngine[_T]"], "TypeEngine[_T]"],
+    *args: SchemaEventTarget,
+    nullable: Union[Literal[None], Literal[False]] = ...,
+    primary_key: Literal[True] = True,
+    **kw: Any,
+) -> "Mapped[_T]":
+    ...
+
+
+@overload
+def mapped_column(
+    __name: str,
+    __type: Union[Type["TypeEngine[_T]"], "TypeEngine[_T]"],
+    *args: SchemaEventTarget,
+    nullable: Literal[False] = ...,
+    primary_key: bool = ...,
+    **kw: Any,
+) -> "Mapped[_T]":
+    ...
+
+
+def mapped_column(*args, **kw) -> "Mapped":
+    """construct a new ORM-mapped :class:`_schema.Column` construct.
+
+    The :func:`_orm.mapped_column` function is shorthand for the construction
+    of a Core :class:`_schema.Column` object delivered within a
+    :func:`_orm.column_property` construct, which provides for consistent
+    typing information to be delivered to the class so that it works under
+    static type checkers such as mypy and delivers useful information in
+    IDE related type checkers such as pylance.   The function can be used
+    in declarative mappings anywhere that :class:`_schema.Column` is normally
+    used::
+
+        from sqlalchemy.orm import mapped_column
+
+        class User(Base):
+            __tablename__ = 'user'
+
+            id = mapped_column(Integer)
+            name = mapped_column(String)
+
+
+    .. versionadded:: 2.0
+
+    """
+    return column_property(Column(*args, **kw))
+
+
 def column_property(
     column: sql.ColumnElement[_T], *additional_columns, **kwargs
-) -> "ColumnProperty[_T]":
+) -> "Mapped[_T]":
     r"""Provide a column-level property for use with a mapping.
 
     Column-based properties can normally be applied to the mapper's
@@ -130,7 +269,7 @@ def column_property(
     return ColumnProperty(column, *additional_columns, **kwargs)
 
 
-def composite(class_: Type[_T], *attrs, **kwargs) -> "CompositeProperty[_T]":
+def composite(class_: Type[_T], *attrs, **kwargs) -> "Mapped[_T]":
     r"""Return a composite column-based property for use with a Mapper.
 
     See the mapping documentation section :ref:`mapper_composite` for a
@@ -359,13 +498,106 @@ def with_loader_criteria(
     )
 
 
+@overload
+def relationship(
+    argument: Union[str, Type[_RC], Callable[[], Type[_RC]]],
+    secondary=None,
+    *,
+    uselist: Literal[True] = None,
+    primaryjoin=None,
+    secondaryjoin=None,
+    foreign_keys=None,
+    order_by=False,
+    backref=None,
+    back_populates=None,
+    overlaps=None,
+    post_update=False,
+    cascade=False,
+    viewonly=False,
+    lazy="select",
+    collection_class=None,
+    passive_deletes=RelationshipProperty._persistence_only["passive_deletes"],
+    passive_updates=RelationshipProperty._persistence_only["passive_updates"],
+    remote_side=None,
+    enable_typechecks=RelationshipProperty._persistence_only[
+        "enable_typechecks"
+    ],
+    join_depth=None,
+    comparator_factory=None,
+    single_parent=False,
+    innerjoin=False,
+    distinct_target_key=None,
+    doc=None,
+    active_history=RelationshipProperty._persistence_only["active_history"],
+    cascade_backrefs=RelationshipProperty._persistence_only[
+        "cascade_backrefs"
+    ],
+    load_on_pending=False,
+    bake_queries=True,
+    _local_remote_pairs=None,
+    query_class=None,
+    info=None,
+    omit_join=None,
+    sync_backref=None,
+    _legacy_inactive_history_style=False,
+) -> Mapped[Collection[_RC]]:
+    ...
+
+
+@overload
+def relationship(
+    argument: Union[str, Type[_RC], Callable[[], Type[_RC]]],
+    secondary=None,
+    *,
+    uselist: Optional[bool] = None,
+    primaryjoin=None,
+    secondaryjoin=None,
+    foreign_keys=None,
+    order_by=False,
+    backref=None,
+    back_populates=None,
+    overlaps=None,
+    post_update=False,
+    cascade=False,
+    viewonly=False,
+    lazy="select",
+    collection_class=None,
+    passive_deletes=RelationshipProperty._persistence_only["passive_deletes"],
+    passive_updates=RelationshipProperty._persistence_only["passive_updates"],
+    remote_side=None,
+    enable_typechecks=RelationshipProperty._persistence_only[
+        "enable_typechecks"
+    ],
+    join_depth=None,
+    comparator_factory=None,
+    single_parent=False,
+    innerjoin=False,
+    distinct_target_key=None,
+    doc=None,
+    active_history=RelationshipProperty._persistence_only["active_history"],
+    cascade_backrefs=RelationshipProperty._persistence_only[
+        "cascade_backrefs"
+    ],
+    load_on_pending=False,
+    bake_queries=True,
+    _local_remote_pairs=None,
+    query_class=None,
+    info=None,
+    omit_join=None,
+    sync_backref=None,
+    _legacy_inactive_history_style=False,
+) -> Mapped[_RC]:
+    ...
+
+
 def relationship(
     argument: Union[str, Type[_RC], Callable[[], Type[_RC]]],
     secondary=None,
+    *,
     primaryjoin=None,
     secondaryjoin=None,
     foreign_keys=None,
-    uselist=None,
+    uselist: Optional[bool] = None,
     order_by=False,
     backref=None,
     back_populates=None,
@@ -399,7 +631,7 @@ def relationship(
     omit_join=None,
     sync_backref=None,
     _legacy_inactive_history_style=False,
-) -> RelationshipProperty[_RC]:
+) -> Mapped[_RC]:
     """Provide a relationship between two mapped classes.
 
     This corresponds to a parent-child or associative table relationship.
@@ -1261,7 +1493,7 @@ def synonym(
     comparator_factory=None,
     doc=None,
     info=None,
-) -> "SynonymProperty":
+) -> "Mapped":
     """Denote an attribute name as a synonym to a mapped property,
     in that the attribute will mirror the value and expression behavior
     of another attribute.
diff --git a/lib/sqlalchemy/orm/attributes.py b/lib/sqlalchemy/orm/attributes.py
index d24250ea0..5a605b7c6 100644
--- a/lib/sqlalchemy/orm/attributes.py
+++ b/lib/sqlalchemy/orm/attributes.py
@@ -13,10 +13,14 @@ defines a large part of the ORM's interactivity.
 
 
 """
-
+from collections import namedtuple
 import operator
-from typing import Generic
+from typing import Any
+from typing import List
+from typing import NamedTuple
+from typing import Tuple
 from typing import TypeVar
+from typing import Union
 
 from . import collections
 from . import exc as orm_exc
@@ -57,6 +61,8 @@ from ..sql import roles
 from ..sql import traversals
 from ..sql import visitors
 
+_T = TypeVar("_T")
+
 
 class NoKey(str):
     pass
@@ -67,9 +73,9 @@ NO_KEY = NoKey("no name")
 
 @inspection._self_inspects
 class QueryableAttribute(
-    interfaces._MappedAttribute,
+    interfaces._MappedAttribute[_T],
     interfaces.InspectionAttr,
-    interfaces.PropComparator,
+    interfaces.PropComparator[_T],
     traversals.HasCopyInternals,
     roles.JoinTargetRole,
     roles.OnClauseRole,
@@ -362,80 +368,7 @@ def _queryable_attribute_unreduce(key, mapped_class, parententity, entity):
         return getattr(entity, key)
 
 
-_T = TypeVar("_T")
-_Generic_T = Generic[_T]
-
-
-class Mapped(QueryableAttribute, _Generic_T):
-    """Represent an ORM mapped :term:`descriptor` attribute for typing purposes.
-
-    This class represents the complete descriptor interface for any class
-    attribute that will have been :term:`instrumented` by the ORM
-    :class:`_orm.Mapper` class. When used with typing stubs, it is the final
-    type that would be used by a type checker such as mypy to provide the full
-    behavioral contract for the attribute.
-
-    .. tip::
-
-        The :class:`_orm.Mapped` class represents attributes that are handled
-        directly by the :class:`_orm.Mapper` class. It does not include other
-        Python descriptor classes that are provided as extensions, including
-        :ref:`hybrids_toplevel` and the :ref:`associationproxy_toplevel`.
-        While these systems still make use of ORM-specific superclasses
-        and structures, they are not :term:`instrumented` by the
-        :class:`_orm.Mapper` and instead provide their own functionality
-        when they are accessed on a class.
-
-    When using the :ref:`SQLAlchemy Mypy plugin <mypy_toplevel>`, the
-    :class:`_orm.Mapped` construct is used in typing annotations to indicate to
-    the plugin those attributes that are expected to be mapped; the plugin also
-    applies :class:`_orm.Mapped` as an annotation automatically when it scans
-    through declarative mappings in :ref:`orm_declarative_table` style. For
-    more indirect mapping styles such as
-    :ref:`imperative table <orm_imperative_table_configuration>` it is
-    typically applied explicitly to class level attributes that expect
-    to be mapped based on a given :class:`_schema.Table` configuration.
-
-    :class:`_orm.Mapped` is defined in the
-    `sqlalchemy2-stubs <https://pypi.org/project/sqlalchemy2-stubs>`_ project
-    as a :pep:`484` generic class which may subscribe to any arbitrary Python
-    type, which represents the Python type handled by the attribute::
-
-        class MyMappedClass(Base):
-            __table_ = Table(
-                "some_table", Base.metadata,
-                Column("id", Integer, primary_key=True),
-                Column("data", String(50)),
-                Column("created_at", DateTime)
-            )
-
-            id : Mapped[int]
-            data: Mapped[str]
-            created_at: Mapped[datetime]
-
-    For complete background on how to use :class:`_orm.Mapped` with
-    pep-484 tools like Mypy, see the link below for background on SQLAlchemy's
-    Mypy plugin.
-
-    .. versionadded:: 1.4
-
-    .. seealso::
-
-        :ref:`mypy_toplevel` - complete background on Mypy integration
-
-    """
-
-    def __get__(self, instance, owner):
-        raise NotImplementedError()
-
-    def __set__(self, instance, value):
-        raise NotImplementedError()
-
-    def __delete__(self, instance):
-        raise NotImplementedError()
-
-
-class InstrumentedAttribute(Mapped):
+class InstrumentedAttribute(QueryableAttribute[_T]):
     """Class bound instrumented attribute which adds basic
     :term:`descriptor` methods.
 
@@ -469,9 +402,7 @@ class InstrumentedAttribute(Mapped):
             return self.impl.get(state, dict_)
 
 
-HasEntityNamespace = util.namedtuple(
-    "HasEntityNamespace", ["entity_namespace"]
-)
+HasEntityNamespace = namedtuple("HasEntityNamespace", ["entity_namespace"])
 HasEntityNamespace.is_mapper = HasEntityNamespace.is_aliased_class = False
 
 
@@ -1837,7 +1768,7 @@ _NO_HISTORY = util.symbol("NO_HISTORY")
 _NO_STATE_SYMBOLS = frozenset([id(PASSIVE_NO_RESULT), id(NO_VALUE)])
 
 
-class History(util.namedtuple("History", ["added", "unchanged", "deleted"])):
+class History(NamedTuple):
     """A 3-tuple of added, unchanged and deleted values,
     representing the changes which have occurred on an instrumented
     attribute.
@@ -1862,11 +1793,13 @@ class History(util.namedtuple("History", ["added", "unchanged", "deleted"])):
 
     """
 
+    added: Union[Tuple[()], List[Any]]
+    unchanged: Union[Tuple[()], List[Any]]
+    deleted: Union[Tuple[()], List[Any]]
+
     def __bool__(self):
         return self != HISTORY_BLANK
 
-    __nonzero__ = __bool__
-
     def empty(self):
         """Return True if this :class:`.History` has no changes
         and no existing, unchanged state.
@@ -2012,7 +1945,7 @@ class History(util.namedtuple("History", ["added", "unchanged", "deleted"])):
             )
 
 
-HISTORY_BLANK = History(None, None, None)
+HISTORY_BLANK = History((), (), ())
 
 
 def get_history(obj, key, passive=PASSIVE_OFF):
diff --git a/lib/sqlalchemy/orm/base.py b/lib/sqlalchemy/orm/base.py
index 93e2d609a..7ab4b7737 100644
--- a/lib/sqlalchemy/orm/base.py
+++ b/lib/sqlalchemy/orm/base.py
@@ -13,17 +13,25 @@ import operator
 import typing
 from typing import Any
 from typing import Generic
+from typing import overload
 from typing import TypeVar
+from typing import Union
 
 from . import exc
 from .. import exc as sa_exc
 from .. import inspection
 from .. import util
+from ..sql.elements import SQLCoreOperations
 from ..util import typing as compat_typing
+from ..util.langhelpers import TypingOnly
 
 
+if typing.TYPE_CHECKING:
+    from .attributes import InstrumentedAttribute
+
 _T = TypeVar("_T", bound=Any)
 
+
 PASSIVE_NO_RESULT = util.symbol(
     "PASSIVE_NO_RESULT",
     """Symbol returned by a loader callable or other attribute/history
@@ -579,7 +587,88 @@ class InspectionAttrInfo(InspectionAttr):
         return {}
 
 
-class _MappedAttribute(Generic[_T]):
+class SQLORMOperations(SQLCoreOperations[_T], TypingOnly):
+    __slots__ = ()
+
+    if typing.TYPE_CHECKING:
+
+        def of_type(self, class_):
+            ...
+
+        def and_(self, *criteria):
+            ...
+
+        def any(self, criterion=None, **kwargs):  # noqa A001
+            ...
+
+        def has(self, criterion=None, **kwargs):
+            ...
+
+
+class Mapped(Generic[_T], util.TypingOnly):
+    """Represent an ORM mapped attribute for typing purposes.
+
+    This class represents the complete descriptor interface for any class
+    attribute that will have been :term:`instrumented` by the ORM
+    :class:`_orm.Mapper` class.   Provides appropriate information to type
+    checkers such as pylance and mypy so that ORM-mapped attributes
+    are correctly typed.
+
+    .. tip::
+
+        The :class:`_orm.Mapped` class represents attributes that are handled
+        directly by the :class:`_orm.Mapper` class. It does not include other
+        Python descriptor classes that are provided as extensions, including
+        :ref:`hybrids_toplevel` and the :ref:`associationproxy_toplevel`.
+        While these systems still make use of ORM-specific superclasses
+        and structures, they are not :term:`instrumented` by the
+        :class:`_orm.Mapper` and instead provide their own functionality
+        when they are accessed on a class.
+
+    .. versionadded:: 1.4
+
+
+    """
+
+    __slots__ = ()
+
+    if typing.TYPE_CHECKING:
+
+        @overload
+        def __get__(
+            self, instance: None, owner: Any
+        ) -> "InstrumentedAttribute[_T]":
+            ...
+
+        @overload
+        def __get__(self, instance: object, owner: Any) -> _T:
+            ...
+
+        def __get__(
+            self, instance: object, owner: Any
+        ) -> Union["InstrumentedAttribute[_T]", _T]:
+            ...
+
+        @classmethod
+        def _empty_constructor(cls, arg1: Any) -> "SQLORMOperations[_T]":
+            ...
+
+        @overload
+        def __set__(self, instance: Any, value: _T) -> None:
+            ...
+
+        @overload
+        def __set__(self, instance: Any, value: SQLCoreOperations) -> None:
+            ...
+
+        def __set__(self, instance, value):
+            ...
+
+        def __delete__(self, instance: Any):
+            ...
+
+
+class _MappedAttribute(Mapped[_T], TypingOnly):
     """Mixin for attributes which should be replaced by mapper-assigned
     attributes.
 
diff --git a/lib/sqlalchemy/orm/decl_api.py b/lib/sqlalchemy/orm/decl_api.py
index 99b2e9b6f..00c5574fa 100644
--- a/lib/sqlalchemy/orm/decl_api.py
+++ b/lib/sqlalchemy/orm/decl_api.py
@@ -7,6 +7,13 @@
 """Public API functions and helpers for declarative."""
 import itertools
 import re
+import typing
+from typing import Any
+from typing import Callable
+from typing import ClassVar
+from typing import Optional
+from typing import TypeVar
+from typing import Union
 import weakref
 
 from . import attributes
@@ -15,7 +22,9 @@ from . import exc as orm_exc
 from . import instrumentation
 from . import interfaces
 from . import mapperlib
+from .attributes import InstrumentedAttribute
 from .base import _inspect_mapped_class
+from .base import Mapped
 from .decl_base import _add_attribute
 from .decl_base import _as_declarative
 from .decl_base import _declarative_constructor
@@ -23,13 +32,18 @@ from .decl_base import _DeferredMapperConfig
 from .decl_base import _del_attribute
 from .decl_base import _mapper
 from .descriptor_props import SynonymProperty as _orm_synonym
+from .mapper import Mapper
 from .. import exc
 from .. import inspection
 from .. import util
+from ..sql.elements import SQLCoreOperations
 from ..sql.schema import MetaData
+from ..sql.selectable import FromClause
 from ..util import hybridmethod
 from ..util import hybridproperty
 
+_T = TypeVar("_T", bound=Any)
+
 
 def has_inherited_table(cls):
     """Given a class, return True if any of the classes it inherits from has a
@@ -50,11 +64,21 @@ def has_inherited_table(cls):
     return False
 
 
-class DeclarativeMeta(type):
-    # DeclarativeMeta could be replaced by __subclass_init__()
-    # except for the class-level __setattr__() and __delattr__ hooks,
-    # which are still very important.
+class DeclarativeAttributeIntercept(type):
+    """Metaclass that may be used in conjunction with the
+    :class:`_orm.DeclarativeBase` class to support addition of class
+    attributes dynamically.
+
+    """
+
+    def __setattr__(cls, key, value):
+        _add_attribute(cls, key, value)
+
+    def __delattr__(cls, key):
+        _del_attribute(cls, key)
 
+
+class DeclarativeMeta(type):
     def __init__(cls, classname, bases, dict_, **kw):
         # early-consume registry from the initial declarative base,
         # assign privately to not conflict with subclass attributes named
@@ -121,7 +145,7 @@ def synonym_for(name, map_column=False):
     return decorate
 
 
-class declared_attr(interfaces._MappedAttribute, property):
+class declared_attr(interfaces._MappedAttribute[_T]):
     """Mark a class-level method as representing the definition of
     a mapped property or special declarative member name.
 
@@ -204,39 +228,52 @@ class declared_attr(interfaces._MappedAttribute, property):
 
     """  # noqa E501
 
-    def __init__(self, fget, cascading=False):
-        super(declared_attr, self).__init__(fget)
-        self.__doc__ = fget.__doc__
+    if typing.TYPE_CHECKING:
+
+        def __set__(self, instance, value):
+            ...
+
+        def __delete__(self, instance: Any):
+            ...
+
+    def __init__(
+        self,
+        fn: Callable[..., Union[Mapped[_T], SQLCoreOperations[_T]]],
+        cascading=False,
+    ):
+        self.fget = fn
         self._cascading = cascading
+        self.__doc__ = fn.__doc__
 
-    def __get__(desc, self, cls):
+    def __get__(self, instance, owner) -> InstrumentedAttribute[_T]:
         # the declared_attr needs to make use of a cache that exists
         # for the span of the declarative scan_attributes() phase.
         # to achieve this we look at the class manager that's configured.
+        cls = owner
         manager = attributes.manager_of_class(cls)
         if manager is None:
-            if not re.match(r"^__.+__$", desc.fget.__name__):
+            if not re.match(r"^__.+__$", self.fget.__name__):
                 # if there is no manager at all, then this class hasn't been
                 # run through declarative or mapper() at all, emit a warning.
                 util.warn(
                     "Unmanaged access of declarative attribute %s from "
-                    "non-mapped class %s" % (desc.fget.__name__, cls.__name__)
+                    "non-mapped class %s" % (self.fget.__name__, cls.__name__)
                 )
-            return desc.fget(cls)
+            return self.fget(cls)
         elif manager.is_mapped:
             # the class is mapped, which means we're outside of the declarative
             # scan setup, just run the function.
-            return desc.fget(cls)
+            return self.fget(cls)
 
         # here, we are inside of the declarative scan.  use the registry
         # that is tracking the values of these attributes.
         declarative_scan = manager.declarative_scan
         reg = declarative_scan.declared_attr_reg
 
-        if desc in reg:
-            return reg[desc]
+        if self in reg:
+            return reg[self]
         else:
-            reg[desc] = obj = desc.fget(cls)
+            reg[self] = obj = self.fget(cls)
             return obj
 
     @hybridmethod
@@ -361,6 +398,115 @@ def declarative_mixin(cls):
     return cls
 
 
+def _setup_declarative_base(cls):
+    if "metadata" in cls.__dict__:
+        metadata = cls.metadata
+    else:
+        metadata = None
+
+    reg = cls.__dict__.get("registry", None)
+    if reg is not None:
+        if not isinstance(reg, registry):
+            raise exc.InvalidRequestError(
+                "Declarative base class has a 'registry' attribute that is "
+                "not an instance of sqlalchemy.orm.registry()"
+            )
+    else:
+        reg = registry(metadata=metadata)
+        cls.registry = reg
+
+    cls._sa_registry = reg
+
+    if "metadata" not in cls.__dict__:
+        cls.metadata = cls.registry.metadata
+
+
+class DeclarativeBaseNoMeta:
+    """Same as :class:`_orm.DeclarativeBase`, but does not use a metaclass
+    to intercept new attributes.
+
+    The :class:`_orm.DeclarativeBaseNoMeta` base may be used when use of
+    custom metaclasses is desirable.
+
+    .. versionadded:: 2.0
+
+
+    """
+
+    registry: ClassVar["registry"]
+    _sa_registry: ClassVar["registry"]
+    metadata: ClassVar[MetaData]
+    __mapper__: ClassVar[Mapper]
+    __table__: Optional[FromClause]
+
+    if typing.TYPE_CHECKING:
+
+        def __init__(self, **kw: Any):
+            ...
+
+    def __init_subclass__(cls) -> None:
+        if DeclarativeBaseNoMeta in cls.__bases__:
+            _setup_declarative_base(cls)
+        else:
+            cls._sa_registry.map_declaratively(cls)
+
+
+class DeclarativeBase(metaclass=DeclarativeAttributeIntercept):
+    """Base class used for declarative class definitions.
+
+    The :class:`_orm.DeclarativeBase` allows for the creation of new
+    declarative bases in such a way that is compatible with type checkers::
+
+
+        from sqlalchemy.orm import DeclarativeBase
+
+        class Base(DeclarativeBase):
+            pass
+
+
+    The above ``Base`` class is now usable as the base for new declarative
+    mappings.  The superclass makes use of the ``__init_subclass__()``
+    method to set up new classes and metaclasses aren't used.
+
+    .. versionadded:: 2.0
+
+    """
+
+    registry: ClassVar["registry"]
+    _sa_registry: ClassVar["registry"]
+    metadata: ClassVar[MetaData]
+    __mapper__: ClassVar[Mapper]
+    __table__: Optional[FromClause]
+
+    if typing.TYPE_CHECKING:
+
+        def __init__(self, **kw: Any):
+            ...
+
+    def __init_subclass__(cls) -> None:
+        if DeclarativeBase in cls.__bases__:
+            _setup_declarative_base(cls)
+        else:
+            cls._sa_registry.map_declaratively(cls)
+
+
+def add_mapped_attribute(target, key, attr):
+    """Add a new mapped attribute to an ORM mapped class.
+
+    E.g.::
+
+        add_mapped_attribute(User, "addresses", relationship(Address))
+
+    This may be used for ORM mappings that aren't using a declarative
+    metaclass that intercepts attribute set operations.
+
+    .. versionadded:: 2.0
+
+
+    """
+    _add_attribute(target, key, attr)
+
+
 def declarative_base(
     metadata=None,
     mapper=None,
@@ -369,7 +515,7 @@ def declarative_base(
     constructor=_declarative_constructor,
     class_registry=None,
     metaclass=DeclarativeMeta,
-):
+) -> Any:
     r"""Construct a base class for declarative class definitions.
 
     The new base class will be given a metaclass that produces
@@ -1010,7 +1156,9 @@ def as_declarative(**kw):
     ).as_declarative_base(**kw)
 
 
-@inspection._inspects(DeclarativeMeta)
+@inspection._inspects(
+    DeclarativeMeta, DeclarativeBase, DeclarativeAttributeIntercept
+)
 def _inspect_decl_meta(cls):
     mp = _inspect_mapped_class(cls)
     if mp is None:
diff --git a/lib/sqlalchemy/orm/descriptor_props.py b/lib/sqlalchemy/orm/descriptor_props.py
index 80fce86d0..5e67b64cd 100644
--- a/lib/sqlalchemy/orm/descriptor_props.py
+++ b/lib/sqlalchemy/orm/descriptor_props.py
@@ -28,6 +28,7 @@ from ..sql import expression
 from ..sql import operators
 
 _T = TypeVar("_T", bound=Any)
+_PT = TypeVar("_PT", bound=Any)
 
 
 class DescriptorProperty(MapperProperty[_T]):
@@ -362,7 +363,7 @@ class CompositeProperty(DescriptorProperty[_T]):
 
             return proc
 
-    class Comparator(PropComparator):
+    class Comparator(PropComparator[_PT]):
         """Produce boolean, comparison, and other operators for
         :class:`.CompositeProperty` attributes.
 
@@ -448,7 +449,7 @@ class CompositeProperty(DescriptorProperty[_T]):
         return str(self.parent.class_.__name__) + "." + self.key
 
 
-class ConcreteInheritedProperty(DescriptorProperty):
+class ConcreteInheritedProperty(DescriptorProperty[_T]):
     """A 'do nothing' :class:`.MapperProperty` that disables
     an attribute on a concrete subclass that is only present
     on the inherited mapper, not the concrete classes' mapper.
@@ -501,7 +502,7 @@ class ConcreteInheritedProperty(DescriptorProperty):
         self.descriptor = NoninheritedConcreteProp()
 
 
-class SynonymProperty(DescriptorProperty):
+class SynonymProperty(DescriptorProperty[_T]):
     def __init__(
         self,
         name,
diff --git a/lib/sqlalchemy/orm/interfaces.py b/lib/sqlalchemy/orm/interfaces.py
index df265db57..08189a1b7 100644
--- a/lib/sqlalchemy/orm/interfaces.py
+++ b/lib/sqlalchemy/orm/interfaces.py
@@ -17,7 +17,9 @@ are exposed when inspecting mappings.
 """
 
 import collections
+import typing
 from typing import Any
+from typing import cast
 from typing import TypeVar
 
 from . import exc as orm_exc
@@ -32,6 +34,7 @@ from .base import MANYTOMANY
 from .base import MANYTOONE
 from .base import NOT_EXTENSION
 from .base import ONETOMANY
+from .base import SQLORMOperations
 from .. import inspect
 from .. import inspection
 from .. import util
@@ -307,8 +310,10 @@ class MapperProperty(
 
 
 @inspection._self_inspects
-class PropComparator(operators.ColumnOperators):
-    r"""Defines SQL operators for :class:`.MapperProperty` objects.
+class PropComparator(
+    SQLORMOperations[_T], operators.ColumnOperators[SQLORMOperations]
+):
+    r"""Defines SQL operations for ORM mapped attributes.
 
     SQLAlchemy allows for operators to
     be redefined at both the Core and ORM level.  :class:`.PropComparator`
@@ -316,12 +321,6 @@ class PropComparator(operators.ColumnOperators):
     including those of :class:`.ColumnProperty`,
     :class:`.RelationshipProperty`, and :class:`.CompositeProperty`.
 
-    .. note:: With the advent of Hybrid properties introduced in SQLAlchemy
-       0.7, as well as Core-level operator redefinition in
-       SQLAlchemy 0.8, the use case for user-defined :class:`.PropComparator`
-       instances is extremely rare.  See :ref:`hybrids_toplevel` as well
-       as :ref:`types_operators`.
-
     User-defined subclasses of :class:`.PropComparator` may be created. The
     built-in Python comparison and math operator methods, such as
     :meth:`.operators.ColumnOperators.__eq__`,
@@ -463,18 +462,34 @@ class PropComparator(operators.ColumnOperators):
         return self.property.info
 
     @staticmethod
-    def any_op(a, b, **kwargs):
+    def _any_op(a, b, **kwargs):
         return a.any(b, **kwargs)
 
     @staticmethod
-    def has_op(a, b, **kwargs):
-        return a.has(b, **kwargs)
+    def _has_op(left, other, **kwargs):
+        return left.has(other, **kwargs)
 
     @staticmethod
-    def of_type_op(a, class_):
+    def _of_type_op(a, class_):
         return a.of_type(class_)
 
-    def of_type(self, class_):
+    any_op = cast(operators.OperatorType, _any_op)
+    has_op = cast(operators.OperatorType, _has_op)
+    of_type_op = cast(operators.OperatorType, _of_type_op)
+
+    if typing.TYPE_CHECKING:
+
+        def operate(
+            self, op: operators.OperatorType, *other: Any, **kwargs: Any
+        ) -> "SQLORMOperations":
+            ...
+
+        def reverse_operate(
+            self, op: operators.OperatorType, other: Any, **kwargs: Any
+        ) -> "SQLORMOperations":
+            ...
+
+    def of_type(self, class_) -> "SQLORMOperations[_T]":
         r"""Redefine this object in terms of a polymorphic subclass,
         :func:`_orm.with_polymorphic` construct, or :func:`_orm.aliased`
         construct.
@@ -500,7 +515,7 @@ class PropComparator(operators.ColumnOperators):
 
         return self.operate(PropComparator.of_type_op, class_)
 
-    def and_(self, *criteria):
+    def and_(self, *criteria) -> "SQLORMOperations[_T]":
         """Add additional criteria to the ON clause that's represented by this
         relationship attribute.
 
@@ -528,7 +543,7 @@ class PropComparator(operators.ColumnOperators):
         """
         return self.operate(operators.and_, *criteria)
 
-    def any(self, criterion=None, **kwargs):
+    def any(self, criterion=None, **kwargs) -> "SQLORMOperations[_T]":
         r"""Return true if this collection contains any member that meets the
         given criterion.
 
@@ -546,7 +561,7 @@ class PropComparator(operators.ColumnOperators):
 
         return self.operate(PropComparator.any_op, criterion, **kwargs)
 
-    def has(self, criterion=None, **kwargs):
+    def has(self, criterion=None, **kwargs) -> "SQLORMOperations[_T]":
         r"""Return true if this element references a member which meets the
         given criterion.
 
@@ -565,7 +580,7 @@ class PropComparator(operators.ColumnOperators):
         return self.operate(PropComparator.has_op, criterion, **kwargs)
 
 
-class StrategizedProperty(MapperProperty):
+class StrategizedProperty(MapperProperty[_T]):
     """A MapperProperty which uses selectable strategies to affect
     loading behavior.
 
diff --git a/lib/sqlalchemy/orm/properties.py b/lib/sqlalchemy/orm/properties.py
index 8ee26315e..c4aac5a38 100644
--- a/lib/sqlalchemy/orm/properties.py
+++ b/lib/sqlalchemy/orm/properties.py
@@ -13,7 +13,6 @@ mapped attributes.
 """
 
 from typing import Any
-from typing import Generic
 from typing import TypeVar
 
 from . import attributes
@@ -32,6 +31,7 @@ from ..sql import coercions
 from ..sql import roles
 
 _T = TypeVar("_T", bound=Any)
+_PT = TypeVar("_PT", bound=Any)
 
 __all__ = [
     "ColumnProperty",
@@ -43,7 +43,7 @@ __all__ = [
 
 
 @log.class_logger
-class ColumnProperty(StrategizedProperty, Generic[_T]):
+class ColumnProperty(StrategizedProperty[_T]):
     """Describes an object attribute that corresponds to a table column.
 
     Public constructor is the :func:`_orm.column_property` function.
@@ -90,6 +90,7 @@ class ColumnProperty(StrategizedProperty, Generic[_T]):
             )
             for c in columns
         ]
+        self.parent = self.key = None
         self.group = kwargs.pop("group", None)
         self.deferred = kwargs.pop("deferred", False)
         self.raiseload = kwargs.pop("raiseload", False)
@@ -253,7 +254,7 @@ class ColumnProperty(StrategizedProperty, Generic[_T]):
                 dest_dict, [self.key], no_loader=True
             )
 
-    class Comparator(util.MemoizedSlots, PropComparator):
+    class Comparator(util.MemoizedSlots, PropComparator[_PT]):
         """Produce boolean, comparison, and other operators for
         :class:`.ColumnProperty` attributes.
 
@@ -361,4 +362,6 @@ class ColumnProperty(StrategizedProperty, Generic[_T]):
             return op(col._bind_param(op, other), col, **kwargs)
 
     def __str__(self):
+        if not self.parent or not self.key:
+            return object.__repr__(self)
         return str(self.parent.class_.__name__) + "." + self.key
diff --git a/lib/sqlalchemy/orm/relationships.py b/lib/sqlalchemy/orm/relationships.py
index 330d45430..c5ea07051 100644
--- a/lib/sqlalchemy/orm/relationships.py
+++ b/lib/sqlalchemy/orm/relationships.py
@@ -15,8 +15,8 @@ and `secondaryjoin` aspects of :func:`_orm.relationship`.
 """
 import collections
 import re
+from typing import Any
 from typing import Callable
-from typing import Generic
 from typing import Type
 from typing import TypeVar
 from typing import Union
@@ -53,7 +53,8 @@ from ..sql.util import join_condition
 from ..sql.util import selectables_overlap
 from ..sql.util import visit_binary_product
 
-_RC = TypeVar("_RC")
+_T = TypeVar("_T", bound=Any)
+_PT = TypeVar("_PT", bound=Any)
 
 
 def remote(expr):
@@ -96,7 +97,7 @@ def foreign(expr):
 
 
 @log.class_logger
-class RelationshipProperty(StrategizedProperty, Generic[_RC]):
+class RelationshipProperty(StrategizedProperty[_T]):
     """Describes an object property that holds a single item or list
     of items that correspond to a related database table.
 
@@ -125,7 +126,7 @@ class RelationshipProperty(StrategizedProperty, Generic[_RC]):
 
     def __init__(
         self,
-        argument: Union[str, Type[_RC], Callable[[], Type[_RC]]],
+        argument: Union[str, Type[_T], Callable[[], Type[_T]]],
         secondary=None,
         primaryjoin=None,
         secondaryjoin=None,
@@ -285,7 +286,7 @@ class RelationshipProperty(StrategizedProperty, Generic[_RC]):
             doc=self.doc,
         )
 
-    class Comparator(PropComparator):
+    class Comparator(PropComparator[_PT]):
         """Produce boolean, comparison, and other operators for
         :class:`.RelationshipProperty` attributes.
 
@@ -861,6 +862,8 @@ class RelationshipProperty(StrategizedProperty, Generic[_RC]):
             self.prop.parent._check_configure()
             return self.prop
 
+    comparator: Comparator[_T]
+
     def _with_parent(self, instance, alias_secondary=True, from_entity=None):
         assert instance is not None
         adapt_source = None
diff --git a/lib/sqlalchemy/sql/base.py b/lib/sqlalchemy/sql/base.py
index 6ab9a75f6..7841ce88a 100644
--- a/lib/sqlalchemy/sql/base.py
+++ b/lib/sqlalchemy/sql/base.py
@@ -31,11 +31,12 @@ from .. import util
 from ..util import HasMemoized
 from ..util import hybridmethod
 from ..util import typing as compat_typing
+from ..util._has_cy import HAS_CYEXTENSION
 
-try:
-    from sqlalchemy.cyextension.util import prefix_anon_map  # noqa
-except ImportError:
+if typing.TYPE_CHECKING or not HAS_CYEXTENSION:
     from ._py_util import prefix_anon_map  # noqa
+else:
+    from sqlalchemy.cyextension.util import prefix_anon_map  # noqa
 
 coercions = None
 elements = None
diff --git a/lib/sqlalchemy/sql/coercions.py b/lib/sqlalchemy/sql/coercions.py
index 3bec73f7d..d5a75a165 100644
--- a/lib/sqlalchemy/sql/coercions.py
+++ b/lib/sqlalchemy/sql/coercions.py
@@ -10,12 +10,10 @@ import numbers
 import re
 import typing
 from typing import Any
-from typing import Callable
+from typing import Any as TODO_Any
 from typing import Optional
-from typing import overload
 from typing import Type
 from typing import TypeVar
-from typing import Union
 
 from . import operators
 from . import roles
@@ -42,7 +40,6 @@ if typing.TYPE_CHECKING:
     from . import selectable
     from . import traversals
     from .elements import ClauseElement
-    from .elements import ColumnElement
 
 _SR = TypeVar("_SR", bound=roles.SQLRole)
 _StringOnlyR = TypeVar("_StringOnlyR", bound=roles.StringRole)
@@ -129,59 +126,10 @@ def _expression_collection_was_a_list(attrname, fnname, args):
     return args
 
 
-@overload
-def expect(
-    role: Type[roles.InElementRole],
-    element: Any,
-    *,
-    apply_propagate_attrs: Optional["ClauseElement"] = None,
-    argname: Optional[str] = None,
-    post_inspect: bool = False,
-    **kw: Any,
-) -> Union["elements.ColumnElement", "selectable.Select"]:
-    ...
-
-
-@overload
-def expect(
-    role: Type[roles.HasCTERole],
-    element: Any,
-    *,
-    apply_propagate_attrs: Optional["ClauseElement"] = None,
-    argname: Optional[str] = None,
-    post_inspect: bool = False,
-    **kw: Any,
-) -> "selectable.HasCTE":
-    ...
+# TODO; would like to have overloads here, however mypy is being extremely
+# pedantic about them. not sure why pylance is OK with them.
 
 
-@overload
-def expect(
-    role: Type[roles.ExpressionElementRole],
-    element: Any,
-    *,
-    apply_propagate_attrs: Optional["ClauseElement"] = None,
-    argname: Optional[str] = None,
-    post_inspect: bool = False,
-    **kw: Any,
-) -> "ColumnElement":
-    ...
-
-
-@overload
-def expect(
-    role: "Type[_StringOnlyR]",
-    element: Any,
-    *,
-    apply_propagate_attrs: Optional["ClauseElement"] = None,
-    argname: Optional[str] = None,
-    post_inspect: bool = False,
-    **kw: Any,
-) -> str:
-    ...
-
-
-@overload
 def expect(
     role: Type[_SR],
     element: Any,
@@ -190,32 +138,7 @@ def expect(
     argname: Optional[str] = None,
     post_inspect: bool = False,
     **kw: Any,
-) -> _SR:
-    ...
-
-
-@overload
-def expect(
-    role: Type[_SR],
-    element: Callable[..., Any],
-    *,
-    apply_propagate_attrs: Optional["ClauseElement"] = None,
-    argname: Optional[str] = None,
-    post_inspect: bool = False,
-    **kw: Any,
-) -> "lambdas.LambdaElement":
-    ...
-
-
-def expect(
-    role: Type[_SR],
-    element: Any,
-    *,
-    apply_propagate_attrs: Optional["ClauseElement"] = None,
-    argname: Optional[str] = None,
-    post_inspect: bool = False,
-    **kw: Any,
-) -> Union[str, _SR, "lambdas.LambdaElement"]:
+) -> TODO_Any:
     if (
         role.allows_lambda
         # note callable() will not invoke a __getattr__() method, whereas
@@ -350,7 +273,9 @@ class RoleImpl:
         self.name = role_class._role_name
         self._use_inspection = issubclass(role_class, roles.UsesInspection)
 
-    def _implicit_coercions(self, element, resolved, argname=None, **kw):
+    def _implicit_coercions(
+        self, element, resolved, argname=None, **kw
+    ) -> Any:
         self._raise_for_expected(element, argname, resolved)
 
     def _raise_for_expected(
@@ -422,9 +347,8 @@ class _ColumnCoercions:
             "subquery.",
         )
 
-    def _implicit_coercions(
-        self, original_element, resolved, argname=None, **kw
-    ):
+    def _implicit_coercions(self, element, resolved, argname=None, **kw):
+        original_element = element
         if not getattr(resolved, "is_clause_element", False):
             self._raise_for_expected(original_element, argname, resolved)
         elif resolved._is_select_statement:
diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index 705a89889..65f345fb3 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -50,6 +50,7 @@ from .visitors import Traversible
 from .. import exc
 from .. import inspection
 from .. import util
+from ..util.langhelpers import TypingOnly
 
 if typing.TYPE_CHECKING:
     from decimal import Decimal
@@ -572,262 +573,8 @@ class CompilerColumnElement(
     __slots__ = ()
 
 
-class ColumnElement(
-    roles.ColumnArgumentOrKeyRole,
-    roles.StatementOptionRole,
-    roles.WhereHavingRole,
-    roles.BinaryElementRole,
-    roles.OrderByRole,
-    roles.ColumnsClauseRole,
-    roles.LimitOffsetRole,
-    roles.DMLColumnRole,
-    roles.DDLConstraintColumnRole,
-    roles.DDLExpressionRole,
-    operators.ColumnOperators["ColumnElement"],
-    ClauseElement,
-    Generic[_T],
-):
-    """Represent a column-oriented SQL expression suitable for usage in the
-    "columns" clause, WHERE clause etc. of a statement.
-
-    While the most familiar kind of :class:`_expression.ColumnElement` is the
-    :class:`_schema.Column` object, :class:`_expression.ColumnElement`
-    serves as the basis
-    for any unit that may be present in a SQL expression, including
-    the expressions themselves, SQL functions, bound parameters,
-    literal expressions, keywords such as ``NULL``, etc.
-    :class:`_expression.ColumnElement`
-    is the ultimate base class for all such elements.
-
-    A wide variety of SQLAlchemy Core functions work at the SQL expression
-    level, and are intended to accept instances of
-    :class:`_expression.ColumnElement` as
-    arguments.  These functions will typically document that they accept a
-    "SQL expression" as an argument.  What this means in terms of SQLAlchemy
-    usually refers to an input which is either already in the form of a
-    :class:`_expression.ColumnElement` object,
-    or a value which can be **coerced** into
-    one.  The coercion rules followed by most, but not all, SQLAlchemy Core
-    functions with regards to SQL expressions are as follows:
-
-        * a literal Python value, such as a string, integer or floating
-          point value, boolean, datetime, ``Decimal`` object, or virtually
-          any other Python object, will be coerced into a "literal bound
-          value".  This generally means that a :func:`.bindparam` will be
-          produced featuring the given value embedded into the construct; the
-          resulting :class:`.BindParameter` object is an instance of
-          :class:`_expression.ColumnElement`.
-          The Python value will ultimately be sent
-          to the DBAPI at execution time as a parameterized argument to the
-          ``execute()`` or ``executemany()`` methods, after SQLAlchemy
-          type-specific converters (e.g. those provided by any associated
-          :class:`.TypeEngine` objects) are applied to the value.
-
-        * any special object value, typically ORM-level constructs, which
-          feature an accessor called ``__clause_element__()``.  The Core
-          expression system looks for this method when an object of otherwise
-          unknown type is passed to a function that is looking to coerce the
-          argument into a :class:`_expression.ColumnElement` and sometimes a
-          :class:`_expression.SelectBase` expression.
-          It is used within the ORM to
-          convert from ORM-specific objects like mapped classes and
-          mapped attributes into Core expression objects.
-
-        * The Python ``None`` value is typically interpreted as ``NULL``,
-          which in SQLAlchemy Core produces an instance of :func:`.null`.
-
-    A :class:`_expression.ColumnElement` provides the ability to generate new
-    :class:`_expression.ColumnElement`
-    objects using Python expressions.  This means that Python operators
-    such as ``==``, ``!=`` and ``<`` are overloaded to mimic SQL operations,
-    and allow the instantiation of further :class:`_expression.ColumnElement`
-    instances
-    which are composed from other, more fundamental
-    :class:`_expression.ColumnElement`
-    objects.  For example, two :class:`.ColumnClause` objects can be added
-    together with the addition operator ``+`` to produce
-    a :class:`.BinaryExpression`.
-    Both :class:`.ColumnClause` and :class:`.BinaryExpression` are subclasses
-    of :class:`_expression.ColumnElement`::
-
-        >>> from sqlalchemy.sql import column
-        >>> column('a') + column('b')
-        <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
-        >>> print(column('a') + column('b'))
-        a + b
-
-    .. seealso::
-
-        :class:`_schema.Column`
-
-        :func:`_expression.column`
-
-    """
-
-    __visit_name__ = "column_element"
-
-    primary_key = False
-    foreign_keys = []
-    _proxies = ()
-
-    _tq_label = None
-    """The named label that can be used to target
-    this column in a result set in a "table qualified" context.
-
-    This label is almost always the label used when
-    rendering <expr> AS <label> in a SELECT statement when using
-    the LABEL_STYLE_TABLENAME_PLUS_COL label style, which is what the legacy
-    ORM ``Query`` object uses as well.
-
-    For a regular Column bound to a Table, this is typically the label
-    <tablename>_<columnname>.  For other constructs, different rules
-    may apply, such as anonymized labels and others.
-
-    .. versionchanged:: 1.4.21 renamed from ``._label``
-
-    """
-
-    key = None
-    """The 'key' that in some circumstances refers to this object in a
-    Python namespace.
-
-    This typically refers to the "key" of the column as present in the
-    ``.c`` collection of a selectable, e.g. ``sometable.c["somekey"]`` would
-    return a :class:`_schema.Column` with a ``.key`` of "somekey".
-
-    """
-
-    @HasMemoized.memoized_attribute
-    def _tq_key_label(self):
-        """A label-based version of 'key' that in some circumstances refers
-        to this object in a Python namespace.
-
-
-        _tq_key_label comes into play when a select() statement is constructed
-        with apply_labels(); in this case, all Column objects in the ``.c``
-        collection are rendered as <tablename>_<columnname> in SQL; this is
-        essentially the value of ._label. But to locate those columns in the
-        ``.c`` collection, the name is along the lines of <tablename>_<key>;
-        that's the typical value of .key_label.
-
-        .. versionchanged:: 1.4.21 renamed from ``._key_label``
-
-        """
-        return self._proxy_key
-
-    @property
-    def _key_label(self):
-        """legacy; renamed to _tq_key_label"""
-        return self._tq_key_label
-
-    @property
-    def _label(self):
-        """legacy; renamed to _tq_label"""
-        return self._tq_label
-
-    @property
-    def _non_anon_label(self):
-        """the 'name' that naturally applies this element when rendered in
-        SQL.
-
-        Concretely, this is the "name" of a column or a label in a
-        SELECT statement; ``<columnname>`` and ``<labelname>`` below::
-
-            SELECT <columnmame> FROM table
-
-            SELECT column AS <labelname> FROM table
-
-        Above, the two names noted will be what's present in the DBAPI
-        ``cursor.description`` as the names.
-
-        If this attribute returns ``None``, it means that the SQL element as
-        written does not have a 100% fully predictable "name" that would appear
-        in the ``cursor.description``. Examples include SQL functions, CAST
-        functions, etc. While such things do return names in
-        ``cursor.description``, they are only predictable on a
-        database-specific basis; e.g. an expression like ``MAX(table.col)`` may
-        appear as the string ``max`` on one database (like PostgreSQL) or may
-        appear as the whole expression ``max(table.col)`` on SQLite.
-
-        The default implementation looks for a ``.name`` attribute on the
-        object, as has been the precedent established in SQLAlchemy for many
-        years.  An exception is made on the ``FunctionElement`` subclass
-        so that the return value is always ``None``.
-
-        .. versionadded:: 1.4.21
-
-
-
-        """
-        return getattr(self, "name", None)
-
-    _render_label_in_columns_clause = True
-    """A flag used by select._columns_plus_names that helps to determine
-    we are actually going to render in terms of "SELECT <col> AS <label>".
-    This flag can be returned as False for some Column objects that want
-    to be rendered as simple "SELECT <col>"; typically columns that don't have
-    any parent table and are named the same as what the label would be
-    in any case.
-
-    """
-
-    _allow_label_resolve = True
-    """A flag that can be flipped to prevent a column from being resolvable
-    by string label name.
-
-    The joined eager loader strategy in the ORM uses this, for example.
-
-    """
-
-    _is_implicitly_boolean = False
-
-    _alt_names = ()
-
-    def self_group(self, against=None):
-        if (
-            against in (operators.and_, operators.or_, operators._asbool)
-            and self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity
-        ):
-            return AsBoolean(self, operators.is_true, operators.is_false)
-        elif against in (operators.any_op, operators.all_op):
-            return Grouping(self)
-        else:
-            return self
-
-    def _negate(self):
-        if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
-            return AsBoolean(self, operators.is_false, operators.is_true)
-        else:
-            return super(ColumnElement, self)._negate()
-
-    @util.memoized_property
-    def type(self) -> "TypeEngine[_T]":
-        return type_api.NULLTYPE
-
-    @HasMemoized.memoized_attribute
-    def comparator(self) -> "TypeEngine.Comparator[_T]":
-        try:
-            comparator_factory = self.type.comparator_factory
-        except AttributeError as err:
-            raise TypeError(
-                "Object %r associated with '.type' attribute "
-                "is not a TypeEngine class or object" % self.type
-            ) from err
-        else:
-            return comparator_factory(self)
-
-    def __getattr__(self, key):
-        try:
-            return getattr(self.comparator, key)
-        except AttributeError as err:
-            raise AttributeError(
-                "Neither %r object nor %r object has an attribute %r"
-                % (
-                    type(self).__name__,
-                    type(self.comparator).__name__,
-                    key,
-                )
-            ) from err
+class SQLCoreOperations(Generic[_T], TypingOnly):
+    __slots__ = ()
 
     # annotations for comparison methods
     # these are from operators->Operators / ColumnOperators,
@@ -894,7 +641,9 @@ class ColumnElement(
             ...
 
         @overload
-        def concat(self, other: Any) -> "BinaryExpression[_ST]":
+        def concat(
+            self: "SQLCoreOperations[_ST]", other: Any
+        ) -> "BinaryExpression[_ST]":
             ...
 
         @overload
@@ -986,7 +735,7 @@ class ColumnElement(
         ) -> "BinaryExpression[bool]":
             ...
 
-        def distinct(self: "ColumnElement[_T]") -> "UnaryExpression[_T]":
+        def distinct(self: "SQLCoreOperations[_T]") -> "UnaryExpression[_T]":
             ...
 
         def any_(self) -> "CollectionAggregate":
@@ -996,22 +745,28 @@ class ColumnElement(
             ...
 
         # numeric overloads.  These need more tweaking
+        # in particular they all need to have a variant for Optiona[_T]
+        # because Optional only applies to the data side, not the expression
+        # side
 
         @overload
         def __add__(
-            self: "ColumnElement[_NT]", other: "Union[ColumnElement[_NT], _NT]"
+            self: "Union[_SQO[_NT], _SQO[Optional[_NT]]]",
+            other: "Union[_SQO[Optional[_NT]], _SQO[_NT], _NT]",
         ) -> "BinaryExpression[_NT]":
             ...
 
         @overload
         def __add__(
-            self: "ColumnElement[_NT]", other: Any
+            self: "Union[_SQO[_NT], _SQO[Optional[_NT]]]",
+            other: Any,
         ) -> "BinaryExpression[_NUMERIC]":
             ...
 
         @overload
         def __add__(
-            self: "ColumnElement[_ST]", other: Any
+            self: "Union[_SQO[_ST], _SQO[Optional[_ST]]]",
+            other: Any,
         ) -> "BinaryExpression[_ST]":
             ...
 
@@ -1031,7 +786,8 @@ class ColumnElement(
 
         @overload
         def __sub__(
-            self: "ColumnElement[_NT]", other: "Union[ColumnElement[_NT], _NT]"
+            self: "SQLCoreOperations[_NT]",
+            other: "Union[SQLCoreOperations[_NT], _NT]",
         ) -> "BinaryExpression[_NT]":
             ...
 
@@ -1044,7 +800,7 @@ class ColumnElement(
 
         @overload
         def __rsub__(
-            self: "ColumnElement[_NT]", other: Any
+            self: "SQLCoreOperations[_NT]", other: Any
         ) -> "BinaryExpression[_NUMERIC]":
             ...
 
@@ -1057,7 +813,7 @@ class ColumnElement(
 
         @overload
         def __mul__(
-            self: "ColumnElement[_NT]", other: Any
+            self: "SQLCoreOperations[_NT]", other: Any
         ) -> "BinaryExpression[_NUMERIC]":
             ...
 
@@ -1070,7 +826,7 @@ class ColumnElement(
 
         @overload
         def __rmul__(
-            self: "ColumnElement[_NT]", other: Any
+            self: "SQLCoreOperations[_NT]", other: Any
         ) -> "BinaryExpression[_NUMERIC]":
             ...
 
@@ -1083,7 +839,7 @@ class ColumnElement(
 
         @overload
         def __mod__(
-            self: "ColumnElement[_NT]", other: Any
+            self: "SQLCoreOperations[_NT]", other: Any
         ) -> "BinaryExpression[_NUMERIC]":
             ...
 
@@ -1096,7 +852,7 @@ class ColumnElement(
 
         @overload
         def __rmod__(
-            self: "ColumnElement[_NT]", other: Any
+            self: "SQLCoreOperations[_NT]", other: Any
         ) -> "BinaryExpression[_NUMERIC]":
             ...
 
@@ -1109,7 +865,7 @@ class ColumnElement(
 
         @overload
         def __truediv__(
-            self: "ColumnElement[_NT]", other: Any
+            self: "SQLCoreOperations[_NT]", other: Any
         ) -> "BinaryExpression[_NUMERIC]":
             ...
 
@@ -1122,7 +878,7 @@ class ColumnElement(
 
         @overload
         def __rtruediv__(
-            self: "ColumnElement[_NT]", other: Any
+            self: "SQLCoreOperations[_NT]", other: Any
         ) -> "BinaryExpression[_NUMERIC]":
             ...
 
@@ -1135,7 +891,7 @@ class ColumnElement(
 
         @overload
         def __floordiv__(
-            self: "ColumnElement[_NT]", other: Any
+            self: "SQLCoreOperations[_NT]", other: Any
         ) -> "BinaryExpression[_NUMERIC]":
             ...
 
@@ -1148,7 +904,7 @@ class ColumnElement(
 
         @overload
         def __rfloordiv__(
-            self: "ColumnElement[_NT]", other: Any
+            self: "SQLCoreOperations[_NT]", other: Any
         ) -> "BinaryExpression[_NUMERIC]":
             ...
 
@@ -1159,6 +915,267 @@ class ColumnElement(
         def __rfloordiv__(self, other: Any) -> "BinaryExpression":
             ...
 
+
+_SQO = SQLCoreOperations
+
+
+class ColumnElement(
+    roles.ColumnArgumentOrKeyRole,
+    roles.StatementOptionRole,
+    roles.WhereHavingRole,
+    roles.BinaryElementRole,
+    roles.OrderByRole,
+    roles.ColumnsClauseRole,
+    roles.LimitOffsetRole,
+    roles.DMLColumnRole,
+    roles.DDLConstraintColumnRole,
+    roles.DDLExpressionRole,
+    SQLCoreOperations[_T],
+    operators.ColumnOperators[SQLCoreOperations],
+    ClauseElement,
+):
+    """Represent a column-oriented SQL expression suitable for usage in the
+    "columns" clause, WHERE clause etc. of a statement.
+
+    While the most familiar kind of :class:`_expression.ColumnElement` is the
+    :class:`_schema.Column` object, :class:`_expression.ColumnElement`
+    serves as the basis
+    for any unit that may be present in a SQL expression, including
+    the expressions themselves, SQL functions, bound parameters,
+    literal expressions, keywords such as ``NULL``, etc.
+    :class:`_expression.ColumnElement`
+    is the ultimate base class for all such elements.
+
+    A wide variety of SQLAlchemy Core functions work at the SQL expression
+    level, and are intended to accept instances of
+    :class:`_expression.ColumnElement` as
+    arguments.  These functions will typically document that they accept a
+    "SQL expression" as an argument.  What this means in terms of SQLAlchemy
+    usually refers to an input which is either already in the form of a
+    :class:`_expression.ColumnElement` object,
+    or a value which can be **coerced** into
+    one.  The coercion rules followed by most, but not all, SQLAlchemy Core
+    functions with regards to SQL expressions are as follows:
+
+        * a literal Python value, such as a string, integer or floating
+          point value, boolean, datetime, ``Decimal`` object, or virtually
+          any other Python object, will be coerced into a "literal bound
+          value".  This generally means that a :func:`.bindparam` will be
+          produced featuring the given value embedded into the construct; the
+          resulting :class:`.BindParameter` object is an instance of
+          :class:`_expression.ColumnElement`.
+          The Python value will ultimately be sent
+          to the DBAPI at execution time as a parameterized argument to the
+          ``execute()`` or ``executemany()`` methods, after SQLAlchemy
+          type-specific converters (e.g. those provided by any associated
+          :class:`.TypeEngine` objects) are applied to the value.
+
+        * any special object value, typically ORM-level constructs, which
+          feature an accessor called ``__clause_element__()``.  The Core
+          expression system looks for this method when an object of otherwise
+          unknown type is passed to a function that is looking to coerce the
+          argument into a :class:`_expression.ColumnElement` and sometimes a
+          :class:`_expression.SelectBase` expression.
+          It is used within the ORM to
+          convert from ORM-specific objects like mapped classes and
+          mapped attributes into Core expression objects.
+
+        * The Python ``None`` value is typically interpreted as ``NULL``,
+          which in SQLAlchemy Core produces an instance of :func:`.null`.
+
+    A :class:`_expression.ColumnElement` provides the ability to generate new
+    :class:`_expression.ColumnElement`
+    objects using Python expressions.  This means that Python operators
+    such as ``==``, ``!=`` and ``<`` are overloaded to mimic SQL operations,
+    and allow the instantiation of further :class:`_expression.ColumnElement`
+    instances
+    which are composed from other, more fundamental
+    :class:`_expression.ColumnElement`
+    objects.  For example, two :class:`.ColumnClause` objects can be added
+    together with the addition operator ``+`` to produce
+    a :class:`.BinaryExpression`.
+    Both :class:`.ColumnClause` and :class:`.BinaryExpression` are subclasses
+    of :class:`_expression.ColumnElement`::
+
+        >>> from sqlalchemy.sql import column
+        >>> column('a') + column('b')
+        <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
+        >>> print(column('a') + column('b'))
+        a + b
+
+    .. seealso::
+
+        :class:`_schema.Column`
+
+        :func:`_expression.column`
+
+    """
+
+    __visit_name__ = "column_element"
+
+    primary_key = False
+    foreign_keys = []
+    _proxies = ()
+
+    _tq_label = None
+    """The named label that can be used to target
+    this column in a result set in a "table qualified" context.
+
+    This label is almost always the label used when
+    rendering <expr> AS <label> in a SELECT statement when using
+    the LABEL_STYLE_TABLENAME_PLUS_COL label style, which is what the legacy
+    ORM ``Query`` object uses as well.
+
+    For a regular Column bound to a Table, this is typically the label
+    <tablename>_<columnname>.  For other constructs, different rules
+    may apply, such as anonymized labels and others.
+
+    .. versionchanged:: 1.4.21 renamed from ``._label``
+
+    """
+
+    key = None
+    """The 'key' that in some circumstances refers to this object in a
+    Python namespace.
+
+    This typically refers to the "key" of the column as present in the
+    ``.c`` collection of a selectable, e.g. ``sometable.c["somekey"]`` would
+    return a :class:`_schema.Column` with a ``.key`` of "somekey".
+
+    """
+
+    @HasMemoized.memoized_attribute
+    def _tq_key_label(self):
+        """A label-based version of 'key' that in some circumstances refers
+        to this object in a Python namespace.
+
+
+        _tq_key_label comes into play when a select() statement is constructed
+        with apply_labels(); in this case, all Column objects in the ``.c``
+        collection are rendered as <tablename>_<columnname> in SQL; this is
+        essentially the value of ._label. But to locate those columns in the
+        ``.c`` collection, the name is along the lines of <tablename>_<key>;
+        that's the typical value of .key_label.
+
+        .. versionchanged:: 1.4.21 renamed from ``._key_label``
+
+        """
+        return self._proxy_key
+
+    @property
+    def _key_label(self):
+        """legacy; renamed to _tq_key_label"""
+        return self._tq_key_label
+
+    @property
+    def _label(self):
+        """legacy; renamed to _tq_label"""
+        return self._tq_label
+
+    @property
+    def _non_anon_label(self):
+        """the 'name' that naturally applies this element when rendered in
+        SQL.
+
+        Concretely, this is the "name" of a column or a label in a
+        SELECT statement; ``<columnname>`` and ``<labelname>`` below::
+
+            SELECT <columnmame> FROM table
+
+            SELECT column AS <labelname> FROM table
+
+        Above, the two names noted will be what's present in the DBAPI
+        ``cursor.description`` as the names.
+
+        If this attribute returns ``None``, it means that the SQL element as
+        written does not have a 100% fully predictable "name" that would appear
+        in the ``cursor.description``. Examples include SQL functions, CAST
+        functions, etc. While such things do return names in
+        ``cursor.description``, they are only predictable on a
+        database-specific basis; e.g. an expression like ``MAX(table.col)`` may
+        appear as the string ``max`` on one database (like PostgreSQL) or may
+        appear as the whole expression ``max(table.col)`` on SQLite.
+
+        The default implementation looks for a ``.name`` attribute on the
+        object, as has been the precedent established in SQLAlchemy for many
+        years.  An exception is made on the ``FunctionElement`` subclass
+        so that the return value is always ``None``.
+
+        .. versionadded:: 1.4.21
+
+
+
+        """
+        return getattr(self, "name", None)
+
+    _render_label_in_columns_clause = True
+    """A flag used by select._columns_plus_names that helps to determine
+    we are actually going to render in terms of "SELECT <col> AS <label>".
+    This flag can be returned as False for some Column objects that want
+    to be rendered as simple "SELECT <col>"; typically columns that don't have
+    any parent table and are named the same as what the label would be
+    in any case.
+
+    """
+
+    _allow_label_resolve = True
+    """A flag that can be flipped to prevent a column from being resolvable
+    by string label name.
+
+    The joined eager loader strategy in the ORM uses this, for example.
+
+    """
+
+    _is_implicitly_boolean = False
+
+    _alt_names = ()
+
+    def self_group(self, against=None):
+        if (
+            against in (operators.and_, operators.or_, operators._asbool)
+            and self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity
+        ):
+            return AsBoolean(self, operators.is_true, operators.is_false)
+        elif against in (operators.any_op, operators.all_op):
+            return Grouping(self)
+        else:
+            return self
+
+    def _negate(self):
+        if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
+            return AsBoolean(self, operators.is_false, operators.is_true)
+        else:
+            return super(ColumnElement, self)._negate()
+
+    @util.memoized_property
+    def type(self) -> "TypeEngine[_T]":
+        return type_api.NULLTYPE
+
+    @HasMemoized.memoized_attribute
+    def comparator(self) -> "TypeEngine.Comparator[_T]":
+        try:
+            comparator_factory = self.type.comparator_factory
+        except AttributeError as err:
+            raise TypeError(
+                "Object %r associated with '.type' attribute "
+                "is not a TypeEngine class or object" % self.type
+            ) from err
+        else:
+            return comparator_factory(self)
+
+    def __getattr__(self, key):
+        try:
+            return getattr(self.comparator, key)
+        except AttributeError as err:
+            raise AttributeError(
+                "Neither %r object nor %r object has an attribute %r"
+                % (
+                    type(self).__name__,
+                    type(self.comparator).__name__,
+                    key,
+                )
+            ) from err
+
     def operate(
         self,
         op: operators.OperatorType,
diff --git a/lib/sqlalchemy/sql/functions.py b/lib/sqlalchemy/sql/functions.py
index 7a1e80889..2e6d64c55 100644
--- a/lib/sqlalchemy/sql/functions.py
+++ b/lib/sqlalchemy/sql/functions.py
@@ -9,6 +9,9 @@
 
 """
 
+from typing import Any
+from typing import TypeVar
+
 from . import annotation
 from . import coercions
 from . import operators
@@ -42,6 +45,8 @@ from .visitors import InternalTraversal
 from .. import util
 
 
+_T = TypeVar("_T", bound=Any)
+
 _registry = util.defaultdict(dict)
 
 
@@ -67,7 +72,7 @@ def register_function(identifier, fn, package="_default"):
     reg[identifier] = fn
 
 
-class FunctionElement(Executable, ColumnElement, FromClause, Generative):
+class FunctionElement(Executable, ColumnElement[_T], FromClause, Generative):
     """Base for SQL function-oriented constructs.
 
     .. seealso::
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index fd1abd71b..00e20e3fb 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -15,6 +15,9 @@ import collections
 import itertools
 from operator import attrgetter
 import typing
+from typing import Any as TODO_Any
+from typing import Optional
+from typing import Tuple
 from typing import Type
 from typing import Union
 
@@ -53,6 +56,7 @@ from .elements import BooleanClauseList
 from .elements import ClauseElement
 from .elements import ClauseList
 from .elements import ColumnClause
+from .elements import ColumnElement
 from .elements import GroupedElement
 from .elements import Grouping
 from .elements import literal_column
@@ -2648,8 +2652,7 @@ class SelectBase(
         """
         return self.selected_columns
 
-    @property
-    @util.deprecated(
+    @util.deprecated_property(
         "1.4",
         "The :attr:`_expression.SelectBase.c` and "
         ":attr:`_expression.SelectBase.columns` attributes "
@@ -4039,16 +4042,16 @@ class Select(
 
     __visit_name__ = "select"
 
-    _setup_joins = ()
-    _memoized_select_entities = ()
+    _setup_joins: Tuple[TODO_Any, ...] = ()
+    _memoized_select_entities: Tuple[TODO_Any, ...] = ()
 
     _distinct = False
-    _distinct_on = ()
-    _correlate = ()
-    _correlate_except = None
-    _where_criteria = ()
-    _having_criteria = ()
-    _from_obj = ()
+    _distinct_on: Tuple[ColumnElement, ...] = ()
+    _correlate: Tuple[FromClause, ...] = ()
+    _correlate_except: Optional[Tuple[FromClause, ...]] = None
+    _where_criteria: Tuple[ColumnElement, ...] = ()
+    _having_criteria: Tuple[ColumnElement, ...] = ()
+    _from_obj: Tuple[FromClause, ...] = ()
     _auto_correlate = True
 
     _compile_options = SelectState.default_select_compile_options
@@ -4417,8 +4420,7 @@ class Select(
         """
         return self._compile_state_factory(self, None)._get_display_froms()
 
-    @property
-    @util.deprecated(
+    @util.deprecated_property(
         "1.4.23",
         "The :attr:`_expression.Select.froms` attribute is moved to "
         "the :meth:`_expression.Select.get_final_froms` method.",
diff --git a/lib/sqlalchemy/sql/sqltypes.py b/lib/sqlalchemy/sql/sqltypes.py
index 81434fbb9..42fad5e04 100644
--- a/lib/sqlalchemy/sql/sqltypes.py
+++ b/lib/sqlalchemy/sql/sqltypes.py
@@ -65,7 +65,7 @@ class _LookupExpressionAdapter:
     def _expression_adaptations(self):
         raise NotImplementedError()
 
-    class Comparator(TypeEngine.Comparator):
+    class Comparator(TypeEngine.Comparator[_T]):
         _blank_dict = util.immutabledict()
 
         def _adapt_expression(self, op, other_comparator):
@@ -88,7 +88,7 @@ class Concatenable:
     """A mixin that marks a type as supporting 'concatenation',
     typically strings."""
 
-    class Comparator(TypeEngine.Comparator):
+    class Comparator(TypeEngine.Comparator[_T]):
         def _adapt_expression(self, op, other_comparator):
             if op is operators.add and isinstance(
                 other_comparator,
@@ -113,7 +113,7 @@ class Indexable:
 
     """
 
-    class Comparator(TypeEngine.Comparator):
+    class Comparator(TypeEngine.Comparator[_T]):
         def _setup_getitem(self, index):
             raise NotImplementedError()
 
@@ -1377,7 +1377,7 @@ class Enum(Emulated, String, TypeEngine[Union[str, enum.Enum]], SchemaType):
                     )
                 ) from err
 
-    class Comparator(String.Comparator):
+    class Comparator(String.Comparator[_T]):
         def _adapt_expression(self, op, other_comparator):
             op, typ = super(Enum.Comparator, self)._adapt_expression(
                 op, other_comparator
@@ -2204,7 +2204,7 @@ class JSON(Indexable, TypeEngine[Any]):
 
         """
 
-    class Comparator(Indexable.Comparator, Concatenable.Comparator):
+    class Comparator(Indexable.Comparator[_T], Concatenable.Comparator[_T]):
         """Define comparison operations for :class:`_types.JSON`."""
 
         def _setup_getitem(self, index):
@@ -2523,7 +2523,7 @@ class ARRAY(
     """If True, Python zero-based indexes should be interpreted as one-based
     on the SQL expression side."""
 
-    class Comparator(Indexable.Comparator, Concatenable.Comparator):
+    class Comparator(Indexable.Comparator[_T], Concatenable.Comparator[_T]):
 
         """Define comparison operations for :class:`_types.ARRAY`.
 
@@ -2967,7 +2967,7 @@ class NullType(TypeEngine):
 
         return process
 
-    class Comparator(TypeEngine.Comparator):
+    class Comparator(TypeEngine.Comparator[_T]):
         def _adapt_expression(self, op, other_comparator):
             if isinstance(
                 other_comparator, NullType.Comparator
diff --git a/lib/sqlalchemy/sql/type_api.py b/lib/sqlalchemy/sql/type_api.py
index 75eb1b8c5..dd29b2c3a 100644
--- a/lib/sqlalchemy/sql/type_api.py
+++ b/lib/sqlalchemy/sql/type_api.py
@@ -11,6 +11,7 @@
 
 import typing
 from typing import Any
+from typing import Callable
 from typing import Generic
 from typing import Tuple
 from typing import Type
@@ -1400,7 +1401,7 @@ class TypeDecorator(ExternalType, SchemaEventTarget, TypeEngine[_T]):
 
     """
 
-    class Comparator(TypeEngine.Comparator):
+    class Comparator(TypeEngine.Comparator[_CT]):
         """A :class:`.TypeEngine.Comparator` that is specific to
         :class:`.TypeDecorator`.
 
@@ -1425,7 +1426,7 @@ class TypeDecorator(ExternalType, SchemaEventTarget, TypeEngine[_T]):
             )
 
     @property
-    def comparator_factory(self):
+    def comparator_factory(self) -> Callable[..., TypeEngine.Comparator[_T]]:
         if TypeDecorator.Comparator in self.impl.comparator_factory.__mro__:
             return self.impl.comparator_factory
         else:
diff --git a/lib/sqlalchemy/util/__init__.py b/lib/sqlalchemy/util/__init__.py
index 7c03bcd4b..203460c26 100644
--- a/lib/sqlalchemy/util/__init__.py
+++ b/lib/sqlalchemy/util/__init__.py
@@ -57,7 +57,6 @@ from .compat import dottedgetter
 from .compat import has_refcount_gc
 from .compat import inspect_getfullargspec
 from .compat import local_dataclass_fields
-from .compat import namedtuple
 from .compat import next
 from .compat import osx
 from .compat import py38
@@ -74,6 +73,7 @@ from .deprecations import became_legacy_20
 from .deprecations import deprecated
 from .deprecations import deprecated_cls
 from .deprecations import deprecated_params
+from .deprecations import deprecated_property
 from .deprecations import inject_docstring_text
 from .deprecations import moved_20
 from .deprecations import warn_deprecated
@@ -128,6 +128,7 @@ from .langhelpers import safe_reraise
 from .langhelpers import set_creation_order
 from .langhelpers import string_or_unprintable
 from .langhelpers import symbol
+from .langhelpers import TypingOnly
 from .langhelpers import unbound_method_to_callable
 from .langhelpers import walk_subclasses
 from .langhelpers import warn
diff --git a/lib/sqlalchemy/util/_has_cy.py b/lib/sqlalchemy/util/_has_cy.py
new file mode 100644
index 000000000..bf251b5b5
--- /dev/null
+++ b/lib/sqlalchemy/util/_has_cy.py
@@ -0,0 +1,11 @@
+import typing
+
+if not typing.TYPE_CHECKING:
+    try:
+        from ..cyextension import util  # noqa
+    except ImportError:
+        HAS_CYEXTENSION = False
+    else:
+        HAS_CYEXTENSION = True
+else:
+    HAS_CYEXTENSION = False
diff --git a/lib/sqlalchemy/util/compat.py b/lib/sqlalchemy/util/compat.py
index f3a3debe0..679df73c7 100644
--- a/lib/sqlalchemy/util/compat.py
+++ b/lib/sqlalchemy/util/compat.py
@@ -30,7 +30,6 @@ arm = "aarch" in platform.machine().lower()
 has_refcount_gc = bool(cpython)
 
 dottedgetter = operator.attrgetter
-namedtuple = collections.namedtuple
 next = next  # noqa
 
 FullArgSpec = collections.namedtuple(
diff --git a/lib/sqlalchemy/util/deprecations.py b/lib/sqlalchemy/util/deprecations.py
index c587b82f9..e5d5d5461 100644
--- a/lib/sqlalchemy/util/deprecations.py
+++ b/lib/sqlalchemy/util/deprecations.py
@@ -9,6 +9,9 @@
 functionality."""
 
 import re
+from typing import Any
+from typing import Callable
+from typing import TypeVar
 
 from . import compat
 from .langhelpers import _hash_limit_string
@@ -16,8 +19,11 @@ from .langhelpers import _warnings_warn
 from .langhelpers import decorator
 from .langhelpers import inject_docstring_text
 from .langhelpers import inject_param_text
+from .typing import ReadOnlyInstanceDescriptor
 from .. import exc
 
+_T = TypeVar("_T", bound=Any)
+
 
 def _warn_with_version(msg, version, type_, stacklevel, code=None):
     warn = type_(msg, code=code)
@@ -60,6 +66,50 @@ def deprecated_cls(version, message, constructor="__init__"):
     return decorate
 
 
+def deprecated_property(
+    version,
+    message=None,
+    add_deprecation_to_docstring=True,
+    warning=None,
+    enable_warnings=True,
+) -> Callable[[Callable[..., _T]], ReadOnlyInstanceDescriptor[_T]]:
+    """the @deprecated decorator with a @property.
+
+    E.g.::
+
+        class Foo:
+            @deprecated_property("1.4", "thing is deprecated")
+            def thing(self):
+                return "thing"
+
+    is equivalent to::
+
+        class Foo:
+            @property
+            @deprecated("1.4", "thing is deprecated")
+            def thing(self):
+                return "thing"
+
+    How come?
+
+    Because::
+
+        mypy: error: Decorated property not supported
+
+    great!   now it is.
+
+    """
+    return lambda fn: property(
+        deprecated(
+            version,
+            message=message,
+            add_deprecation_to_docstring=add_deprecation_to_docstring,
+            warning=warning,
+            enable_warnings=enable_warnings,
+        )(fn)
+    )
+
+
 def deprecated(
     version,
     message=None,
diff --git a/lib/sqlalchemy/util/langhelpers.py b/lib/sqlalchemy/util/langhelpers.py
index d85b1261b..80ef3458c 100644
--- a/lib/sqlalchemy/util/langhelpers.py
+++ b/lib/sqlalchemy/util/langhelpers.py
@@ -1669,6 +1669,32 @@ def attrsetter(attrname):
     return env["set"]
 
 
+class TypingOnly:
+    """A mixin class that marks a class as 'typing only', meaning it has
+    absolutely no methods, attributes, or runtime functionality whatsoever.
+
+    """
+
+    __slots__ = ()
+
+    def __init_subclass__(cls) -> None:
+        if TypingOnly in cls.__bases__:
+            remaining = set(cls.__dict__).difference(
+                {
+                    "__module__",
+                    "__doc__",
+                    "__slots__",
+                    "__orig_bases__",
+                }
+            )
+            if remaining:
+                raise AssertionError(
+                    f"Class {cls} directly inherits TypingOnly but has "
+                    f"additional attributes {remaining}."
+                )
+        super().__init_subclass__()
+
+
 class EnsureKWArg:
     r"""Apply translation of functions to accept \**kw arguments if they
     don't already.
diff --git a/lib/sqlalchemy/util/typing.py b/lib/sqlalchemy/util/typing.py
index e735ce531..e2d42db6e 100644
--- a/lib/sqlalchemy/util/typing.py
+++ b/lib/sqlalchemy/util/typing.py
@@ -4,9 +4,12 @@ from typing import Generic
 from typing import overload
 from typing import Type
 from typing import TypeVar
+from typing import Union
 
 from . import compat
 
+_T = TypeVar("_T", bound=Any)
+
 if compat.py38:
     from typing import Literal
     from typing import Protocol
@@ -48,3 +51,22 @@ class _TypeToInstance(Generic[_T]):
     @overload
     def __set__(self, instance: object, value: _T) -> None:
         ...
+
+
+class ReadOnlyInstanceDescriptor(Protocol[_T]):
+    """protocol representing an instance-only descriptor"""
+
+    @overload
+    def __get__(
+        self, instance: None, owner: Any
+    ) -> "ReadOnlyInstanceDescriptor[_T]":
+        ...
+
+    @overload
+    def __get__(self, instance: object, owner: Any) -> _T:
+        ...
+
+    def __get__(
+        self, instance: object, owner: Any
+    ) -> Union["ReadOnlyInstanceDescriptor[_T]", _T]:
+        ...