--- /dev/null
+.. change::
+ :tags: typing, bug
+ :tickets: 9254
+
+ Remove ``typing.Self`` workaround, now using :pep:`673` for most methods
+ that return ``Self``. Pull request courtesy Yurii Karabas.
# mypy: ignore-errors
-import typing
-
from ... import exc
from ... import util
from ...sql.base import _exclusive_against
from ...sql.dml import Insert as StandardInsert
from ...sql.elements import ClauseElement
from ...sql.expression import alias
+from ...util.typing import Self
__all__ = ("Insert", "insert")
return Insert(table)
-SelfInsert = typing.TypeVar("SelfInsert", bound="Insert")
-
-
class Insert(StandardInsert):
"""MySQL-specific implementation of INSERT.
"has an ON DUPLICATE KEY clause present"
},
)
- def on_duplicate_key_update(self: SelfInsert, *args, **kw) -> SelfInsert:
+ def on_duplicate_key_update(self, *args, **kw) -> Self:
r"""
Specifies the ON DUPLICATE KEY UPDATE clause.
# mypy: ignore-errors
-import typing
-
from ... import exc
from ... import util
from ...sql import coercions
from ...sql import roles
from ...sql.base import _generative
from ...sql.base import Generative
-
-
-Selfmatch = typing.TypeVar("Selfmatch", bound="match")
+from ...util.typing import Self
class match(Generative, elements.BinaryExpression):
super().__init__(left, against, operators.match_op, modifiers=flags)
@_generative
- def in_boolean_mode(self: Selfmatch) -> Selfmatch:
+ def in_boolean_mode(self) -> Self:
"""Apply the "IN BOOLEAN MODE" modifier to the MATCH expression.
:return: a new :class:`_mysql.match` instance with modifications
return self
@_generative
- def in_natural_language_mode(self: Selfmatch) -> Selfmatch:
+ def in_natural_language_mode(self) -> Self:
"""Apply the "IN NATURAL LANGUAGE MODE" modifier to the MATCH
expression.
return self
@_generative
- def with_query_expansion(self: Selfmatch) -> Selfmatch:
+ def with_query_expansion(self) -> Self:
"""Apply the "WITH QUERY EXPANSION" modifier to the MATCH expression.
:return: a new :class:`_mysql.match` instance with modifications
# the MIT License: https://www.opensource.org/licenses/mit-license.php
# mypy: ignore-errors
-import typing
-
from . import ext
from ... import util
from ...sql import coercions
from ...sql.dml import Insert as StandardInsert
from ...sql.elements import ClauseElement
from ...sql.expression import alias
+from ...util.typing import Self
__all__ = ("Insert", "insert")
return Insert(table)
-SelfInsert = typing.TypeVar("SelfInsert", bound="Insert")
-
-
class Insert(StandardInsert):
"""PostgreSQL-specific implementation of INSERT.
@_generative
@_on_conflict_exclusive
def on_conflict_do_update(
- self: SelfInsert,
+ self,
constraint=None,
index_elements=None,
index_where=None,
set_=None,
where=None,
- ) -> SelfInsert:
+ ) -> Self:
r"""
Specifies a DO UPDATE SET action for ON CONFLICT clause.
@_generative
@_on_conflict_exclusive
def on_conflict_do_nothing(
- self: SelfInsert,
+ self,
constraint=None,
index_elements=None,
index_where=None,
- ) -> SelfInsert:
+ ) -> Self:
"""
Specifies a DO NOTHING action for ON CONFLICT clause.
# mypy: ignore-errors
-import typing
-
from ... import util
from ...sql import coercions
from ...sql import roles
from ...sql.dml import Insert as StandardInsert
from ...sql.elements import ClauseElement
from ...sql.expression import alias
+from ...util.typing import Self
__all__ = ("Insert", "insert")
return Insert(table)
-SelfInsert = typing.TypeVar("SelfInsert", bound="Insert")
-
-
class Insert(StandardInsert):
"""SQLite-specific implementation of INSERT.
@_generative
@_on_conflict_exclusive
def on_conflict_do_update(
- self: SelfInsert,
+ self,
index_elements=None,
index_where=None,
set_=None,
where=None,
- ) -> SelfInsert:
+ ) -> Self:
r"""
Specifies a DO UPDATE SET action for ON CONFLICT clause.
@_generative
@_on_conflict_exclusive
def on_conflict_do_nothing(
- self: SelfInsert, index_elements=None, index_where=None
- ) -> SelfInsert:
+ self, index_elements=None, index_where=None
+ ) -> Self:
"""
Specifies a DO NOTHING action for ON CONFLICT clause.
from ..sql.type_api import TypeEngine
from ..util import compat
from ..util.typing import Literal
+from ..util.typing import Self
_UNPICKLED = util.symbol("unpickled")
_NO_RESULT_METADATA = _NoResultMetaData()
-SelfCursorResult = TypeVar("SelfCursorResult", bound="CursorResult[Any]")
-
-
def null_dml_result() -> IteratorResult[Any]:
it: IteratorResult[Any] = IteratorResult(_NoResultMetaData(), iter([]))
it._soft_close()
self._soft_close(hard=True)
@_generative
- def yield_per(self: SelfCursorResult, num: int) -> SelfCursorResult:
+ def yield_per(self, num: int) -> Self:
self._yield_per = num
self.cursor_strategy.yield_per(self, self.cursor, num)
return self
from ..util import HasMemoized_ro_memoized_attribute
from ..util._has_cy import HAS_CYEXTENSION
from ..util.typing import Literal
+from ..util.typing import Self
if typing.TYPE_CHECKING or not HAS_CYEXTENSION:
from ._py_row import tuplegetter as tuplegetter
_NO_ROW = _NoRow._NO_ROW
-SelfResultInternal = TypeVar("SelfResultInternal", bound="ResultInternal[Any]")
-
class ResultInternal(InPlaceGenerative, Generic[_R]):
__slots__ = ()
return row
@_generative
- def _column_slices(
- self: SelfResultInternal, indexes: Sequence[_KeyIndexType]
- ) -> SelfResultInternal:
+ def _column_slices(self, indexes: Sequence[_KeyIndexType]) -> Self:
real_result = (
self._real_result
if self._real_result
return self._metadata.keys
-SelfResult = TypeVar("SelfResult", bound="Result[Any]")
-
-
class Result(_WithKeys, ResultInternal[Row[_TP]]):
"""Represent a set of database results.
def __init__(self, cursor_metadata: ResultMetaData):
self._metadata = cursor_metadata
- def __enter__(self: SelfResult) -> SelfResult:
+ def __enter__(self) -> Self:
return self
def __exit__(self, type_: Any, value: Any, traceback: Any) -> None:
raise NotImplementedError()
@_generative
- def yield_per(self: SelfResult, num: int) -> SelfResult:
+ def yield_per(self, num: int) -> Self:
"""Configure the row-fetching strategy to fetch ``num`` rows at a time.
This impacts the underlying behavior of the result when iterating over
return self
@_generative
- def unique(
- self: SelfResult, strategy: Optional[_UniqueFilterType] = None
- ) -> SelfResult:
+ def unique(self, strategy: Optional[_UniqueFilterType] = None) -> Self:
"""Apply unique filtering to the objects returned by this
:class:`_engine.Result`.
self._unique_filter_state = (set(), strategy)
return self
- def columns(
- self: SelfResultInternal, *col_expressions: _KeyIndexType
- ) -> SelfResultInternal:
+ def columns(self, *col_expressions: _KeyIndexType) -> Self:
r"""Establish the columns that should be returned in each row.
This method may be used to limit the columns returned as well
return MergedResult(self._metadata, (self,) + others)
-SelfFilterResult = TypeVar("SelfFilterResult", bound="FilterResult[Any]")
-
-
class FilterResult(ResultInternal[_R]):
"""A wrapper for a :class:`_engine.Result` that returns objects other than
:class:`_engine.Row` objects, such as dictionaries or scalar objects.
_real_result: Result[Any]
- def __enter__(self: SelfFilterResult) -> SelfFilterResult:
+ def __enter__(self) -> Self:
return self
def __exit__(self, type_: Any, value: Any, traceback: Any) -> None:
self._real_result.__exit__(type_, value, traceback)
@_generative
- def yield_per(self: SelfFilterResult, num: int) -> SelfFilterResult:
+ def yield_per(self, num: int) -> Self:
"""Configure the row-fetching strategy to fetch ``num`` rows at a time.
The :meth:`_engine.FilterResult.yield_per` method is a pass through
return self._real_result._fetchmany_impl(size=size)
-SelfScalarResult = TypeVar("SelfScalarResult", bound="ScalarResult[Any]")
-
-
class ScalarResult(FilterResult[_R]):
"""A wrapper for a :class:`_engine.Result` that returns scalar values
rather than :class:`_row.Row` values.
self._unique_filter_state = real_result._unique_filter_state
- def unique(
- self: SelfScalarResult, strategy: Optional[_UniqueFilterType] = None
- ) -> SelfScalarResult:
+ def unique(self, strategy: Optional[_UniqueFilterType] = None) -> Self:
"""Apply unique filtering to the objects returned by this
:class:`_engine.ScalarResult`.
)
-SelfTupleResult = TypeVar("SelfTupleResult", bound="TupleResult[Any]")
-
-
class TupleResult(FilterResult[_R], util.TypingOnly):
"""A :class:`_engine.Result` that's typed as returning plain
Python tuples instead of rows.
...
-SelfMappingResult = TypeVar("SelfMappingResult", bound="MappingResult")
-
-
class MappingResult(_WithKeys, FilterResult[RowMapping]):
"""A wrapper for a :class:`_engine.Result` that returns dictionary values
rather than :class:`_engine.Row` values.
if result._source_supports_scalars:
self._metadata = self._metadata._reduce([0])
- def unique(
- self: SelfMappingResult, strategy: Optional[_UniqueFilterType] = None
- ) -> SelfMappingResult:
+ def unique(self, strategy: Optional[_UniqueFilterType] = None) -> Self:
"""Apply unique filtering to the objects returned by this
:class:`_engine.MappingResult`.
self._unique_filter_state = (set(), strategy)
return self
- def columns(
- self: SelfMappingResult, *col_expressions: _KeyIndexType
- ) -> SelfMappingResult:
+ def columns(self, *col_expressions: _KeyIndexType) -> Self:
r"""Establish the columns that should be returned in each row."""
return self._column_slices(col_expressions)
return IteratorResult(SimpleResultMetaData([]), iter([]))
-SelfChunkedIteratorResult = TypeVar(
- "SelfChunkedIteratorResult", bound="ChunkedIteratorResult[Any]"
-)
-
-
class ChunkedIteratorResult(IteratorResult[_TP]):
"""An :class:`_engine.IteratorResult` that works from an
iterator-producing callable.
self.dynamic_yield_per = dynamic_yield_per
@_generative
- def yield_per(
- self: SelfChunkedIteratorResult, num: int
- ) -> SelfChunkedIteratorResult:
+ def yield_per(self, num: int) -> Self:
# TODO: this throws away the iterator which may be holding
# onto a chunk. the yield_per cannot be changed once any
# rows have been fetched. either find a way to enforce this,
...
-_SelfAssociationProxy = TypeVar(
- "_SelfAssociationProxy", bound="AssociationProxy[Any]"
-)
-
-
class AssociationProxy(
interfaces.InspectionAttrInfo,
ORMDescriptor[_T],
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
)
-_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]):
return self.parent.info
@overload
- def get(self: Self, obj: Literal[None]) -> Self:
+ def get(self: _Self, obj: Literal[None]) -> _Self:
...
@overload
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
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()
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:
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)
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)
_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]]]
@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:
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
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
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.
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`.
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
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.
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`.
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.
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`.
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)
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.
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
_T = TypeVar("_T", bound=Any)
-SelfShardedQuery = TypeVar("SelfShardedQuery", bound="ShardedQuery[Any]")
ShardIdentifier = str
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
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
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.
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
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
from ..sql import roles
from ..sql import visitors
from ..util.typing import Literal
+from ..util.typing import Self
from ..util.typing import TypeGuard
if TYPE_CHECKING:
Tuple[Optional["InstanceState[Any]"], Optional[object]]
]
-SelfQueryableAttribute = TypeVar(
- "SelfQueryableAttribute", bound="QueryableAttribute[Any]"
-)
_UNKNOWN_ATTR_KEY = object()
return self.comparator._bulk_update_tuples(value)
- def adapt_to_entity(
- self: SelfQueryableAttribute, adapt_to_entity: AliasedInsp[Any]
- ) -> SelfQueryableAttribute:
+ def adapt_to_entity(self, adapt_to_entity: AliasedInsp[Any]) -> Self:
assert not self._of_type
return self.__class__(
adapt_to_entity.entity,
util.set_creation_order(self)
- def _copy(self: Self, **kw: Any) -> Self:
- new = cast(Self, self.__class__.__new__(self.__class__))
+ def _copy(self, **kw: Any) -> Self:
+ new = self.__class__.__new__(self.__class__)
new.column = self.column._copy(**kw)
new.deferred = self.deferred
new.foreign_keys = new.column.foreign_keys
from ..util.typing import Literal
from ..util.typing import Self
+
if TYPE_CHECKING:
from ._typing import _EntityType
from ._typing import _ExternalEntityType
_T = TypeVar("_T", bound=Any)
-SelfQuery = TypeVar("SelfQuery", bound="Query[Any]")
-
@inspection._self_inspects
@log.class_logger
self.session = session # type: ignore
self._set_entities(entities)
- def _set_propagate_attrs(
- self: SelfQuery, values: Mapping[str, Any]
- ) -> SelfQuery:
+ def _set_propagate_attrs(self, values: Mapping[str, Any]) -> Self:
self._propagate_attrs = util.immutabledict(values) # type: ignore
return self
self._from_obj = tuple(fa)
@_generative
- def _set_lazyload_from(
- self: SelfQuery, state: InstanceState[Any]
- ) -> SelfQuery:
+ def _set_lazyload_from(self, state: InstanceState[Any]) -> Self:
self.load_options += {"_lazy_loaded_from": state}
return self
)
def _get_options(
- self: SelfQuery,
+ self,
populate_existing: Optional[bool] = None,
version_check: Optional[bool] = None,
only_load_props: Optional[Sequence[str]] = None,
refresh_state: Optional[InstanceState[Any]] = None,
identity_token: Optional[Any] = None,
- ) -> SelfQuery:
+ ) -> Self:
load_options: Dict[str, Any] = {}
compile_options: Dict[str, Any] = {}
return self
- def _clone(self: Self, **kw: Any) -> Self:
+ def _clone(self, **kw: Any) -> Self:
return self._generate() # type: ignore
def _get_select_statement_only(self) -> Select[_T]:
)
@_generative
- def enable_eagerloads(self: SelfQuery, value: bool) -> SelfQuery:
+ def enable_eagerloads(self, value: bool) -> Self:
"""Control whether or not eager joins and subqueries are
rendered.
return self
@_generative
- def _with_compile_options(self: SelfQuery, **opt: Any) -> SelfQuery:
+ def _with_compile_options(self, **opt: Any) -> Self:
self._compile_options += opt
return self
alternative="Use set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL) "
"instead.",
)
- def with_labels(self: SelfQuery) -> SelfQuery:
+ def with_labels(self) -> Self:
return self.set_label_style(
SelectLabelStyle.LABEL_STYLE_TABLENAME_PLUS_COL
)
"""
return self._label_style
- def set_label_style(self: SelfQuery, style: SelectLabelStyle) -> SelfQuery:
+ def set_label_style(self, style: SelectLabelStyle) -> Self:
"""Apply column labels to the return value of Query.statement.
Indicates that this Query's `statement` accessor should return
return self
@_generative
- def enable_assertions(self: SelfQuery, value: bool) -> SelfQuery:
+ def enable_assertions(self, value: bool) -> Self:
"""Control whether assertions are generated.
When set to False, the returned Query will
)
@_generative
- def _with_current_path(self: SelfQuery, path: PathRegistry) -> SelfQuery:
+ def _with_current_path(self, path: PathRegistry) -> Self:
"""indicate that this query applies to objects loaded
within a certain path.
return self
@_generative
- def yield_per(self: SelfQuery, count: int) -> SelfQuery:
+ def yield_per(self, count: int) -> Self:
r"""Yield only ``count`` rows at a time.
The purpose of this method is when fetching very large result sets
@_generative
def correlate(
- self: SelfQuery,
+ self,
*fromclauses: Union[Literal[None, False], _FromClauseArgument],
- ) -> SelfQuery:
+ ) -> Self:
"""Return a :class:`.Query` construct which will correlate the given
FROM clauses to that of an enclosing :class:`.Query` or
:func:`~.expression.select`.
return self
@_generative
- def autoflush(self: SelfQuery, setting: bool) -> SelfQuery:
+ def autoflush(self, setting: bool) -> Self:
"""Return a Query with a specific 'autoflush' setting.
As of SQLAlchemy 1.4, the :meth:`_orm.Query.autoflush` method
return self
@_generative
- def populate_existing(self: SelfQuery) -> SelfQuery:
+ def populate_existing(self) -> Self:
"""Return a :class:`_query.Query`
that will expire and refresh all instances
as they are loaded, or reused from the current :class:`.Session`.
return self
@_generative
- def _with_invoke_all_eagers(self: SelfQuery, value: bool) -> SelfQuery:
+ def _with_invoke_all_eagers(self, value: bool) -> Self:
"""Set the 'invoke all eagers' flag which causes joined- and
subquery loaders to traverse into already-loaded related objects
and collections.
)
@util.preload_module("sqlalchemy.orm.relationships")
def with_parent(
- self: SelfQuery,
+ self,
instance: object,
property: Optional[ # noqa: A002
attributes.QueryableAttribute[Any]
] = None,
from_entity: Optional[_ExternalEntityType[Any]] = None,
- ) -> SelfQuery:
+ ) -> Self:
"""Add filtering criterion that relates the given instance
to a child object or collection, using its attribute state
as well as an established :func:`_orm.relationship()`
return self
@_generative
- def with_session(self: SelfQuery, session: Session) -> SelfQuery:
+ def with_session(self, session: Session) -> Self:
"""Return a :class:`_query.Query` that will use the given
:class:`.Session`.
return self
def _legacy_from_self(
- self: SelfQuery, *entities: _ColumnsClauseArgument[Any]
- ) -> SelfQuery:
+ self, *entities: _ColumnsClauseArgument[Any]
+ ) -> Self:
# used for query.count() as well as for the same
# function in BakedQuery, as well as some old tests in test_baked.py.
return q
@_generative
- def _set_enable_single_crit(self: SelfQuery, val: bool) -> SelfQuery:
+ def _set_enable_single_crit(self, val: bool) -> Self:
self._compile_options += {"_enable_single_crit": val}
return self
@_generative
def _from_selectable(
- self: SelfQuery, fromclause: FromClause, set_entity_from: bool = True
- ) -> SelfQuery:
+ self, fromclause: FromClause, set_entity_from: bool = True
+ ) -> Self:
for attr in (
"_where_criteria",
"_order_by_clauses",
return self.add_columns(column)
@_generative
- def options(self: SelfQuery, *args: ExecutableOption) -> SelfQuery:
+ def options(self, *args: ExecutableOption) -> Self:
"""Return a new :class:`_query.Query` object,
applying the given list of
mapper options.
@overload
def execution_options(
- self: SelfQuery,
+ self,
*,
compiled_cache: Optional[CompiledCacheType] = ...,
logging_token: str = ...,
populate_existing: bool = False,
autoflush: bool = False,
**opt: Any,
- ) -> SelfQuery:
+ ) -> Self:
...
@overload
- def execution_options(self: SelfQuery, **opt: Any) -> SelfQuery:
+ def execution_options(self, **opt: Any) -> Self:
...
@_generative
- def execution_options(self: SelfQuery, **kwargs: Any) -> SelfQuery:
+ def execution_options(self, **kwargs: Any) -> Self:
"""Set non-SQL options which take effect during execution.
Options allowed here include all of those accepted by
@_generative
def with_for_update(
- self: SelfQuery,
+ self,
*,
nowait: bool = False,
read: bool = False,
] = None,
skip_locked: bool = False,
key_share: bool = False,
- ) -> SelfQuery:
+ ) -> Self:
"""return a new :class:`_query.Query`
with the specified options for the
``FOR UPDATE`` clause.
@_generative
def params(
- self: SelfQuery, __params: Optional[Dict[str, Any]] = None, **kw: Any
- ) -> SelfQuery:
+ self, __params: Optional[Dict[str, Any]] = None, **kw: Any
+ ) -> Self:
r"""Add values for bind parameters which may have been
specified in filter().
self._params = self._params.union(kw)
return self
- def where(
- self: SelfQuery, *criterion: _ColumnExpressionArgument[bool]
- ) -> SelfQuery:
+ def where(self, *criterion: _ColumnExpressionArgument[bool]) -> Self:
"""A synonym for :meth:`.Query.filter`.
.. versionadded:: 1.4
@_generative
@_assertions(_no_statement_condition, _no_limit_offset)
- def filter(
- self: SelfQuery, *criterion: _ColumnExpressionArgument[bool]
- ) -> SelfQuery:
+ def filter(self, *criterion: _ColumnExpressionArgument[bool]) -> Self:
r"""Apply the given filtering criterion to a copy
of this :class:`_query.Query`, using SQL expressions.
return self._raw_columns[0]
- def filter_by(self: SelfQuery, **kwargs: Any) -> SelfQuery:
+ def filter_by(self, **kwargs: Any) -> Self:
r"""Apply the given filtering criterion to a copy
of this :class:`_query.Query`, using keyword expressions.
@_generative
def order_by(
- self: SelfQuery,
+ self,
__first: Union[
Literal[None, False, _NoArg.NO_ARG],
_ColumnExpressionOrStrLabelArgument[Any],
] = _NoArg.NO_ARG,
*clauses: _ColumnExpressionOrStrLabelArgument[Any],
- ) -> SelfQuery:
+ ) -> Self:
"""Apply one or more ORDER BY criteria to the query and return
the newly resulting :class:`_query.Query`.
@_generative
def group_by(
- self: SelfQuery,
+ self,
__first: Union[
Literal[None, False, _NoArg.NO_ARG],
_ColumnExpressionOrStrLabelArgument[Any],
] = _NoArg.NO_ARG,
*clauses: _ColumnExpressionOrStrLabelArgument[Any],
- ) -> SelfQuery:
+ ) -> Self:
"""Apply one or more GROUP BY criterion to the query and return
the newly resulting :class:`_query.Query`.
@_generative
@_assertions(_no_statement_condition, _no_limit_offset)
- def having(
- self: SelfQuery, *having: _ColumnExpressionArgument[bool]
- ) -> SelfQuery:
+ def having(self, *having: _ColumnExpressionArgument[bool]) -> Self:
r"""Apply a HAVING criterion to the query and return the
newly resulting :class:`_query.Query`.
self._having_criteria += (having_criteria,)
return self
- def _set_op(self: SelfQuery, expr_fn: Any, *q: Query[Any]) -> SelfQuery:
+ def _set_op(self, expr_fn: Any, *q: Query[Any]) -> Self:
list_of_queries = (self,) + q
return self._from_selectable(expr_fn(*(list_of_queries)).subquery())
- def union(self: SelfQuery, *q: Query[Any]) -> SelfQuery:
+ def union(self, *q: Query[Any]) -> Self:
"""Produce a UNION of this Query against one or more queries.
e.g.::
"""
return self._set_op(expression.union, *q)
- def union_all(self: SelfQuery, *q: Query[Any]) -> SelfQuery:
+ def union_all(self, *q: Query[Any]) -> Self:
"""Produce a UNION ALL of this Query against one or more queries.
Works the same way as :meth:`~sqlalchemy.orm.query.Query.union`. See
"""
return self._set_op(expression.union_all, *q)
- def intersect(self: SelfQuery, *q: Query[Any]) -> SelfQuery:
+ def intersect(self, *q: Query[Any]) -> Self:
"""Produce an INTERSECT of this Query against one or more queries.
Works the same way as :meth:`~sqlalchemy.orm.query.Query.union`. See
"""
return self._set_op(expression.intersect, *q)
- def intersect_all(self: SelfQuery, *q: Query[Any]) -> SelfQuery:
+ def intersect_all(self, *q: Query[Any]) -> Self:
"""Produce an INTERSECT ALL of this Query against one or more queries.
Works the same way as :meth:`~sqlalchemy.orm.query.Query.union`. See
"""
return self._set_op(expression.intersect_all, *q)
- def except_(self: SelfQuery, *q: Query[Any]) -> SelfQuery:
+ def except_(self, *q: Query[Any]) -> Self:
"""Produce an EXCEPT of this Query against one or more queries.
Works the same way as :meth:`~sqlalchemy.orm.query.Query.union`. See
"""
return self._set_op(expression.except_, *q)
- def except_all(self: SelfQuery, *q: Query[Any]) -> SelfQuery:
+ def except_all(self, *q: Query[Any]) -> Self:
"""Produce an EXCEPT ALL of this Query against one or more queries.
Works the same way as :meth:`~sqlalchemy.orm.query.Query.union`. See
@_generative
@_assertions(_no_statement_condition, _no_limit_offset)
def join(
- self: SelfQuery,
+ self,
target: _JoinTargetArgument,
onclause: Optional[_OnClauseArgument] = None,
*,
isouter: bool = False,
full: bool = False,
- ) -> SelfQuery:
+ ) -> Self:
r"""Create a SQL JOIN against this :class:`_query.Query`
object's criterion
and apply generatively, returning the newly resulting
return self
def outerjoin(
- self: SelfQuery,
+ self,
target: _JoinTargetArgument,
onclause: Optional[_OnClauseArgument] = None,
*,
full: bool = False,
- ) -> SelfQuery:
+ ) -> Self:
"""Create a left outer join against this ``Query`` object's criterion
and apply generatively, returning the newly resulting ``Query``.
@_generative
@_assertions(_no_statement_condition)
- def reset_joinpoint(self: SelfQuery) -> SelfQuery:
+ def reset_joinpoint(self) -> Self:
"""Return a new :class:`.Query`, where the "join point" has
been reset back to the base FROM entities of the query.
@_generative
@_assertions(_no_clauseelement_condition)
- def select_from(
- self: SelfQuery, *from_obj: _FromClauseArgument
- ) -> SelfQuery:
+ def select_from(self, *from_obj: _FromClauseArgument) -> Self:
r"""Set the FROM clause of this :class:`.Query` explicitly.
:meth:`.Query.select_from` is often used in conjunction with
@_generative
@_assertions(_no_statement_condition)
def slice(
- self: SelfQuery,
+ self,
start: int,
stop: int,
- ) -> SelfQuery:
+ ) -> Self:
"""Computes the "slice" of the :class:`_query.Query` represented by
the given indices and returns the resulting :class:`_query.Query`.
@_generative
@_assertions(_no_statement_condition)
- def limit(self: SelfQuery, limit: _LimitOffsetType) -> SelfQuery:
+ def limit(self, limit: _LimitOffsetType) -> Self:
"""Apply a ``LIMIT`` to the query and return the newly resulting
``Query``.
@_generative
@_assertions(_no_statement_condition)
- def offset(self: SelfQuery, offset: _LimitOffsetType) -> SelfQuery:
+ def offset(self, offset: _LimitOffsetType) -> Self:
"""Apply an ``OFFSET`` to the query and return the newly resulting
``Query``.
@_generative
@_assertions(_no_statement_condition)
- def distinct(
- self: SelfQuery, *expr: _ColumnExpressionArgument[Any]
- ) -> SelfQuery:
+ def distinct(self, *expr: _ColumnExpressionArgument[Any]) -> Self:
r"""Apply a ``DISTINCT`` to the query and return the newly resulting
``Query``.
@_generative
@_assertions(_no_clauseelement_condition)
- def from_statement(
- self: SelfQuery, statement: ExecutableReturnsRows
- ) -> SelfQuery:
+ def from_statement(self, statement: ExecutableReturnsRows) -> Self:
"""Execute the given SELECT statement and return results.
This method bypasses all internal statement compilation, and the
from ..sql.base import _generative
from ..util.typing import Final
from ..util.typing import Literal
+from ..util.typing import Self
_RELATIONSHIP_TOKEN: Final[Literal["relationship"]] = "relationship"
_COLUMN_TOKEN: Final[Literal["column"]] = "column"
from ..sql.cache_key import _CacheKeyTraversalType
from ..sql.cache_key import CacheKey
-Self_AbstractLoad = TypeVar("Self_AbstractLoad", bound="_AbstractLoad")
_AttrType = Union[str, "QueryableAttribute[Any]"]
propagate_to_loaders: bool
def contains_eager(
- self: Self_AbstractLoad,
+ self,
attr: _AttrType,
alias: Optional[_FromClauseArgument] = None,
_is_chain: bool = False,
- ) -> Self_AbstractLoad:
+ ) -> Self:
r"""Indicate that the given attribute should be eagerly loaded from
columns stated manually in the query.
)
return cloned
- def load_only(
- self: Self_AbstractLoad, *attrs: _AttrType, raiseload: bool = False
- ) -> Self_AbstractLoad:
+ def load_only(self, *attrs: _AttrType, raiseload: bool = False) -> Self:
r"""Indicate that for a particular entity, only the given list
of column-based attribute names should be loaded; all others will be
deferred.
return cloned
def joinedload(
- self: Self_AbstractLoad,
+ self,
attr: _AttrType,
innerjoin: Optional[bool] = None,
- ) -> Self_AbstractLoad:
+ ) -> Self:
"""Indicate that the given attribute should be loaded using joined
eager loading.
)
return loader
- def subqueryload(
- self: Self_AbstractLoad, attr: _AttrType
- ) -> Self_AbstractLoad:
+ def subqueryload(self, attr: _AttrType) -> Self:
"""Indicate that the given attribute should be loaded using
subquery eager loading.
return self._set_relationship_strategy(attr, {"lazy": "subquery"})
def selectinload(
- self: Self_AbstractLoad,
+ self,
attr: _AttrType,
recursion_depth: Optional[int] = None,
- ) -> Self_AbstractLoad:
+ ) -> Self:
"""Indicate that the given attribute should be loaded using
SELECT IN eager loading.
opts={"recursion_depth": recursion_depth},
)
- def lazyload(
- self: Self_AbstractLoad, attr: _AttrType
- ) -> Self_AbstractLoad:
+ def lazyload(self, attr: _AttrType) -> Self:
"""Indicate that the given attribute should be loaded using "lazy"
loading.
return self._set_relationship_strategy(attr, {"lazy": "select"})
def immediateload(
- self: Self_AbstractLoad,
+ self,
attr: _AttrType,
recursion_depth: Optional[int] = None,
- ) -> Self_AbstractLoad:
+ ) -> Self:
"""Indicate that the given attribute should be loaded using
an immediate load with a per-attribute SELECT statement.
)
return loader
- def noload(self: Self_AbstractLoad, attr: _AttrType) -> Self_AbstractLoad:
+ def noload(self, attr: _AttrType) -> Self:
"""Indicate that the given relationship attribute should remain
unloaded.
return self._set_relationship_strategy(attr, {"lazy": "noload"})
- def raiseload(
- self: Self_AbstractLoad, attr: _AttrType, sql_only: bool = False
- ) -> Self_AbstractLoad:
+ def raiseload(self, attr: _AttrType, sql_only: bool = False) -> Self:
"""Indicate that the given attribute should raise an error if accessed.
A relationship attribute configured with :func:`_orm.raiseload` will
attr, {"lazy": "raise_on_sql" if sql_only else "raise"}
)
- def defaultload(
- self: Self_AbstractLoad, attr: _AttrType
- ) -> Self_AbstractLoad:
+ def defaultload(self, attr: _AttrType) -> Self:
"""Indicate an attribute should load using its default loader style.
This method is used to link to other loader options further into
"""
return self._set_relationship_strategy(attr, None)
- def defer(
- self: Self_AbstractLoad, key: _AttrType, raiseload: bool = False
- ) -> Self_AbstractLoad:
+ def defer(self, key: _AttrType, raiseload: bool = False) -> Self:
r"""Indicate that the given column-oriented attribute should be
deferred, e.g. not loaded until accessed.
strategy["raiseload"] = True
return self._set_column_strategy((key,), strategy)
- def undefer(self: Self_AbstractLoad, key: _AttrType) -> Self_AbstractLoad:
+ def undefer(self, key: _AttrType) -> Self:
r"""Indicate that the given column-oriented attribute should be
undeferred, e.g. specified within the SELECT statement of the entity
as a whole.
(key,), {"deferred": False, "instrument": True}
)
- def undefer_group(self: Self_AbstractLoad, name: str) -> Self_AbstractLoad:
+ def undefer_group(self, name: str) -> Self:
"""Indicate that columns within the given deferred group name should be
undeferred.
)
def with_expression(
- self: Self_AbstractLoad,
+ self,
key: _AttrType,
expression: _ColumnExpressionArgument[Any],
- ) -> Self_AbstractLoad:
+ ) -> Self:
r"""Apply an ad-hoc SQL expression to a "deferred expression"
attribute.
(key,), {"query_expression": True}, opts={"expression": expression}
)
- def selectin_polymorphic(
- self: Self_AbstractLoad, classes: Iterable[Type[Any]]
- ) -> Self_AbstractLoad:
+ def selectin_polymorphic(self, classes: Iterable[Type[Any]]) -> Self:
"""Indicate an eager load should take place for all attributes
specific to a subclass.
@_generative
def _set_relationship_strategy(
- self: Self_AbstractLoad,
+ self,
attr: _AttrType,
strategy: Optional[_StrategySpec],
propagate_to_loaders: bool = True,
opts: Optional[_OptsType] = None,
_reconcile_to_other: Optional[bool] = None,
- ) -> Self_AbstractLoad:
+ ) -> Self:
strategy_key = self._coerce_strat(strategy)
self._clone_for_bind_strategy(
@_generative
def _set_column_strategy(
- self: Self_AbstractLoad,
+ self,
attrs: Tuple[_AttrType, ...],
strategy: Optional[_StrategySpec],
opts: Optional[_OptsType] = None,
- ) -> Self_AbstractLoad:
+ ) -> Self:
strategy_key = self._coerce_strat(strategy)
self._clone_for_bind_strategy(
@_generative
def _set_generic_strategy(
- self: Self_AbstractLoad,
+ self,
attrs: Tuple[_AttrType, ...],
strategy: _StrategySpec,
_reconcile_to_other: Optional[bool] = None,
- ) -> Self_AbstractLoad:
+ ) -> Self:
strategy_key = self._coerce_strat(strategy)
self._clone_for_bind_strategy(
attrs,
@_generative
def _set_class_strategy(
- self: Self_AbstractLoad, strategy: _StrategySpec, opts: _OptsType
- ) -> Self_AbstractLoad:
+ self, strategy: _StrategySpec, opts: _OptsType
+ ) -> Self:
strategy_key = self._coerce_strat(strategy)
self._clone_for_bind_strategy(None, strategy_key, None, opts=opts)
"""
raise NotImplementedError()
- def options(
- self: Self_AbstractLoad, *opts: _AbstractLoad
- ) -> Self_AbstractLoad:
+ def options(self, *opts: _AbstractLoad) -> Self:
r"""Apply a series of options as sub-options to this
:class:`_orm._AbstractLoad` object.
raise NotImplementedError()
def _clone_for_bind_strategy(
- self: Self_AbstractLoad,
+ self,
attrs: Optional[Tuple[_AttrType, ...]],
strategy: Optional[_StrategyKey],
wildcard_key: Optional[_WildcardKeyType],
attr_group: Optional[_AttrGroupType] = None,
propagate_to_loaders: bool = True,
reconcile_to_other: Optional[bool] = None,
- ) -> Self_AbstractLoad:
+ ) -> Self:
raise NotImplementedError()
def process_compile_state_replaced_entities(
return to_chop[i + 1 :]
-SelfLoad = TypeVar("SelfLoad", bound="Load")
-
-
class Load(_AbstractLoad):
"""Represents loader options which modify the state of a
ORM-enabled :class:`_sql.Select` or a legacy :class:`_query.Query` in
parent.context += cloned.context
@_generative
- def options(self: SelfLoad, *opts: _AbstractLoad) -> SelfLoad:
+ def options(self, *opts: _AbstractLoad) -> Self:
r"""Apply a series of options as sub-options to this
:class:`_orm.Load`
object.
return self
def _clone_for_bind_strategy(
- self: SelfLoad,
+ self,
attrs: Optional[Tuple[_AttrType, ...]],
strategy: Optional[_StrategyKey],
wildcard_key: Optional[_WildcardKeyType],
attr_group: Optional[_AttrGroupType] = None,
propagate_to_loaders: bool = True,
reconcile_to_other: Optional[bool] = None,
- ) -> SelfLoad:
+ ) -> Self:
# for individual strategy that needs to propagate, set the whole
# Load container to also propagate, so that it shows up in
# InstanceState.load_options
self._shallow_from_dict(state)
-SelfWildcardLoad = TypeVar("SelfWildcardLoad", bound="_WildcardLoad")
-
-
class _WildcardLoad(_AbstractLoad):
"""represent a standalone '*' load operation"""
if opts:
self.local_opts = util.immutabledict(opts)
- def options(
- self: SelfWildcardLoad, *opts: _AbstractLoad
- ) -> SelfWildcardLoad:
+ def options(self, *opts: _AbstractLoad) -> Self:
raise NotImplementedError("Star option does not support sub-options")
def _apply_to_parent(self, parent: Load) -> None:
from .visitors import InternalTraversal
from .. import util
from ..util.typing import Literal
+from ..util.typing import Self
if TYPE_CHECKING:
from .base import _EntityNamespace
EMPTY_ANNOTATIONS: util.immutabledict[str, Any] = util.EMPTY_DICT
-SelfSupportsAnnotations = TypeVar(
- "SelfSupportsAnnotations", bound="SupportsAnnotations"
-)
-
-
class SupportsAnnotations(ExternallyTraversible):
__slots__ = ()
_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
)
-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
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
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.
__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__(
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)
@overload
def _deannotate(
- self: SelfAnnotated,
+ self,
values: Literal[None] = ...,
clone: bool = ...,
- ) -> SelfAnnotated:
+ ) -> Self:
...
@overload
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
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.
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(
class _GenerativeType(compat_typing.Protocol):
- def _generate(self: _SelfGenerativeType) -> _SelfGenerativeType:
+ def _generate(self) -> Self:
...
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)
return c
-SelfExecutable = TypeVar("SelfExecutable", bound="Executable")
-
-
class Executable(roles.StatementRole):
"""Mark a :class:`_expression.ClauseElement` as supporting execution.
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
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
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
@_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
@overload
def execution_options(
- self: SelfExecutable,
+ self,
*,
compiled_cache: Optional[CompiledCacheType] = ...,
logging_token: str = ...,
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.
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
return True
-SelfExecutableDDLElement = typing.TypeVar(
- "SelfExecutableDDLElement", bound="ExecutableDDLElement"
-)
-
-
class ExecutableDDLElement(roles.DDLRole, Executable, BaseDDLElement):
"""Base class for standalone executable DDL expression constructs.
)
@_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.
@_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.
import collections.abc as collections_abc
import operator
-import typing
from typing import Any
from typing import cast
from typing import Dict
from .visitors import InternalTraversal
from .. import exc
from .. import util
+from ..util.typing import Self
from ..util.typing import TypeGuard
if TYPE_CHECKING:
self.is_multitable = ef
-SelfUpdateBase = typing.TypeVar("SelfUpdateBase", bound="UpdateBase")
-
-
class UpdateBase(
roles.DMLRole,
HasCTE,
)
@_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.::
@_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.
@_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.
return meth(self)
-SelfValuesBase = typing.TypeVar("SelfValuesBase", bound="ValuesBase")
-
-
class ValuesBase(UpdateBase):
"""Supplies support for :meth:`.ValuesBase.values` to
INSERT and UPDATE constructs."""
},
)
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.
return self
-SelfInsert = typing.TypeVar("SelfInsert", bound="Insert")
-
-
class Insert(ValuesBase):
"""Represent an INSERT construct.
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
@_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.
"""
-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.
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
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.
)
-SelfUpdate = typing.TypeVar("SelfUpdate", bound="Update")
-
-
class Update(DMLWhereBase, ValuesBase):
"""Represent an Update construct.
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.
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`
"""
-SelfDelete = typing.TypeVar("SelfDelete", bound="Delete")
-
-
class Delete(DMLWhereBase, UpdateBase):
"""Represent a DELETE construct.
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
return str(self.compile())
-SelfClauseElement = TypeVar("SelfClauseElement", bound="ClauseElement")
-
-
@inspection._self_inspects
class ClauseElement(
SupportsWrappingAnnotations,
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
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
).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.
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.
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)
_SQO = SQLCoreOperations
-SelfColumnElement = TypeVar("SelfColumnElement", bound="ColumnElement[Any]")
-
class ColumnElement(
roles.ColumnArgumentOrKeyRole,
return super()._proxy_key
-SelfBindParameter = TypeVar("SelfBindParameter", bound="BindParameter[Any]")
-
-
class BindParameter(roles.InElementRole, KeyedColumnElement[_T]):
r"""Represent a "bound expression".
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
self.type = type_
-SelfTextClause = typing.TypeVar("SelfTextClause", bound="TextClause")
-
-
class TextClause(
roles.DDLConstraintColumnRole,
roles.DDLExpressionRole,
@_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.
from .. import inspection
from .. import util
from ..util.typing import Literal
-from ..util.typing import Self
+
if TYPE_CHECKING:
from .elements import BindParameter
return expr
def _copy_internals(
- self: Self,
+ self,
clone: _CloneCallableType = _clone,
deferred_copy_internals: Optional[_CloneCallableType] = None,
**kw: Any,
_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.
_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
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)
"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()
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
"""base for executable statements that return rows."""
-SelfSelectable = TypeVar("SelfSelectable", bound="Selectable")
-
-
class Selectable(ReturnsRows):
"""Mark a class as being selectable."""
"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`.
)
-SelfHasPrefixes = typing.TypeVar("SelfHasPrefixes", bound="HasPrefixes")
-
-
class HasPrefixes:
_prefixes: Tuple[Tuple[DQLDMLClauseElement, str], ...] = ()
":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.
return self
-SelfHasSuffixes = typing.TypeVar("SelfHasSuffixes", bound="HasSuffixes")
-
-
class HasSuffixes:
_suffixes: Tuple[Tuple[DQLDMLClauseElement, str], ...] = ()
":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
return self
-SelfHasHints = typing.TypeVar("SelfHasHints", bound="HasHints")
-
-
class HasHints:
_hints: util.immutabledict[
Tuple[FromClause, str], str
("_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.
@_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.
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:
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.
if TYPE_CHECKING:
def self_group(
- self: Self, against: Optional[OperatorType] = None
+ self, against: Optional[OperatorType] = None
) -> Union[FromGrouping, Self]:
...
"""mark a FROM clause as being able to render directly as LATERAL"""
-_SelfAliasedReturnsRows = TypeVar(
- "_SelfAliasedReturnsRows", bound="AliasedReturnsRows"
-)
-
# FromClause ->
# AliasedReturnsRows
# -> Alias only for FromClause
@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
return self._restates if self._restates is not None else self
-SelfHasCTE = typing.TypeVar("SelfHasCTE", bound="HasCTE")
-
-
class _CTEOpts(NamedTuple):
nesting: bool
_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
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.
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
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.
return self
-SelfSelectBase = TypeVar("SelfSelectBase", bound=Any)
-
-
class SelectBase(
roles.SelectStatementRole,
roles.DMLSelectRole,
"""
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.
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.
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
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.
@_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`.
"""
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,
)
@_generative
- def limit(
- self: SelfGenerativeSelect, limit: _LimitOffsetType
- ) -> SelfGenerativeSelect:
+ def limit(self, limit: _LimitOffsetType) -> Self:
"""Return a new selectable with the given LIMIT criterion
applied.
@_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.
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.
@_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
@_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.
@_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.
yield element
-Self_MemoizedSelectEntities = TypeVar("Self_MemoizedSelectEntities", bound=Any)
-
-
class _MemoizedSelectEntities(
cache_key.HasCacheKey, traversals.HasCopyInternals, visitors.Traversible
):
_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()}
select_stmt._setup_joins = select_stmt._with_options = ()
-SelfSelect = typing.TypeVar("SelfSelect", bound="Select[Any]")
-
-
class Select(
HasPrefixes,
HasSuffixes,
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)
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.
@_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
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`.
@_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
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`.
_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.
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.
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.
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.
@_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`.
@_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.
)
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
return CompoundSelect._create_intersect_all(self, *other)
-SelfScalarSelect = typing.TypeVar(
- "SelfScalarSelect", bound="ScalarSelect[Any]"
-)
-
-
class ScalarSelect(
roles.InElementRole, Generative, GroupedElement, ColumnElement[_T]
):
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`.
@_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`.
@_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.
return self
-SelfExists = TypeVar("SelfExists", bound="Exists")
-
-
class Exists(UnaryExpression[bool]):
"""Represent an ``EXISTS`` clause.
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`.
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`.
)
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`
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.
return e
-SelfTextualSelect = typing.TypeVar("SelfTextualSelect", bound="TextualSelect")
-
-
class TextualSelect(SelectBase, Executable, Generative):
"""Wrap a :class:`_expression.TextClause` construct within a
:class:`_expression.SelectBase`
@_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
from typing import Set
from typing import Tuple
from typing import Type
-from typing import TypeVar
from . import operators
from .cache_key import HasCacheKey
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
)
-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.
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(
@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
@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
@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
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:
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.
__slots__ = ()
- def _generate(
- self: SelfGenerativeOnTraversal,
- ) -> SelfGenerativeOnTraversal:
+ def _generate(self) -> Self:
cls = self.__class__
s = cls.__new__(cls)
self._shallow_copy_to(s)
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
_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]):
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.
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(
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.
)
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.
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.
# ...
-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.
"""
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
_generate_traversal_dispatch()
-SelfExternallyTraversible = TypeVar(
- "SelfExternallyTraversible", bound="ExternallyTraversible"
-)
-
-
class ExternallyTraversible(HasTraverseInternals, Visitable):
__slots__ = ()
if typing.TYPE_CHECKING:
- def _annotate(
- self: SelfExternallyTraversible, values: _AnnotationDict
- ) -> SelfExternallyTraversible:
+ def _annotate(self, values: _AnnotationDict) -> Self:
...
def get_children(
) -> Iterable[ExternallyTraversible]:
...
- def _clone(self: Self, **kw: Any) -> Self:
+ def _clone(self, **kw: Any) -> Self:
"""clone this element"""
raise NotImplementedError()
from typing_extensions import TypeAlias as TypeAlias # 3.10
from typing_extensions import TypedDict as TypedDict # 3.8
from typing_extensions import TypeGuard as TypeGuard # 3.10
+ from typing_extensions import Self as Self # 3.11
_T = TypeVar("_T", bound=Any)
_VT = TypeVar("_VT")
_VT_co = TypeVar("_VT_co", covariant=True)
-Self = TypeVar("Self", bound=Any)
if compat.py310:
# why they took until py310 to put this in stdlib is beyond me,
deps=
greenlet != 0.4.17
importlib_metadata; python_version < '3.8'
- mypy >= 0.981
+ mypy >= 1.0.0
commands =
mypy {env:MYPY_COLOR} ./lib/sqlalchemy
# pyright changes too often with not-exactly-correct errors
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
