from enum import Enum
import itertools
import typing
+from typing import AbstractSet
from typing import Any as TODO_Any
from typing import Any
+from typing import Callable
+from typing import cast
+from typing import Dict
from typing import Iterable
+from typing import Iterator
from typing import List
from typing import NamedTuple
+from typing import NoReturn
from typing import Optional
+from typing import overload
from typing import Sequence
+from typing import Set
from typing import Tuple
+from typing import Type
from typing import TYPE_CHECKING
from typing import TypeVar
+from typing import Union
from . import cache_key
from . import coercions
from . import visitors
from ._typing import _ColumnsClauseArgument
from ._typing import is_column_element
+from ._typing import is_select_statement
+from ._typing import is_subquery
+from ._typing import is_table
from .annotation import Annotated
from .annotation import SupportsCloneAnnotations
from .base import _clone
from .elements import ColumnElement
from .elements import DQLDMLClauseElement
from .elements import GroupedElement
-from .elements import Grouping
from .elements import literal_column
from .elements import TableValuedColumn
from .elements import UnaryExpression
+from .operators import OperatorType
+from .visitors import _TraverseInternalsType
from .visitors import InternalTraversal
from .visitors import prefix_anon_map
from .. import exc
from .. import util
+from ..util import HasMemoized_ro_memoized_attribute
+from ..util.typing import Literal
+from ..util.typing import Protocol
+from ..util.typing import Self
and_ = BooleanClauseList.and_
_T = TypeVar("_T", bound=Any)
if TYPE_CHECKING:
- from ._typing import _SelectIterable
+ from ._typing import _ColumnExpressionArgument
+ from ._typing import _FromClauseArgument
+ from ._typing import _JoinTargetArgument
+ from ._typing import _OnClauseArgument
+ from ._typing import _SelectStatementForCompoundArgument
+ from ._typing import _TextCoercedExpressionArgument
+ from ._typing import _TypeEngineArgument
+ from .base import _AmbiguousTableNameMap
+ from .base import ExecutableOption
from .base import ReadOnlyColumnCollection
+ from .cache_key import _CacheKeyTraversalType
+ from .compiler import SQLCompiler
+ from .dml import Delete
+ from .dml import Insert
+ from .dml import Update
from .elements import NamedColumn
+ from .elements import TextClause
+ from .functions import Function
+ from .schema import Column
from .schema import ForeignKey
- from .schema import PrimaryKeyConstraint
+ from .schema import ForeignKeyConstraint
+ from .type_api import TypeEngine
+ from .util import ClauseAdapter
+ from .visitors import _CloneCallableType
-class _OffsetLimitParam(BindParameter):
+_ColumnsClauseElement = Union["FromClause", ColumnElement[Any], "TextClause"]
+
+
+class _JoinTargetProtocol(Protocol):
+ @util.ro_non_memoized_property
+ def _from_objects(self) -> List[FromClause]:
+ ...
+
+
+_JoinTargetElement = Union["FromClause", _JoinTargetProtocol]
+_OnClauseElement = Union["ColumnElement[bool]", _JoinTargetProtocol]
+
+
+_SetupJoinsElement = Tuple[
+ _JoinTargetElement,
+ Optional[_OnClauseElement],
+ Optional["FromClause"],
+ Dict[str, Any],
+]
+
+
+_SelectIterable = Iterable[Union["ColumnElement[Any]", "TextClause"]]
+
+
+class _OffsetLimitParam(BindParameter[int]):
inherit_cache = True
@property
- def _limit_offset_value(self):
+ def _limit_offset_value(self) -> Optional[int]:
return self.effective_value
# sub-elements of returns_rows
_is_from_clause = False
+ _is_select_base = False
_is_select_statement = False
_is_lateral = False
@property
- def selectable(self):
+ def selectable(self) -> ReturnsRows:
return self
@util.non_memoized_property
"""
raise NotImplementedError()
+ def is_derived_from(self, fromclause: FromClause) -> bool:
+ """Return ``True`` if this :class:`.ReturnsRows` is
+ 'derived' from the given :class:`.FromClause`.
+
+ An example would be an Alias of a Table is derived from that Table.
+
+ """
+ raise NotImplementedError()
+
+ def _generate_fromclause_column_proxies(
+ self, fromclause: FromClause
+ ) -> None:
+ """Populate columns into an :class:`.AliasedReturnsRows` object."""
+
+ raise NotImplementedError()
+
+ def _refresh_for_new_column(self, column: ColumnElement[Any]) -> None:
+ """reset internal collections for an incoming column being added."""
+ raise NotImplementedError()
+
@property
- def exported_columns(self):
+ def exported_columns(self) -> ReadOnlyColumnCollection[Any, Any]:
"""A :class:`_expression.ColumnCollection`
that represents the "exported"
columns of this :class:`_expression.ReturnsRows`.
raise NotImplementedError()
+SelfSelectable = TypeVar("SelfSelectable", bound="Selectable")
+
+
class Selectable(ReturnsRows):
"""Mark a class as being selectable."""
is_selectable = True
- def _refresh_for_new_column(self, column):
+ def _refresh_for_new_column(self, column: ColumnElement[Any]) -> None:
raise NotImplementedError()
- def lateral(self, name=None):
+ def lateral(self, name: Optional[str] = None) -> LateralFromClause:
"""Return a LATERAL alias of this :class:`_expression.Selectable`.
The return value is the :class:`_expression.Lateral` construct also
"functionality is available via the sqlalchemy.sql.visitors module.",
)
@util.preload_module("sqlalchemy.sql.util")
- def replace_selectable(self, old, alias):
+ def replace_selectable(
+ self: SelfSelectable, old: FromClause, alias: Alias
+ ) -> SelfSelectable:
"""Replace all occurrences of :class:`_expression.FromClause`
'old' with the given :class:`_expression.Alias`
object, returning a copy of this :class:`_expression.FromClause`.
"""
- return util.preloaded.sql_util.ClauseAdapter(alias).traverse(self)
+ return util.preloaded.sql_util.ClauseAdapter(alias).traverse( # type: ignore # noqa E501
+ self
+ )
- def corresponding_column(self, column, require_embedded=False):
+ def corresponding_column(
+ self, column: ColumnElement[Any], require_embedded: bool = False
+ ) -> Optional[ColumnElement[Any]]:
"""Given a :class:`_expression.ColumnElement`, return the exported
:class:`_expression.ColumnElement` object from the
:attr:`_expression.Selectable.exported_columns`
class HasPrefixes:
- _prefixes = ()
+ _prefixes: Tuple[Tuple[DQLDMLClauseElement, str], ...] = ()
- _has_prefixes_traverse_internals = [
+ _has_prefixes_traverse_internals: _TraverseInternalsType = [
("_prefixes", InternalTraversal.dp_prefix_sequence)
]
@_generative
@_document_text_coercion(
- "expr",
+ "prefixes",
":meth:`_expression.HasPrefixes.prefix_with`",
- ":paramref:`.HasPrefixes.prefix_with.*expr`",
+ ":paramref:`.HasPrefixes.prefix_with.*prefixes`",
)
- def prefix_with(self: SelfHasPrefixes, *expr, **kw) -> SelfHasPrefixes:
+ def prefix_with(
+ self: SelfHasPrefixes,
+ *prefixes: _TextCoercedExpressionArgument[Any],
+ dialect: str = "*",
+ ) -> SelfHasPrefixes:
r"""Add one or more expressions following the statement keyword, i.e.
SELECT, INSERT, UPDATE, or DELETE. Generative.
Multiple prefixes can be specified by multiple calls
to :meth:`_expression.HasPrefixes.prefix_with`.
- :param \*expr: textual or :class:`_expression.ClauseElement`
+ :param \*prefixes: textual or :class:`_expression.ClauseElement`
construct which
will be rendered following the INSERT, UPDATE, or DELETE
keyword.
- :param \**kw: A single keyword 'dialect' is accepted. This is an
- optional string dialect name which will
+ :param dialect: optional string dialect name which will
limit rendering of this prefix to only that dialect.
"""
- dialect = kw.pop("dialect", None)
- if kw:
- raise exc.ArgumentError(
- "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(
[
(coercions.expect(roles.StatementOptionRole, p), dialect)
for p in prefixes
]
)
+ return self
SelfHasSuffixes = typing.TypeVar("SelfHasSuffixes", bound="HasSuffixes")
class HasSuffixes:
- _suffixes = ()
+ _suffixes: Tuple[Tuple[DQLDMLClauseElement, str], ...] = ()
- _has_suffixes_traverse_internals = [
+ _has_suffixes_traverse_internals: _TraverseInternalsType = [
("_suffixes", InternalTraversal.dp_prefix_sequence)
]
@_generative
@_document_text_coercion(
- "expr",
+ "suffixes",
":meth:`_expression.HasSuffixes.suffix_with`",
- ":paramref:`.HasSuffixes.suffix_with.*expr`",
+ ":paramref:`.HasSuffixes.suffix_with.*suffixes`",
)
- def suffix_with(self: SelfHasSuffixes, *expr, **kw) -> SelfHasSuffixes:
+ def suffix_with(
+ self: SelfHasSuffixes,
+ *suffixes: _TextCoercedExpressionArgument[Any],
+ dialect: str = "*",
+ ) -> SelfHasSuffixes:
r"""Add one or more expressions following the statement as a whole.
This is used to support backend-specific suffix keywords on
Multiple suffixes can be specified by multiple calls
to :meth:`_expression.HasSuffixes.suffix_with`.
- :param \*expr: textual or :class:`_expression.ClauseElement`
+ :param \*suffixes: textual or :class:`_expression.ClauseElement`
construct which
will be rendered following the target clause.
- :param \**kw: A single keyword 'dialect' is accepted. This is an
- optional string dialect name which will
+ :param dialect: Optional string dialect name which will
limit rendering of this suffix to only that dialect.
"""
- dialect = kw.pop("dialect", None)
- if kw:
- raise exc.ArgumentError(
- "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(
[
(coercions.expect(roles.StatementOptionRole, p), dialect)
for p in suffixes
]
)
+ return self
SelfHasHints = typing.TypeVar("SelfHasHints", bound="HasHints")
class HasHints:
- _hints = util.immutabledict()
- _statement_hints = ()
+ _hints: util.immutabledict[
+ Tuple[FromClause, str], str
+ ] = util.immutabledict()
+ _statement_hints: Tuple[Tuple[str, str], ...] = ()
- _has_hints_traverse_internals = [
+ _has_hints_traverse_internals: _TraverseInternalsType = [
("_statement_hints", InternalTraversal.dp_statement_hint_list),
("_hints", InternalTraversal.dp_table_hint_list),
]
- def with_statement_hint(self, text, dialect_name="*"):
+ def with_statement_hint(
+ self: SelfHasHints, text: str, dialect_name: str = "*"
+ ) -> SelfHasHints:
"""Add a statement hint to this :class:`_expression.Select` or
other selectable object.
MySQL optimizer hints
"""
- return self.with_hint(None, text, dialect_name)
+ return self._with_hint(None, text, dialect_name)
@_generative
def with_hint(
- self: SelfHasHints, selectable, text, dialect_name="*"
+ self: SelfHasHints,
+ selectable: _FromClauseArgument,
+ text: str,
+ dialect_name: str = "*",
) -> SelfHasHints:
r"""Add an indexing or other executional context hint for the given
selectable to this :class:`_expression.Select` or other selectable
:meth:`_expression.Select.with_statement_hint`
"""
+
+ return self._with_hint(selectable, text, dialect_name)
+
+ def _with_hint(
+ self: SelfHasHints,
+ selectable: Optional[_FromClauseArgument],
+ text: str,
+ dialect_name: str,
+ ) -> SelfHasHints:
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.
_is_clone_of: Optional[FromClause]
+ _columns: ColumnCollection[Any, Any]
+
schema: Optional[str] = None
"""Define the 'schema' attribute for this :class:`_expression.FromClause`.
_use_schema_map = False
- def select(self) -> "Select":
+ def select(self) -> Select:
r"""Return a SELECT of this :class:`_expression.FromClause`.
"""
return Select(self)
- def join(self, right, onclause=None, isouter=False, full=False):
+ def join(
+ self,
+ right: _FromClauseArgument,
+ onclause: Optional[_ColumnExpressionArgument[bool]] = None,
+ isouter: bool = False,
+ full: bool = False,
+ ) -> Join:
"""Return a :class:`_expression.Join` from this
:class:`_expression.FromClause`
to another :class:`FromClause`.
return Join(self, right, onclause, isouter, full)
- def outerjoin(self, right, onclause=None, full=False):
+ def outerjoin(
+ self,
+ right: _FromClauseArgument,
+ onclause: Optional[_ColumnExpressionArgument[bool]] = None,
+ full: bool = False,
+ ) -> Join:
"""Return a :class:`_expression.Join` from this
:class:`_expression.FromClause`
to another :class:`FromClause`, with the "isouter" flag set to
return Join(self, right, onclause, True, full)
- def alias(self, name=None, flat=False):
+ def alias(
+ self, name: Optional[str] = None, flat: bool = False
+ ) -> NamedFromClause:
"""Return an alias of this :class:`_expression.FromClause`.
E.g.::
return Alias._construct(self, name)
- @util.preload_module("sqlalchemy.sql.sqltypes")
- def table_valued(self):
- """Return a :class:`_sql.TableValuedColumn` object for this
- :class:`_expression.FromClause`.
-
- A :class:`_sql.TableValuedColumn` is a :class:`_sql.ColumnElement` that
- represents a complete row in a table. Support for this construct is
- backend dependent, and is supported in various forms by backends
- such as PostgreSQL, Oracle and SQL Server.
-
- E.g.::
-
- >>> from sqlalchemy import select, column, func, table
- >>> a = table("a", column("id"), column("x"), column("y"))
- >>> stmt = select(func.row_to_json(a.table_valued()))
- >>> print(stmt)
- SELECT row_to_json(a) AS row_to_json_1
- FROM a
-
- .. versionadded:: 1.4.0b2
-
- .. seealso::
-
- :ref:`tutorial_functions` - in the :ref:`unified_tutorial`
-
- """
- return TableValuedColumn(self, type_api.TABLEVALUE)
-
- def tablesample(self, sampling, name=None, seed=None):
+ def tablesample(
+ self,
+ sampling: Union[float, Function[Any]],
+ name: Optional[str] = None,
+ seed: Optional[roles.ExpressionElementRole[Any]] = None,
+ ) -> TableSample:
"""Return a TABLESAMPLE alias of this :class:`_expression.FromClause`.
The return value is the :class:`_expression.TableSample`
"""
return TableSample._construct(self, sampling, name, seed)
- def is_derived_from(self, fromclause):
+ def is_derived_from(self, fromclause: FromClause) -> bool:
"""Return ``True`` if this :class:`_expression.FromClause` is
'derived' from the given ``FromClause``.
# contained elements.
return fromclause in self._cloned_set
- def _is_lexical_equivalent(self, other):
+ def _is_lexical_equivalent(self, other: FromClause) -> bool:
"""Return ``True`` if this :class:`_expression.FromClause` and
the other represent the same lexical identity.
if they are the same via annotation identity.
"""
- return self._cloned_set.intersection(other._cloned_set)
+ return bool(self._cloned_set.intersection(other._cloned_set))
- @util.non_memoized_property
+ @util.ro_non_memoized_property
def description(self) -> str:
"""A brief description of this :class:`_expression.FromClause`.
"""
return getattr(self, "name", self.__class__.__name__ + " object")
- def _generate_fromclause_column_proxies(self, fromclause):
+ def _generate_fromclause_column_proxies(
+ self, fromclause: FromClause
+ ) -> None:
fromclause._columns._populate_separate_keys(
col._make_proxy(fromclause) for col in self.c
)
@property
- def exported_columns(self):
+ def exported_columns(self) -> ReadOnlyColumnCollection[str, Any]:
"""A :class:`_expression.ColumnCollection`
that represents the "exported"
columns of this :class:`_expression.Selectable`.
self._populate_column_collection()
return self.foreign_keys
- def _reset_column_collection(self):
+ def _reset_column_collection(self) -> None:
"""Reset the attributes linked to the ``FromClause.c`` attribute.
This collection is separate from all the other memoized things
def _select_iterable(self) -> _SelectIterable:
return self.c
- def _init_collections(self):
+ def _init_collections(self) -> None:
assert "_columns" not in self.__dict__
assert "primary_key" not in self.__dict__
assert "foreign_keys" not in self.__dict__
self.foreign_keys = set() # type: ignore
@property
- def _cols_populated(self):
+ def _cols_populated(self) -> bool:
return "_columns" in self.__dict__
- def _populate_column_collection(self):
+ def _populate_column_collection(self) -> None:
"""Called on subclasses to establish the .c collection.
Each implementation has a different way of establishing
"""
- def _refresh_for_new_column(self, column):
+ def _refresh_for_new_column(self, column: ColumnElement[Any]) -> None:
"""Given a column added to the .c collection of an underlying
selectable, produce the local version of that column, assuming this
selectable ultimately should proxy this column.
"""
self._reset_column_collection()
- def _anonymous_fromclause(self, name=None, flat=False):
+ def _anonymous_fromclause(
+ self, name: Optional[str] = None, flat: bool = False
+ ) -> NamedFromClause:
return self.alias(name=name)
+ if TYPE_CHECKING:
+
+ def self_group(
+ self: Self, against: Optional[OperatorType] = None
+ ) -> Union[FromGrouping, Self]:
+ ...
+
class NamedFromClause(FromClause):
+ """A :class:`.FromClause` that has a name.
+
+ Examples include tables, subqueries, CTEs, aliased tables.
+
+ .. versionadded:: 2.0
+
+ """
+
named_with_column = True
name: str
+ @util.preload_module("sqlalchemy.sql.sqltypes")
+ def table_valued(self) -> TableValuedColumn[Any]:
+ """Return a :class:`_sql.TableValuedColumn` object for this
+ :class:`_expression.FromClause`.
+
+ A :class:`_sql.TableValuedColumn` is a :class:`_sql.ColumnElement` that
+ represents a complete row in a table. Support for this construct is
+ backend dependent, and is supported in various forms by backends
+ such as PostgreSQL, Oracle and SQL Server.
+
+ E.g.::
+
+ >>> from sqlalchemy import select, column, func, table
+ >>> a = table("a", column("id"), column("x"), column("y"))
+ >>> stmt = select(func.row_to_json(a.table_valued()))
+ >>> print(stmt)
+ SELECT row_to_json(a) AS row_to_json_1
+ FROM a
+
+ .. versionadded:: 1.4.0b2
+
+ .. seealso::
+
+ :ref:`tutorial_functions` - in the :ref:`unified_tutorial`
+
+ """
+ return TableValuedColumn(self, type_api.TABLEVALUE)
+
class SelectLabelStyle(Enum):
"""Label style constants that may be passed to
__visit_name__ = "join"
- _traverse_internals = [
+ _traverse_internals: _TraverseInternalsType = [
("left", InternalTraversal.dp_clauseelement),
("right", InternalTraversal.dp_clauseelement),
("onclause", InternalTraversal.dp_clauseelement),
_is_join = True
- def __init__(self, left, right, onclause=None, isouter=False, full=False):
+ left: FromClause
+ right: FromClause
+ onclause: Optional[ColumnElement[bool]]
+ isouter: bool
+ full: bool
+
+ def __init__(
+ self,
+ left: _FromClauseArgument,
+ right: _FromClauseArgument,
+ onclause: Optional[_OnClauseArgument] = None,
+ isouter: bool = False,
+ full: bool = False,
+ ):
"""Construct a new :class:`_expression.Join`.
The usual entrypoint here is the :func:`_expression.join`
:class:`_expression.FromClause` object.
"""
+
+ # when deannotate was removed here, callcounts went up for ORM
+ # compilation of eager joins, since there were more comparisons of
+ # annotated objects. test_orm.py -> test_fetch_results
+ # was therefore changed to show a more real-world use case, where the
+ # compilation is cached; there's no change in post-cache callcounts.
+ # callcounts for a single compilation in that particular test
+ # that includes about eight joins about 1100 extra fn calls, from
+ # 29200 -> 30373
+
self.left = coercions.expect(
- roles.FromClauseRole, left, deannotate=True
+ roles.FromClauseRole,
+ left,
)
self.right = coercions.expect(
- roles.FromClauseRole, right, deannotate=True
+ roles.FromClauseRole,
+ right,
).self_group()
if onclause is None:
self.isouter = isouter
self.full = full
- @property
+ @util.ro_non_memoized_property
def description(self) -> str:
return "Join object on %s(%d) and %s(%d)" % (
self.left.description,
id(self.right),
)
- def is_derived_from(self, fromclause):
+ def is_derived_from(self, fromclause: FromClause) -> bool:
return (
# use hash() to ensure direct comparison to annotated works
# as well
or self.right.is_derived_from(fromclause)
)
- def self_group(self, against=None):
+ def self_group(
+ self, against: Optional[OperatorType] = None
+ ) -> FromGrouping:
+ ...
return FromGrouping(self)
@util.preload_module("sqlalchemy.sql.util")
sqlutil = util.preloaded.sql_util
columns = [c for c in self.left.c] + [c for c in self.right.c]
- self.primary_key.extend(
+ self.primary_key.extend( # type: ignore
sqlutil.reduce_columns(
(c for c in columns if c.primary_key), self.onclause
)
self._columns._populate_separate_keys(
(col._tq_key_label, col) for col in columns
)
- self.foreign_keys.update(
+ self.foreign_keys.update( # type: ignore
itertools.chain(*[col.foreign_keys for col in columns])
)
- def _copy_internals(self, clone=_clone, **kw):
+ def _copy_internals(
+ self, clone: _CloneCallableType = _clone, **kw: Any
+ ) -> None:
# see Select._copy_internals() for similar concept
# here we pre-clone "left" and "right" so that we can
self._reset_memoizations()
- def _refresh_for_new_column(self, column):
+ def _refresh_for_new_column(self, column: ColumnElement[Any]) -> None:
super(Join, self)._refresh_for_new_column(column)
self.left._refresh_for_new_column(column)
self.right._refresh_for_new_column(column)
- def _match_primaries(self, left, right):
+ def _match_primaries(
+ self, left: FromClause, right: FromClause
+ ) -> ColumnElement[bool]:
if isinstance(left, Join):
left_right = left.right
else:
@classmethod
def _join_condition(
- cls, a, b, a_subset=None, consider_as_foreign_keys=None
- ):
+ cls,
+ a: FromClause,
+ b: FromClause,
+ *,
+ a_subset: Optional[FromClause] = None,
+ consider_as_foreign_keys: Optional[
+ AbstractSet[ColumnClause[Any]]
+ ] = None,
+ ) -> ColumnElement[bool]:
"""Create a join condition between two tables or selectables.
See sqlalchemy.sql.util.join_condition() for full docs.
return and_(*crit)
@classmethod
- def _can_join(cls, left, right, consider_as_foreign_keys=None):
+ def _can_join(
+ cls,
+ left: FromClause,
+ right: FromClause,
+ *,
+ consider_as_foreign_keys: Optional[
+ AbstractSet[ColumnClause[Any]]
+ ] = None,
+ ) -> bool:
if isinstance(left, Join):
left_right = left.right
else:
@classmethod
@util.preload_module("sqlalchemy.sql.util")
def _joincond_scan_left_right(
- cls, a, a_subset, b, consider_as_foreign_keys
- ):
+ cls,
+ a: FromClause,
+ a_subset: Optional[FromClause],
+ b: FromClause,
+ consider_as_foreign_keys: Optional[AbstractSet[ColumnClause[Any]]],
+ ) -> collections.defaultdict[
+ Optional[ForeignKeyConstraint],
+ List[Tuple[ColumnClause[Any], ColumnClause[Any]]],
+ ]:
sql_util = util.preloaded.sql_util
a = coercions.expect(roles.FromClauseRole, a)
b = coercions.expect(roles.FromClauseRole, b)
- constraints = collections.defaultdict(list)
+ constraints: collections.defaultdict[
+ Optional[ForeignKeyConstraint],
+ List[Tuple[ColumnClause[Any], ColumnClause[Any]]],
+ ] = collections.defaultdict(list)
for left in (a_subset, a):
if left is None:
continue
for fk in sorted(
- b.foreign_keys, key=lambda fk: fk.parent._creation_order
+ b.foreign_keys,
+ key=lambda fk: fk.parent._creation_order, # type: ignore
):
if (
consider_as_foreign_keys is not None
constraints[fk.constraint].append((col, fk.parent))
if left is not b:
for fk in sorted(
- left.foreign_keys, key=lambda fk: fk.parent._creation_order
+ left.foreign_keys,
+ key=lambda fk: fk.parent._creation_order, # type: ignore
):
if (
consider_as_foreign_keys is not None
@util.ro_non_memoized_property
def _from_objects(self) -> List[FromClause]:
- return [self] + self.left._from_objects + self.right._from_objects
+ self_list: List[FromClause] = [self]
+ return self_list + self.left._from_objects + self.right._from_objects
class NoInit:
)
+class LateralFromClause(NamedFromClause):
+ """mark a FROM clause as being able to render directly as LATERAL"""
+
+
+_SelfAliasedReturnsRows = TypeVar(
+ "_SelfAliasedReturnsRows", bound="AliasedReturnsRows"
+)
+
# FromClause ->
# AliasedReturnsRows
# -> Alias only for FromClause
# -> Lateral -> FromClause, but we accept SelectBase
# w/ non-deprecated coercion
# -> TableSample -> only for FromClause
+
+
class AliasedReturnsRows(NoInit, NamedFromClause):
"""Base class of aliases against tables, subqueries, and other
selectables."""
_supports_derived_columns = False
- element: ClauseElement
+ element: ReturnsRows
- _traverse_internals = [
+ _traverse_internals: _TraverseInternalsType = [
("element", InternalTraversal.dp_clauseelement),
("name", InternalTraversal.dp_anon_name),
]
@classmethod
- def _construct(cls, *arg, **kw):
+ def _construct(
+ cls: Type[_SelfAliasedReturnsRows], *arg: Any, **kw: Any
+ ) -> _SelfAliasedReturnsRows:
obj = cls.__new__(cls)
obj._init(*arg, **kw)
return obj
- @classmethod
- def _factory(cls, returnsrows, name=None):
- """Base factory method. Subclasses need to provide this."""
- raise NotImplementedError()
-
def _init(self, selectable, name=None):
self.element = coercions.expect(
roles.ReturnsRowsRole, selectable, apply_propagate_attrs=self
name = _anonymous_label.safe_construct(id(self), name or "anon")
self.name = name
- def _refresh_for_new_column(self, column):
+ def _refresh_for_new_column(self, column: ColumnElement[Any]) -> None:
super(AliasedReturnsRows, self)._refresh_for_new_column(column)
self.element._refresh_for_new_column(column)
- @property
+ def _populate_column_collection(self):
+ self.element._generate_fromclause_column_proxies(self)
+
+ @util.ro_non_memoized_property
def description(self) -> str:
name = self.name
if isinstance(name, _anonymous_label):
"""Legacy for dialects that are referring to Alias.original."""
return self.element
- def is_derived_from(self, fromclause):
+ def is_derived_from(self, fromclause: FromClause) -> bool:
if fromclause in self._cloned_set:
return True
return self.element.is_derived_from(fromclause)
- def _populate_column_collection(self):
- self.element._generate_fromclause_column_proxies(self)
-
- def _copy_internals(self, clone=_clone, **kw):
+ def _copy_internals(
+ self, clone: _CloneCallableType = _clone, **kw: Any
+ ) -> None:
existing_element = self.element
super(AliasedReturnsRows, self)._copy_internals(clone=clone, **kw)
return [self]
-class Alias(roles.DMLTableRole, AliasedReturnsRows):
+class FromClauseAlias(AliasedReturnsRows):
+ element: FromClause
+
+
+class Alias(roles.DMLTableRole, FromClauseAlias):
"""Represents an table or selectable alias (AS).
Represents an alias, as typically applied to any table or
element: FromClause
@classmethod
- def _factory(cls, selectable, name=None, flat=False):
+ def _factory(
+ cls,
+ selectable: FromClause,
+ name: Optional[str] = None,
+ flat: bool = False,
+ ) -> NamedFromClause:
return coercions.expect(
roles.FromClauseRole, selectable, allow_select=True
).alias(name=name, flat=flat)
-class TableValuedAlias(Alias):
+class TableValuedAlias(LateralFromClause, Alias):
"""An alias against a "table valued" SQL function.
This construct provides for a SQL function that returns columns
_render_derived_w_types = False
joins_implicitly = False
- _traverse_internals = [
+ _traverse_internals: _TraverseInternalsType = [
("element", InternalTraversal.dp_clauseelement),
("name", InternalTraversal.dp_anon_name),
("_tableval_type", InternalTraversal.dp_type),
return TableValuedColumn(self, self._tableval_type)
- def alias(self, name=None):
+ def alias(
+ self, name: Optional[str] = None, flat: bool = False
+ ) -> TableValuedAlias:
"""Return a new alias of this :class:`_sql.TableValuedAlias`.
This creates a distinct FROM object that will be distinguished
return tva
- def lateral(self, name=None):
+ def lateral(self, name: Optional[str] = None) -> LateralFromClause:
"""Return a new :class:`_sql.TableValuedAlias` with the lateral flag set,
so that it renders as LATERAL.
return new_alias
-class Lateral(AliasedReturnsRows):
+class Lateral(FromClauseAlias, LateralFromClause):
"""Represent a LATERAL subquery.
This object is constructed from the :func:`_expression.lateral` module
inherit_cache = True
@classmethod
- def _factory(cls, selectable, name=None):
+ def _factory(
+ cls,
+ selectable: Union[SelectBase, _FromClauseArgument],
+ name: Optional[str] = None,
+ ) -> LateralFromClause:
return coercions.expect(
roles.FromClauseRole, selectable, explicit_subquery=True
).lateral(name=name)
-class TableSample(AliasedReturnsRows):
+class TableSample(FromClauseAlias):
"""Represent a TABLESAMPLE clause.
This object is constructed from the :func:`_expression.tablesample` module
__visit_name__ = "tablesample"
- _traverse_internals = AliasedReturnsRows._traverse_internals + [
- ("sampling", InternalTraversal.dp_clauseelement),
- ("seed", InternalTraversal.dp_clauseelement),
- ]
+ _traverse_internals: _TraverseInternalsType = (
+ AliasedReturnsRows._traverse_internals
+ + [
+ ("sampling", InternalTraversal.dp_clauseelement),
+ ("seed", InternalTraversal.dp_clauseelement),
+ ]
+ )
@classmethod
- def _factory(cls, selectable, sampling, name=None, seed=None):
+ def _factory(
+ cls,
+ selectable: _FromClauseArgument,
+ sampling: Union[float, Function[Any]],
+ name: Optional[str] = None,
+ seed: Optional[roles.ExpressionElementRole[Any]] = None,
+ ) -> TableSample:
return coercions.expect(roles.FromClauseRole, selectable).tablesample(
sampling, name=name, seed=seed
)
__visit_name__ = "cte"
- _traverse_internals = (
+ _traverse_internals: _TraverseInternalsType = (
AliasedReturnsRows._traverse_internals
+ [
("_cte_alias", InternalTraversal.dp_clauseelement),
element: HasCTE
@classmethod
- def _factory(cls, selectable, name=None, recursive=False):
+ def _factory(
+ cls,
+ selectable: HasCTE,
+ name: Optional[str] = None,
+ recursive: bool = False,
+ ) -> CTE:
r"""Return a new :class:`_expression.CTE`,
or Common Table Expression instance.
else:
self.element._generate_fromclause_column_proxies(self)
- def alias(self, name=None, flat=False):
+ def alias(
+ self, name: Optional[str] = None, flat: bool = False
+ ) -> NamedFromClause:
"""Return an :class:`_expression.Alias` of this
:class:`_expression.CTE`.
:meth:`_sql.HasCTE.cte` - examples of calling styles
"""
+ assert is_select_statement(
+ self.element
+ ), f"CTE element f{self.element} does not support union()"
+
return CTE._construct(
self.element.union(*other),
name=self.name,
:meth:`_sql.HasCTE.cte` - examples of calling styles
"""
+
+ assert is_select_statement(
+ self.element
+ ), f"CTE element f{self.element} does not support union_all()"
+
return CTE._construct(
self.element.union_all(*other),
name=self.name,
nesting: bool
-class HasCTE(roles.HasCTERole, ClauseElement):
- """Mixin that declares a class to include CTE support.
+class _ColumnsPlusNames(NamedTuple):
+ required_label_name: Optional[str]
+ """
+ string label name, if non-None, must be rendered as a
+ label, i.e. "AS <name>"
+ """
- .. versionadded:: 1.1
+ proxy_key: Optional[str]
+ """
+ proxy_key that is to be part of the result map for this
+ col. this is also the key in a fromclause.c or
+ select.selected_columns collection
+ """
+ fallback_label_name: Optional[str]
+ """
+ name that can be used to render an "AS <name>" when
+ we have to render a label even though
+ required_label_name was not given
"""
- _has_ctes_traverse_internals = [
- ("_independent_ctes", InternalTraversal.dp_clauseelement_list),
- ("_independent_ctes_opts", InternalTraversal.dp_plain_obj),
- ]
+ column: Union[ColumnElement[Any], TextClause]
+ """
+ the ColumnElement itself
+ """
- _independent_ctes = ()
- _independent_ctes_opts = ()
+ repeated: bool
+ """
+ True if this is a duplicate of a previous column
+ in the list of columns
+ """
- @_generative
- def add_cte(self: SelfHasCTE, *ctes, nest_here=False) -> SelfHasCTE:
- r"""Add one or more :class:`_sql.CTE` constructs to this statement.
- This method will associate the given :class:`_sql.CTE` constructs with
- the parent statement such that they will each be unconditionally
- rendered in the WITH clause of the final statement, even if not
- referenced elsewhere within the statement or any sub-selects.
+class SelectsRows(ReturnsRows):
+ """Sub-base of ReturnsRows for elements that deliver rows
+ directly, namely SELECT and INSERT/UPDATE/DELETE..RETURNING"""
- The optional :paramref:`.HasCTE.add_cte.nest_here` parameter when set
- to True will have the effect that each given :class:`_sql.CTE` will
- render in a WITH clause rendered directly along with this statement,
- rather than being moved to the top of the ultimate rendered statement,
- even if this statement is rendered as a subquery within a larger
- statement.
+ _label_style: SelectLabelStyle = LABEL_STYLE_NONE
- This method has two general uses. One is to embed CTE statements that
- serve some purpose without being referenced explicitly, such as the use
- case of embedding a DML statement such as an INSERT or UPDATE as a CTE
- inline with a primary statement that may draw from its results
- indirectly. The other is to provide control over the exact placement
- of a particular series of CTE constructs that should remain rendered
- directly in terms of a particular statement that may be nested in a
- larger statement.
+ def _generate_columns_plus_names(
+ self, anon_for_dupe_key: bool
+ ) -> List[_ColumnsPlusNames]:
+ """Generate column names as rendered in a SELECT statement by
+ the compiler.
- E.g.::
+ This is distinct from the _column_naming_convention generator that's
+ intended for population of .c collections and similar, which has
+ different rules. the collection returned here calls upon the
+ _column_naming_convention as well.
- from sqlalchemy import table, column, select
- t = table('t', column('c1'), column('c2'))
+ """
+ cols = self._all_selected_columns
- ins = t.insert().values({"c1": "x", "c2": "y"}).cte()
+ key_naming_convention = SelectState._column_naming_convention(
+ self._label_style
+ )
- stmt = select(t).add_cte(ins)
+ names = {}
- Would render::
+ result: List[_ColumnsPlusNames] = []
+ result_append = result.append
- WITH anon_1 AS
- (INSERT INTO t (c1, c2) VALUES (:param_1, :param_2))
- SELECT t.c1, t.c2
- FROM t
+ table_qualified = self._label_style is LABEL_STYLE_TABLENAME_PLUS_COL
+ label_style_none = self._label_style is LABEL_STYLE_NONE
- Above, the "anon_1" CTE is not referred towards in the SELECT
- statement, however still accomplishes the task of running an INSERT
- statement.
+ # a counter used for "dedupe" labels, which have double underscores
+ # in them and are never referred by name; they only act
+ # as positional placeholders. they need only be unique within
+ # the single columns clause they're rendered within (required by
+ # some dbs such as mysql). So their anon identity is tracked against
+ # a fixed counter rather than hash() identity.
+ dedupe_hash = 1
- Similarly in a DML-related context, using the PostgreSQL
+ for c in cols:
+ repeated = False
+
+ if not c._render_label_in_columns_clause:
+ effective_name = (
+ required_label_name
+ ) = fallback_label_name = None
+ elif label_style_none:
+ if TYPE_CHECKING:
+ assert is_column_element(c)
+
+ effective_name = required_label_name = None
+ fallback_label_name = c._non_anon_label or c._anon_name_label
+ else:
+ if TYPE_CHECKING:
+ assert is_column_element(c)
+
+ if table_qualified:
+ required_label_name = (
+ effective_name
+ ) = fallback_label_name = c._tq_label
+ else:
+ effective_name = fallback_label_name = c._non_anon_label
+ required_label_name = None
+
+ if effective_name is None:
+ # it seems like this could be _proxy_key and we would
+ # not need _expression_label but it isn't
+ # giving us a clue when to use anon_label instead
+ expr_label = c._expression_label
+ if expr_label is None:
+ repeated = c._anon_name_label in names
+ names[c._anon_name_label] = c
+ effective_name = required_label_name = None
+
+ if repeated:
+ # here, "required_label_name" is sent as
+ # "None" and "fallback_label_name" is sent.
+ if table_qualified:
+ fallback_label_name = (
+ c._dedupe_anon_tq_label_idx(dedupe_hash)
+ )
+ dedupe_hash += 1
+ else:
+ fallback_label_name = c._dedupe_anon_label_idx(
+ dedupe_hash
+ )
+ dedupe_hash += 1
+ else:
+ fallback_label_name = c._anon_name_label
+ else:
+ required_label_name = (
+ effective_name
+ ) = fallback_label_name = expr_label
+
+ if effective_name is not None:
+ if TYPE_CHECKING:
+ assert is_column_element(c)
+
+ if effective_name in names:
+ # when looking to see if names[name] is the same column as
+ # c, use hash(), so that an annotated version of the column
+ # is seen as the same as the non-annotated
+ if hash(names[effective_name]) != hash(c):
+
+ # different column under the same name. apply
+ # disambiguating label
+ if table_qualified:
+ required_label_name = (
+ fallback_label_name
+ ) = c._anon_tq_label
+ else:
+ required_label_name = (
+ fallback_label_name
+ ) = c._anon_name_label
+
+ if anon_for_dupe_key and required_label_name in names:
+ # here, c._anon_tq_label is definitely unique to
+ # that column identity (or annotated version), so
+ # this should always be true.
+ # this is also an infrequent codepath because
+ # you need two levels of duplication to be here
+ assert hash(names[required_label_name]) == hash(c)
+
+ # the column under the disambiguating label is
+ # already present. apply the "dedupe" label to
+ # subsequent occurrences of the column so that the
+ # original stays non-ambiguous
+ if table_qualified:
+ required_label_name = (
+ fallback_label_name
+ ) = c._dedupe_anon_tq_label_idx(dedupe_hash)
+ dedupe_hash += 1
+ else:
+ required_label_name = (
+ fallback_label_name
+ ) = c._dedupe_anon_label_idx(dedupe_hash)
+ dedupe_hash += 1
+ repeated = True
+ else:
+ names[required_label_name] = c
+ elif anon_for_dupe_key:
+ # same column under the same name. apply the "dedupe"
+ # label so that the original stays non-ambiguous
+ if table_qualified:
+ required_label_name = (
+ fallback_label_name
+ ) = c._dedupe_anon_tq_label_idx(dedupe_hash)
+ dedupe_hash += 1
+ else:
+ required_label_name = (
+ fallback_label_name
+ ) = c._dedupe_anon_label_idx(dedupe_hash)
+ dedupe_hash += 1
+ repeated = True
+ else:
+ names[effective_name] = c
+
+ result_append(
+ _ColumnsPlusNames(
+ required_label_name,
+ key_naming_convention(c),
+ fallback_label_name,
+ c,
+ repeated,
+ )
+ )
+
+ return result
+
+
+class HasCTE(roles.HasCTERole, SelectsRows):
+ """Mixin that declares a class to include CTE support.
+
+ .. versionadded:: 1.1
+
+ """
+
+ _has_ctes_traverse_internals: _TraverseInternalsType = [
+ ("_independent_ctes", InternalTraversal.dp_clauseelement_list),
+ ("_independent_ctes_opts", InternalTraversal.dp_plain_obj),
+ ]
+
+ _independent_ctes: Tuple[CTE, ...] = ()
+ _independent_ctes_opts: Tuple[_CTEOpts, ...] = ()
+
+ @_generative
+ def add_cte(
+ self: SelfHasCTE, *ctes: CTE, nest_here: bool = False
+ ) -> SelfHasCTE:
+ r"""Add one or more :class:`_sql.CTE` constructs to this statement.
+
+ This method will associate the given :class:`_sql.CTE` constructs with
+ the parent statement such that they will each be unconditionally
+ rendered in the WITH clause of the final statement, even if not
+ referenced elsewhere within the statement or any sub-selects.
+
+ The optional :paramref:`.HasCTE.add_cte.nest_here` parameter when set
+ to True will have the effect that each given :class:`_sql.CTE` will
+ render in a WITH clause rendered directly along with this statement,
+ rather than being moved to the top of the ultimate rendered statement,
+ even if this statement is rendered as a subquery within a larger
+ statement.
+
+ This method has two general uses. One is to embed CTE statements that
+ serve some purpose without being referenced explicitly, such as the use
+ case of embedding a DML statement such as an INSERT or UPDATE as a CTE
+ inline with a primary statement that may draw from its results
+ indirectly. The other is to provide control over the exact placement
+ of a particular series of CTE constructs that should remain rendered
+ directly in terms of a particular statement that may be nested in a
+ larger statement.
+
+ E.g.::
+
+ from sqlalchemy import table, column, select
+ t = table('t', column('c1'), column('c2'))
+
+ ins = t.insert().values({"c1": "x", "c2": "y"}).cte()
+
+ stmt = select(t).add_cte(ins)
+
+ Would render::
+
+ WITH anon_1 AS
+ (INSERT INTO t (c1, c2) VALUES (:param_1, :param_2))
+ SELECT t.c1, t.c2
+ FROM t
+
+ Above, the "anon_1" CTE is not referred towards in the SELECT
+ statement, however still accomplishes the task of running an INSERT
+ statement.
+
+ Similarly in a DML-related context, using the PostgreSQL
:class:`_postgresql.Insert` construct to generate an "upsert"::
from sqlalchemy import table, column
self._independent_ctes_opts += (opt,)
return self
- def cte(self, name=None, recursive=False, nesting=False):
+ def cte(
+ self,
+ name: Optional[str] = None,
+ recursive: bool = False,
+ nesting: bool = False,
+ ) -> CTE:
r"""Return a new :class:`_expression.CTE`,
or Common Table Expression instance.
inherit_cache = True
- element: Select
+ element: SelectBase
@classmethod
- def _factory(cls, selectable, name=None):
+ def _factory(
+ cls, selectable: SelectBase, name: Optional[str] = None
+ ) -> Subquery:
"""Return a :class:`.Subquery` object."""
return coercions.expect(
roles.SelectStatementRole, selectable
class FromGrouping(GroupedElement, FromClause):
"""Represent a grouping of a FROM clause"""
- _traverse_internals = [("element", InternalTraversal.dp_clauseelement)]
+ _traverse_internals: _TraverseInternalsType = [
+ ("element", InternalTraversal.dp_clauseelement)
+ ]
element: FromClause
- def __init__(self, element):
+ def __init__(self, element: FromClause):
self.element = coercions.expect(roles.FromClauseRole, element)
def _init_collections(self):
def foreign_keys(self):
return self.element.foreign_keys
- def is_derived_from(self, element):
- return self.element.is_derived_from(element)
+ def is_derived_from(self, fromclause: FromClause) -> bool:
+ return self.element.is_derived_from(fromclause)
- def alias(self, **kw):
- return FromGrouping(self.element.alias(**kw))
+ def alias(
+ self, name: Optional[str] = None, flat: bool = False
+ ) -> NamedFromGrouping:
+ return NamedFromGrouping(self.element.alias(name=name, flat=flat))
def _anonymous_fromclause(self, **kw):
return FromGrouping(self.element._anonymous_fromclause(**kw))
self.element = state["element"]
+class NamedFromGrouping(FromGrouping, NamedFromClause):
+ """represent a grouping of a named FROM clause
+
+ .. versionadded:: 2.0
+
+ """
+
+ inherit_cache = True
+
+
class TableClause(roles.DMLTableRole, Immutable, NamedFromClause):
"""Represents a minimal "table" construct.
__visit_name__ = "table"
- _traverse_internals = [
+ _traverse_internals: _TraverseInternalsType = [
(
"columns",
InternalTraversal.dp_fromclause_canonical_column_collection,
doesn't support having a primary key or column
-level defaults, so implicit returning doesn't apply."""
- _autoincrement_column = None
- """No PK or default support so no autoincrement column."""
+ @util.ro_memoized_property
+ def _autoincrement_column(self) -> Optional[ColumnClause[Any]]:
+ """No PK or default support so no autoincrement column."""
+ return None
- def __init__(self, name, *columns, **kw):
+ def __init__(self, name: str, *columns: ColumnClause[Any], **kw: Any):
super(TableClause, self).__init__()
self.name = name
self._columns = DedupeColumnCollection()
- self.primary_key = ColumnSet()
- self.foreign_keys = set()
+ self.primary_key = ColumnSet() # type: ignore
+ self.foreign_keys = set() # type: ignore
for c in columns:
self.append_column(c)
def c(self) -> ReadOnlyColumnCollection[str, ColumnClause[Any]]:
...
- def __str__(self):
+ def __str__(self) -> str:
if self.schema is not None:
return self.schema + "." + self.name
else:
return self.name
- def _refresh_for_new_column(self, column):
+ def _refresh_for_new_column(self, column: ColumnElement[Any]) -> None:
pass
- def _init_collections(self):
+ def _init_collections(self) -> None:
pass
- @util.memoized_property
+ @util.ro_memoized_property
def description(self) -> str:
return self.name
- def append_column(self, c, **kw):
+ def append_column(self, c: ColumnClause[Any]) -> None:
existing = c.table
if existing is not None and existing is not self:
raise exc.ArgumentError(
c.table = self
@util.preload_module("sqlalchemy.sql.dml")
- def insert(self):
+ def insert(self) -> Insert:
"""Generate an :func:`_expression.insert` construct against this
:class:`_expression.TableClause`.
See :func:`_expression.insert` for argument and usage information.
"""
+
return util.preloaded.sql_dml.Insert(self)
@util.preload_module("sqlalchemy.sql.dml")
- def update(self):
+ def update(self) -> Update:
"""Generate an :func:`_expression.update` construct against this
:class:`_expression.TableClause`.
)
@util.preload_module("sqlalchemy.sql.dml")
- def delete(self):
+ def delete(self) -> Delete:
"""Generate a :func:`_expression.delete` construct against this
:class:`_expression.TableClause`.
class ForUpdateArg(ClauseElement):
- _traverse_internals = [
+ _traverse_internals: _TraverseInternalsType = [
("of", InternalTraversal.dp_clauseelement_list),
("nowait", InternalTraversal.dp_boolean),
("read", InternalTraversal.dp_boolean),
("skip_locked", InternalTraversal.dp_boolean),
]
+ of: Optional[Sequence[ClauseElement]]
+ nowait: bool
+ read: bool
+ skip_locked: bool
+
@classmethod
def _from_argument(cls, with_for_update):
if isinstance(with_for_update, ForUpdateArg):
SelfValues = typing.TypeVar("SelfValues", bound="Values")
-class Values(Generative, NamedFromClause):
+class Values(Generative, LateralFromClause):
"""Represent a ``VALUES`` construct that can be used as a FROM element
in a statement.
__visit_name__ = "values"
- _data = ()
+ _data: Tuple[List[Tuple[Any, ...]], ...] = ()
- _traverse_internals = [
+ _unnamed: bool
+ _traverse_internals: _TraverseInternalsType = [
("_column_args", InternalTraversal.dp_clauseelement_list),
("_data", InternalTraversal.dp_dml_multi_values),
("name", InternalTraversal.dp_string),
("literal_binds", InternalTraversal.dp_boolean),
]
- def __init__(self, *columns, name=None, literal_binds=False):
+ def __init__(
+ self,
+ *columns: ColumnClause[Any],
+ name: Optional[str] = None,
+ literal_binds: bool = False,
+ ):
super(Values, self).__init__()
self._column_args = columns
- self.name = name
+ if name is None:
+ self._unnamed = True
+ self.name = _anonymous_label.safe_construct(id(self), "anon")
+ else:
+ self._unnamed = False
+ self.name = name
self.literal_binds = literal_binds
- self.named_with_column = self.name is not None
+ self.named_with_column = not self._unnamed
@property
def _column_types(self):
return [col.type for col in self._column_args]
@_generative
- def alias(self: SelfValues, name, **kw) -> SelfValues:
+ def alias(
+ self: SelfValues, name: Optional[str] = None, flat: bool = False
+ ) -> SelfValues:
+
"""Return a new :class:`_expression.Values`
construct that is a copy of this
one with the given name.
:func:`_expression.alias`
"""
- self.name = name
- self.named_with_column = self.name is not None
+ non_none_name: str
+
+ if name is None:
+ non_none_name = _anonymous_label.safe_construct(id(self), "anon")
+ else:
+ non_none_name = name
+
+ self.name = non_none_name
+ self.named_with_column = True
+ self._unnamed = False
return self
@_generative
- def lateral(self: SelfValues, name=None) -> SelfValues:
+ def lateral(self, name: Optional[str] = None) -> LateralFromClause:
"""Return a new :class:`_expression.Values` with the lateral flag set,
so that
it renders as LATERAL.
:func:`_expression.lateral`
"""
+ non_none_name: str
+
+ if name is None:
+ non_none_name = self.name
+ else:
+ non_none_name = name
+
self._is_lateral = True
- if name is not None:
- self.name = name
+ self.name = non_none_name
+ self._unnamed = False
return self
@_generative
- def data(self: SelfValues, values) -> SelfValues:
+ def data(self: SelfValues, values: List[Tuple[Any, ...]]) -> 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):
+ def _populate_column_collection(self) -> None:
for c in self._column_args:
self._columns.add(c)
c.table = self
"""
- _is_select_statement = True
+ _is_select_base = True
is_select = True
- def _generate_fromclause_column_proxies(
- self, fromclause: FromClause
- ) -> None:
- raise NotImplementedError()
+ _label_style: SelectLabelStyle = LABEL_STYLE_NONE
def _refresh_for_new_column(self, column: ColumnElement[Any]) -> None:
self._reset_memoizations()
- def _generate_columns_plus_names(
- self, anon_for_dupe_key: bool
- ) -> List[Tuple[str, str, str, ColumnElement[Any], bool]]:
- raise NotImplementedError()
-
- def set_label_style(
- self: SelfSelectBase, label_style: SelectLabelStyle
- ) -> SelfSelectBase:
- raise NotImplementedError()
-
- def get_label_style(self) -> SelectLabelStyle:
- raise NotImplementedError()
-
- @property
- def selected_columns(self):
+ @util.ro_non_memoized_property
+ def selected_columns(self) -> ColumnCollection[str, ColumnElement[Any]]:
"""A :class:`_expression.ColumnCollection`
representing the columns that
this SELECT statement or similar construct returns in its result set.
raise NotImplementedError()
@property
- def exported_columns(self):
+ def exported_columns(self) -> ReadOnlyColumnCollection[str, Any]:
"""A :class:`_expression.ColumnCollection`
that represents the "exported"
columns of this :class:`_expression.Selectable`, not including
"""
- return self.selected_columns
+ return self.selected_columns.as_readonly()
@util.deprecated_property(
"1.4",
def columns(self):
return self.c
+ def get_label_style(self) -> SelectLabelStyle:
+ """
+ Retrieve the current label style.
+
+ Implemented by subclasses.
+
+ """
+ raise NotImplementedError()
+
+ def set_label_style(
+ self: SelfSelectBase, style: SelectLabelStyle
+ ) -> SelfSelectBase:
+ """Return a new selectable with the specified label style.
+
+ Implemented by subclasses.
+
+ """
+
+ raise NotImplementedError()
+
@util.deprecated(
"1.4",
"The :meth:`_expression.SelectBase.select` method is deprecated "
def _implicit_subquery(self):
return self.subquery()
+ def _scalar_type(self) -> TypeEngine[Any]:
+ raise NotImplementedError()
+
@util.deprecated(
"1.4",
"The :meth:`_expression.SelectBase.as_scalar` "
"""
return self.scalar_subquery().label(name)
- def lateral(self, name=None):
+ def lateral(self, name: Optional[str] = None) -> LateralFromClause:
"""Return a LATERAL alias of this :class:`_expression.Selectable`.
The return value is the :class:`_expression.Lateral` construct also
"""
return Lateral._factory(self, name)
- @util.ro_non_memoized_property
- def _from_objects(self) -> List[FromClause]:
- return [self]
-
- def subquery(self, name=None):
+ def subquery(self, name: Optional[str] = None) -> Subquery:
"""Return a subquery of this :class:`_expression.SelectBase`.
A subquery is from a SQL perspective a parenthesized, named
raise NotImplementedError()
- def alias(self, name=None, flat=False):
+ def alias(
+ self, name: Optional[str] = None, flat: bool = False
+ ) -> Subquery:
"""Return a named subquery against this
:class:`_expression.SelectBase`.
"""
__visit_name__ = "select_statement_grouping"
- _traverse_internals = [("element", InternalTraversal.dp_clauseelement)]
+ _traverse_internals: _TraverseInternalsType = [
+ ("element", InternalTraversal.dp_clauseelement)
+ ]
_is_select_container = True
def select_statement(self):
return self.element
- def self_group(self, against=None):
+ def self_group(self: Self, against: Optional[OperatorType] = None) -> Self:
+ ...
return self
- def _generate_columns_plus_names(
- self, anon_for_dupe_key: bool
- ) -> List[Tuple[str, str, str, ColumnElement[Any], bool]]:
- return self.element._generate_columns_plus_names(anon_for_dupe_key)
+ # def _generate_columns_plus_names(
+ # self, anon_for_dupe_key: bool
+ # ) -> List[Tuple[str, str, str, ColumnElement[Any], bool]]:
+ # return self.element._generate_columns_plus_names(anon_for_dupe_key)
def _generate_fromclause_column_proxies(
self, subquery: FromClause
def _all_selected_columns(self) -> _SelectIterable:
return self.element._all_selected_columns
- @property
- def selected_columns(self):
+ @util.ro_non_memoized_property
+ def selected_columns(self) -> ColumnCollection[str, ColumnElement[Any]]:
"""A :class:`_expression.ColumnCollection`
representing the columns that
the embedded SELECT statement returns in its result set, not including
"""
- _order_by_clauses = ()
- _group_by_clauses = ()
- _limit_clause = None
- _offset_clause = None
- _fetch_clause = None
- _fetch_clause_options = None
- _for_update_arg = None
+ _order_by_clauses: Tuple[ColumnElement[Any], ...] = ()
+ _group_by_clauses: Tuple[ColumnElement[Any], ...] = ()
+ _limit_clause: Optional[ColumnElement[Any]] = None
+ _offset_clause: Optional[ColumnElement[Any]] = None
+ _fetch_clause: Optional[ColumnElement[Any]] = None
+ _fetch_clause_options: Optional[Dict[str, bool]] = None
+ _for_update_arg: Optional[ForUpdateArg] = None
- def __init__(self, _label_style=LABEL_STYLE_DEFAULT):
+ def __init__(self, _label_style: SelectLabelStyle = LABEL_STYLE_DEFAULT):
self._label_style = _label_style
@_generative
def with_for_update(
self: SelfGenerativeSelect,
- nowait=False,
- read=False,
- of=None,
- skip_locked=False,
- key_share=False,
+ nowait: bool = False,
+ read: bool = False,
+ of: Optional[
+ Union[
+ _ColumnExpressionArgument[Any],
+ Sequence[_ColumnExpressionArgument[Any]],
+ ]
+ ] = None,
+ skip_locked: bool = False,
+ key_share: bool = False,
) -> SelfGenerativeSelect:
"""Specify a ``FOR UPDATE`` clause for this
:class:`_expression.GenerativeSelect`.
return self
@property
- def _group_by_clause(self):
+ def _group_by_clause(self) -> ClauseList:
"""ClauseList access to group_by_clauses for legacy dialects"""
return ClauseList._construct_raw(
operators.comma_op, self._group_by_clauses
)
@property
- def _order_by_clause(self):
+ def _order_by_clause(self) -> ClauseList:
"""ClauseList access to order_by_clauses for legacy dialects"""
return ClauseList._construct_raw(
operators.comma_op, self._order_by_clauses
)
- def _offset_or_limit_clause(self, element, name=None, type_=None):
+ def _offset_or_limit_clause(
+ self,
+ element: Union[int, _ColumnExpressionArgument[Any]],
+ name: Optional[str] = None,
+ type_: Optional[_TypeEngineArgument[int]] = None,
+ ) -> ColumnElement[Any]:
"""Convert the given value to an "offset or limit" clause.
This handles incoming integers and converts to an expression; if
roles.LimitOffsetRole, element, name=name, type_=type_
)
- def _offset_or_limit_clause_asint(self, clause, attrname):
+ @overload
+ def _offset_or_limit_clause_asint(
+ self, clause: ColumnElement[Any], attrname: str
+ ) -> NoReturn:
+ ...
+
+ @overload
+ def _offset_or_limit_clause_asint(
+ self, clause: Optional[_OffsetLimitParam], attrname: str
+ ) -> Optional[int]:
+ ...
+
+ def _offset_or_limit_clause_asint(
+ self, clause: Optional[ColumnElement[Any]], attrname: str
+ ) -> Union[NoReturn, Optional[int]]:
"""Convert the "offset or limit" clause of a select construct to an
integer.
return util.asint(value)
@property
- def _limit(self):
+ def _limit(self) -> Optional[int]:
"""Get an integer value for the limit. This should only be used
by code that cannot support a limit as a BindParameter or
other custom clause as it will throw an exception if the limit
"""
return self._offset_or_limit_clause_asint(self._limit_clause, "limit")
- def _simple_int_clause(self, clause):
+ def _simple_int_clause(self, clause: ClauseElement) -> bool:
"""True if the clause is a simple integer, False
if it is not present or is a SQL expression.
"""
return isinstance(clause, _OffsetLimitParam)
@property
- def _offset(self):
+ def _offset(self) -> Optional[int]:
"""Get an integer value for the offset. This should only be used
by code that cannot support an offset as a BindParameter or
other custom clause as it will throw an exception if the
)
@property
- def _has_row_limiting_clause(self):
+ def _has_row_limiting_clause(self) -> bool:
return (
self._limit_clause is not None
or self._offset_clause is not None
)
@_generative
- def limit(self: SelfGenerativeSelect, limit) -> SelfGenerativeSelect:
+ def limit(
+ self: SelfGenerativeSelect,
+ limit: Union[int, _ColumnExpressionArgument[int]],
+ ) -> SelfGenerativeSelect:
"""Return a new selectable with the given LIMIT criterion
applied.
@_generative
def fetch(
- self: SelfGenerativeSelect, count, with_ties=False, percent=False
+ self: SelfGenerativeSelect,
+ count: Union[int, _ColumnExpressionArgument[int]],
+ with_ties: bool = False,
+ percent: bool = False,
) -> SelfGenerativeSelect:
"""Return a new selectable with the given FETCH FIRST criterion
applied.
return self
@_generative
- def offset(self: SelfGenerativeSelect, offset) -> SelfGenerativeSelect:
+ def offset(
+ self: SelfGenerativeSelect,
+ offset: Union[int, _ColumnExpressionArgument[int]],
+ ) -> SelfGenerativeSelect:
"""Return a new selectable with the given OFFSET criterion
applied.
@_generative
@util.preload_module("sqlalchemy.sql.util")
- def slice(self: SelfGenerativeSelect, start, stop) -> SelfGenerativeSelect:
+ def slice(
+ self: SelfGenerativeSelect,
+ start: int,
+ stop: int,
+ ) -> SelfGenerativeSelect:
"""Apply LIMIT / OFFSET to this statement based on a slice.
The start and stop indices behave like the argument to Python's
return self
@_generative
- def order_by(self: SelfGenerativeSelect, *clauses) -> SelfGenerativeSelect:
+ def order_by(
+ self: SelfGenerativeSelect, *clauses: _ColumnExpressionArgument[Any]
+ ) -> SelfGenerativeSelect:
r"""Return a new selectable with the given list of ORDER BY
criteria applied.
return self
@_generative
- def group_by(self: SelfGenerativeSelect, *clauses) -> SelfGenerativeSelect:
+ def group_by(
+ self: SelfGenerativeSelect, *clauses: _ColumnExpressionArgument[Any]
+ ) -> SelfGenerativeSelect:
r"""Return a new selectable with the given list of GROUP BY
criterion applied.
return d, d, d
+class _CompoundSelectKeyword(Enum):
+ UNION = "UNION"
+ UNION_ALL = "UNION ALL"
+ EXCEPT = "EXCEPT"
+ EXCEPT_ALL = "EXCEPT ALL"
+ INTERSECT = "INTERSECT"
+ INTERSECT_ALL = "INTERSECT ALL"
+
+
class CompoundSelect(HasCompileState, GenerativeSelect):
"""Forms the basis of ``UNION``, ``UNION ALL``, and other
SELECT-based set operations.
__visit_name__ = "compound_select"
- _traverse_internals = [
+ _traverse_internals: _TraverseInternalsType = [
("selects", InternalTraversal.dp_clauseelement_list),
("_limit_clause", InternalTraversal.dp_clauseelement),
("_offset_clause", InternalTraversal.dp_clauseelement),
("keyword", InternalTraversal.dp_string),
] + SupportsCloneAnnotations._clone_annotations_traverse_internals
- UNION = util.symbol("UNION")
- UNION_ALL = util.symbol("UNION ALL")
- EXCEPT = util.symbol("EXCEPT")
- EXCEPT_ALL = util.symbol("EXCEPT ALL")
- INTERSECT = util.symbol("INTERSECT")
- INTERSECT_ALL = util.symbol("INTERSECT ALL")
+ selects: List[SelectBase]
_is_from_container = True
_auto_correlate = False
- def __init__(self, keyword, *selects):
+ def __init__(
+ self,
+ keyword: _CompoundSelectKeyword,
+ *selects: _SelectStatementForCompoundArgument,
+ ):
self.keyword = keyword
self.selects = [
coercions.expect(roles.CompoundElementRole, s).self_group(
GenerativeSelect.__init__(self)
@classmethod
- def _create_union(cls, *selects, **kwargs):
- return CompoundSelect(CompoundSelect.UNION, *selects, **kwargs)
+ def _create_union(
+ cls, *selects: _SelectStatementForCompoundArgument
+ ) -> CompoundSelect:
+ return CompoundSelect(_CompoundSelectKeyword.UNION, *selects)
@classmethod
- def _create_union_all(cls, *selects):
- return CompoundSelect(CompoundSelect.UNION_ALL, *selects)
+ def _create_union_all(
+ cls, *selects: _SelectStatementForCompoundArgument
+ ) -> CompoundSelect:
+ return CompoundSelect(_CompoundSelectKeyword.UNION_ALL, *selects)
@classmethod
- def _create_except(cls, *selects):
- return CompoundSelect(CompoundSelect.EXCEPT, *selects)
+ def _create_except(
+ cls, *selects: _SelectStatementForCompoundArgument
+ ) -> CompoundSelect:
+ return CompoundSelect(_CompoundSelectKeyword.EXCEPT, *selects)
@classmethod
- def _create_except_all(cls, *selects):
- return CompoundSelect(CompoundSelect.EXCEPT_ALL, *selects)
+ def _create_except_all(
+ cls, *selects: _SelectStatementForCompoundArgument
+ ) -> CompoundSelect:
+ return CompoundSelect(_CompoundSelectKeyword.EXCEPT_ALL, *selects)
@classmethod
- def _create_intersect(cls, *selects):
- return CompoundSelect(CompoundSelect.INTERSECT, *selects)
+ def _create_intersect(
+ cls, *selects: _SelectStatementForCompoundArgument
+ ) -> CompoundSelect:
+ return CompoundSelect(_CompoundSelectKeyword.INTERSECT, *selects)
@classmethod
- def _create_intersect_all(cls, *selects):
- return CompoundSelect(CompoundSelect.INTERSECT_ALL, *selects)
+ def _create_intersect_all(
+ cls, *selects: _SelectStatementForCompoundArgument
+ ) -> CompoundSelect:
+ return CompoundSelect(_CompoundSelectKeyword.INTERSECT_ALL, *selects)
- def _scalar_type(self):
+ def _scalar_type(self) -> TypeEngine[Any]:
return self.selects[0]._scalar_type()
- def self_group(self, against=None):
+ def self_group(
+ self, against: Optional[OperatorType] = None
+ ) -> GroupedElement:
return SelectStatementGrouping(self)
- def is_derived_from(self, fromclause):
+ def is_derived_from(self, fromclause: FromClause) -> bool:
for s in self.selects:
if s.is_derived_from(fromclause):
return True
return self
- def _generate_fromclause_column_proxies(self, subquery):
+ def _generate_fromclause_column_proxies(
+ self, subquery: FromClause
+ ) -> None:
# this is a slightly hacky thing - the union exports a
# column that resembles just that of the *first* selectable.
def _all_selected_columns(self) -> _SelectIterable:
return self.selects[0]._all_selected_columns
- @property
- def selected_columns(self):
+ @util.ro_non_memoized_property
+ def selected_columns(self) -> ColumnCollection[str, ColumnElement[Any]]:
"""A :class:`_expression.ColumnCollection`
representing the columns that
this SELECT statement or similar construct returns in its result set,
return self.selects[0].selected_columns
+# backwards compat
+for elem in _CompoundSelectKeyword:
+ setattr(CompoundSelect, elem.name, elem)
+
+
@CompileState.plugin_for("default", "select")
class SelectState(util.MemoizedSlots, CompileState):
__slots__ = (
if TYPE_CHECKING:
@classmethod
- def get_plugin_class(cls, statement: Select) -> SelectState:
+ def get_plugin_class(cls, statement: Executable) -> Type[SelectState]:
...
- def __init__(self, statement, compiler, **kw):
+ def __init__(
+ self, statement: Select, compiler: Optional[SQLCompiler], **kw: Any
+ ):
self.statement = statement
self.from_clauses = statement._from_obj
self.columns_plus_names = statement._generate_columns_plus_names(True)
@classmethod
- def _plugin_not_implemented(cls):
+ def _plugin_not_implemented(cls) -> NoReturn:
raise NotImplementedError(
"The default SELECT construct without plugins does not "
"implement this method."
)
@classmethod
- def get_column_descriptions(cls, statement):
+ def get_column_descriptions(
+ cls, statement: Select
+ ) -> List[Dict[str, Any]]:
return [
{
"name": name,
]
@classmethod
- def from_statement(cls, statement, from_statement):
+ def from_statement(
+ cls, statement: Select, from_statement: ReturnsRows
+ ) -> Any:
cls._plugin_not_implemented()
@classmethod
- def get_columns_clause_froms(cls, statement):
+ def get_columns_clause_froms(cls, statement: Select) -> List[FromClause]:
return cls._normalize_froms(
itertools.chain.from_iterable(
element._from_objects for element in statement._raw_columns
)
@classmethod
- def _column_naming_convention(cls, label_style):
+ def _column_naming_convention(
+ cls, label_style: SelectLabelStyle
+ ) -> Callable[[Union[ColumnElement[Any], TextClause]], Optional[str]]:
table_qualified = label_style is LABEL_STYLE_TABLENAME_PLUS_COL
dedupe = label_style is not LABEL_STYLE_NONE
return go
- def _get_froms(self, statement):
+ def _get_froms(self, statement: Select) -> List[FromClause]:
+ ambiguous_table_name_map: _AmbiguousTableNameMap
self._ambiguous_table_name_map = ambiguous_table_name_map = {}
return self._normalize_froms(
@classmethod
def _normalize_froms(
cls,
- iterable_of_froms,
- check_statement=None,
- ambiguous_table_name_map=None,
- ):
+ iterable_of_froms: Iterable[FromClause],
+ check_statement: Optional[Select] = None,
+ ambiguous_table_name_map: Optional[_AmbiguousTableNameMap] = None,
+ ) -> List[FromClause]:
"""given an iterable of things to select FROM, reduce them to what
would actually render in the FROM clause of a SELECT.
etc.
"""
- seen = set()
- froms = []
+ seen: Set[FromClause] = set()
+ froms: List[FromClause] = []
for item in iterable_of_froms:
- if item._is_subquery and item.element is check_statement:
+ if is_subquery(item) and item.element is check_statement:
raise exc.InvalidRequestError(
"select() construct refers to itself as a FROM"
)
)
for item in froms
for fr in item._from_objects
- if fr._is_table
+ if is_table(fr)
and fr.schema
and fr.name not in ambiguous_table_name_map
)
return froms
def _get_display_froms(
- self, explicit_correlate_froms=None, implicit_correlate_froms=None
- ):
+ self,
+ explicit_correlate_froms: Optional[Sequence[FromClause]] = None,
+ implicit_correlate_froms: Optional[Sequence[FromClause]] = None,
+ ) -> List[FromClause]:
"""Return the full list of 'from' clauses to be displayed.
Takes into account a set of existing froms which may be
return froms
- def _memoized_attr__label_resolve_dict(self):
- with_cols = dict(
- (c._tq_label or c.key, c)
+ def _memoized_attr__label_resolve_dict(
+ self,
+ ) -> Tuple[
+ Dict[str, ColumnElement[Any]],
+ Dict[str, ColumnElement[Any]],
+ Dict[str, ColumnElement[Any]],
+ ]:
+ with_cols: Dict[str, ColumnElement[Any]] = dict(
+ (c._tq_label or c.key, c) # type: ignore
for c in self.statement._all_selected_columns
if c._allow_label_resolve
)
- only_froms = dict(
- (c.key, c)
+ only_froms: Dict[str, ColumnElement[Any]] = dict(
+ (c.key, c) # type: ignore
for c in _select_iterables(self.froms)
if c._allow_label_resolve
)
- only_cols = with_cols.copy()
+ only_cols: Dict[str, ColumnElement[Any]] = with_cols.copy()
for key, value in only_froms.items():
with_cols.setdefault(key, value)
return with_cols, only_froms, only_cols
@classmethod
- def determine_last_joined_entity(cls, stmt):
+ def determine_last_joined_entity(
+ cls, stmt: Select
+ ) -> Optional[_JoinTargetElement]:
if stmt._setup_joins:
return stmt._setup_joins[-1][0]
else:
def all_selected_columns(cls, statement: Select) -> _SelectIterable:
return [c for c in _select_iterables(statement._raw_columns)]
- def _setup_joins(self, args, raw_columns):
+ def _setup_joins(
+ self,
+ args: Tuple[_SetupJoinsElement, ...],
+ raw_columns: List[_ColumnsClauseElement],
+ ) -> None:
for (right, onclause, left, flags) in args:
+ if TYPE_CHECKING:
+ if onclause is not None:
+ assert isinstance(onclause, ColumnElement)
+
isouter = flags["isouter"]
full = flags["full"]
left
)
+ # these assertions can be made here, as if the right/onclause
+ # contained ORM elements, the select() statement would have been
+ # upgraded to an ORM select, and this method would not be called;
+ # orm.context.ORMSelectCompileState._join() would be
+ # used instead.
+ if TYPE_CHECKING:
+ assert isinstance(right, FromClause)
+ if onclause is not None:
+ assert isinstance(onclause, ColumnElement)
+
if replace_from_obj_index is not None:
# splice into an existing element in the
# self._from_obj list
+ self.from_clauses[replace_from_obj_index + 1 :]
)
else:
-
+ assert left is not None
self.from_clauses = self.from_clauses + (
Join(left, right, onclause, isouter=isouter, full=full),
)
@util.preload_module("sqlalchemy.sql.util")
def _join_determine_implicit_left_side(
- self, raw_columns, left, right, onclause
- ):
+ self,
+ raw_columns: List[_ColumnsClauseElement],
+ left: Optional[FromClause],
+ right: _JoinTargetElement,
+ onclause: Optional[ColumnElement[Any]],
+ ) -> Tuple[Optional[FromClause], Optional[int]]:
"""When join conditions don't express the left side explicitly,
determine if an existing FROM or entity in this query
can serve as the left hand side.
sql_util = util.preloaded.sql_util
- replace_from_obj_index = None
+ replace_from_obj_index: Optional[int] = None
from_clauses = self.from_clauses
if from_clauses:
- indexes = sql_util.find_left_clause_to_join_from(
+ indexes: List[int] = sql_util.find_left_clause_to_join_from(
from_clauses, right, onclause
)
return left, replace_from_obj_index
@util.preload_module("sqlalchemy.sql.util")
- def _join_place_explicit_left_side(self, left):
- replace_from_obj_index = None
+ def _join_place_explicit_left_side(
+ self, left: FromClause
+ ) -> Optional[int]:
+ replace_from_obj_index: Optional[int] = None
sql_util = util.preloaded.sql_util
from_clauses = list(self.statement._iterate_from_elements())
if from_clauses:
- indexes = sql_util.find_left_clause_that_matches_given(
+ indexes: List[int] = sql_util.find_left_clause_that_matches_given(
self.from_clauses, left
)
else:
class _SelectFromElements:
- def _iterate_from_elements(self):
+ __slots__ = ()
+
+ _raw_columns: List[_ColumnsClauseElement]
+ _where_criteria: Tuple[ColumnElement[Any], ...]
+ _from_obj: Tuple[FromClause, ...]
+
+ def _iterate_from_elements(self) -> Iterator[FromClause]:
# note this does not include elements
# in _setup_joins
yield element
+Self_MemoizedSelectEntities = TypeVar("Self_MemoizedSelectEntities", bound=Any)
+
+
class _MemoizedSelectEntities(
cache_key.HasCacheKey, traversals.HasCopyInternals, visitors.Traversible
):
+ """represents partial state from a Select object, for the case
+ where Select.columns() has redefined the set of columns/entities the
+ statement will be SELECTing from. This object represents
+ the entities from the SELECT before that transformation was applied,
+ so that transformations that were made in terms of the SELECT at that
+ time, such as join() as well as options(), can access the correct context.
+
+ In previous SQLAlchemy versions, this wasn't needed because these
+ constructs calculated everything up front, like when you called join()
+ or options(), it did everything to figure out how that would translate
+ into specific SQL constructs that would be ready to send directly to the
+ SQL compiler when needed. But as of
+ 1.4, all of that stuff is done in the compilation phase, during the
+ "compile state" portion of the process, so that the work can all be
+ cached. So it needs to be able to resolve joins/options2 based on what
+ the list of entities was when those methods were called.
+
+
+ """
+
__visit_name__ = "memoized_select_entities"
- _traverse_internals = [
+ _traverse_internals: _TraverseInternalsType = [
("_raw_columns", InternalTraversal.dp_clauseelement_list),
("_setup_joins", InternalTraversal.dp_setup_join_tuple),
("_with_options", InternalTraversal.dp_executable_options),
]
+ _is_clone_of: Optional[ClauseElement]
+ _raw_columns: List[_ColumnsClauseElement]
+ _setup_joins: Tuple[_SetupJoinsElement, ...]
+ _with_options: Tuple[ExecutableOption, ...]
+
_annotations = util.EMPTY_DICT
- def _clone(self, **kw):
+ def _clone(
+ self: Self_MemoizedSelectEntities, **kw: Any
+ ) -> Self_MemoizedSelectEntities:
c = self.__class__.__new__(self.__class__)
c.__dict__ = {k: v for k, v in self.__dict__.items()}
c._is_clone_of = self.__dict__.get("_is_clone_of", self)
- return c
+ return c # type: ignore
@classmethod
- def _generate_for_statement(cls, select_stmt):
+ def _generate_for_statement(cls, select_stmt: Select) -> None:
if select_stmt._setup_joins or select_stmt._with_options:
self = _MemoizedSelectEntities()
self._raw_columns = select_stmt._raw_columns
self._with_options = select_stmt._with_options
select_stmt._memoized_select_entities += (self,)
- select_stmt._raw_columns = (
- select_stmt._setup_joins
- ) = select_stmt._with_options = ()
+ select_stmt._raw_columns = []
+ select_stmt._setup_joins = select_stmt._with_options = ()
-# TODO: use pep-673 when feasible
SelfSelect = typing.TypeVar("SelfSelect", bound="Select")
__visit_name__ = "select"
- _setup_joins: Tuple[TODO_Any, ...] = ()
+ _setup_joins: Tuple[_SetupJoinsElement, ...] = ()
_memoized_select_entities: Tuple[TODO_Any, ...] = ()
+ _raw_columns: List[_ColumnsClauseElement]
+
_distinct = False
_distinct_on: Tuple[ColumnElement[Any], ...] = ()
_correlate: Tuple[FromClause, ...] = ()
_having_criteria: Tuple[ColumnElement[Any], ...] = ()
_from_obj: Tuple[FromClause, ...] = ()
_auto_correlate = True
-
+ _is_select_statement = True
_compile_options: CacheableOptions = (
SelectState.default_select_compile_options
)
- _traverse_internals = (
+ _traverse_internals: _TraverseInternalsType = (
[
("_raw_columns", InternalTraversal.dp_clauseelement_list),
(
+ Executable._executable_traverse_internals
)
- _cache_key_traversal = _traverse_internals + [
+ _cache_key_traversal: _CacheKeyTraversalType = _traverse_internals + [
("_compile_options", InternalTraversal.dp_has_cache_key)
]
+ _compile_state_factory: Type[SelectState]
+
@classmethod
- def _create_raw_select(cls, **kw) -> "Select":
+ def _create_raw_select(cls, **kw: Any) -> Select:
"""Create a :class:`.Select` using raw ``__new__`` with no coercions.
Used internally to build up :class:`.Select` constructs with
:func:`_sql.select` function.
"""
+ things = [
+ coercions.expect(
+ roles.ColumnsClauseRole, ent, apply_propagate_attrs=self
+ )
+ for ent in entities
+ ]
self._raw_columns = [
coercions.expect(
GenerativeSelect.__init__(self)
- def _scalar_type(self):
+ def _scalar_type(self) -> TypeEngine[Any]:
elem = self._raw_columns[0]
cols = list(elem._select_iterable)
return cols[0].type
@_generative
def join(
- self: SelfSelect, target, onclause=None, *, isouter=False, full=False
+ self: SelfSelect,
+ target: _JoinTargetArgument,
+ onclause: Optional[_OnClauseArgument] = None,
+ *,
+ isouter: bool = False,
+ full: bool = False,
) -> SelfSelect:
r"""Create a SQL JOIN against this :class:`_expression.Select`
object's criterion
:meth:`_expression.Select.outerjoin`
""" # noqa: E501
- target = coercions.expect(
+ join_target = coercions.expect(
roles.JoinTargetRole, target, apply_propagate_attrs=self
)
if onclause is not None:
- onclause = coercions.expect(roles.OnClauseRole, onclause)
+ onclause_element = coercions.expect(roles.OnClauseRole, onclause)
+ else:
+ onclause_element = None
+
self._setup_joins += (
- (target, onclause, None, {"isouter": isouter, "full": full}),
+ (
+ join_target,
+ onclause_element,
+ None,
+ {"isouter": isouter, "full": full},
+ ),
)
return self
- def outerjoin_from(self, from_, target, onclause=None, *, full=False):
+ def outerjoin_from(
+ self: SelfSelect,
+ from_: _FromClauseArgument,
+ target: _JoinTargetArgument,
+ onclause: Optional[_OnClauseArgument] = None,
+ *,
+ full: bool = False,
+ ) -> SelfSelect:
r"""Create a SQL LEFT OUTER JOIN against this :class:`_expression.Select`
object's criterion
and apply generatively, returning the newly resulting
@_generative
def join_from(
self: SelfSelect,
- from_,
- target,
- onclause=None,
+ from_: _FromClauseArgument,
+ target: _JoinTargetArgument,
+ onclause: Optional[_OnClauseArgument] = None,
*,
- isouter=False,
- full=False,
+ isouter: bool = False,
+ full: bool = False,
) -> SelfSelect:
r"""Create a SQL JOIN against this :class:`_expression.Select`
object's criterion
from_ = coercions.expect(
roles.FromClauseRole, from_, apply_propagate_attrs=self
)
- target = coercions.expect(
+ join_target = coercions.expect(
roles.JoinTargetRole, target, apply_propagate_attrs=self
)
if onclause is not None:
- onclause = coercions.expect(roles.OnClauseRole, onclause)
+ onclause_element = coercions.expect(roles.OnClauseRole, onclause)
+ else:
+ onclause_element = None
self._setup_joins += (
- (target, onclause, from_, {"isouter": isouter, "full": full}),
+ (
+ join_target,
+ onclause_element,
+ from_,
+ {"isouter": isouter, "full": full},
+ ),
)
return self
- def outerjoin(self, target, onclause=None, *, full=False):
+ def outerjoin(
+ self: SelfSelect,
+ target: _JoinTargetArgument,
+ onclause: Optional[_OnClauseArgument] = None,
+ *,
+ full: bool = False,
+ ) -> SelfSelect:
"""Create a left outer join.
Parameters are the same as that of :meth:`_expression.Select.join`.
"""
return self.join(target, onclause=onclause, isouter=True, full=full)
- def get_final_froms(self):
+ def get_final_froms(self) -> Sequence[FromClause]:
"""Compute the final displayed list of :class:`_expression.FromClause`
elements.
:attr:`_sql.Select.columns_clause_froms`
"""
+
return self._compile_state_factory(self, None)._get_display_froms()
@util.deprecated_property(
"The :attr:`_expression.Select.froms` attribute is moved to "
"the :meth:`_expression.Select.get_final_froms` method.",
)
- def froms(self):
+ def froms(self) -> Sequence[FromClause]:
"""Return the displayed list of :class:`_expression.FromClause`
elements.
return self.get_final_froms()
@property
- def columns_clause_froms(self):
+ def columns_clause_froms(self) -> List[FromClause]:
"""Return the set of :class:`_expression.FromClause` objects implied
by the columns clause of this SELECT statement.
return iter(self._all_selected_columns)
- def is_derived_from(self, fromclause):
+ def is_derived_from(self, fromclause: FromClause) -> bool:
if self in fromclause._cloned_set:
return True
return True
return False
- def _copy_internals(self, clone=_clone, **kw):
+ def _copy_internals(
+ self, clone: _CloneCallableType = _clone, **kw: Any
+ ) -> None:
# Select() object has been cloned and probably adapted by the
# given clone function. Apply the cloning function to internal
# objects
def get_children(self, **kwargs):
return itertools.chain(
super(Select, self).get_children(
- omit_attrs=["_from_obj", "_correlate", "_correlate_except"]
+ omit_attrs=("_from_obj", "_correlate", "_correlate_except")
),
self._iterate_from_elements(),
)
@_generative
- def add_columns(self: SelfSelect, *columns) -> SelfSelect:
+ def add_columns(
+ self: SelfSelect, *columns: _ColumnsClauseArgument
+ ) -> SelfSelect:
"""Return a new :func:`_expression.select` construct with
the given column expressions added to its columns clause.
]
return self
- def _set_entities(self, entities):
+ def _set_entities(
+ self, entities: Iterable[_ColumnsClauseArgument]
+ ) -> None:
self._raw_columns = [
coercions.expect(
roles.ColumnsClauseRole, ent, apply_propagate_attrs=self
"be removed in a future release. Please use "
":meth:`_expression.Select.add_columns`",
)
- def column(self, column):
+ def column(self: SelfSelect, column: _ColumnsClauseArgument) -> SelfSelect:
"""Return a new :func:`_expression.select` construct with
the given column expression added to its columns clause.
return self.add_columns(column)
@util.preload_module("sqlalchemy.sql.util")
- def reduce_columns(self, only_synonyms=True):
+ def reduce_columns(
+ self: SelfSelect, only_synonyms: bool = True
+ ) -> SelfSelect:
"""Return a new :func:`_expression.select` construct with redundantly
named, equivalently-valued columns removed from the columns clause.
@_generative
def with_only_columns(
- self: SelfSelect, *columns, maintain_column_froms=False
+ self: SelfSelect,
+ *columns: _ColumnsClauseArgument,
+ maintain_column_froms: bool = False,
) -> SelfSelect:
r"""Return a new :func:`_expression.select` construct with its columns
clause replaced with the given columns.
self._assert_no_memoizations()
if maintain_column_froms:
- self.select_from.non_generative(self, *self.columns_clause_froms)
+ self.select_from.non_generative( # type: ignore
+ self, *self.columns_clause_froms
+ )
# then memoize the FROMs etc.
_MemoizedSelectEntities._generate_for_statement(self)
_whereclause = whereclause
@_generative
- def where(self: SelfSelect, *whereclause) -> SelfSelect:
+ def where(
+ self: SelfSelect, *whereclause: _ColumnExpressionArgument[bool]
+ ) -> 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.
assert isinstance(self._where_criteria, tuple)
for criterion in whereclause:
- where_criteria = coercions.expect(roles.WhereHavingRole, criterion)
+ where_criteria: ColumnElement[Any] = coercions.expect(
+ roles.WhereHavingRole, criterion
+ )
self._where_criteria += (where_criteria,)
return self
@_generative
- def having(self: SelfSelect, having) -> SelfSelect:
+ def having(
+ self: SelfSelect, *having: _ColumnExpressionArgument[bool]
+ ) -> 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),
- )
+
+ for criterion in having:
+ having_criteria = coercions.expect(
+ roles.WhereHavingRole, criterion
+ )
+ self._having_criteria += (having_criteria,)
return self
@_generative
- def distinct(self: SelfSelect, *expr) -> SelfSelect:
+ def distinct(
+ self: SelfSelect, *expr: _ColumnExpressionArgument[Any]
+ ) -> SelfSelect:
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) -> SelfSelect:
+ def select_from(
+ self: SelfSelect, *froms: _FromClauseArgument
+ ) -> SelfSelect:
r"""Return a new :func:`_expression.select` construct with the
given FROM expression(s)
merged into its list of FROM objects.
return self
@_generative
- def correlate(self: SelfSelect, *fromclauses) -> SelfSelect:
+ def correlate(
+ self: SelfSelect,
+ *fromclauses: Union[Literal[None, False], _FromClauseArgument],
+ ) -> SelfSelect:
r"""Return a new :class:`_expression.Select`
which will correlate the given FROM
clauses to that of an enclosing :class:`_expression.Select`.
none of its FROM entries, and all will render unconditionally
in the local FROM clause.
- :param \*fromclauses: a list of one or more
- :class:`_expression.FromClause`
- constructs, or other compatible constructs (i.e. ORM-mapped
- classes) to become part of the correlate collection.
+ :param \*fromclauses: one or more :class:`.FromClause` or other
+ FROM-compatible construct such as an ORM mapped entity to become part
+ of the correlate collection; alternatively pass a single value
+ ``None`` to remove all existing correlations.
.. seealso::
"""
+ # tests failing when we try to change how these
+ # arguments are passed
+
self._auto_correlate = False
- if fromclauses and fromclauses[0] in {None, False}:
+ if not fromclauses or fromclauses[0] in {None, False}:
+ if len(fromclauses) > 1:
+ raise exc.ArgumentError(
+ "additional FROM objects not accepted when "
+ "passing None/False to correlate()"
+ )
self._correlate = ()
else:
self._correlate = self._correlate + tuple(
return self
@_generative
- def correlate_except(self: SelfSelect, *fromclauses) -> SelfSelect:
+ def correlate_except(
+ self: SelfSelect,
+ *fromclauses: Union[Literal[None, False], _FromClauseArgument],
+ ) -> SelfSelect:
r"""Return a new :class:`_expression.Select`
which will omit the given FROM
clauses from the auto-correlation process.
all other FROM elements remain subject to normal auto-correlation
behaviors.
- If ``None`` is passed, the :class:`_expression.Select`
- object will correlate
- all of its FROM entries.
+ If ``None`` is passed, or no arguments are passed,
+ the :class:`_expression.Select` object will correlate all of its
+ FROM entries.
:param \*fromclauses: a list of one or more
:class:`_expression.FromClause`
"""
self._auto_correlate = False
- if fromclauses and fromclauses[0] in {None, False}:
+ if not fromclauses or fromclauses[0] in {None, False}:
+ if len(fromclauses) > 1:
+ raise exc.ArgumentError(
+ "additional FROM objects not accepted when "
+ "passing None/False to correlate_except()"
+ )
self._correlate_except = ()
else:
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):
+ @HasMemoized_ro_memoized_attribute
+ def selected_columns(self) -> ColumnCollection[str, ColumnElement[Any]]:
"""A :class:`_expression.ColumnCollection`
representing the columns that
this SELECT statement or similar construct returns in its result set,
# generates the actual names used in the SELECT string. that
# method is more complex because it also renders columns that are
# fully ambiguous, e.g. same column more than once.
- conv = SelectState._column_naming_convention(self._label_style)
+ conv = cast(
+ "Callable[[Any], str]",
+ SelectState._column_naming_convention(self._label_style),
+ )
- return ColumnCollection(
+ cc: ColumnCollection[str, ColumnElement[Any]] = ColumnCollection(
[
(conv(c), c)
for c in self._all_selected_columns
if is_column_element(c)
]
- ).as_readonly()
+ )
+ return cc.as_readonly()
@HasMemoized.memoized_attribute
- def _all_selected_columns(self) -> Sequence[ColumnElement[Any]]:
+ def _all_selected_columns(self) -> _SelectIterable:
meth = SelectState.get_plugin_class(self).all_selected_columns
return list(meth(self))
self = self.set_label_style(LABEL_STYLE_DISAMBIGUATE_ONLY)
return self
- def _generate_columns_plus_names(
- self, anon_for_dupe_key: bool
- ) -> List[Tuple[str, str, str, ColumnElement[Any], bool]]:
- """Generate column names as rendered in a SELECT statement by
- the compiler.
-
- This is distinct from the _column_naming_convention generator that's
- intended for population of .c collections and similar, which has
- different rules. the collection returned here calls upon the
- _column_naming_convention as well.
-
- """
- cols = self._all_selected_columns
-
- key_naming_convention = SelectState._column_naming_convention(
- self._label_style
- )
-
- names = {}
-
- result = []
- result_append = result.append
-
- table_qualified = self._label_style is LABEL_STYLE_TABLENAME_PLUS_COL
- label_style_none = self._label_style is LABEL_STYLE_NONE
-
- # a counter used for "dedupe" labels, which have double underscores
- # in them and are never referred by name; they only act
- # as positional placeholders. they need only be unique within
- # the single columns clause they're rendered within (required by
- # some dbs such as mysql). So their anon identity is tracked against
- # a fixed counter rather than hash() identity.
- dedupe_hash = 1
-
- for c in cols:
- repeated = False
-
- if not c._render_label_in_columns_clause:
- effective_name = (
- required_label_name
- ) = fallback_label_name = None
- elif label_style_none:
- effective_name = required_label_name = None
- fallback_label_name = c._non_anon_label or c._anon_name_label
- else:
- if table_qualified:
- required_label_name = (
- effective_name
- ) = fallback_label_name = c._tq_label
- else:
- effective_name = fallback_label_name = c._non_anon_label
- required_label_name = None
-
- if effective_name is None:
- # it seems like this could be _proxy_key and we would
- # not need _expression_label but it isn't
- # giving us a clue when to use anon_label instead
- expr_label = c._expression_label
- if expr_label is None:
- repeated = c._anon_name_label in names
- names[c._anon_name_label] = c
- effective_name = required_label_name = None
-
- if repeated:
- # here, "required_label_name" is sent as
- # "None" and "fallback_label_name" is sent.
- if table_qualified:
- fallback_label_name = (
- c._dedupe_anon_tq_label_idx(dedupe_hash)
- )
- dedupe_hash += 1
- else:
- fallback_label_name = c._dedupe_anon_label_idx(
- dedupe_hash
- )
- dedupe_hash += 1
- else:
- fallback_label_name = c._anon_name_label
- else:
- required_label_name = (
- effective_name
- ) = fallback_label_name = expr_label
-
- if effective_name is not None:
- if effective_name in names:
- # when looking to see if names[name] is the same column as
- # c, use hash(), so that an annotated version of the column
- # is seen as the same as the non-annotated
- if hash(names[effective_name]) != hash(c):
-
- # different column under the same name. apply
- # disambiguating label
- if table_qualified:
- required_label_name = (
- fallback_label_name
- ) = c._anon_tq_label
- else:
- required_label_name = (
- fallback_label_name
- ) = c._anon_name_label
-
- if anon_for_dupe_key and required_label_name in names:
- # here, c._anon_tq_label is definitely unique to
- # that column identity (or annotated version), so
- # this should always be true.
- # this is also an infrequent codepath because
- # you need two levels of duplication to be here
- assert hash(names[required_label_name]) == hash(c)
-
- # the column under the disambiguating label is
- # already present. apply the "dedupe" label to
- # subsequent occurrences of the column so that the
- # original stays non-ambiguous
- if table_qualified:
- required_label_name = (
- fallback_label_name
- ) = c._dedupe_anon_tq_label_idx(dedupe_hash)
- dedupe_hash += 1
- else:
- required_label_name = (
- fallback_label_name
- ) = c._dedupe_anon_label_idx(dedupe_hash)
- dedupe_hash += 1
- repeated = True
- else:
- names[required_label_name] = c
- elif anon_for_dupe_key:
- # same column under the same name. apply the "dedupe"
- # label so that the original stays non-ambiguous
- if table_qualified:
- required_label_name = (
- fallback_label_name
- ) = c._dedupe_anon_tq_label_idx(dedupe_hash)
- dedupe_hash += 1
- else:
- required_label_name = (
- fallback_label_name
- ) = c._dedupe_anon_label_idx(dedupe_hash)
- dedupe_hash += 1
- repeated = True
- else:
- names[effective_name] = c
-
- result_append(
- (
- # string label name, if non-None, must be rendered as a
- # label, i.e. "AS <name>"
- required_label_name,
- # proxy_key that is to be part of the result map for this
- # col. this is also the key in a fromclause.c or
- # select.selected_columns collection
- key_naming_convention(c),
- # name that can be used to render an "AS <name>" when
- # we have to render a label even though
- # required_label_name was not given
- fallback_label_name,
- # the ColumnElement itself
- c,
- # True if this is a duplicate of a previous column
- # in the list of columns
- repeated,
- )
- )
-
- return result
-
- def _generate_fromclause_column_proxies(self, subquery):
+ def _generate_fromclause_column_proxies(
+ self, subquery: FromClause
+ ) -> None:
"""Generate column proxies to place in the exported ``.c``
collection of a subquery."""
c,
repeated,
) in (self._generate_columns_plus_names(False))
- if not c._is_text_clause
+ if is_column_element(c)
]
subquery._columns._populate_separate_keys(prox)
self._order_by_clause.clauses
)
- def self_group(self, against=None):
+ def self_group(
+ self: Self, against: Optional[OperatorType] = None
+ ) -> Union[SelectStatementGrouping, Self]:
+ ...
"""Return a 'grouping' construct as per the
:class:`_expression.ClauseElement` specification.
else:
return SelectStatementGrouping(self)
- def union(self, *other, **kwargs):
+ def union(
+ self, *other: _SelectStatementForCompoundArgument
+ ) -> CompoundSelect:
r"""Return a SQL ``UNION`` of this select() construct against
the given selectables provided as positional arguments.
for the newly created :class:`_sql.CompoundSelect` object.
"""
- return CompoundSelect._create_union(self, *other, **kwargs)
+ return CompoundSelect._create_union(self, *other)
- def union_all(self, *other, **kwargs):
+ def union_all(
+ self, *other: _SelectStatementForCompoundArgument
+ ) -> CompoundSelect:
r"""Return a SQL ``UNION ALL`` of this select() construct against
the given selectables provided as positional arguments.
for the newly created :class:`_sql.CompoundSelect` object.
"""
- return CompoundSelect._create_union_all(self, *other, **kwargs)
+ return CompoundSelect._create_union_all(self, *other)
- def except_(self, *other, **kwargs):
+ def except_(
+ self, *other: _SelectStatementForCompoundArgument
+ ) -> CompoundSelect:
r"""Return a SQL ``EXCEPT`` of this select() construct against
the given selectable provided as positional arguments.
multiple elements are now accepted.
- :param \**kwargs: keyword arguments are forwarded to the constructor
- for the newly created :class:`_sql.CompoundSelect` object.
-
"""
- return CompoundSelect._create_except(self, *other, **kwargs)
+ return CompoundSelect._create_except(self, *other)
- def except_all(self, *other, **kwargs):
+ def except_all(
+ self, *other: _SelectStatementForCompoundArgument
+ ) -> CompoundSelect:
r"""Return a SQL ``EXCEPT ALL`` of this select() construct against
the given selectables provided as positional arguments.
multiple elements are now accepted.
- :param \**kwargs: keyword arguments are forwarded to the constructor
- for the newly created :class:`_sql.CompoundSelect` object.
-
"""
- return CompoundSelect._create_except_all(self, *other, **kwargs)
+ return CompoundSelect._create_except_all(self, *other)
- def intersect(self, *other, **kwargs):
+ def intersect(
+ self, *other: _SelectStatementForCompoundArgument
+ ) -> CompoundSelect:
r"""Return a SQL ``INTERSECT`` of this select() construct against
the given selectables provided as positional arguments.
for the newly created :class:`_sql.CompoundSelect` object.
"""
- return CompoundSelect._create_intersect(self, *other, **kwargs)
+ return CompoundSelect._create_intersect(self, *other)
- def intersect_all(self, *other, **kwargs):
+ def intersect_all(
+ self, *other: _SelectStatementForCompoundArgument
+ ) -> CompoundSelect:
r"""Return a SQL ``INTERSECT ALL`` of this select() construct
against the given selectables provided as positional arguments.
for the newly created :class:`_sql.CompoundSelect` object.
"""
- return CompoundSelect._create_intersect_all(self, *other, **kwargs)
+ return CompoundSelect._create_intersect_all(self, *other)
-SelfScalarSelect = typing.TypeVar("SelfScalarSelect", bound="ScalarSelect")
+SelfScalarSelect = typing.TypeVar(
+ "SelfScalarSelect", bound="ScalarSelect[Any]"
+)
-class ScalarSelect(roles.InElementRole, Generative, Grouping):
+class ScalarSelect(
+ roles.InElementRole, Generative, GroupedElement, ColumnElement[_T]
+):
"""Represent a scalar subquery.
"""
- _from_objects = []
+ _traverse_internals: _TraverseInternalsType = [
+ ("element", InternalTraversal.dp_clauseelement),
+ ("type", InternalTraversal.dp_type),
+ ]
+
+ _from_objects: List[FromClause] = []
_is_from_container = True
- _is_implicitly_boolean = False
+ if not TYPE_CHECKING:
+ _is_implicitly_boolean = False
inherit_cache = True
+ element: SelectBase
+
def __init__(self, element):
self.element = element
self.type = element._scalar_type()
+ def __getattr__(self, attr):
+ return getattr(self.element, attr)
+
+ def __getstate__(self):
+ return {"element": self.element, "type": self.type}
+
+ def __setstate__(self, state):
+ self.element = state["element"]
+ self.type = state["type"]
+
@property
def columns(self):
raise exc.InvalidRequestError(
c = columns
@_generative
- def where(self: SelfScalarSelect, crit) -> SelfScalarSelect:
+ def where(
+ self: SelfScalarSelect, crit: _ColumnExpressionArgument[bool]
+ ) -> SelfScalarSelect:
"""Apply a WHERE clause to the SELECT statement referred to
by this :class:`_expression.ScalarSelect`.
"""
- self.element = self.element.where(crit)
+ self.element = cast(Select, self.element).where(crit)
return self
- def self_group(self, **kwargs):
+ @overload
+ def self_group(
+ self: ScalarSelect[Any], against: Optional[OperatorType] = None
+ ) -> ScalarSelect[Any]:
+ ...
+
+ @overload
+ def self_group(
+ self: ColumnElement[Any], against: Optional[OperatorType] = None
+ ) -> ColumnElement[Any]:
+ ...
+
+ def self_group(
+ self, against: Optional[OperatorType] = None
+ ) -> ColumnElement[Any]:
+
return self
@_generative
- def correlate(self: SelfScalarSelect, *fromclauses) -> SelfScalarSelect:
+ def correlate(
+ self: SelfScalarSelect,
+ *fromclauses: Union[Literal[None, False], _FromClauseArgument],
+ ) -> 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)
+ self.element = cast(Select, self.element).correlate(*fromclauses)
return self
@_generative
def correlate_except(
- self: SelfScalarSelect, *fromclauses
+ self: SelfScalarSelect,
+ *fromclauses: Union[Literal[None, False], _FromClauseArgument],
) -> SelfScalarSelect:
r"""Return a new :class:`_expression.ScalarSelect`
which will omit the given FROM
"""
- self.element = self.element.correlate_except(*fromclauses)
+ self.element = cast(Select, self.element).correlate_except(
+ *fromclauses
+ )
return self
-class Exists(UnaryExpression[_T]):
+SelfExists = TypeVar("SelfExists", bound="Exists")
+
+
+class Exists(UnaryExpression[bool]):
"""Represent an ``EXISTS`` clause.
See :func:`_sql.exists` for a description of usage.
"""
- _from_objects = ()
inherit_cache = True
- def __init__(self, __argument=None):
+ def __init__(
+ self,
+ __argument: Optional[
+ Union[_ColumnsClauseArgument, SelectBase, ScalarSelect[bool]]
+ ] = None,
+ ):
if __argument is None:
s = Select(literal_column("*")).scalar_subquery()
elif isinstance(__argument, (SelectBase, ScalarSelect)):
wraps_column_expression=True,
)
+ @util.ro_non_memoized_property
+ def _from_objects(self) -> List[FromClause]:
+ return []
+
def _regroup(self, fn):
element = self.element._ungroup()
element = fn(element)
return element.self_group(against=operators.exists)
- def select(self) -> "Select":
+ def select(self) -> Select:
r"""Return a SELECT of this :class:`_expression.Exists`.
e.g.::
return Select(self)
- def correlate(self, *fromclause):
+ def correlate(
+ self: SelfExists,
+ *fromclauses: Union[Literal[None, False], _FromClauseArgument],
+ ) -> SelfExists:
"""Apply correlation to the subquery noted by this :class:`_sql.Exists`.
.. seealso::
"""
e = self._clone()
e.element = self._regroup(
- lambda element: element.correlate(*fromclause)
+ lambda element: element.correlate(*fromclauses)
)
return e
- def correlate_except(self, *fromclause):
+ def correlate_except(
+ self: SelfExists,
+ *fromclauses: Union[Literal[None, False], _FromClauseArgument],
+ ) -> SelfExists:
"""Apply correlation to the subquery noted by this :class:`_sql.Exists`.
.. seealso::
e = self._clone()
e.element = self._regroup(
- lambda element: element.correlate_except(*fromclause)
+ lambda element: element.correlate_except(*fromclauses)
)
return e
- def select_from(self, *froms):
+ def select_from(self: SelfExists, *froms: FromClause) -> SelfExists:
"""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, *clause):
+ def where(
+ self: SelfExists, *clause: _ColumnExpressionArgument[bool]
+ ) -> SelfExists:
"""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.
_label_style = LABEL_STYLE_NONE
- _traverse_internals = [
+ _traverse_internals: _TraverseInternalsType = [
("element", InternalTraversal.dp_clauseelement),
("column_args", InternalTraversal.dp_clauseelement_list),
] + SupportsCloneAnnotations._clone_annotations_traverse_internals
]
self.positional = positional
- @HasMemoized.memoized_attribute
- def selected_columns(self):
+ @HasMemoized_ro_memoized_attribute
+ def selected_columns(self) -> ColumnCollection[str, ColumnElement[Any]]:
"""A :class:`_expression.ColumnCollection`
representing the columns that
this SELECT statement or similar construct returns in its result set,
(c.key, c) for c in self.column_args
).as_readonly()
+ # def _generate_columns_plus_names(
+ # self, anon_for_dupe_key: bool
+ # ) -> List[Tuple[str, str, str, ColumnElement[Any], bool]]:
+ # return Select._generate_columns_plus_names(
+ # self, anon_for_dupe_key=anon_for_dupe_key
+ # )
+
@util.non_memoized_property
def _all_selected_columns(self) -> _SelectIterable:
return self.column_args
@_generative
def bindparams(
- self: SelfTextualSelect, *binds, **bind_as_values
+ self: SelfTextualSelect,
+ *binds: BindParameter[Any],
+ **bind_as_values: Any,
) -> SelfTextualSelect:
self.element = self.element.bindparams(*binds, **bind_as_values)
return self
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
