Remove `typing.Self` workaround

Remove ``typing.Self`` workaround, now using :pep:`673` for most methods
that return ``Self``. Pull request courtesy Yurii Karabas.

Fixes: #9254
Closes: #9255
Pull-request: https://github.com/sqlalchemy/sqlalchemy/pull/9255
Pull-request-sha: 2947df8ada79f5c3afe9c838e65993302199c2f7

Change-Id: Ic32015ad52e95a61f3913d43ea436aa9402804df

5 files changed, 36 insertions, 87 deletions

diff --git a/lib/sqlalchemy/ext/associationproxy.py b/lib/sqlalchemy/ext/associationproxy.py
index 24f348bd1..243c140f8 100644
--- a/lib/sqlalchemy/ext/associationproxy.py
+++ b/lib/sqlalchemy/ext/associationproxy.py
@@ -339,11 +339,6 @@ class _AssociationProxyProtocol(Protocol[_T]):
         ...
 
 
-_SelfAssociationProxy = TypeVar(
-    "_SelfAssociationProxy", bound="AssociationProxy[Any]"
-)
-
-
 class AssociationProxy(
     interfaces.InspectionAttrInfo,
     ORMDescriptor[_T],
@@ -404,9 +399,7 @@ class AssociationProxy(
             self._attribute_options = _DEFAULT_ATTRIBUTE_OPTIONS
 
     @overload
-    def __get__(
-        self: _SelfAssociationProxy, instance: Any, owner: Literal[None]
-    ) -> _SelfAssociationProxy:
+    def __get__(self, instance: Any, owner: Literal[None]) -> Self:
         ...
 
     @overload
@@ -549,9 +542,10 @@ class AssociationProxy(
         )
 
 
-_SelfAssociationProxyInstance = TypeVar(
-    "_SelfAssociationProxyInstance", bound="AssociationProxyInstance[Any]"
-)
+# the pep-673 Self type does not work in Mypy for a "hybrid"
+# style method that returns type or Self, so for one specific case
+# we still need to use the pre-pep-673 workaround.
+_Self = TypeVar("_Self", bound="AssociationProxyInstance[Any]")
 
 
 class AssociationProxyInstance(SQLORMOperations[_T]):
@@ -847,7 +841,7 @@ class AssociationProxyInstance(SQLORMOperations[_T]):
         return self.parent.info
 
     @overload
-    def get(self: Self, obj: Literal[None]) -> Self:
+    def get(self: _Self, obj: Literal[None]) -> _Self:
         ...
 
     @overload
@@ -1589,11 +1583,11 @@ class _AssociationList(_AssociationSingleItem[_T], MutableSequence[_T]):
             return NotImplemented
         return n * list(self)
 
-    def __iadd__(self: Self, iterable: Iterable[_T]) -> Self:
+    def __iadd__(self, iterable: Iterable[_T]) -> Self:
         self.extend(iterable)
         return self
 
-    def __imul__(self: Self, n: SupportsIndex) -> Self:
+    def __imul__(self, n: SupportsIndex) -> Self:
         # unlike a regular list *=, proxied __imul__ will generate unique
         # backing objects for each copy.  *= on proxied lists is a bit of
         # a stretch anyhow, and this interpretation of the __imul__ contract
@@ -1875,7 +1869,7 @@ class _AssociationSet(_AssociationSingleItem[_T], MutableSet[_T]):
             remover(member)
 
     def __ior__(  # type: ignore
-        self: Self, other: AbstractSet[_S]
+        self, other: AbstractSet[_S]
     ) -> MutableSet[Union[_T, _S]]:
         if not collections._set_binops_check_strict(self, other):
             raise NotImplementedError()
@@ -1903,7 +1897,7 @@ class _AssociationSet(_AssociationSingleItem[_T], MutableSet[_T]):
             for value in other:
                 self.discard(value)
 
-    def __isub__(self: Self, s: AbstractSet[Any]) -> Self:
+    def __isub__(self, s: AbstractSet[Any]) -> Self:
         if not collections._set_binops_check_strict(self, s):
             raise NotImplementedError()
         for value in s:
@@ -1927,7 +1921,7 @@ class _AssociationSet(_AssociationSingleItem[_T], MutableSet[_T]):
             for value in add:
                 self.add(value)
 
-    def __iand__(self: Self, s: AbstractSet[Any]) -> Self:
+    def __iand__(self, s: AbstractSet[Any]) -> Self:
         if not collections._set_binops_check_strict(self, s):
             raise NotImplementedError()
         want = self.intersection(s)
diff --git a/lib/sqlalchemy/ext/asyncio/base.py b/lib/sqlalchemy/ext/asyncio/base.py
index 765ee94a0..368c9b055 100644
--- a/lib/sqlalchemy/ext/asyncio/base.py
+++ b/lib/sqlalchemy/ext/asyncio/base.py
@@ -22,13 +22,13 @@ from typing import NoReturn
 from typing import Optional
 from typing import overload
 from typing import Tuple
-from typing import Type
 from typing import TypeVar
 import weakref
 
 from . import exc as async_exc
 from ... import util
 from ...util.typing import Literal
+from ...util.typing import Self
 
 _T = TypeVar("_T", bound=Any)
 _T_co = TypeVar("_T_co", bound=Any, covariant=True)
@@ -37,11 +37,6 @@ _T_co = TypeVar("_T_co", bound=Any, covariant=True)
 _PT = TypeVar("_PT", bound=Any)
 
 
-SelfReversibleProxy = TypeVar(
-    "SelfReversibleProxy", bound="ReversibleProxy[Any]"
-)
-
-
 class ReversibleProxy(Generic[_PT]):
     _proxy_objects: ClassVar[
         Dict[weakref.ref[Any], weakref.ref[ReversibleProxy[Any]]]
@@ -73,38 +68,36 @@ class ReversibleProxy(Generic[_PT]):
 
     @classmethod
     def _target_gced(
-        cls: Type[SelfReversibleProxy],
+        cls,
         ref: weakref.ref[_PT],
-        proxy_ref: Optional[weakref.ref[SelfReversibleProxy]] = None,
+        proxy_ref: Optional[weakref.ref[Self]] = None,
     ) -> None:
         cls._proxy_objects.pop(ref, None)
 
     @classmethod
-    def _regenerate_proxy_for_target(
-        cls: Type[SelfReversibleProxy], target: _PT
-    ) -> SelfReversibleProxy:
+    def _regenerate_proxy_for_target(cls, target: _PT) -> Self:
         raise NotImplementedError()
 
     @overload
     @classmethod
     def _retrieve_proxy_for_target(
-        cls: Type[SelfReversibleProxy],
+        cls,
         target: _PT,
         regenerate: Literal[True] = ...,
-    ) -> SelfReversibleProxy:
+    ) -> Self:
         ...
 
     @overload
     @classmethod
     def _retrieve_proxy_for_target(
-        cls: Type[SelfReversibleProxy], target: _PT, regenerate: bool = True
-    ) -> Optional[SelfReversibleProxy]:
+        cls, target: _PT, regenerate: bool = True
+    ) -> Optional[Self]:
         ...
 
     @classmethod
     def _retrieve_proxy_for_target(
-        cls: Type[SelfReversibleProxy], target: _PT, regenerate: bool = True
-    ) -> Optional[SelfReversibleProxy]:
+        cls, target: _PT, regenerate: bool = True
+    ) -> Optional[Self]:
         try:
             proxy_ref = cls._proxy_objects[weakref.ref(target)]
         except KeyError:
@@ -120,24 +113,17 @@ class ReversibleProxy(Generic[_PT]):
             return None
 
 
-SelfStartableContext = TypeVar(
-    "SelfStartableContext", bound="StartableContext[Any]"
-)
-
-
 class StartableContext(Awaitable[_T_co], abc.ABC):
     __slots__ = ()
 
     @abc.abstractmethod
-    async def start(
-        self: SelfStartableContext, is_ctxmanager: bool = False
-    ) -> _T_co:
+    async def start(self, is_ctxmanager: bool = False) -> _T_co:
         raise NotImplementedError()
 
     def __await__(self) -> Generator[Any, Any, _T_co]:
         return self.start().__await__()
 
-    async def __aenter__(self: SelfStartableContext) -> _T_co:
+    async def __aenter__(self) -> _T_co:
         return await self.start(is_ctxmanager=True)  # type: ignore
 
     @abc.abstractmethod
diff --git a/lib/sqlalchemy/ext/asyncio/result.py b/lib/sqlalchemy/ext/asyncio/result.py
index aaed638e3..3dcb1cfd0 100644
--- a/lib/sqlalchemy/ext/asyncio/result.py
+++ b/lib/sqlalchemy/ext/asyncio/result.py
@@ -30,6 +30,7 @@ from ...engine.row import RowMapping
 from ...sql.base import _generative
 from ...util.concurrency import greenlet_spawn
 from ...util.typing import Literal
+from ...util.typing import Self
 
 if TYPE_CHECKING:
     from ...engine import CursorResult
@@ -62,9 +63,6 @@ class AsyncCommon(FilterResult[_R]):
         return self._real_result.closed  # type: ignore
 
 
-SelfAsyncResult = TypeVar("SelfAsyncResult", bound="AsyncResult[Any]")
-
-
 class AsyncResult(_WithKeys, AsyncCommon[Row[_TP]]):
     """An asyncio wrapper around a :class:`_result.Result` object.
 
@@ -143,9 +141,7 @@ class AsyncResult(_WithKeys, AsyncCommon[Row[_TP]]):
         return self  # type: ignore
 
     @_generative
-    def unique(
-        self: SelfAsyncResult, strategy: Optional[_UniqueFilterType] = None
-    ) -> SelfAsyncResult:
+    def unique(self, strategy: Optional[_UniqueFilterType] = None) -> Self:
         """Apply unique filtering to the objects returned by this
         :class:`_asyncio.AsyncResult`.
 
@@ -156,9 +152,7 @@ class AsyncResult(_WithKeys, AsyncCommon[Row[_TP]]):
         self._unique_filter_state = (set(), strategy)
         return self
 
-    def columns(
-        self: SelfAsyncResult, *col_expressions: _KeyIndexType
-    ) -> SelfAsyncResult:
+    def columns(self, *col_expressions: _KeyIndexType) -> Self:
         r"""Establish the columns that should be returned in each row.
 
         Refer to :meth:`_engine.Result.columns` in the synchronous
@@ -501,11 +495,6 @@ class AsyncResult(_WithKeys, AsyncCommon[Row[_TP]]):
         return AsyncMappingResult(self._real_result)
 
 
-SelfAsyncScalarResult = TypeVar(
-    "SelfAsyncScalarResult", bound="AsyncScalarResult[Any]"
-)
-
-
 class AsyncScalarResult(AsyncCommon[_R]):
     """A wrapper for a :class:`_asyncio.AsyncResult` that returns scalar values
     rather than :class:`_row.Row` values.
@@ -537,9 +526,9 @@ class AsyncScalarResult(AsyncCommon[_R]):
         self._unique_filter_state = real_result._unique_filter_state
 
     def unique(
-        self: SelfAsyncScalarResult,
+        self,
         strategy: Optional[_UniqueFilterType] = None,
-    ) -> SelfAsyncScalarResult:
+    ) -> Self:
         """Apply unique filtering to the objects returned by this
         :class:`_asyncio.AsyncScalarResult`.
 
@@ -635,11 +624,6 @@ class AsyncScalarResult(AsyncCommon[_R]):
         return await greenlet_spawn(self._only_one_row, True, True, False)
 
 
-SelfAsyncMappingResult = TypeVar(
-    "SelfAsyncMappingResult", bound="AsyncMappingResult"
-)
-
-
 class AsyncMappingResult(_WithKeys, AsyncCommon[RowMapping]):
     """A wrapper for a :class:`_asyncio.AsyncResult` that returns dictionary
     values rather than :class:`_engine.Row` values.
@@ -668,9 +652,9 @@ class AsyncMappingResult(_WithKeys, AsyncCommon[RowMapping]):
             self._metadata = self._metadata._reduce([0])
 
     def unique(
-        self: SelfAsyncMappingResult,
+        self,
         strategy: Optional[_UniqueFilterType] = None,
-    ) -> SelfAsyncMappingResult:
+    ) -> Self:
         """Apply unique filtering to the objects returned by this
         :class:`_asyncio.AsyncMappingResult`.
 
@@ -680,9 +664,7 @@ class AsyncMappingResult(_WithKeys, AsyncCommon[RowMapping]):
         self._unique_filter_state = (set(), strategy)
         return self
 
-    def columns(
-        self: SelfAsyncMappingResult, *col_expressions: _KeyIndexType
-    ) -> SelfAsyncMappingResult:
+    def columns(self, *col_expressions: _KeyIndexType) -> Self:
         r"""Establish the columns that should be returned in each row."""
         return self._column_slices(col_expressions)
 
@@ -791,11 +773,6 @@ class AsyncMappingResult(_WithKeys, AsyncCommon[RowMapping]):
         return await greenlet_spawn(self._only_one_row, True, True, False)
 
 
-SelfAsyncTupleResult = TypeVar(
-    "SelfAsyncTupleResult", bound="AsyncTupleResult[Any]"
-)
-
-
 class AsyncTupleResult(AsyncCommon[_R], util.TypingOnly):
     """A :class:`_asyncio.AsyncResult` that's typed as returning plain
     Python tuples instead of rows.
diff --git a/lib/sqlalchemy/ext/horizontal_shard.py b/lib/sqlalchemy/ext/horizontal_shard.py
index 0bcc5628f..e1741fe52 100644
--- a/lib/sqlalchemy/ext/horizontal_shard.py
+++ b/lib/sqlalchemy/ext/horizontal_shard.py
@@ -49,6 +49,7 @@ from ..orm.session import _BindArguments
 from ..orm.session import _PKIdentityArgument
 from ..orm.session import Session
 from ..util.typing import Protocol
+from ..util.typing import Self
 
 if TYPE_CHECKING:
     from ..engine.base import Connection
@@ -72,7 +73,6 @@ __all__ = ["ShardedSession", "ShardedQuery"]
 
 _T = TypeVar("_T", bound=Any)
 
-SelfShardedQuery = TypeVar("SelfShardedQuery", bound="ShardedQuery[Any]")
 
 ShardIdentifier = str
 
@@ -118,9 +118,7 @@ class ShardedQuery(Query[_T]):
         self.execute_chooser = self.session.execute_chooser
         self._shard_id = None
 
-    def set_shard(
-        self: SelfShardedQuery, shard_id: ShardIdentifier
-    ) -> SelfShardedQuery:
+    def set_shard(self, shard_id: ShardIdentifier) -> Self:
         """Return a new query, limited to a single shard ID.
 
         All subsequent operations with the returned query will
diff --git a/lib/sqlalchemy/ext/hybrid.py b/lib/sqlalchemy/ext/hybrid.py
index 115c1cb85..d3bd69996 100644
--- a/lib/sqlalchemy/ext/hybrid.py
+++ b/lib/sqlalchemy/ext/hybrid.py
@@ -711,6 +711,7 @@ from ..util.typing import Concatenate
 from ..util.typing import Literal
 from ..util.typing import ParamSpec
 from ..util.typing import Protocol
+from ..util.typing import Self
 
 if TYPE_CHECKING:
     from ..orm.interfaces import MapperProperty
@@ -858,11 +859,6 @@ class hybrid_method(interfaces.InspectionAttrInfo, Generic[_P, _R]):
         return self
 
 
-Selfhybrid_property = TypeVar(
-    "Selfhybrid_property", bound="hybrid_property[Any]"
-)
-
-
 class hybrid_property(interfaces.InspectionAttrInfo, ORMDescriptor[_T]):
     """A decorator which allows definition of a Python descriptor with both
     instance-level and class-level behavior.
@@ -908,9 +904,7 @@ class hybrid_property(interfaces.InspectionAttrInfo, ORMDescriptor[_T]):
         util.update_wrapper(self, fget)
 
     @overload
-    def __get__(
-        self: Selfhybrid_property, instance: Any, owner: Literal[None]
-    ) -> Selfhybrid_property:
+    def __get__(self, instance: Any, owner: Literal[None]) -> Self:
         ...
 
     @overload
@@ -953,7 +947,7 @@ class hybrid_property(interfaces.InspectionAttrInfo, ORMDescriptor[_T]):
         return type(self)(**defaults)
 
     @property
-    def overrides(self: Selfhybrid_property) -> Selfhybrid_property:
+    def overrides(self) -> Self:
         """Prefix for a method that is overriding an existing attribute.
 
         The :attr:`.hybrid_property.overrides` accessor just returns