# include the pyx and pxd extensions, which otherwise
# don't come in if --with-cextensions isn't specified.
-recursive-include lib *.pyx *.pxd *.txt
+recursive-include lib *.pyx *.pxd *.txt *.typed
include README* AUTHORS LICENSE CHANGES* tox.ini
prune doc/build/output
+# Copyright (C) 2005-2021 the SQLAlchemy authors and contributors
+# <see AUTHORS file>
+#
+# This module is part of SQLAlchemy and is released under
+# the MIT License: https://www.opensource.org/licenses/mit-license.php
+
+import typing
+
from ... import exc
from ... import util
from ...sql.base import _exclusive_against
__all__ = ("Insert", "insert")
+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, *args, **kw):
+ def on_duplicate_key_update(self: SelfInsert, *args, **kw) -> SelfInsert:
r"""
Specifies the ON DUPLICATE KEY UPDATE clause.
inserted_alias = getattr(self, "inserted_alias", None)
self._post_values_clause = OnDuplicateClause(inserted_alias, values)
+ return self
insert = public_factory(
+# Copyright (C) 2005-2021 the SQLAlchemy authors and contributors
+# <see AUTHORS file>
+#
+# This module is part of SQLAlchemy and is released under
+# the MIT License: https://www.opensource.org/licenses/mit-license.php
+
+import typing
+
from ... import exc
from ... import util
from ...sql import coercions
from ...sql.base import Generative
+Selfmatch = typing.TypeVar("Selfmatch", bound="match")
+
+
class match(Generative, elements.BinaryExpression):
"""Produce a ``MATCH (X, Y) AGAINST ('TEXT')`` clause.
)
@_generative
- def in_boolean_mode(self):
+ def in_boolean_mode(self: Selfmatch) -> Selfmatch:
"""Apply the "IN BOOLEAN MODE" modifier to the MATCH expression.
:return: a new :class:`_mysql.match` instance with modifications
"""
self.modifiers = self.modifiers.union({"mysql_boolean_mode": True})
+ return self
@_generative
- def in_natural_language_mode(self):
+ def in_natural_language_mode(self: Selfmatch) -> Selfmatch:
"""Apply the "IN NATURAL LANGUAGE MODE" modifier to the MATCH
expression.
"""
self.modifiers = self.modifiers.union({"mysql_natural_language": True})
+ return self
@_generative
- def with_query_expansion(self):
+ def with_query_expansion(self: Selfmatch) -> Selfmatch:
"""Apply the "WITH QUERY EXPANSION" modifier to the MATCH expression.
:return: a new :class:`_mysql.match` instance with modifications
"""
self.modifiers = self.modifiers.union({"mysql_query_expansion": True})
+ return self
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
+import typing
from . import ext
from ... import util
__all__ = ("Insert", "insert")
+SelfInsert = typing.TypeVar("SelfInsert", bound="Insert")
+
class Insert(StandardInsert):
"""PostgreSQL-specific implementation of INSERT.
@_generative
@_on_conflict_exclusive
def on_conflict_do_update(
- self,
+ self: SelfInsert,
constraint=None,
index_elements=None,
index_where=None,
set_=None,
where=None,
- ):
+ ) -> SelfInsert:
r"""
Specifies a DO UPDATE SET action for ON CONFLICT clause.
self._post_values_clause = OnConflictDoUpdate(
constraint, index_elements, index_where, set_, where
)
+ return self
@_generative
@_on_conflict_exclusive
def on_conflict_do_nothing(
- self, constraint=None, index_elements=None, index_where=None
- ):
+ self: SelfInsert,
+ constraint=None,
+ index_elements=None,
+ index_where=None,
+ ) -> SelfInsert:
"""
Specifies a DO NOTHING action for ON CONFLICT clause.
self._post_values_clause = OnConflictDoNothing(
constraint, index_elements, index_where
)
+ return self
insert = public_factory(
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
+import typing
+
from ... import util
from ...sql import coercions
from ...sql import roles
__all__ = ("Insert", "insert")
+SelfInsert = typing.TypeVar("SelfInsert", bound="Insert")
+
class Insert(StandardInsert):
"""SQLite-specific implementation of INSERT.
@_generative
@_on_conflict_exclusive
def on_conflict_do_update(
- self,
+ self: SelfInsert,
index_elements=None,
index_where=None,
set_=None,
where=None,
- ):
+ ) -> SelfInsert:
r"""
Specifies a DO UPDATE SET action for ON CONFLICT clause.
self._post_values_clause = OnConflictDoUpdate(
index_elements, index_where, set_, where
)
+ return self
@_generative
@_on_conflict_exclusive
- def on_conflict_do_nothing(self, index_elements=None, index_where=None):
+ def on_conflict_do_nothing(
+ self: SelfInsert, index_elements=None, index_where=None
+ ) -> SelfInsert:
"""
Specifies a DO NOTHING action for ON CONFLICT clause.
self._post_values_clause = OnConflictDoNothing(
index_elements, index_where
)
+ return self
insert = public_factory(
def yield_per(self, num):
self._yield_per = num
self.cursor_strategy.yield_per(self, self.cursor, num)
+ return self
ResultProxy = CursorResult
"""Define generic result set constructs."""
-
import collections.abc as collections_abc
import functools
import itertools
import operator
+import typing
from .row import Row
from .. import exc
# filter is applied to rows.
_NO_ROW = util.symbol("NO_ROW")
+SelfResultInternal = typing.TypeVar(
+ "SelfResultInternal", bound="ResultInternal"
+)
+
class ResultInternal(InPlaceGenerative):
_real_result = None
return row
@_generative
- def _column_slices(self, indexes):
+ def _column_slices(
+ self: SelfResultInternal, indexes
+ ) -> SelfResultInternal:
real_result = self._real_result if self._real_result else self
if real_result._source_supports_scalars and len(indexes) == 1:
self._generate_rows = True
self._metadata = self._metadata._reduce(indexes)
+ return self
+
@HasMemoized.memoized_attribute
def _unique_strategy(self):
uniques, strategy = self._unique_filter_state
return self._metadata.keys
+SelfResult = typing.TypeVar("SelfResult", bound="Result")
+
+
class Result(_WithKeys, ResultInternal):
"""Represent a set of database results.
self._soft_close(hard=True)
@_generative
- def yield_per(self, num):
+ def yield_per(self: SelfResult, num) -> SelfResult:
"""Configure the row-fetching strategy to fetch num rows at a time.
This impacts the underlying behavior of the result when iterating over
"""
self._yield_per = num
+ return self
@_generative
- def unique(self, strategy=None):
+ def unique(self: SelfResult, strategy=None) -> SelfResult:
"""Apply unique filtering to the objects returned by this
:class:`_engine.Result`.
"""
self._unique_filter_state = (set(), strategy)
+ return self
- def columns(self, *col_expressions):
+ def columns(
+ self: SelfResultInternal, *col_expressions
+ ) -> SelfResultInternal:
r"""Establish the columns that should be returned in each row.
This method may be used to limit the columns returned as well
"""
return self._column_slices(col_expressions)
- def scalars(self, index=0):
+ def scalars(self, index=0) -> "ScalarResult":
"""Return a :class:`_result.ScalarResult` filtering object which
will return single elements rather than :class:`_row.Row` objects.
)
return self._metadata._row_as_tuple_getter(keys)
- def mappings(self):
+ def mappings(self) -> "MappingResult":
"""Apply a mappings filter to returned rows, returning an instance of
:class:`_result.MappingResult`.
return IteratorResult(SimpleResultMetaData([]), iter([]))
+SelfChunkedIteratorResult = typing.TypeVar(
+ "SelfChunkedIteratorResult", bound="ChunkedIteratorResult"
+)
+
+
class ChunkedIteratorResult(IteratorResult):
"""An :class:`.IteratorResult` that works from an iterator-producing callable.
self.dynamic_yield_per = dynamic_yield_per
@_generative
- def yield_per(self, num):
+ def yield_per(
+ self: SelfChunkedIteratorResult, num
+ ) -> SelfChunkedIteratorResult:
# 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,
self._yield_per = num
self.iterator = itertools.chain.from_iterable(self.chunks(num))
+ return self
def _soft_close(self, **kw):
super(ChunkedIteratorResult, self)._soft_close(**kw)
"""
import operator
+import typing
from . import exc
from .. import exc as sa_exc
from .. import inspection
from .. import util
+from ..util import typing as compat_typing
PASSIVE_NO_RESULT = util.symbol(
_RAISE_FOR_STATE = util.symbol("RAISE_FOR_STATE")
-def _assertions(*assertions):
+_Fn = typing.TypeVar("_Fn", bound=typing.Callable)
+_Args = compat_typing.ParamSpec("_Args")
+_Self = typing.TypeVar("_Self")
+
+
+def _assertions(
+ *assertions,
+) -> typing.Callable[
+ [typing.Callable[compat_typing.Concatenate[_Fn, _Args], _Self]],
+ typing.Callable[compat_typing.Concatenate[_Fn, _Args], _Self],
+]:
@util.decorator
- def generate(fn, *args, **kw):
- self = args[0]
+ def generate(
+ fn: _Fn, self: _Self, *args: _Args.args, **kw: _Args.kwargs
+ ) -> _Self:
for assertion in assertions:
assertion(self, fn.__name__)
- fn(self, *args[1:], **kw)
+ fn(self, *args, **kw)
+ return self
return generate
import itertools
import operator
import types
+import typing
from . import exc as orm_exc
from . import interfaces
__all__ = ["Query", "QueryContext", "aliased"]
+SelfQuery = typing.TypeVar("SelfQuery", bound="Query")
+
@inspection._self_inspects
@log.class_logger
self._from_obj = tuple(fa)
@_generative
- def _set_lazyload_from(self, state):
+ def _set_lazyload_from(self: SelfQuery, state) -> SelfQuery:
self.load_options += {"_lazy_loaded_from": state}
+ return self
def _get_condition(self):
return self._no_criterion_condition(
)
@_generative
- def only_return_tuples(self, value):
+ def only_return_tuples(self: SelfQuery, value) -> SelfQuery:
"""When set to True, the query results will always be a tuple.
This is specifically for single element queries. The default is False.
"""
self.load_options += dict(_only_return_tuples=value)
+ return self
@property
def is_single_entity(self):
)
@_generative
- def enable_eagerloads(self, value):
+ def enable_eagerloads(self: SelfQuery, value) -> SelfQuery:
"""Control whether or not eager joins and subqueries are
rendered.
"""
self._compile_options += {"_enable_eagerloads": value}
+ return self
@_generative
- def _with_compile_options(self, **opt):
+ def _with_compile_options(self: SelfQuery, **opt) -> SelfQuery:
self._compile_options += opt
+ return self
@util.deprecated_20(
":meth:`_orm.Query.with_labels` and :meth:`_orm.Query.apply_labels`",
return self
@_generative
- def enable_assertions(self, value):
+ def enable_assertions(self: SelfQuery, value) -> SelfQuery:
"""Control whether assertions are generated.
When set to False, the returned Query will
"""
self._enable_assertions = value
+ return self
@property
def whereclause(self):
)
@_generative
- def _with_current_path(self, path):
+ def _with_current_path(self: SelfQuery, path) -> SelfQuery:
"""indicate that this query applies to objects loaded
within a certain path.
"""
self._compile_options += {"_current_path": path}
+ return self
@_generative
@_assertions(_no_clauseelement_condition)
alternative="Use the orm.with_polymorphic() standalone function",
)
def with_polymorphic(
- self, cls_or_mappers, selectable=None, polymorphic_on=None
- ):
+ self: SelfQuery, cls_or_mappers, selectable=None, polymorphic_on=None
+ ) -> SelfQuery:
"""Load columns for inheriting classes.
This is a legacy method which is replaced by the
self._compile_options = self._compile_options.add_to_element(
"_with_polymorphic_adapt_map", ((entity, inspect(wp)),)
)
+ return self
@_generative
- def yield_per(self, count):
+ def yield_per(self: SelfQuery, count) -> SelfQuery:
r"""Yield only ``count`` rows at a time.
The purpose of this method is when fetching very large result sets
"""
self.load_options += {"_yield_per": count}
+ return self
@util.deprecated_20(
":meth:`_orm.Query.get`",
return self._compile_options._current_path
@_generative
- def correlate(self, *fromclauses):
+ def correlate(self: SelfQuery, *fromclauses) -> SelfQuery:
"""Return a :class:`.Query` construct which will correlate the given
FROM clauses to that of an enclosing :class:`.Query` or
:func:`~.expression.select`.
self._correlate = set(self._correlate).union(
coercions.expect(roles.FromClauseRole, f) for f in fromclauses
)
+ return self
@_generative
- def autoflush(self, setting):
+ def autoflush(self: SelfQuery, setting) -> SelfQuery:
"""Return a Query with a specific 'autoflush' setting.
As of SQLAlchemy 1.4, the :meth:`_orm.Query.autoflush` method
"""
self.load_options += {"_autoflush": setting}
+ return self
@_generative
- def populate_existing(self):
+ def populate_existing(self: SelfQuery) -> SelfQuery:
"""Return a :class:`_query.Query`
that will expire and refresh all instances
as they are loaded, or reused from the current :class:`.Session`.
"""
self.load_options += {"_populate_existing": True}
+ return self
@_generative
- def _with_invoke_all_eagers(self, value):
+ def _with_invoke_all_eagers(self: SelfQuery, value) -> SelfQuery:
"""Set the 'invoke all eagers' flag which causes joined- and
subquery loaders to traverse into already-loaded related objects
and collections.
"""
self.load_options += {"_invoke_all_eagers": value}
+ return self
@util.deprecated_20(
":meth:`_orm.Query.with_parent`",
return self.filter(with_parent(instance, property, entity_zero.entity))
@_generative
- def add_entity(self, entity, alias=None):
+ def add_entity(self: SelfQuery, entity, alias=None) -> SelfQuery:
"""add a mapped entity to the list of result columns
to be returned."""
roles.ColumnsClauseRole, entity, apply_propagate_attrs=self
)
)
+ return self
@_generative
- def with_session(self, session):
+ def with_session(self: SelfQuery, session) -> SelfQuery:
"""Return a :class:`_query.Query` that will use the given
:class:`.Session`.
"""
self.session = session
+ return self
@util.deprecated_20(
":meth:`_query.Query.from_self`",
return q
@_generative
- def _set_enable_single_crit(self, val):
+ def _set_enable_single_crit(self: SelfQuery, val) -> SelfQuery:
self._compile_options += {"_enable_single_crit": val}
+ return self
@_generative
- def _from_selectable(self, fromclause, set_entity_from=True):
+ def _from_selectable(
+ self: SelfQuery, fromclause, set_entity_from=True
+ ) -> SelfQuery:
for attr in (
"_where_criteria",
"_order_by_clauses",
# "oldstyle" tests that rely on this and the corresponding
# "newtyle" that do not.
self._compile_options += {"_orm_only_from_obj_alias": False}
+ return self
@util.deprecated(
"1.4",
return None
@_generative
- def with_entities(self, *entities):
+ def with_entities(self: SelfQuery, *entities) -> SelfQuery:
r"""Return a new :class:`_query.Query`
replacing the SELECT list with the
given entities.
"""
_MemoizedSelectEntities._generate_for_statement(self)
self._set_entities(entities)
+ return self
@_generative
- def add_columns(self, *column):
+ def add_columns(self: SelfQuery, *column) -> SelfQuery:
"""Add one or more column expressions to the list
of result columns to be returned."""
)
for c in column
)
+ return self
@util.deprecated(
"1.4",
return self.add_columns(column)
@_generative
- def options(self, *args):
+ def options(self: SelfQuery, *args) -> SelfQuery:
"""Return a new :class:`_query.Query` object,
applying the given list of
mapper options.
opt.process_query(self)
self._with_options += opts
+ return self
def with_transformation(self, fn):
"""Return a new :class:`_query.Query` object transformed by
return self._execution_options
@_generative
- def execution_options(self, **kwargs):
+ def execution_options(self: SelfQuery, **kwargs) -> SelfQuery:
"""Set non-SQL options which take effect during execution.
Options allowed here include all of those accepted by
"""
self._execution_options = self._execution_options.union(kwargs)
+ return self
@_generative
def with_for_update(
- self,
+ self: SelfQuery,
read=False,
nowait=False,
of=None,
skip_locked=False,
key_share=False,
- ):
+ ) -> SelfQuery:
"""return a new :class:`_query.Query`
with the specified options for the
``FOR UPDATE`` clause.
skip_locked=skip_locked,
key_share=key_share,
)
+ return self
@_generative
- def params(self, *args, **kwargs):
+ def params(self: SelfQuery, *args, **kwargs) -> SelfQuery:
r"""Add values for bind parameters which may have been
specified in filter().
"which is a dictionary."
)
self._params = self._params.union(kwargs)
+ return self
- def where(self, *criterion):
+ def where(self: SelfQuery, *criterion) -> SelfQuery:
"""A synonym for :meth:`.Query.filter`.
.. versionadded:: 1.4
@_generative
@_assertions(_no_statement_condition, _no_limit_offset)
- def filter(self, *criterion):
+ def filter(self: SelfQuery, *criterion) -> SelfQuery:
r"""Apply the given filtering criterion to a copy
of this :class:`_query.Query`, using SQL expressions.
# legacy ^^^^^^^^^^^^^^^^^^^^^^^^^^^
self._where_criteria += (criterion,)
+ return self
@util.memoized_property
def _last_joined_entity(self):
@_generative
@_assertions(_no_statement_condition, _no_limit_offset)
- def order_by(self, *clauses):
+ def order_by(self: SelfQuery, *clauses) -> SelfQuery:
"""Apply one or more ORDER BY criteria to the query and return
the newly resulting :class:`_query.Query`.
# legacy ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
self._order_by_clauses += criterion
+ return self
@_generative
@_assertions(_no_statement_condition, _no_limit_offset)
- def group_by(self, *clauses):
+ def group_by(self: SelfQuery, *clauses) -> SelfQuery:
"""Apply one or more GROUP BY criterion to the query and return
the newly resulting :class:`_query.Query`.
# legacy ^^^^^^^^^^^^^^^^^^^^^^^^^^
self._group_by_clauses += criterion
+ return self
@_generative
@_assertions(_no_statement_condition, _no_limit_offset)
- def having(self, criterion):
+ def having(self: SelfQuery, criterion) -> SelfQuery:
r"""Apply a HAVING criterion to the query and return the
newly resulting :class:`_query.Query`.
roles.WhereHavingRole, criterion, apply_propagate_attrs=self
),
)
+ return self
def _set_op(self, expr_fn, *q):
return self._from_selectable(expr_fn(*([self] + list(q))).subquery())
@_generative
@_assertions(_no_statement_condition, _no_limit_offset)
- def join(self, target, *props, **kwargs):
+ def join(self: SelfQuery, target, *props, **kwargs) -> SelfQuery:
r"""Create a SQL JOIN against this :class:`_query.Query`
object's criterion
and apply generatively, returning the newly resulting
self._legacy_setup_joins += joins_to_add
self.__dict__.pop("_last_joined_entity", None)
+ return self
def outerjoin(self, target, *props, **kwargs):
"""Create a left outer join against this ``Query`` object's criterion
@_generative
@_assertions(_no_statement_condition)
- def reset_joinpoint(self):
+ def reset_joinpoint(self: SelfQuery) -> SelfQuery:
"""Return a new :class:`.Query`, where the "join point" has
been reset back to the base FROM entities of the query.
"""
self._last_joined_entity = None
self._aliased_generation = None
+ return self
@_generative
@_assertions(_no_clauseelement_condition)
- def select_from(self, *from_obj):
+ def select_from(self: SelfQuery, *from_obj) -> SelfQuery:
r"""Set the FROM clause of this :class:`.Query` explicitly.
:meth:`.Query.select_from` is often used in conjunction with
"""
self._set_select_from(from_obj, False)
+ return self
@util.deprecated_20(
":meth:`_orm.Query.select_entity_from`",
)
@_generative
@_assertions(_no_clauseelement_condition)
- def select_entity_from(self, from_obj):
+ def select_entity_from(self: SelfQuery, from_obj) -> SelfQuery:
r"""Set the FROM clause of this :class:`_query.Query` to a
core selectable, applying it as a replacement FROM clause
for corresponding mapped entities.
self._set_select_from([from_obj], True)
self._compile_options += {"_enable_single_crit": False}
+ return self
def __getitem__(self, item):
return orm_util._getitem(
@_generative
@_assertions(_no_statement_condition)
- def slice(self, start, stop):
+ def slice(self: SelfQuery, start, stop) -> SelfQuery:
"""Computes the "slice" of the :class:`_query.Query` represented by
the given indices and returns the resulting :class:`_query.Query`.
self._limit_clause, self._offset_clause = sql_util._make_slice(
self._limit_clause, self._offset_clause, start, stop
)
+ return self
@_generative
@_assertions(_no_statement_condition)
- def limit(self, limit):
+ def limit(self: SelfQuery, limit) -> SelfQuery:
"""Apply a ``LIMIT`` to the query and return the newly resulting
``Query``.
"""
self._limit_clause = sql_util._offset_or_limit_clause(limit)
+ return self
@_generative
@_assertions(_no_statement_condition)
- def offset(self, offset):
+ def offset(self: SelfQuery, offset) -> SelfQuery:
"""Apply an ``OFFSET`` to the query and return the newly resulting
``Query``.
"""
self._offset_clause = sql_util._offset_or_limit_clause(offset)
+ return self
@_generative
@_assertions(_no_statement_condition)
- def distinct(self, *expr):
+ def distinct(self: SelfQuery, *expr) -> SelfQuery:
r"""Apply a ``DISTINCT`` to the query and return the newly resulting
``Query``.
)
else:
self._distinct = True
+ return self
def all(self):
"""Return the results represented by this :class:`_query.Query`
@_generative
@_assertions(_no_clauseelement_condition)
- def from_statement(self, statement):
+ def from_statement(self: SelfQuery, statement) -> SelfQuery:
"""Execute the given SELECT statement and return results.
This method bypasses all internal statement compilation, and the
roles.SelectStatementRole, statement, apply_propagate_attrs=self
)
self._statement = statement
+ return self
def first(self):
"""Return the first result of this ``Query`` or
from typing import Any
from typing import cast
from typing import Mapping
+from typing import NoReturn
from typing import Tuple
from typing import Union
if typing.TYPE_CHECKING:
from .mapper import Mapper
+Self_AbstractLoad = typing.TypeVar("Self_AbstractLoad", bound="_AbstractLoad")
+
class _AbstractLoad(Generative, LoaderOption):
_is_strategy_option = True
@_generative
def _set_relationship_strategy(
- self,
+ self: Self_AbstractLoad,
attr,
strategy,
propagate_to_loaders=True,
opts=None,
_reconcile_to_other=None,
- ) -> "_AbstractLoad":
+ ) -> Self_AbstractLoad:
strategy = self._coerce_strat(strategy)
self._clone_for_bind_strategy(
@_generative
def _set_column_strategy(
- self, attrs, strategy, opts=None
- ) -> "_AbstractLoad":
+ self: Self_AbstractLoad, attrs, strategy, opts=None
+ ) -> Self_AbstractLoad:
strategy = self._coerce_strat(strategy)
self._clone_for_bind_strategy(
@_generative
def _set_generic_strategy(
- self, attrs, strategy, _reconcile_to_other=None
- ) -> "_AbstractLoad":
+ self: Self_AbstractLoad, attrs, strategy, _reconcile_to_other=None
+ ) -> Self_AbstractLoad:
strategy = self._coerce_strat(strategy)
self._clone_for_bind_strategy(
attrs,
return self
@_generative
- def _set_class_strategy(self, strategy, opts) -> "_AbstractLoad":
+ def _set_class_strategy(
+ self: Self_AbstractLoad, strategy, opts
+ ) -> Self_AbstractLoad:
strategy = self._coerce_strat(strategy)
self._clone_for_bind_strategy(None, strategy, None, opts=opts)
"""
raise NotImplementedError()
- def options(self, *opts) -> "_AbstractLoad":
+ def options(self: Self_AbstractLoad, *opts) -> NoReturn:
r"""Apply a series of options as sub-options to this
:class:`_orm._AbstractLoad` object.
return to_chop[i + 1 :]
+SelfLoad = typing.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, *opts) -> "_AbstractLoad":
+ def options(self: SelfLoad, *opts) -> SelfLoad:
r"""Apply a series of options as sub-options to this
:class:`_orm.Load`
object.
self.path = PathRegistry.deserialize(self.path)
+SelfWildcardLoad = typing.TypeVar("SelfWildcardLoad", bound="_WildcardLoad")
+
+
class _WildcardLoad(_AbstractLoad):
"""represent a standalone '*' load operation"""
if opts:
self.local_opts = util.immutabledict(opts)
- def options(self, *opts) -> "_AbstractLoad":
+ def options(self: SelfWildcardLoad, *opts) -> SelfWildcardLoad:
raise NotImplementedError("Star option does not support sub-options")
def _apply_to_parent(self, parent):
return [("loader", cast(PathRegistry, effective_path).natural_path)]
-def _generate_from_keys(meth, keys, chained, kw):
+def _generate_from_keys(meth, keys, chained, kw) -> _AbstractLoad:
lead_element = None
else:
lead_element = meth(lead_element, attr, **kw)
+ assert lead_element
return lead_element
@loader_unbound_fn
-def contains_eager(*keys, **kw):
+def contains_eager(*keys, **kw) -> _AbstractLoad:
return _generate_from_keys(Load.contains_eager, keys, True, kw)
@loader_unbound_fn
-def load_only(*attrs):
+def load_only(*attrs) -> _AbstractLoad:
# TODO: attrs against different classes. we likely have to
# add some extra state to Load of some kind
_, lead_element, _ = _parse_attr_argument(attrs[0])
@loader_unbound_fn
-def joinedload(*keys, **kw):
+def joinedload(*keys, **kw) -> _AbstractLoad:
return _generate_from_keys(Load.joinedload, keys, False, kw)
@loader_unbound_fn
-def subqueryload(*keys):
+def subqueryload(*keys) -> _AbstractLoad:
return _generate_from_keys(Load.subqueryload, keys, False, {})
@loader_unbound_fn
-def selectinload(*keys):
+def selectinload(*keys) -> _AbstractLoad:
return _generate_from_keys(Load.selectinload, keys, False, {})
@loader_unbound_fn
-def lazyload(*keys):
+def lazyload(*keys) -> _AbstractLoad:
return _generate_from_keys(Load.lazyload, keys, False, {})
@loader_unbound_fn
-def immediateload(*keys):
+def immediateload(*keys) -> _AbstractLoad:
return _generate_from_keys(Load.immediateload, keys, False, {})
@loader_unbound_fn
-def noload(*keys):
+def noload(*keys) -> _AbstractLoad:
return _generate_from_keys(Load.noload, keys, False, {})
@loader_unbound_fn
-def raiseload(*keys, **kw):
+def raiseload(*keys, **kw) -> _AbstractLoad:
return _generate_from_keys(Load.raiseload, keys, False, kw)
@loader_unbound_fn
-def defaultload(*keys):
+def defaultload(*keys) -> _AbstractLoad:
return _generate_from_keys(Load.defaultload, keys, False, {})
@loader_unbound_fn
-def defer(key, *addl_attrs, **kw):
+def defer(key, *addl_attrs, **kw) -> _AbstractLoad:
if addl_attrs:
util.warn_deprecated(
"The *addl_attrs on orm.defer is deprecated. Please use "
@loader_unbound_fn
-def undefer(key, *addl_attrs):
+def undefer(key, *addl_attrs) -> _AbstractLoad:
if addl_attrs:
util.warn_deprecated(
"The *addl_attrs on orm.undefer is deprecated. Please use "
@loader_unbound_fn
-def undefer_group(name):
+def undefer_group(name) -> _AbstractLoad:
element = _WildcardLoad()
return element.undefer_group(name)
@loader_unbound_fn
-def with_expression(key, expression):
+def with_expression(key, expression) -> _AbstractLoad:
return _generate_from_keys(
Load.with_expression, (key,), False, {"expression": expression}
)
@loader_unbound_fn
-def selectin_polymorphic(base_cls, classes):
+def selectin_polymorphic(base_cls, classes) -> _AbstractLoad:
ul = Load(base_cls)
return ul.selectin_polymorphic(classes)
from itertools import zip_longest
import operator
import re
+import typing
from . import roles
from . import visitors
from .. import util
from ..util import HasMemoized
from ..util import hybridmethod
+from ..util import typing as compat_typing
try:
from sqlalchemy.cyextension.util import prefix_anon_map # noqa
NO_ARG = util.symbol("NO_ARG")
+# if I use sqlalchemy.util.typing, which has the exact same
+# symbols, mypy reports: "error: _Fn? not callable"
+_Fn = typing.TypeVar("_Fn", bound=typing.Callable)
+
class Immutable:
"""mark a ClauseElement as 'immutable' when expressions are cloned."""
)
-def _generative(fn):
+_Self = typing.TypeVar("_Self", bound="_GenerativeType")
+_Args = compat_typing.ParamSpec("_Args")
+
+
+class _GenerativeType(compat_typing.Protocol):
+ def _generate(self: "_Self") -> "_Self":
+ ...
+
+
+def _generative(fn: _Fn) -> _Fn:
"""non-caching _generative() decorator.
This is basically the legacy decorator that copies the object and
"""
@util.decorator
- def _generative(fn, self, *args, **kw):
+ def _generative(
+ fn: _Fn, self: _Self, *args: _Args.args, **kw: _Args.kwargs
+ ) -> _Self:
"""Mark a method as generative."""
self = self._generate()
x = fn(self, *args, **kw)
- assert (
- x is None or x is self
- ), "generative methods must return None or self"
+ assert x is self, "generative methods must return self"
return self
decorated = _generative(fn)
return c
+SelfExecutable = typing.TypeVar("SelfExecutable", bound="Executable")
+
+
class Executable(roles.StatementRole, Generative):
"""Mark a :class:`_expression.ClauseElement` as supporting execution.
return self.__visit_name__
@_generative
- def options(self, *options):
+ def options(self: SelfExecutable, *options) -> SelfExecutable:
"""Apply options to this statement.
In the general sense, options are any kind of Python object
coercions.expect(roles.ExecutableOptionRole, opt)
for opt in options
)
+ return self
@_generative
- def _set_compile_options(self, compile_options):
+ def _set_compile_options(
+ self: SelfExecutable, compile_options
+ ) -> SelfExecutable:
"""Assign the compile options to a new value.
:param compile_options: appropriate CacheableOptions structure
"""
self._compile_options = compile_options
+ return self
@_generative
- def _update_compile_options(self, options):
+ def _update_compile_options(
+ self: SelfExecutable, options
+ ) -> SelfExecutable:
"""update the _compile_options with new keys."""
self._compile_options += options
+ return self
@_generative
- def _add_context_option(self, callable_, cache_args):
+ def _add_context_option(
+ self: SelfExecutable, callable_, cache_args
+ ) -> SelfExecutable:
"""Add a context option to this statement.
These are callable functions that will
"""
self._with_context_options += ((callable_, cache_args),)
+ return self
@_generative
- def execution_options(self, **kw):
+ def execution_options(self: SelfExecutable, **kw) -> SelfExecutable:
"""Set non-SQL options for the statement which take effect during
execution.
"on Connection.execution_options(), not per statement."
)
self._execution_options = self._execution_options.union(kw)
+ return self
def get_execution_options(self):
"""Get the non-SQL options which will take effect during execution.
to invoke them for a create/drop call.
"""
+import typing
from . import roles
from .base import _generative
raise NotImplementedError()
+SelfDDLElement = typing.TypeVar("SelfDDLElement", bound="DDLElement")
+
+
class DDLElement(roles.DDLRole, Executable, _DDLCompiles):
"""Base class for DDL expression constructs.
)
@_generative
- def against(self, target):
+ def against(self: SelfDDLElement, target) -> SelfDDLElement:
"""Return a copy of this :class:`_schema.DDLElement` which will include
the given target.
"""
self.target = target
+ return self
@_generative
- def execute_if(self, dialect=None, callable_=None, state=None):
+ def execute_if(
+ self: SelfDDLElement, dialect=None, callable_=None, state=None
+ ) -> SelfDDLElement:
r"""Return a callable that will execute this
:class:`_ddl.DDLElement` conditionally within an event handler.
self.dialect = dialect
self.callable_ = callable_
self.state = state
+ return self
def _should_execute(self, target, bind, **kw):
if isinstance(self.dialect, str):
"""
import collections.abc as collections_abc
+import typing
-from sqlalchemy.types import NullType
from . import coercions
from . import roles
from . import util as sql_util
from .selectable import HasCTE
from .selectable import HasPrefixes
from .selectable import ReturnsRows
+from .sqltypes import NullType
from .visitors import InternalTraversal
from .. import exc
from .. import util
self._extra_froms = self._make_extra_froms(statement)
+SelfUpdateBase = typing.TypeVar("SelfUpdateBase", bound="UpdateBase")
+
+
class UpdateBase(
roles.DMLRole,
HasCTE,
)
@_generative
- def with_dialect_options(self, **opt):
+ def with_dialect_options(self: SelfUpdateBase, **opt) -> SelfUpdateBase:
"""Add dialect options to this INSERT/UPDATE/DELETE object.
e.g.::
"""
self._validate_dialect_kwargs(opt)
+ return self
def _validate_dialect_kwargs_deprecated(self, dialect_kw):
util.warn_deprecated_20(
self._validate_dialect_kwargs(dialect_kw)
@_generative
- def returning(self, *cols):
+ def returning(self: SelfUpdateBase, *cols) -> SelfUpdateBase:
r"""Add a :term:`RETURNING` or equivalent clause to this statement.
e.g.:
self._returning += tuple(
coercions.expect(roles.ColumnsClauseRole, c) for c in cols
)
+ return self
@property
def _all_selected_columns(self):
).as_immutable()
@_generative
- def with_hint(self, text, selectable=None, dialect_name="*"):
+ def with_hint(
+ self: SelfUpdateBase, text, selectable=None, dialect_name="*"
+ ) -> SelfUpdateBase:
"""Add a table hint for a single table to this
INSERT/UPDATE/DELETE statement.
selectable = self.table
self._hints = self._hints.union({(selectable, dialect_name): text})
+ return self
+
+
+SelfValuesBase = typing.TypeVar("SelfValuesBase", bound="ValuesBase")
class ValuesBase(UpdateBase):
"values present",
},
)
- def values(self, *args, **kwargs):
+ def values(self: SelfValuesBase, *args, **kwargs) -> SelfValuesBase:
r"""Specify a fixed VALUES clause for an INSERT statement, or the SET
clause for an UPDATE.
if arg and isinstance(arg[0], (list, dict, tuple)):
self._multi_values += (arg,)
- return
+ return self
# tuple values
arg = {c.key: value for c, value in zip(self.table.c, arg)}
self._values = self._values.union(arg)
else:
self._values = util.immutabledict(arg)
+ return self
@_generative
@_exclusive_against(
},
defaults={"_returning": _returning},
)
- def return_defaults(self, *cols):
+ def return_defaults(self: SelfValuesBase, *cols) -> SelfValuesBase:
"""Make use of a :term:`RETURNING` clause for the purpose
of fetching server-side expressions and defaults.
"""
self._return_defaults = True
self._return_defaults_columns = cols
+ return self
+
+
+SelfInsert = typing.TypeVar("SelfInsert", bound="Insert")
class Insert(ValuesBase):
self._return_defaults_columns = return_defaults
@_generative
- def inline(self):
+ def inline(self: SelfInsert) -> SelfInsert:
"""Make this :class:`_expression.Insert` construct "inline" .
When set, no attempt will be made to retrieve the
"""
self._inline = True
+ return self
@_generative
- def from_select(self, names, select, include_defaults=True):
+ def from_select(
+ self: SelfInsert, names, select, include_defaults=True
+ ) -> SelfInsert:
"""Return a new :class:`_expression.Insert` construct which represents
an ``INSERT...FROM SELECT`` statement.
self._inline = True
self.include_insert_from_select_defaults = include_defaults
self.select = coercions.expect(roles.DMLSelectRole, select)
+ return self
+
+
+SelfDMLWhereBase = typing.TypeVar("SelfDMLWhereBase", bound="DMLWhereBase")
class DMLWhereBase:
_where_criteria = ()
@_generative
- def where(self, *whereclause):
+ def where(self: SelfDMLWhereBase, *whereclause) -> SelfDMLWhereBase:
"""Return a new construct with the given expression(s) added to
its WHERE clause, joined to the existing clause via AND, if any.
for criterion in whereclause:
where_criteria = coercions.expect(roles.WhereHavingRole, criterion)
self._where_criteria += (where_criteria,)
+ return self
- def filter(self, *criteria):
+ def filter(self: SelfDMLWhereBase, *criteria) -> SelfDMLWhereBase:
"""A synonym for the :meth:`_dml.DMLWhereBase.where` method.
.. versionadded:: 1.4
def _filter_by_zero(self):
return self.table
- def filter_by(self, **kwargs):
+ def filter_by(self: SelfDMLWhereBase, **kwargs) -> SelfDMLWhereBase:
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.
self._return_defaults = return_defaults
@_generative
- def ordered_values(self, *args):
+ def ordered_values(self: SelfUpdate, *args) -> SelfUpdate:
"""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.
kv_generator = DMLState.get_plugin_class(self)._get_crud_kv_pairs
self._ordered_values = kv_generator(self, args)
+ return self
@_generative
- def inline(self):
+ def inline(self: SelfUpdate) -> SelfUpdate:
"""Make this :class:`_expression.Update` construct "inline" .
When set, SQL defaults present on :class:`_schema.Column`
"""
self._inline = True
+ return self
+
+
+SelfDelete = typing.TypeVar("SelfDelete", bound="Delete")
class Delete(DMLWhereBase, UpdateBase):
import itertools
import operator
import re
+import typing
from . import coercions
from . import operators
self.type = type_
+SelfTextClause = typing.TypeVar("SelfTextClause", bound="TextClause")
+
+
class TextClause(
roles.DDLConstraintColumnRole,
roles.DDLExpressionRole,
return TextClause(text)
@_generative
- def bindparams(self, *binds, **names_to_values):
+ def bindparams(
+ self: SelfTextClause, *binds, **names_to_values
+ ) -> SelfTextClause:
"""Establish the values and/or types of bound parameters within
this :class:`_expression.TextClause` construct.
) from err
else:
new_params[key] = existing._with_value(value, required=False)
+ return self
@util.preload_module("sqlalchemy.sql.selectable")
def columns(self, *cols, **types):
]
+from typing import Callable
+
from .base import _from_objects
from .base import _select_iterables
from .base import ColumnCollection
from .visitors import Visitable
from ..util.langhelpers import public_factory
-# factory functions - these pull class-bound constructors and classmethods
-# from SQL elements and selectables into public functions. This allows
-# the functions to be available in the sqlalchemy.sql.* namespace and
-# to be auto-cross-documenting from the function to the class itself.
+# TODO: proposal is to remove public_factory and replace with traditional
+# functions exported here.
all_ = public_factory(CollectionAggregate._create_all, ".sql.expression.all_")
any_ = public_factory(CollectionAggregate._create_any, ".sql.expression.any_")
import collections
import itertools
from operator import attrgetter
+import typing
+from typing import Type
+from typing import Union
from . import coercions
from . import operators
)
+SelfHasPrefixes = typing.TypeVar("SelfHasPrefixes", bound="HasPrefixes")
+
+
class HasPrefixes:
_prefixes = ()
":meth:`_expression.HasPrefixes.prefix_with`",
":paramref:`.HasPrefixes.prefix_with.*expr`",
)
- def prefix_with(self, *expr, **kw):
+ def prefix_with(self: SelfHasPrefixes, *expr, **kw) -> SelfHasPrefixes:
r"""Add one or more expressions following the statement keyword, i.e.
SELECT, INSERT, UPDATE, or DELETE. Generative.
"Unsupported argument(s): %s" % ",".join(kw)
)
self._setup_prefixes(expr, dialect)
+ return self
def _setup_prefixes(self, prefixes, dialect=None):
self._prefixes = self._prefixes + tuple(
)
+SelfHasSuffixes = typing.TypeVar("SelfHasSuffixes", bound="HasSuffixes")
+
+
class HasSuffixes:
_suffixes = ()
":meth:`_expression.HasSuffixes.suffix_with`",
":paramref:`.HasSuffixes.suffix_with.*expr`",
)
- def suffix_with(self, *expr, **kw):
+ def suffix_with(self: SelfHasSuffixes, *expr, **kw) -> SelfHasSuffixes:
r"""Add one or more expressions following the statement as a whole.
This is used to support backend-specific suffix keywords on
"Unsupported argument(s): %s" % ",".join(kw)
)
self._setup_suffixes(expr, dialect)
+ return self
def _setup_suffixes(self, suffixes, dialect=None):
self._suffixes = self._suffixes + tuple(
)
+SelfHasHints = typing.TypeVar("SelfHasHints", bound="HasHints")
+
+
class HasHints:
_hints = util.immutabledict()
_statement_hints = ()
return self.with_hint(None, text, dialect_name)
@_generative
- def with_hint(self, selectable, text, dialect_name="*"):
+ def with_hint(
+ self: SelfHasHints, selectable, text, dialect_name="*"
+ ) -> SelfHasHints:
r"""Add an indexing or other executional context hint for the given
selectable to this :class:`_expression.Select` or other selectable
object.
): text
}
)
+ return self
class FromClause(roles.AnonymizedFromClauseRole, Selectable):
return self._restates if self._restates is not None else self
+SelfHasCTE = typing.TypeVar("SelfHasCTE", bound="HasCTE")
+
+
class HasCTE(roles.HasCTERole):
"""Mixin that declares a class to include CTE support.
_independent_ctes = ()
@_generative
- def add_cte(self, cte):
+ def add_cte(self: SelfHasCTE, cte) -> SelfHasCTE:
"""Add a :class:`_sql.CTE` to this statement object that will be
independently rendered even if not referenced in the statement
otherwise.
"""
cte = coercions.expect(roles.IsCTERole, cte)
self._independent_ctes += (cte,)
+ return self
def cte(self, name=None, recursive=False, nesting=False):
r"""Return a new :class:`_expression.CTE`,
self.of = None
+SelfValues = typing.TypeVar("SelfValues", bound="Values")
+
+
class Values(Generative, FromClause):
"""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, name, **kw):
+ def alias(self: SelfValues, name, **kw) -> SelfValues:
"""Return a new :class:`_expression.Values`
construct that is a copy of this
one with the given name.
"""
self.name = name
self.named_with_column = self.name is not None
+ return self
@_generative
- def lateral(self, name=None):
+ def lateral(self: SelfValues, name=None) -> SelfValues:
"""Return a new :class:`_expression.Values` with the lateral flag set,
so that
it renders as LATERAL.
self._is_lateral = True
if name is not None:
self.name = name
+ return self
@_generative
- def data(self, values):
+ def data(self: SelfValues, values) -> SelfValues:
"""Return a new :class:`_expression.Values` construct,
adding the given data
to the data list.
"""
self._data += (values,)
+ return self
def _populate_column_collection(self):
for c in self._column_args:
self.group_by.non_generative(self, *clauses)
+SelfGenerativeSelect = typing.TypeVar(
+ "SelfGenerativeSelect", bound="GenerativeSelect"
+)
+
+
class GenerativeSelect(DeprecatedSelectBaseGenerations, SelectBase):
"""Base class for SELECT statements where additional elements can be
added.
@_generative
def with_for_update(
- self,
+ self: SelfGenerativeSelect,
nowait=False,
read=False,
of=None,
skip_locked=False,
key_share=False,
- ):
+ ) -> SelfGenerativeSelect:
"""Specify a ``FOR UPDATE`` clause for this
:class:`_expression.GenerativeSelect`.
skip_locked=skip_locked,
key_share=key_share,
)
+ return self
def get_label_style(self):
"""
)
@_generative
- def limit(self, limit):
+ def limit(self: SelfGenerativeSelect, limit) -> SelfGenerativeSelect:
"""Return a new selectable with the given LIMIT criterion
applied.
self._fetch_clause = self._fetch_clause_options = None
self._limit_clause = self._offset_or_limit_clause(limit)
+ return self
@_generative
- def fetch(self, count, with_ties=False, percent=False):
+ def fetch(
+ self: SelfGenerativeSelect, count, with_ties=False, percent=False
+ ) -> SelfGenerativeSelect:
"""Return a new selectable with the given FETCH FIRST criterion
applied.
"with_ties": with_ties,
"percent": percent,
}
+ return self
@_generative
- def offset(self, offset):
+ def offset(self: SelfGenerativeSelect, offset) -> SelfGenerativeSelect:
"""Return a new selectable with the given OFFSET criterion
applied.
"""
self._offset_clause = self._offset_or_limit_clause(offset)
+ return self
@_generative
@util.preload_module("sqlalchemy.sql.util")
- def slice(self, start, stop):
+ def slice(self: SelfGenerativeSelect, start, stop) -> SelfGenerativeSelect:
"""Apply LIMIT / OFFSET to this statement based on a slice.
The start and stop indices behave like the argument to Python's
self._limit_clause, self._offset_clause = sql_util._make_slice(
self._limit_clause, self._offset_clause, start, stop
)
+ return self
@_generative
- def order_by(self, *clauses):
+ def order_by(self: SelfGenerativeSelect, *clauses) -> SelfGenerativeSelect:
r"""Return a new selectable with the given list of ORDER BY
criteria applied.
coercions.expect(roles.OrderByRole, clause)
for clause in clauses
)
+ return self
@_generative
- def group_by(self, *clauses):
+ def group_by(self: SelfGenerativeSelect, *clauses) -> SelfGenerativeSelect:
r"""Return a new selectable with the given list of GROUP BY
criterion applied.
coercions.expect(roles.GroupByRole, clause)
for clause in clauses
)
+ return self
@CompileState.plugin_for("default", "compound_select")
) = select_stmt._with_options = ()
+# TODO: use pep-673 when feasible
+SelfSelect = typing.TypeVar("SelfSelect", bound="Select")
+
+
class Select(
HasPrefixes,
HasSuffixes,
]
@classmethod
- def _create(cls, *entities) -> "Select":
+ def _create(
+ cls, *entities: Union[roles.ColumnsClauseRole, Type]
+ ) -> "Select":
r"""Construct a new :class:`_expression.Select`.
return self
@classmethod
- def _create_raw_select(cls, **kw):
+ def _create_raw_select(cls, **kw) -> "Select":
"""Create a :class:`.Select` using raw ``__new__`` with no coercions.
Used internally to build up :class:`.Select` constructs with
return meth(self, statement)
@_generative
- def join(self, target, onclause=None, isouter=False, full=False):
+ def join(
+ self: SelfSelect, target, onclause=None, isouter=False, full=False
+ ) -> SelfSelect:
r"""Create a SQL JOIN against this :class:`_expression.Select`
object's criterion
and apply generatively, returning the newly resulting
self._setup_joins += (
(target, onclause, None, {"isouter": isouter, "full": full}),
)
+ return self
def outerjoin_from(self, from_, target, onclause=None, full=False):
r"""Create a SQL LEFT OUTER JOIN against this :class:`_expression.Select`
@_generative
def join_from(
- self, from_, target, onclause=None, isouter=False, full=False
- ):
+ self: SelfSelect,
+ from_,
+ target,
+ onclause=None,
+ isouter=False,
+ full=False,
+ ) -> SelfSelect:
r"""Create a SQL JOIN against this :class:`_expression.Select`
object's criterion
and apply generatively, returning the newly resulting
self._setup_joins += (
(target, onclause, from_, {"isouter": isouter, "full": full}),
)
+ return self
def outerjoin(self, target, onclause=None, full=False):
"""Create a left outer join.
)
@_generative
- def add_columns(self, *columns):
+ def add_columns(self: SelfSelect, *columns) -> SelfSelect:
"""Return a new :func:`_expression.select` construct with
the given column expressions added to its columns clause.
)
for column in columns
]
+ return self
def _set_entities(self, entities):
self._raw_columns = [
)
@_generative
- def with_only_columns(self, *columns, **kw):
+ def with_only_columns(self: SelfSelect, *columns, **kw) -> SelfSelect:
r"""Return a new :func:`_expression.select` construct with its columns
clause replaced with the given columns.
"columns", "Select.with_only_columns", columns
)
]
+ return self
@property
def whereclause(self):
_whereclause = whereclause
@_generative
- def where(self, *whereclause):
+ def where(self: SelfSelect, *whereclause) -> SelfSelect:
"""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.
for criterion in whereclause:
where_criteria = coercions.expect(roles.WhereHavingRole, criterion)
self._where_criteria += (where_criteria,)
+ return self
@_generative
- def having(self, having):
+ def having(self: SelfSelect, having) -> SelfSelect:
"""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.
self._having_criteria += (
coercions.expect(roles.WhereHavingRole, having),
)
+ return self
@_generative
- def distinct(self, *expr):
+ def distinct(self: SelfSelect, *expr) -> SelfSelect:
r"""Return a new :func:`_expression.select` construct which
will apply DISTINCT to its columns clause.
)
else:
self._distinct = True
+ return self
@_generative
- def select_from(self, *froms):
+ def select_from(self: SelfSelect, *froms) -> SelfSelect:
r"""Return a new :func:`_expression.select` construct with the
given FROM expression(s)
merged into its list of FROM objects.
)
for fromclause in froms
)
+ return self
@_generative
- def correlate(self, *fromclauses):
+ def correlate(self: SelfSelect, *fromclauses) -> SelfSelect:
r"""Return a new :class:`_expression.Select`
which will correlate the given FROM
clauses to that of an enclosing :class:`_expression.Select`.
self._correlate = self._correlate + tuple(
coercions.expect(roles.FromClauseRole, f) for f in fromclauses
)
+ return self
@_generative
- def correlate_except(self, *fromclauses):
+ def correlate_except(self: SelfSelect, *fromclauses) -> SelfSelect:
r"""Return a new :class:`_expression.Select`
which will omit the given FROM
clauses from the auto-correlation process.
self._correlate_except = (self._correlate_except or ()) + tuple(
coercions.expect(roles.FromClauseRole, f) for f in fromclauses
)
+ return self
@HasMemoized.memoized_attribute
def selected_columns(self):
return CompoundSelect._create_intersect_all(self, *other, **kwargs)
+SelfScalarSelect = typing.TypeVar("SelfScalarSelect", bound="ScalarSelect")
+
+
class ScalarSelect(roles.InElementRole, Generative, Grouping):
"""Represent a scalar subquery.
c = columns
@_generative
- def where(self, crit):
+ def where(self: SelfScalarSelect, crit) -> SelfScalarSelect:
"""Apply a WHERE clause to the SELECT statement referred to
by this :class:`_expression.ScalarSelect`.
"""
self.element = self.element.where(crit)
+ return self
def self_group(self, **kwargs):
return self
@_generative
- def correlate(self, *fromclauses):
+ def correlate(self: SelfScalarSelect, *fromclauses) -> SelfScalarSelect:
r"""Return a new :class:`_expression.ScalarSelect`
which will correlate the given FROM
clauses to that of an enclosing :class:`_expression.Select`.
"""
self.element = self.element.correlate(*fromclauses)
+ return self
@_generative
- def correlate_except(self, *fromclauses):
+ def correlate_except(
+ self: SelfScalarSelect, *fromclauses
+ ) -> SelfScalarSelect:
r"""Return a new :class:`_expression.ScalarSelect`
which will omit the given FROM
clauses from the auto-correlation process.
"""
self.element = self.element.correlate_except(*fromclauses)
+ return self
class Exists(UnaryExpression):
return e
+SelfTextualSelect = typing.TypeVar("SelfTextualSelect", bound="TextualSelect")
+
+
class TextualSelect(SelectBase):
"""Wrap a :class:`_expression.TextClause` construct within a
:class:`_expression.SelectBase`
return self
@_generative
- def bindparams(self, *binds, **bind_as_values):
+ def bindparams(
+ self: SelfTextualSelect, *binds, **bind_as_values
+ ) -> SelfTextualSelect:
self.element = self.element.bindparams(*binds, **bind_as_values)
+ return self
def _generate_fromclause_column_proxies(self, fromclause):
fromclause._columns._populate_separate_keys(
import operator
import platform
import sys
+import typing
+
py311 = sys.version_info >= (3, 11)
+py310 = sys.version_info >= (3, 10)
py39 = sys.version_info >= (3, 9)
py38 = sys.version_info >= (3, 8)
pypy = platform.python_implementation() == "PyPy"
"""vendored from python 3.7"""
if getattr(annotation, "__module__", None) == "typing":
- return repr(annotation).replace("typing.", "")
+ return f'"{repr(annotation).replace("typing.", "")}"'
if isinstance(annotation, type):
if annotation.__module__ in ("builtins", base_module):
return repr(annotation.__qualname__)
return annotation.__module__ + "." + annotation.__qualname__
+ elif isinstance(annotation, typing.TypeVar):
+ return f'"{annotation}"'
return repr(annotation)
import sys
import textwrap
import types
+import typing
from typing import Any
from typing import Callable
from typing import Generic
from . import _collections
from . import compat
+from . import typing as compat_typing
from .. import exc
_T = TypeVar("_T")
-_MP = TypeVar("_MP", bound="memoized_property[Any]")
def md5_hex(x):
n ^= b
-def decorator(target):
+_Fn = typing.TypeVar("_Fn", bound=typing.Callable)
+_Args = compat_typing.ParamSpec("_Args")
+
+
+def decorator(
+ target: typing.Callable[compat_typing.Concatenate[_Fn, _Args], typing.Any]
+) -> _Fn:
"""A signature-matching decorator factory."""
def decorate(fn):
decorated.__wrapped__ = fn
return update_wrapper(decorated, fn)
- return update_wrapper(decorate, target)
+ return typing.cast(_Fn, update_wrapper(decorate, target))
def _update_argspec_defaults_into_env(spec, env):
return env[fn_name]
-def public_factory(target, location, class_location=None):
+_TE = TypeVar("_TE")
+
+_P = compat_typing.ParamSpec("_P")
+
+
+def public_factory(
+ target: typing.Callable[_P, _TE],
+ location: str,
+ class_location: Optional[str] = None,
+) -> typing.Callable[_P, _TE]:
"""Produce a wrapping function for the given cls or classmethod.
Rationale here is so that the __init__ method of the
"__name__": callable_.__module__,
}
exec(code, env)
+
decorated = env[location_name]
if hasattr(fn, "_linked_to"):
)
+Selfmemoized_property = TypeVar(
+ "Selfmemoized_property", bound="memoized_property[Any]"
+)
+
+
class memoized_property(Generic[_T]):
"""A read-only @property that is only evaluated once."""
self.__name__ = fget.__name__
@overload
- def __get__(self: _MP, obj: None, cls: Any) -> _MP:
+ def __get__(
+ self: Selfmemoized_property, obj: None, cls: Any
+ ) -> Selfmemoized_property:
...
@overload
def __get__(self, obj: Any, cls: Any) -> _T:
...
- def __get__(self: _MP, obj: Any, cls: Any) -> Union[_MP, _T]:
+ def __get__(
+ self: Selfmemoized_property, obj: Any, cls: Any
+ ) -> Union[Selfmemoized_property, _T]:
if obj is None:
return self
obj.__dict__[self.__name__] = result = self.fget(obj)
from typing import Any
+from typing import Callable # noqa
from typing import Generic
from typing import overload
from typing import Type
from typing_extensions import Protocol # noqa
from typing_extensions import TypedDict # noqa
+if compat.py310:
+ from typing import Concatenate
+ from typing import ParamSpec
+else:
+ from typing_extensions import Concatenate # noqa
+ from typing_extensions import ParamSpec # noqa
if compat.py311:
from typing import NotRequired # noqa
[options]
packages = find:
+include_package_data = True
python_requires = >=3.7
package_dir =
=lib
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
