summaryrefslogtreecommitdiff
path: root/lib/sqlalchemy/sql
diff options
context:
space:
mode:
Diffstat (limited to 'lib/sqlalchemy/sql')
-rw-r--r--lib/sqlalchemy/sql/annotation.py138
-rw-r--r--lib/sqlalchemy/sql/base.py40
-rw-r--r--lib/sqlalchemy/sql/ddl.py14
-rw-r--r--lib/sqlalchemy/sql/dml.py58
-rw-r--r--lib/sqlalchemy/sql/elements.py38
-rw-r--r--lib/sqlalchemy/sql/lambdas.py4
-rw-r--r--lib/sqlalchemy/sql/schema.py15
-rw-r--r--lib/sqlalchemy/sql/selectable.py213
-rw-r--r--lib/sqlalchemy/sql/traversals.py39
-rw-r--r--lib/sqlalchemy/sql/type_api.py23
-rw-r--r--lib/sqlalchemy/sql/visitors.py11
11 files changed, 206 insertions, 387 deletions
diff --git a/lib/sqlalchemy/sql/annotation.py b/lib/sqlalchemy/sql/annotation.py
index 046c9bc79..7487e074c 100644
--- a/lib/sqlalchemy/sql/annotation.py
+++ b/lib/sqlalchemy/sql/annotation.py
@@ -39,6 +39,7 @@ from .visitors import ExternallyTraversible
from .visitors import InternalTraversal
from .. import util
from ..util.typing import Literal
+from ..util.typing import Self
if TYPE_CHECKING:
from .base import _EntityNamespace
@@ -49,11 +50,6 @@ _AnnotationDict = Mapping[str, Any]
EMPTY_ANNOTATIONS: util.immutabledict[str, Any] = util.EMPTY_DICT
-SelfSupportsAnnotations = TypeVar(
- "SelfSupportsAnnotations", bound="SupportsAnnotations"
-)
-
-
class SupportsAnnotations(ExternallyTraversible):
__slots__ = ()
@@ -63,17 +59,15 @@ class SupportsAnnotations(ExternallyTraversible):
_is_immutable: bool
- def _annotate(
- self: SelfSupportsAnnotations, values: _AnnotationDict
- ) -> SelfSupportsAnnotations:
+ def _annotate(self, values: _AnnotationDict) -> Self:
raise NotImplementedError()
@overload
def _deannotate(
- self: SelfSupportsAnnotations,
+ self,
values: Literal[None] = ...,
clone: bool = ...,
- ) -> SelfSupportsAnnotations:
+ ) -> Self:
...
@overload
@@ -117,46 +111,37 @@ class SupportsAnnotations(ExternallyTraversible):
)
-class SupportsCloneAnnotations(SupportsAnnotations):
- if not typing.TYPE_CHECKING:
- __slots__ = ()
+class SupportsWrappingAnnotations(SupportsAnnotations):
+ __slots__ = ()
- _clone_annotations_traverse_internals: _TraverseInternalsType = [
- ("_annotations", InternalTraversal.dp_annotations_key)
- ]
+ _constructor: Callable[..., SupportsWrappingAnnotations]
+
+ if TYPE_CHECKING:
+
+ @util.ro_non_memoized_property
+ def entity_namespace(self) -> _EntityNamespace:
+ ...
- def _annotate(
- self: SelfSupportsAnnotations, values: _AnnotationDict
- ) -> SelfSupportsAnnotations:
+ def _annotate(self, values: _AnnotationDict) -> Self:
"""return a copy of this ClauseElement with annotations
updated by the given dictionary.
"""
- new = self._clone()
- new._annotations = new._annotations.union(values)
- new.__dict__.pop("_annotations_cache_key", None)
- new.__dict__.pop("_generate_cache_key", None)
- return new
+ return Annotated._as_annotated_instance(self, values) # type: ignore
- def _with_annotations(
- self: SelfSupportsAnnotations, values: _AnnotationDict
- ) -> SelfSupportsAnnotations:
+ def _with_annotations(self, values: _AnnotationDict) -> Self:
"""return a copy of this ClauseElement with annotations
replaced by the given dictionary.
"""
- new = self._clone()
- new._annotations = util.immutabledict(values)
- new.__dict__.pop("_annotations_cache_key", None)
- new.__dict__.pop("_generate_cache_key", None)
- return new
+ return Annotated._as_annotated_instance(self, values) # type: ignore
@overload
def _deannotate(
- self: SelfSupportsAnnotations,
+ self,
values: Literal[None] = ...,
clone: bool = ...,
- ) -> SelfSupportsAnnotations:
+ ) -> Self:
...
@overload
@@ -180,52 +165,56 @@ class SupportsCloneAnnotations(SupportsAnnotations):
to remove.
"""
- if clone or self._annotations:
- # clone is used when we are also copying
- # the expression for a deep deannotation
- new = self._clone()
- new._annotations = util.immutabledict()
- new.__dict__.pop("_annotations_cache_key", None)
- return new
+ if clone:
+ s = self._clone()
+ return s
else:
return self
-class SupportsWrappingAnnotations(SupportsAnnotations):
- __slots__ = ()
+class SupportsCloneAnnotations(SupportsWrappingAnnotations):
+ # SupportsCloneAnnotations extends from SupportsWrappingAnnotations
+ # to support the structure of having the base ClauseElement
+ # be a subclass of SupportsWrappingAnnotations. Any ClauseElement
+ # subclass that wants to extend from SupportsCloneAnnotations
+ # will inherently also be subclassing SupportsWrappingAnnotations, so
+ # make that specific here.
- _constructor: Callable[..., SupportsWrappingAnnotations]
-
- if TYPE_CHECKING:
+ if not typing.TYPE_CHECKING:
+ __slots__ = ()
- @util.ro_non_memoized_property
- def entity_namespace(self) -> _EntityNamespace:
- ...
+ _clone_annotations_traverse_internals: _TraverseInternalsType = [
+ ("_annotations", InternalTraversal.dp_annotations_key)
+ ]
- def _annotate(
- self: SelfSupportsAnnotations, values: _AnnotationDict
- ) -> SelfSupportsAnnotations:
+ def _annotate(self, values: _AnnotationDict) -> Self:
"""return a copy of this ClauseElement with annotations
updated by the given dictionary.
"""
- return Annotated._as_annotated_instance(self, values) # type: ignore
+ new = self._clone()
+ new._annotations = new._annotations.union(values)
+ new.__dict__.pop("_annotations_cache_key", None)
+ new.__dict__.pop("_generate_cache_key", None)
+ return new
- def _with_annotations(
- self: SelfSupportsAnnotations, values: _AnnotationDict
- ) -> SelfSupportsAnnotations:
+ def _with_annotations(self, values: _AnnotationDict) -> Self:
"""return a copy of this ClauseElement with annotations
replaced by the given dictionary.
"""
- return Annotated._as_annotated_instance(self, values) # type: ignore
+ new = self._clone()
+ new._annotations = util.immutabledict(values)
+ new.__dict__.pop("_annotations_cache_key", None)
+ new.__dict__.pop("_generate_cache_key", None)
+ return new
@overload
def _deannotate(
- self: SelfSupportsAnnotations,
+ self,
values: Literal[None] = ...,
clone: bool = ...,
- ) -> SelfSupportsAnnotations:
+ ) -> Self:
...
@overload
@@ -249,16 +238,17 @@ class SupportsWrappingAnnotations(SupportsAnnotations):
to remove.
"""
- if clone:
- s = self._clone()
- return s
+ if clone or self._annotations:
+ # clone is used when we are also copying
+ # the expression for a deep deannotation
+ new = self._clone()
+ new._annotations = util.immutabledict()
+ new.__dict__.pop("_annotations_cache_key", None)
+ return new
else:
return self
-SelfAnnotated = TypeVar("SelfAnnotated", bound="Annotated")
-
-
class Annotated(SupportsAnnotations):
"""clones a SupportsAnnotations and applies an 'annotations' dictionary.
@@ -295,7 +285,7 @@ class Annotated(SupportsAnnotations):
__element: SupportsWrappingAnnotations
_hash: int
- def __new__(cls: Type[SelfAnnotated], *args: Any) -> SelfAnnotated:
+ def __new__(cls: Type[Self], *args: Any) -> Self:
return object.__new__(cls)
def __init__(
@@ -308,16 +298,12 @@ class Annotated(SupportsAnnotations):
self._annotations = util.immutabledict(values)
self._hash = hash(element)
- def _annotate(
- self: SelfAnnotated, values: _AnnotationDict
- ) -> SelfAnnotated:
+ def _annotate(self, values: _AnnotationDict) -> Self:
_values = self._annotations.union(values)
- new: SelfAnnotated = self._with_annotations(_values) # type: ignore
+ new = self._with_annotations(_values) # type: ignore
return new
- def _with_annotations(
- self: SelfAnnotated, values: _AnnotationDict
- ) -> SupportsAnnotations:
+ def _with_annotations(self, values: _AnnotationDict) -> Self:
clone = self.__class__.__new__(self.__class__)
clone.__dict__ = self.__dict__.copy()
clone.__dict__.pop("_annotations_cache_key", None)
@@ -327,10 +313,10 @@ class Annotated(SupportsAnnotations):
@overload
def _deannotate(
- self: SelfAnnotated,
+ self,
values: Literal[None] = ...,
clone: bool = ...,
- ) -> SelfAnnotated:
+ ) -> Self:
...
@overload
@@ -370,7 +356,7 @@ class Annotated(SupportsAnnotations):
def _constructor(self):
return self.__element._constructor
- def _clone(self: SelfAnnotated, **kw: Any) -> SelfAnnotated:
+ def _clone(self, **kw: Any) -> Self:
clone = self.__element._clone(**kw)
if clone is self.__element:
# detect immutable, don't change anything
diff --git a/lib/sqlalchemy/sql/base.py b/lib/sqlalchemy/sql/base.py
index 96ebc7824..1752a4dc1 100644
--- a/lib/sqlalchemy/sql/base.py
+++ b/lib/sqlalchemy/sql/base.py
@@ -134,6 +134,10 @@ def _is_has_entity_namespace(element: Any) -> TypeGuard[_HasEntityNamespace]:
return hasattr(element, "entity_namespace")
+# Remove when https://github.com/python/mypy/issues/14640 will be fixed
+_Self = TypeVar("_Self", bound=Any)
+
+
class Immutable:
"""mark a ClauseElement as 'immutable' when expressions are cloned.
@@ -157,7 +161,7 @@ class Immutable:
def params(self, *optionaldict, **kwargs):
raise NotImplementedError("Immutable objects do not support copying")
- def _clone(self: Self, **kw: Any) -> Self:
+ def _clone(self: _Self, **kw: Any) -> _Self:
return self
def _copy_internals(
@@ -222,7 +226,7 @@ _SelfGenerativeType = TypeVar("_SelfGenerativeType", bound="_GenerativeType")
class _GenerativeType(compat_typing.Protocol):
- def _generate(self: _SelfGenerativeType) -> _SelfGenerativeType:
+ def _generate(self) -> Self:
...
@@ -692,14 +696,11 @@ class CompileState:
return decorate
-SelfGenerative = TypeVar("SelfGenerative", bound="Generative")
-
-
class Generative(HasMemoized):
"""Provide a method-chaining pattern in conjunction with the
@_generative decorator."""
- def _generate(self: SelfGenerative) -> SelfGenerative:
+ def _generate(self) -> Self:
skip = self._memoized_keys
cls = self.__class__
s = cls.__new__(cls)
@@ -972,9 +973,6 @@ class ExecutableOption(HasCopyInternals):
return c
-SelfExecutable = TypeVar("SelfExecutable", bound="Executable")
-
-
class Executable(roles.StatementRole):
"""Mark a :class:`_expression.ClauseElement` as supporting execution.
@@ -1052,9 +1050,7 @@ class Executable(roles.StatementRole):
return self.__visit_name__
@_generative
- def options(
- self: SelfExecutable, *options: ExecutableOption
- ) -> SelfExecutable:
+ def options(self, *options: ExecutableOption) -> Self:
"""Apply options to this statement.
In the general sense, options are any kind of Python object
@@ -1090,9 +1086,7 @@ class Executable(roles.StatementRole):
return self
@_generative
- def _set_compile_options(
- self: SelfExecutable, compile_options: CacheableOptions
- ) -> SelfExecutable:
+ def _set_compile_options(self, compile_options: CacheableOptions) -> Self:
"""Assign the compile options to a new value.
:param compile_options: appropriate CacheableOptions structure
@@ -1103,9 +1097,7 @@ class Executable(roles.StatementRole):
return self
@_generative
- def _update_compile_options(
- self: SelfExecutable, options: CacheableOptions
- ) -> SelfExecutable:
+ def _update_compile_options(self, options: CacheableOptions) -> Self:
"""update the _compile_options with new keys."""
assert self._compile_options is not None
@@ -1114,10 +1106,10 @@ class Executable(roles.StatementRole):
@_generative
def _add_context_option(
- self: SelfExecutable,
+ self,
callable_: Callable[[CompileState], None],
cache_args: Any,
- ) -> SelfExecutable:
+ ) -> Self:
"""Add a context option to this statement.
These are callable functions that will
@@ -1133,7 +1125,7 @@ class Executable(roles.StatementRole):
@overload
def execution_options(
- self: SelfExecutable,
+ self,
*,
compiled_cache: Optional[CompiledCacheType] = ...,
logging_token: str = ...,
@@ -1151,15 +1143,15 @@ class Executable(roles.StatementRole):
is_delete_using: bool = ...,
is_update_from: bool = ...,
**opt: Any,
- ) -> SelfExecutable:
+ ) -> Self:
...
@overload
- def execution_options(self: SelfExecutable, **opt: Any) -> SelfExecutable:
+ def execution_options(self, **opt: Any) -> Self:
...
@_generative
- def execution_options(self: SelfExecutable, **kw: Any) -> SelfExecutable:
+ def execution_options(self, **kw: Any) -> Self:
"""Set non-SQL options for the statement which take effect during
execution.
diff --git a/lib/sqlalchemy/sql/ddl.py b/lib/sqlalchemy/sql/ddl.py
index 826ff687e..75fa0ba1c 100644
--- a/lib/sqlalchemy/sql/ddl.py
+++ b/lib/sqlalchemy/sql/ddl.py
@@ -32,6 +32,7 @@ from .. import exc
from .. import util
from ..util import topological
from ..util.typing import Protocol
+from ..util.typing import Self
if typing.TYPE_CHECKING:
from .compiler import Compiled
@@ -138,11 +139,6 @@ class DDLIf(typing.NamedTuple):
return True
-SelfExecutableDDLElement = typing.TypeVar(
- "SelfExecutableDDLElement", bound="ExecutableDDLElement"
-)
-
-
class ExecutableDDLElement(roles.DDLRole, Executable, BaseDDLElement):
"""Base class for standalone executable DDL expression constructs.
@@ -187,9 +183,7 @@ class ExecutableDDLElement(roles.DDLRole, Executable, BaseDDLElement):
)
@_generative
- def against(
- self: SelfExecutableDDLElement, target: SchemaItem
- ) -> SelfExecutableDDLElement:
+ def against(self, target: SchemaItem) -> Self:
"""Return a copy of this :class:`_schema.ExecutableDDLElement` which
will include the given target.
@@ -226,11 +220,11 @@ class ExecutableDDLElement(roles.DDLRole, Executable, BaseDDLElement):
@_generative
def execute_if(
- self: SelfExecutableDDLElement,
+ self,
dialect: Optional[str] = None,
callable_: Optional[DDLIfCallable] = None,
state: Optional[Any] = None,
- ) -> SelfExecutableDDLElement:
+ ) -> Self:
r"""Return a callable that will execute this
:class:`_ddl.ExecutableDDLElement` conditionally within an event
handler.
diff --git a/lib/sqlalchemy/sql/dml.py b/lib/sqlalchemy/sql/dml.py
index 1ab7f2ceb..9042fdff7 100644
--- a/lib/sqlalchemy/sql/dml.py
+++ b/lib/sqlalchemy/sql/dml.py
@@ -13,7 +13,6 @@ from __future__ import annotations
import collections.abc as collections_abc
import operator
-import typing
from typing import Any
from typing import cast
from typing import Dict
@@ -66,6 +65,7 @@ from .sqltypes import NullType
from .visitors import InternalTraversal
from .. import exc
from .. import util
+from ..util.typing import Self
from ..util.typing import TypeGuard
if TYPE_CHECKING:
@@ -378,9 +378,6 @@ class DeleteDMLState(DMLState):
self.is_multitable = ef
-SelfUpdateBase = typing.TypeVar("SelfUpdateBase", bound="UpdateBase")
-
-
class UpdateBase(
roles.DMLRole,
HasCTE,
@@ -438,9 +435,7 @@ class UpdateBase(
)
@_generative
- def with_dialect_options(
- self: SelfUpdateBase, **opt: Any
- ) -> SelfUpdateBase:
+ def with_dialect_options(self, **opt: Any) -> Self:
"""Add dialect options to this INSERT/UPDATE/DELETE object.
e.g.::
@@ -457,10 +452,10 @@ class UpdateBase(
@_generative
def return_defaults(
- self: SelfUpdateBase,
+ self,
*cols: _DMLColumnArgument,
supplemental_cols: Optional[Iterable[_DMLColumnArgument]] = None,
- ) -> SelfUpdateBase:
+ ) -> Self:
"""Make use of a :term:`RETURNING` clause for the purpose
of fetching server-side expressions and defaults, for supporting
backends only.
@@ -777,11 +772,11 @@ class UpdateBase(
@_generative
def with_hint(
- self: SelfUpdateBase,
+ self,
text: str,
selectable: Optional[_DMLTableArgument] = None,
dialect_name: str = "*",
- ) -> SelfUpdateBase:
+ ) -> Self:
"""Add a table hint for a single table to this
INSERT/UPDATE/DELETE statement.
@@ -902,9 +897,6 @@ class UpdateBase(
return meth(self)
-SelfValuesBase = typing.TypeVar("SelfValuesBase", bound="ValuesBase")
-
-
class ValuesBase(UpdateBase):
"""Supplies support for :meth:`.ValuesBase.values` to
INSERT and UPDATE constructs."""
@@ -950,13 +942,13 @@ class ValuesBase(UpdateBase):
},
)
def values(
- self: SelfValuesBase,
+ self,
*args: Union[
Dict[_DMLColumnArgument, Any],
Sequence[Any],
],
**kwargs: Any,
- ) -> SelfValuesBase:
+ ) -> Self:
r"""Specify a fixed VALUES clause for an INSERT statement, or the SET
clause for an UPDATE.
@@ -1133,9 +1125,6 @@ class ValuesBase(UpdateBase):
return self
-SelfInsert = typing.TypeVar("SelfInsert", bound="Insert")
-
-
class Insert(ValuesBase):
"""Represent an INSERT construct.
@@ -1182,7 +1171,7 @@ class Insert(ValuesBase):
super().__init__(table)
@_generative
- def inline(self: SelfInsert) -> SelfInsert:
+ def inline(self) -> Self:
"""Make this :class:`_expression.Insert` construct "inline" .
When set, no attempt will be made to retrieve the
@@ -1204,11 +1193,11 @@ class Insert(ValuesBase):
@_generative
def from_select(
- self: SelfInsert,
+ self,
names: List[str],
select: Selectable,
include_defaults: bool = True,
- ) -> SelfInsert:
+ ) -> Self:
"""Return a new :class:`_expression.Insert` construct which represents
an ``INSERT...FROM SELECT`` statement.
@@ -1378,17 +1367,12 @@ class ReturningInsert(Insert, TypedReturnsRows[_TP]):
"""
-SelfDMLWhereBase = typing.TypeVar("SelfDMLWhereBase", bound="DMLWhereBase")
-
-
class DMLWhereBase:
table: _DMLTableElement
_where_criteria: Tuple[ColumnElement[Any], ...] = ()
@_generative
- def where(
- self: SelfDMLWhereBase, *whereclause: _ColumnExpressionArgument[bool]
- ) -> SelfDMLWhereBase:
+ def where(self, *whereclause: _ColumnExpressionArgument[bool]) -> Self:
"""Return a new construct with the given expression(s) added to
its WHERE clause, joined to the existing clause via AND, if any.
@@ -1416,9 +1400,7 @@ class DMLWhereBase:
self._where_criteria += (where_criteria,)
return self
- def filter(
- self: SelfDMLWhereBase, *criteria: roles.ExpressionElementRole[Any]
- ) -> SelfDMLWhereBase:
+ def filter(self, *criteria: roles.ExpressionElementRole[Any]) -> Self:
"""A synonym for the :meth:`_dml.DMLWhereBase.where` method.
.. versionadded:: 1.4
@@ -1430,7 +1412,7 @@ class DMLWhereBase:
def _filter_by_zero(self) -> _DMLTableElement:
return self.table
- def filter_by(self: SelfDMLWhereBase, **kwargs: Any) -> SelfDMLWhereBase:
+ def filter_by(self, **kwargs: Any) -> Self:
r"""apply the given filtering criterion as a WHERE clause
to this select.
@@ -1461,9 +1443,6 @@ class DMLWhereBase:
)
-SelfUpdate = typing.TypeVar("SelfUpdate", bound="Update")
-
-
class Update(DMLWhereBase, ValuesBase):
"""Represent an Update construct.
@@ -1501,9 +1480,7 @@ class Update(DMLWhereBase, ValuesBase):
super().__init__(table)
@_generative
- def ordered_values(
- self: SelfUpdate, *args: Tuple[_DMLColumnArgument, Any]
- ) -> SelfUpdate:
+ def ordered_values(self, *args: Tuple[_DMLColumnArgument, Any]) -> Self:
"""Specify the VALUES clause of this UPDATE statement with an explicit
parameter ordering that will be maintained in the SET clause of the
resulting UPDATE statement.
@@ -1540,7 +1517,7 @@ class Update(DMLWhereBase, ValuesBase):
return self
@_generative
- def inline(self: SelfUpdate) -> SelfUpdate:
+ def inline(self) -> Self:
"""Make this :class:`_expression.Update` construct "inline" .
When set, SQL defaults present on :class:`_schema.Column`
@@ -1666,9 +1643,6 @@ class ReturningUpdate(Update, TypedReturnsRows[_TP]):
"""
-SelfDelete = typing.TypeVar("SelfDelete", bound="Delete")
-
-
class Delete(DMLWhereBase, UpdateBase):
"""Represent a DELETE construct.
diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index 70c65b5a1..37d53b30a 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -77,6 +77,7 @@ from .. import util
from ..util import HasMemoized_ro_memoized_attribute
from ..util import TypingOnly
from ..util.typing import Literal
+from ..util.typing import Self
if typing.TYPE_CHECKING:
from ._typing import _ColumnExpressionArgument
@@ -292,9 +293,6 @@ class CompilerElement(Visitable):
return str(self.compile())
-SelfClauseElement = TypeVar("SelfClauseElement", bound="ClauseElement")
-
-
@inspection._self_inspects
class ClauseElement(
SupportsWrappingAnnotations,
@@ -370,9 +368,7 @@ class ClauseElement(
def _from_objects(self) -> List[FromClause]:
return []
- def _set_propagate_attrs(
- self: SelfClauseElement, values: Mapping[str, Any]
- ) -> SelfClauseElement:
+ def _set_propagate_attrs(self, values: Mapping[str, Any]) -> Self:
# usually, self._propagate_attrs is empty here. one case where it's
# not is a subquery against ORM select, that is then pulled as a
# property of an aliased class. should all be good
@@ -382,7 +378,7 @@ class ClauseElement(
self._propagate_attrs = util.immutabledict(values)
return self
- def _clone(self: SelfClauseElement, **kw: Any) -> SelfClauseElement:
+ def _clone(self, **kw: Any) -> Self:
"""Create a shallow copy of this ClauseElement.
This method may be used by a generative API. Its also used as
@@ -509,10 +505,10 @@ class ClauseElement(
).scalar()
def unique_params(
- self: SelfClauseElement,
+ self,
__optionaldict: Optional[Dict[str, Any]] = None,
**kwargs: Any,
- ) -> SelfClauseElement:
+ ) -> Self:
"""Return a copy with :func:`_expression.bindparam` elements
replaced.
@@ -525,10 +521,10 @@ class ClauseElement(
return self._replace_params(True, __optionaldict, kwargs)
def params(
- self: SelfClauseElement,
+ self,
__optionaldict: Optional[Mapping[str, Any]] = None,
**kwargs: Any,
- ) -> SelfClauseElement:
+ ) -> Self:
"""Return a copy with :func:`_expression.bindparam` elements
replaced.
@@ -546,11 +542,11 @@ class ClauseElement(
return self._replace_params(False, __optionaldict, kwargs)
def _replace_params(
- self: SelfClauseElement,
+ self,
unique: bool,
optionaldict: Optional[Mapping[str, Any]],
kwargs: Dict[str, Any],
- ) -> SelfClauseElement:
+ ) -> Self:
if optionaldict:
kwargs.update(optionaldict)
@@ -1170,8 +1166,6 @@ class SQLColumnExpression(
_SQO = SQLCoreOperations
-SelfColumnElement = TypeVar("SelfColumnElement", bound="ColumnElement[Any]")
-
class ColumnElement(
roles.ColumnArgumentOrKeyRole,
@@ -1872,9 +1866,6 @@ class WrapsColumnExpression(ColumnElement[_T]):
return super()._proxy_key
-SelfBindParameter = TypeVar("SelfBindParameter", bound="BindParameter[Any]")
-
-
class BindParameter(roles.InElementRole, KeyedColumnElement[_T]):
r"""Represent a "bound expression".
@@ -2083,9 +2074,7 @@ class BindParameter(roles.InElementRole, KeyedColumnElement[_T]):
c.type = type_
return c
- def _clone(
- self: SelfBindParameter, maintain_key: bool = False, **kw: Any
- ) -> SelfBindParameter:
+ def _clone(self, maintain_key: bool = False, **kw: Any) -> Self:
c = ClauseElement._clone(self, **kw)
# ensure all the BindParameter objects stay in cloned set.
# in #7823, we changed "clone" so that a clone only keeps a reference
@@ -2176,9 +2165,6 @@ class TypeClause(DQLDMLClauseElement):
self.type = type_
-SelfTextClause = typing.TypeVar("SelfTextClause", bound="TextClause")
-
-
class TextClause(
roles.DDLConstraintColumnRole,
roles.DDLExpressionRole,
@@ -2266,10 +2252,10 @@ class TextClause(
@_generative
def bindparams(
- self: SelfTextClause,
+ self,
*binds: BindParameter[Any],
**names_to_values: Any,
- ) -> SelfTextClause:
+ ) -> Self:
"""Establish the values and/or types of bound parameters within
this :class:`_expression.TextClause` construct.
diff --git a/lib/sqlalchemy/sql/lambdas.py b/lib/sqlalchemy/sql/lambdas.py
index d737b1bcb..04bf86ee6 100644
--- a/lib/sqlalchemy/sql/lambdas.py
+++ b/lib/sqlalchemy/sql/lambdas.py
@@ -43,7 +43,7 @@ from .. import exc
from .. import inspection
from .. import util
from ..util.typing import Literal
-from ..util.typing import Self
+
if TYPE_CHECKING:
from .elements import BindParameter
@@ -368,7 +368,7 @@ class LambdaElement(elements.ClauseElement):
return expr
def _copy_internals(
- self: Self,
+ self,
clone: _CloneCallableType = _clone,
deferred_copy_internals: Optional[_CloneCallableType] = None,
**kw: Any,
diff --git a/lib/sqlalchemy/sql/schema.py b/lib/sqlalchemy/sql/schema.py
index 5083940f0..976432721 100644
--- a/lib/sqlalchemy/sql/schema.py
+++ b/lib/sqlalchemy/sql/schema.py
@@ -247,11 +247,6 @@ class SchemaItem(SchemaEventTarget, visitors.Visitable):
_use_schema_map = True
-SelfHasConditionalDDL = TypeVar(
- "SelfHasConditionalDDL", bound="HasConditionalDDL"
-)
-
-
class HasConditionalDDL:
"""define a class that includes the :meth:`.HasConditionalDDL.ddl_if`
method, allowing for conditional rendering of DDL.
@@ -266,11 +261,11 @@ class HasConditionalDDL:
_ddl_if: Optional[ddl.DDLIf] = None
def ddl_if(
- self: SelfHasConditionalDDL,
+ self,
dialect: Optional[str] = None,
callable_: Optional[ddl.DDLIfCallable] = None,
state: Optional[Any] = None,
- ) -> SelfHasConditionalDDL:
+ ) -> Self:
r"""apply a conditional DDL rule to this schema item.
These rules work in a similar manner to the
@@ -3707,7 +3702,7 @@ class FetchedValue(SchemaEventTarget):
def _copy(self) -> FetchedValue:
return FetchedValue(self.for_update)
- def _clone(self, for_update: bool) -> Any:
+ def _clone(self, for_update: bool) -> Self:
n = self.__class__.__new__(self.__class__)
n.__dict__.update(self.__dict__)
n.__dict__.pop("column", None)
@@ -3881,10 +3876,10 @@ class Constraint(DialectKWArgs, HasConditionalDDL, SchemaItem):
"The :meth:`_schema.Constraint.copy` method is deprecated "
"and will be removed in a future release.",
)
- def copy(self: Self, **kw: Any) -> Self:
+ def copy(self, **kw: Any) -> Self:
return self._copy(**kw) # type: ignore
- def _copy(self: Self, **kw: Any) -> Self:
+ def _copy(self, **kw: Any) -> Self:
raise NotImplementedError()
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index 47cf68357..21b83d556 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -16,7 +16,6 @@ from __future__ import annotations
import collections
from enum import Enum
import itertools
-import typing
from typing import AbstractSet
from typing import Any as TODO_Any
from typing import Any
@@ -286,9 +285,6 @@ class TypedReturnsRows(ExecutableReturnsRows, Generic[_TP]):
"""base for executable statements that return rows."""
-SelfSelectable = TypeVar("SelfSelectable", bound="Selectable")
-
-
class Selectable(ReturnsRows):
"""Mark a class as being selectable."""
@@ -321,9 +317,7 @@ class Selectable(ReturnsRows):
"functionality is available via the sqlalchemy.sql.visitors module.",
)
@util.preload_module("sqlalchemy.sql.util")
- def replace_selectable(
- self: SelfSelectable, old: FromClause, alias: Alias
- ) -> SelfSelectable:
+ def replace_selectable(self, old: FromClause, alias: Alias) -> Self:
"""Replace all occurrences of :class:`_expression.FromClause`
'old' with the given :class:`_expression.Alias`
object, returning a copy of this :class:`_expression.FromClause`.
@@ -373,9 +367,6 @@ class Selectable(ReturnsRows):
)
-SelfHasPrefixes = typing.TypeVar("SelfHasPrefixes", bound="HasPrefixes")
-
-
class HasPrefixes:
_prefixes: Tuple[Tuple[DQLDMLClauseElement, str], ...] = ()
@@ -390,10 +381,10 @@ class HasPrefixes:
":paramref:`.HasPrefixes.prefix_with.*prefixes`",
)
def prefix_with(
- self: SelfHasPrefixes,
+ self,
*prefixes: _TextCoercedExpressionArgument[Any],
dialect: str = "*",
- ) -> SelfHasPrefixes:
+ ) -> Self:
r"""Add one or more expressions following the statement keyword, i.e.
SELECT, INSERT, UPDATE, or DELETE. Generative.
@@ -428,9 +419,6 @@ class HasPrefixes:
return self
-SelfHasSuffixes = typing.TypeVar("SelfHasSuffixes", bound="HasSuffixes")
-
-
class HasSuffixes:
_suffixes: Tuple[Tuple[DQLDMLClauseElement, str], ...] = ()
@@ -445,10 +433,10 @@ class HasSuffixes:
":paramref:`.HasSuffixes.suffix_with.*suffixes`",
)
def suffix_with(
- self: SelfHasSuffixes,
+ self,
*suffixes: _TextCoercedExpressionArgument[Any],
dialect: str = "*",
- ) -> SelfHasSuffixes:
+ ) -> Self:
r"""Add one or more expressions following the statement as a whole.
This is used to support backend-specific suffix keywords on
@@ -478,9 +466,6 @@ class HasSuffixes:
return self
-SelfHasHints = typing.TypeVar("SelfHasHints", bound="HasHints")
-
-
class HasHints:
_hints: util.immutabledict[
Tuple[FromClause, str], str
@@ -492,9 +477,7 @@ class HasHints:
("_hints", InternalTraversal.dp_table_hint_list),
]
- def with_statement_hint(
- self: SelfHasHints, text: str, dialect_name: str = "*"
- ) -> SelfHasHints:
+ def with_statement_hint(self, text: str, dialect_name: str = "*") -> Self:
"""Add a statement hint to this :class:`_expression.Select` or
other selectable object.
@@ -522,11 +505,11 @@ class HasHints:
@_generative
def with_hint(
- self: SelfHasHints,
+ self,
selectable: _FromClauseArgument,
text: str,
dialect_name: str = "*",
- ) -> SelfHasHints:
+ ) -> Self:
r"""Add an indexing or other executional context hint for the given
selectable to this :class:`_expression.Select` or other selectable
object.
@@ -565,11 +548,11 @@ class HasHints:
return self._with_hint(selectable, text, dialect_name)
def _with_hint(
- self: SelfHasHints,
+ self,
selectable: Optional[_FromClauseArgument],
text: str,
dialect_name: str,
- ) -> SelfHasHints:
+ ) -> Self:
if selectable is None:
self._statement_hints += ((dialect_name, text),)
else:
@@ -584,9 +567,6 @@ class HasHints:
return self
-SelfFromClause = TypeVar("SelfFromClause", bound="FromClause")
-
-
class FromClause(roles.AnonymizedFromClauseRole, Selectable):
"""Represent an element that can be used within the ``FROM``
clause of a ``SELECT`` statement.
@@ -1017,7 +997,7 @@ class FromClause(roles.AnonymizedFromClauseRole, Selectable):
if TYPE_CHECKING:
def self_group(
- self: Self, against: Optional[OperatorType] = None
+ self, against: Optional[OperatorType] = None
) -> Union[FromGrouping, Self]:
...
@@ -1605,10 +1585,6 @@ class LateralFromClause(NamedFromClause):
"""mark a FROM clause as being able to render directly as LATERAL"""
-_SelfAliasedReturnsRows = TypeVar(
- "_SelfAliasedReturnsRows", bound="AliasedReturnsRows"
-)
-
# FromClause ->
# AliasedReturnsRows
# -> Alias only for FromClause
@@ -1636,12 +1612,12 @@ class AliasedReturnsRows(NoInit, NamedFromClause):
@classmethod
def _construct(
- cls: Type[_SelfAliasedReturnsRows],
+ cls,
selectable: Any,
*,
name: Optional[str] = None,
**kw: Any,
- ) -> _SelfAliasedReturnsRows:
+ ) -> Self:
obj = cls.__new__(cls)
obj._init(selectable, name=name, **kw)
return obj
@@ -2209,9 +2185,6 @@ class CTE(
return self._restates if self._restates is not None else self
-SelfHasCTE = typing.TypeVar("SelfHasCTE", bound="HasCTE")
-
-
class _CTEOpts(NamedTuple):
nesting: bool
@@ -2440,9 +2413,7 @@ class HasCTE(roles.HasCTERole, SelectsRows):
_independent_ctes_opts: Tuple[_CTEOpts, ...] = ()
@_generative
- def add_cte(
- self: SelfHasCTE, *ctes: CTE, nest_here: bool = False
- ) -> SelfHasCTE:
+ def add_cte(self, *ctes: CTE, nest_here: bool = False) -> Self:
r"""Add one or more :class:`_sql.CTE` constructs to this statement.
This method will associate the given :class:`_sql.CTE` constructs with
@@ -3184,9 +3155,6 @@ class ForUpdateArg(ClauseElement):
self.of = None
-SelfValues = typing.TypeVar("SelfValues", bound="Values")
-
-
class Values(roles.InElementRole, Generative, LateralFromClause):
"""Represent a ``VALUES`` construct that can be used as a FROM element
in a statement.
@@ -3232,9 +3200,7 @@ class Values(roles.InElementRole, Generative, LateralFromClause):
return [col.type for col in self._column_args]
@_generative
- def alias(
- self: SelfValues, name: Optional[str] = None, flat: bool = False
- ) -> SelfValues:
+ def alias(self, name: Optional[str] = None, flat: bool = False) -> Self:
"""Return a new :class:`_expression.Values`
construct that is a copy of this
@@ -3286,7 +3252,7 @@ class Values(roles.InElementRole, Generative, LateralFromClause):
return self
@_generative
- def data(self: SelfValues, values: List[Tuple[Any, ...]]) -> SelfValues:
+ def data(self, values: List[Tuple[Any, ...]]) -> Self:
"""Return a new :class:`_expression.Values` construct,
adding the given data to the data list.
@@ -3362,9 +3328,6 @@ class ScalarValues(roles.InElementRole, GroupedElement, ColumnElement[Any]):
return self
-SelfSelectBase = TypeVar("SelfSelectBase", bound=Any)
-
-
class SelectBase(
roles.SelectStatementRole,
roles.DMLSelectRole,
@@ -3508,9 +3471,7 @@ class SelectBase(
"""
raise NotImplementedError()
- def set_label_style(
- self: SelfSelectBase, style: SelectLabelStyle
- ) -> SelfSelectBase:
+ def set_label_style(self, style: SelectLabelStyle) -> Self:
"""Return a new selectable with the specified label style.
Implemented by subclasses.
@@ -3664,7 +3625,7 @@ class SelectBase(
self._ensure_disambiguated_names(), name=name
)
- def _ensure_disambiguated_names(self: SelfSelectBase) -> SelfSelectBase:
+ def _ensure_disambiguated_names(self) -> Self:
"""Ensure that the names generated by this selectbase will be
disambiguated in some way, if possible.
@@ -3739,7 +3700,7 @@ class SelectStatementGrouping(GroupedElement, SelectBase, Generic[_SB]):
def select_statement(self) -> _SB:
return self.element
- def self_group(self: Self, against: Optional[OperatorType] = None) -> Self:
+ def self_group(self, against: Optional[OperatorType] = None) -> Self:
...
return self
@@ -3790,11 +3751,6 @@ class SelectStatementGrouping(GroupedElement, SelectBase, Generic[_SB]):
return self.element._from_objects
-SelfGenerativeSelect = typing.TypeVar(
- "SelfGenerativeSelect", bound="GenerativeSelect"
-)
-
-
class GenerativeSelect(SelectBase, Generative):
"""Base class for SELECT statements where additional elements can be
added.
@@ -3824,14 +3780,14 @@ class GenerativeSelect(SelectBase, Generative):
@_generative
def with_for_update(
- self: SelfGenerativeSelect,
+ self,
*,
nowait: bool = False,
read: bool = False,
of: Optional[_ForUpdateOfArgument] = None,
skip_locked: bool = False,
key_share: bool = False,
- ) -> SelfGenerativeSelect:
+ ) -> Self:
"""Specify a ``FOR UPDATE`` clause for this
:class:`_expression.GenerativeSelect`.
@@ -3895,9 +3851,7 @@ class GenerativeSelect(SelectBase, Generative):
"""
return self._label_style
- def set_label_style(
- self: SelfGenerativeSelect, style: SelectLabelStyle
- ) -> SelfGenerativeSelect:
+ def set_label_style(self, style: SelectLabelStyle) -> Self:
"""Return a new selectable with the specified label style.
There are three "label styles" available,
@@ -4041,9 +3995,7 @@ class GenerativeSelect(SelectBase, Generative):
)
@_generative
- def limit(
- self: SelfGenerativeSelect, limit: _LimitOffsetType
- ) -> SelfGenerativeSelect:
+ def limit(self, limit: _LimitOffsetType) -> Self:
"""Return a new selectable with the given LIMIT criterion
applied.
@@ -4077,11 +4029,11 @@ class GenerativeSelect(SelectBase, Generative):
@_generative
def fetch(
- self: SelfGenerativeSelect,
+ self,
count: _LimitOffsetType,
with_ties: bool = False,
percent: bool = False,
- ) -> SelfGenerativeSelect:
+ ) -> Self:
"""Return a new selectable with the given FETCH FIRST criterion
applied.
@@ -4132,9 +4084,7 @@ class GenerativeSelect(SelectBase, Generative):
return self
@_generative
- def offset(
- self: SelfGenerativeSelect, offset: _LimitOffsetType
- ) -> SelfGenerativeSelect:
+ def offset(self, offset: _LimitOffsetType) -> Self:
"""Return a new selectable with the given OFFSET criterion
applied.
@@ -4165,10 +4115,10 @@ class GenerativeSelect(SelectBase, Generative):
@_generative
@util.preload_module("sqlalchemy.sql.util")
def slice(
- self: SelfGenerativeSelect,
+ self,
start: int,
stop: int,
- ) -> SelfGenerativeSelect:
+ ) -> Self:
"""Apply LIMIT / OFFSET to this statement based on a slice.
The start and stop indices behave like the argument to Python's
@@ -4216,13 +4166,13 @@ class GenerativeSelect(SelectBase, Generative):
@_generative
def order_by(
- self: SelfGenerativeSelect,
+ self,
__first: Union[
Literal[None, _NoArg.NO_ARG],
_ColumnExpressionOrStrLabelArgument[Any],
] = _NoArg.NO_ARG,
*clauses: _ColumnExpressionOrStrLabelArgument[Any],
- ) -> SelfGenerativeSelect:
+ ) -> Self:
r"""Return a new selectable with the given list of ORDER BY
criteria applied.
@@ -4261,13 +4211,13 @@ class GenerativeSelect(SelectBase, Generative):
@_generative
def group_by(
- self: SelfGenerativeSelect,
+ self,
__first: Union[
Literal[None, _NoArg.NO_ARG],
_ColumnExpressionOrStrLabelArgument[Any],
] = _NoArg.NO_ARG,
*clauses: _ColumnExpressionOrStrLabelArgument[Any],
- ) -> SelfGenerativeSelect:
+ ) -> Self:
r"""Return a new selectable with the given list of GROUP BY
criterion applied.
@@ -5049,9 +4999,6 @@ class _SelectFromElements:
yield element
-Self_MemoizedSelectEntities = TypeVar("Self_MemoizedSelectEntities", bound=Any)
-
-
class _MemoizedSelectEntities(
cache_key.HasCacheKey, traversals.HasCopyInternals, visitors.Traversible
):
@@ -5090,9 +5037,7 @@ class _MemoizedSelectEntities(
_annotations = util.EMPTY_DICT
- def _clone(
- self: Self_MemoizedSelectEntities, **kw: Any
- ) -> Self_MemoizedSelectEntities:
+ def _clone(self, **kw: Any) -> Self:
c = self.__class__.__new__(self.__class__)
c.__dict__ = {k: v for k, v in self.__dict__.items()}
@@ -5112,9 +5057,6 @@ class _MemoizedSelectEntities(
select_stmt._setup_joins = select_stmt._with_options = ()
-SelfSelect = typing.TypeVar("SelfSelect", bound="Select[Any]")
-
-
class Select(
HasPrefixes,
HasSuffixes,
@@ -5229,9 +5171,7 @@ class Select(
cols = list(elem._select_iterable)
return cols[0].type
- def filter(
- self: SelfSelect, *criteria: _ColumnExpressionArgument[bool]
- ) -> SelfSelect:
+ def filter(self, *criteria: _ColumnExpressionArgument[bool]) -> Self:
"""A synonym for the :meth:`_sql.Select.where` method."""
return self.where(*criteria)
@@ -5275,7 +5215,7 @@ class Select(
def scalar_subquery(self) -> ScalarSelect[Any]:
...
- def filter_by(self: SelfSelect, **kwargs: Any) -> SelfSelect:
+ def filter_by(self, **kwargs: Any) -> Self:
r"""apply the given filtering criterion as a WHERE clause
to this select.
@@ -5359,13 +5299,13 @@ class Select(
@_generative
def join(
- self: SelfSelect,
+ self,
target: _JoinTargetArgument,
onclause: Optional[_OnClauseArgument] = None,
*,
isouter: bool = False,
full: bool = False,
- ) -> SelfSelect:
+ ) -> Self:
r"""Create a SQL JOIN against this :class:`_expression.Select`
object's criterion
and apply generatively, returning the newly resulting
@@ -5442,13 +5382,13 @@ class Select(
return self
def outerjoin_from(
- self: SelfSelect,
+ self,
from_: _FromClauseArgument,
target: _JoinTargetArgument,
onclause: Optional[_OnClauseArgument] = None,
*,
full: bool = False,
- ) -> SelfSelect:
+ ) -> Self:
r"""Create a SQL LEFT OUTER JOIN against this
:class:`_expression.Select` object's criterion and apply generatively,
returning the newly resulting :class:`_expression.Select`.
@@ -5462,14 +5402,14 @@ class Select(
@_generative
def join_from(
- self: SelfSelect,
+ self,
from_: _FromClauseArgument,
target: _JoinTargetArgument,
onclause: Optional[_OnClauseArgument] = None,
*,
isouter: bool = False,
full: bool = False,
- ) -> SelfSelect:
+ ) -> Self:
r"""Create a SQL JOIN against this :class:`_expression.Select`
object's criterion
and apply generatively, returning the newly resulting
@@ -5537,12 +5477,12 @@ class Select(
return self
def outerjoin(
- self: SelfSelect,
+ self,
target: _JoinTargetArgument,
onclause: Optional[_OnClauseArgument] = None,
*,
full: bool = False,
- ) -> SelfSelect:
+ ) -> Self:
"""Create a left outer join.
Parameters are the same as that of :meth:`_expression.Select.join`.
@@ -6038,9 +5978,7 @@ class Select(
_whereclause = whereclause
@_generative
- def where(
- self: SelfSelect, *whereclause: _ColumnExpressionArgument[bool]
- ) -> SelfSelect:
+ def where(self, *whereclause: _ColumnExpressionArgument[bool]) -> Self:
"""Return a new :func:`_expression.select` construct with
the given expression added to
its WHERE clause, joined to the existing clause via AND, if any.
@@ -6057,9 +5995,7 @@ class Select(
return self
@_generative
- def having(
- self: SelfSelect, *having: _ColumnExpressionArgument[bool]
- ) -> SelfSelect:
+ def having(self, *having: _ColumnExpressionArgument[bool]) -> Self:
"""Return a new :func:`_expression.select` construct with
the given expression added to
its HAVING clause, joined to the existing clause via AND, if any.
@@ -6074,9 +6010,7 @@ class Select(
return self
@_generative
- def distinct(
- self: SelfSelect, *expr: _ColumnExpressionArgument[Any]
- ) -> SelfSelect:
+ def distinct(self, *expr: _ColumnExpressionArgument[Any]) -> Self:
r"""Return a new :func:`_expression.select` construct which
will apply DISTINCT to its columns clause.
@@ -6098,9 +6032,7 @@ class Select(
return self
@_generative
- def select_from(
- self: SelfSelect, *froms: _FromClauseArgument
- ) -> SelfSelect:
+ def select_from(self, *froms: _FromClauseArgument) -> Self:
r"""Return a new :func:`_expression.select` construct with the
given FROM expression(s)
merged into its list of FROM objects.
@@ -6145,9 +6077,9 @@ class Select(
@_generative
def correlate(
- self: SelfSelect,
+ self,
*fromclauses: Union[Literal[None, False], _FromClauseArgument],
- ) -> SelfSelect:
+ ) -> Self:
r"""Return a new :class:`_expression.Select`
which will correlate the given FROM
clauses to that of an enclosing :class:`_expression.Select`.
@@ -6218,9 +6150,9 @@ class Select(
@_generative
def correlate_except(
- self: SelfSelect,
+ self,
*fromclauses: Union[Literal[None, False], _FromClauseArgument],
- ) -> SelfSelect:
+ ) -> Self:
r"""Return a new :class:`_expression.Select`
which will omit the given FROM
clauses from the auto-correlation process.
@@ -6402,7 +6334,7 @@ class Select(
)
def self_group(
- self: Self, against: Optional[OperatorType] = None
+ self, against: Optional[OperatorType] = None
) -> Union[SelectStatementGrouping[Self], Self]:
...
"""Return a 'grouping' construct as per the
@@ -6530,11 +6462,6 @@ class Select(
return CompoundSelect._create_intersect_all(self, *other)
-SelfScalarSelect = typing.TypeVar(
- "SelfScalarSelect", bound="ScalarSelect[Any]"
-)
-
-
class ScalarSelect(
roles.InElementRole, Generative, GroupedElement, ColumnElement[_T]
):
@@ -6592,9 +6519,7 @@ class ScalarSelect(
c = columns
@_generative
- def where(
- self: SelfScalarSelect, crit: _ColumnExpressionArgument[bool]
- ) -> SelfScalarSelect:
+ def where(self, crit: _ColumnExpressionArgument[bool]) -> Self:
"""Apply a WHERE clause to the SELECT statement referred to
by this :class:`_expression.ScalarSelect`.
@@ -6627,9 +6552,9 @@ class ScalarSelect(
@_generative
def correlate(
- self: SelfScalarSelect,
+ self,
*fromclauses: Union[Literal[None, False], _FromClauseArgument],
- ) -> SelfScalarSelect:
+ ) -> Self:
r"""Return a new :class:`_expression.ScalarSelect`
which will correlate the given FROM
clauses to that of an enclosing :class:`_expression.Select`.
@@ -6663,9 +6588,9 @@ class ScalarSelect(
@_generative
def correlate_except(
- self: SelfScalarSelect,
+ self,
*fromclauses: Union[Literal[None, False], _FromClauseArgument],
- ) -> SelfScalarSelect:
+ ) -> Self:
r"""Return a new :class:`_expression.ScalarSelect`
which will omit the given FROM
clauses from the auto-correlation process.
@@ -6700,9 +6625,6 @@ class ScalarSelect(
return self
-SelfExists = TypeVar("SelfExists", bound="Exists")
-
-
class Exists(UnaryExpression[bool]):
"""Represent an ``EXISTS`` clause.
@@ -6778,9 +6700,9 @@ class Exists(UnaryExpression[bool]):
return Select(self)
def correlate(
- self: SelfExists,
+ self,
*fromclauses: Union[Literal[None, False], _FromClauseArgument],
- ) -> SelfExists:
+ ) -> Self:
"""Apply correlation to the subquery noted by this
:class:`_sql.Exists`.
@@ -6796,9 +6718,9 @@ class Exists(UnaryExpression[bool]):
return e
def correlate_except(
- self: SelfExists,
+ self,
*fromclauses: Union[Literal[None, False], _FromClauseArgument],
- ) -> SelfExists:
+ ) -> Self:
"""Apply correlation to the subquery noted by this
:class:`_sql.Exists`.
@@ -6814,7 +6736,7 @@ class Exists(UnaryExpression[bool]):
)
return e
- def select_from(self: SelfExists, *froms: FromClause) -> SelfExists:
+ def select_from(self, *froms: FromClause) -> Self:
"""Return a new :class:`_expression.Exists` construct,
applying the given
expression to the :meth:`_expression.Select.select_from`
@@ -6831,9 +6753,7 @@ class Exists(UnaryExpression[bool]):
e.element = self._regroup(lambda element: element.select_from(*froms))
return e
- def where(
- self: SelfExists, *clause: _ColumnExpressionArgument[bool]
- ) -> SelfExists:
+ def where(self, *clause: _ColumnExpressionArgument[bool]) -> Self:
"""Return a new :func:`_expression.exists` construct with the
given expression added to
its WHERE clause, joined to the existing clause via AND, if any.
@@ -6850,9 +6770,6 @@ class Exists(UnaryExpression[bool]):
return e
-SelfTextualSelect = typing.TypeVar("SelfTextualSelect", bound="TextualSelect")
-
-
class TextualSelect(SelectBase, Executable, Generative):
"""Wrap a :class:`_expression.TextClause` construct within a
:class:`_expression.SelectBase`
@@ -6948,10 +6865,10 @@ class TextualSelect(SelectBase, Executable, Generative):
@_generative
def bindparams(
- self: SelfTextualSelect,
+ self,
*binds: BindParameter[Any],
**bind_as_values: Any,
- ) -> SelfTextualSelect:
+ ) -> Self:
self.element = self.element.bindparams(*binds, **bind_as_values)
return self
diff --git a/lib/sqlalchemy/sql/traversals.py b/lib/sqlalchemy/sql/traversals.py
index 983cd8ebd..4b55560ec 100644
--- a/lib/sqlalchemy/sql/traversals.py
+++ b/lib/sqlalchemy/sql/traversals.py
@@ -23,7 +23,6 @@ from typing import Optional
from typing import Set
from typing import Tuple
from typing import Type
-from typing import TypeVar
from . import operators
from .cache_key import HasCacheKey
@@ -34,6 +33,8 @@ from .visitors import HasTraversalDispatch
from .visitors import HasTraverseInternals
from .. import util
from ..util import langhelpers
+from ..util.typing import Self
+
SKIP_TRAVERSE = util.symbol("skip_traverse")
COMPARE_FAILED = False
@@ -68,9 +69,6 @@ def _preconfigure_traversals(target_hierarchy: Type[Any]) -> None:
)
-SelfHasShallowCopy = TypeVar("SelfHasShallowCopy", bound="HasShallowCopy")
-
-
class HasShallowCopy(HasTraverseInternals):
"""attribute-wide operations that are useful for classes that use
__slots__ and therefore can't operate on their attributes in a dictionary.
@@ -82,9 +80,7 @@ class HasShallowCopy(HasTraverseInternals):
if typing.TYPE_CHECKING:
- def _generated_shallow_copy_traversal(
- self: SelfHasShallowCopy, other: SelfHasShallowCopy
- ) -> None:
+ def _generated_shallow_copy_traversal(self, other: Self) -> None:
...
def _generated_shallow_from_dict_traversal(
@@ -97,10 +93,10 @@ class HasShallowCopy(HasTraverseInternals):
@classmethod
def _generate_shallow_copy(
- cls: Type[SelfHasShallowCopy],
+ cls,
internal_dispatch: _TraverseInternalsType,
method_name: str,
- ) -> Callable[[SelfHasShallowCopy, SelfHasShallowCopy], None]:
+ ) -> Callable[[Self, Self], None]:
code = "\n".join(
f" other.{attrname} = self.{attrname}"
for attrname, _ in internal_dispatch
@@ -110,10 +106,10 @@ class HasShallowCopy(HasTraverseInternals):
@classmethod
def _generate_shallow_to_dict(
- cls: Type[SelfHasShallowCopy],
+ cls,
internal_dispatch: _TraverseInternalsType,
method_name: str,
- ) -> Callable[[SelfHasShallowCopy], Dict[str, Any]]:
+ ) -> Callable[[Self], Dict[str, Any]]:
code = ",\n".join(
f" '{attrname}': self.{attrname}"
for attrname, _ in internal_dispatch
@@ -123,10 +119,10 @@ class HasShallowCopy(HasTraverseInternals):
@classmethod
def _generate_shallow_from_dict(
- cls: Type[SelfHasShallowCopy],
+ cls,
internal_dispatch: _TraverseInternalsType,
method_name: str,
- ) -> Callable[[SelfHasShallowCopy, Dict[str, Any]], None]:
+ ) -> Callable[[Self, Dict[str, Any]], None]:
code = "\n".join(
f" self.{attrname} = d['{attrname}']"
for attrname, _ in internal_dispatch
@@ -169,12 +165,10 @@ class HasShallowCopy(HasTraverseInternals):
cls._generated_shallow_to_dict_traversal = shallow_to_dict # type: ignore # noqa: E501
return shallow_to_dict(self)
- def _shallow_copy_to(
- self: SelfHasShallowCopy, other: SelfHasShallowCopy
- ) -> None:
+ def _shallow_copy_to(self, other: Self) -> None:
cls = self.__class__
- shallow_copy: Callable[[SelfHasShallowCopy, SelfHasShallowCopy], None]
+ shallow_copy: Callable[[Self, Self], None]
try:
shallow_copy = cls.__dict__["_generated_shallow_copy_traversal"]
except KeyError:
@@ -185,18 +179,13 @@ class HasShallowCopy(HasTraverseInternals):
cls._generated_shallow_copy_traversal = shallow_copy # type: ignore # noqa: E501
shallow_copy(self, other)
- def _clone(self: SelfHasShallowCopy, **kw: Any) -> SelfHasShallowCopy:
+ def _clone(self, **kw: Any) -> Self:
"""Create a shallow copy"""
c = self.__class__.__new__(self.__class__)
self._shallow_copy_to(c)
return c
-SelfGenerativeOnTraversal = TypeVar(
- "SelfGenerativeOnTraversal", bound="GenerativeOnTraversal"
-)
-
-
class GenerativeOnTraversal(HasShallowCopy):
"""Supplies Generative behavior but making use of traversals to shallow
copy.
@@ -210,9 +199,7 @@ class GenerativeOnTraversal(HasShallowCopy):
__slots__ = ()
- def _generate(
- self: SelfGenerativeOnTraversal,
- ) -> SelfGenerativeOnTraversal:
+ def _generate(self) -> Self:
cls = self.__class__
s = cls.__new__(cls)
self._shallow_copy_to(s)
diff --git a/lib/sqlalchemy/sql/type_api.py b/lib/sqlalchemy/sql/type_api.py
index db89bfe8b..af7ed21c4 100644
--- a/lib/sqlalchemy/sql/type_api.py
+++ b/lib/sqlalchemy/sql/type_api.py
@@ -37,6 +37,7 @@ from .visitors import Visitable
from .. import exc
from .. import util
from ..util.typing import Protocol
+from ..util.typing import Self
from ..util.typing import TypedDict
from ..util.typing import TypeGuard
@@ -68,7 +69,6 @@ _CT = TypeVar("_CT", bound=Any)
_MatchedOnType = Union["GenericProtocol[Any]", NewType, Type[Any]]
# replace with pep-673 when applicable
-SelfTypeEngine = typing.TypeVar("SelfTypeEngine", bound="TypeEngine[Any]")
class _LiteralProcessorType(Protocol[_T_co]):
@@ -293,7 +293,7 @@ class TypeEngine(Visitable, Generic[_T]):
str, TypeEngine[Any]
] = util.EMPTY_DICT
- def evaluates_none(self: SelfTypeEngine) -> SelfTypeEngine:
+ def evaluates_none(self) -> Self:
"""Return a copy of this type which has the
:attr:`.should_evaluate_none` flag set to True.
@@ -345,7 +345,7 @@ class TypeEngine(Visitable, Generic[_T]):
typ.should_evaluate_none = True
return typ
- def copy(self: SelfTypeEngine, **kw: Any) -> SelfTypeEngine:
+ def copy(self, **kw: Any) -> Self:
return self.adapt(self.__class__)
def compare_against_backend(
@@ -646,10 +646,10 @@ class TypeEngine(Visitable, Generic[_T]):
raise NotImplementedError()
def with_variant(
- self: SelfTypeEngine,
+ self,
type_: _TypeEngineArgument[Any],
*dialect_names: str,
- ) -> SelfTypeEngine:
+ ) -> Self:
r"""Produce a copy of this type object that will utilize the given
type when applied to the dialect of the given name.
@@ -711,9 +711,7 @@ class TypeEngine(Visitable, Generic[_T]):
)
return new_type
- def _resolve_for_literal(
- self: SelfTypeEngine, value: Any
- ) -> SelfTypeEngine:
+ def _resolve_for_literal(self, value: Any) -> Self:
"""adjust this type given a literal Python value that will be
stored in a bound parameter.
@@ -731,11 +729,11 @@ class TypeEngine(Visitable, Generic[_T]):
return self
def _resolve_for_python_type(
- self: SelfTypeEngine,
+ self,
python_type: Type[Any],
matched_on: _MatchedOnType,
matched_on_flattened: Type[Any],
- ) -> Optional[SelfTypeEngine]:
+ ) -> Optional[Self]:
"""given a Python type (e.g. ``int``, ``str``, etc. ) return an
instance of this :class:`.TypeEngine` that's appropriate for this type.
@@ -1515,9 +1513,6 @@ class NativeForEmulated(TypeEngineMixin):
# ...
-SelfTypeDecorator = TypeVar("SelfTypeDecorator", bound="TypeDecorator[Any]")
-
-
class TypeDecorator(SchemaEventTarget, ExternalType, TypeEngine[_T]):
"""Allows the creation of types which add additional functionality
to an existing type.
@@ -2229,7 +2224,7 @@ class TypeDecorator(SchemaEventTarget, ExternalType, TypeEngine[_T]):
"""
return self
- def copy(self: SelfTypeDecorator, **kw: Any) -> SelfTypeDecorator:
+ def copy(self, **kw: Any) -> Self:
"""Produce a copy of this :class:`.TypeDecorator` instance.
This is a shallow copy and is provided to fulfill part of
diff --git a/lib/sqlalchemy/sql/visitors.py b/lib/sqlalchemy/sql/visitors.py
index 1a6719cce..79a163f67 100644
--- a/lib/sqlalchemy/sql/visitors.py
+++ b/lib/sqlalchemy/sql/visitors.py
@@ -594,11 +594,6 @@ _dispatch_lookup = HasTraversalDispatch._dispatch_lookup
_generate_traversal_dispatch()
-SelfExternallyTraversible = TypeVar(
- "SelfExternallyTraversible", bound="ExternallyTraversible"
-)
-
-
class ExternallyTraversible(HasTraverseInternals, Visitable):
__slots__ = ()
@@ -606,9 +601,7 @@ class ExternallyTraversible(HasTraverseInternals, Visitable):
if typing.TYPE_CHECKING:
- def _annotate(
- self: SelfExternallyTraversible, values: _AnnotationDict
- ) -> SelfExternallyTraversible:
+ def _annotate(self, values: _AnnotationDict) -> Self:
...
def get_children(
@@ -616,7 +609,7 @@ class ExternallyTraversible(HasTraverseInternals, Visitable):
) -> Iterable[ExternallyTraversible]:
...
- def _clone(self: Self, **kw: Any) -> Self:
+ def _clone(self, **kw: Any) -> Self:
"""clone this element"""
raise NotImplementedError()
View Lorry Controller Status