diff --git a/lib/sqlalchemy/sql/base.py b/lib/sqlalchemy/sql/base.py
index 1752a4dc1ab..2eaf0bcde77 100644
--- a/lib/sqlalchemy/sql/base.py
+++ b/lib/sqlalchemy/sql/base.py
@@ -185,7 +185,7 @@ def proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
raise NotImplementedError()
@classmethod
- def _create_singleton(cls):
+ def _create_singleton(cls) -> None:
obj = object.__new__(cls)
obj.__init__() # type: ignore
diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index a416b6ac096..078ce2bad72 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -4,7 +4,6 @@
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
-# mypy: allow-untyped-defs, allow-untyped-calls
"""Core SQL expression elements, including :class:`_expression.ClauseElement`,
:class:`_expression.ColumnElement`, and derived classes.
@@ -80,11 +79,16 @@
from ..util.typing import Self
if typing.TYPE_CHECKING:
+ from re import Match
+ from typing import NoReturn
+
from ._typing import _ColumnExpressionArgument
from ._typing import _ColumnExpressionOrStrLabelArgument
from ._typing import _InfoType
from ._typing import _PropagateAttrsType
from ._typing import _TypeEngineArgument
+ from .annotation import SupportsAnnotations
+ from .base import _EntityNamespace
from .cache_key import _CacheKeyTraversalType
from .cache_key import CacheKey
from .compiler import Compiled
@@ -100,10 +104,14 @@
from .selectable import FromClause
from .selectable import NamedFromClause
from .selectable import TextualSelect
+ from .sqltypes import Boolean
+ from .sqltypes import NullType
+ from .sqltypes import String
from .sqltypes import TupleType
from .type_api import TypeEngine
from .visitors import _CloneCallableType
from .visitors import _TraverseInternalsType
+ from .visitors import anon_map
from ..engine import Connection
from ..engine import Dialect
from ..engine import Engine
@@ -408,7 +416,11 @@ def _clone(self, **kw: Any) -> Self:
c._is_clone_of = cc if cc is not None else self
return c
- def _negate_in_binary(self, negated_op, original_op):
+ def _negate_in_binary(
+ self,
+ negated_op: OperatorType,
+ original_op: OperatorType,
+ ) -> Self:
"""a hook to allow the right side of a binary expression to respond
to a negation of the binary expression.
@@ -417,7 +429,7 @@ def _negate_in_binary(self, negated_op, original_op):
"""
return self
- def _with_binary_element_type(self, type_):
+ def _with_binary_element_type(self, type_: Any) -> Any:
"""in the context of binary expression, convert the type of this
object to the one given.
@@ -427,7 +439,7 @@ def _with_binary_element_type(self, type_):
return self
@property
- def _constructor(self):
+ def _constructor(self) -> Any:
"""return the 'constructor' for this ClauseElement.
This is for the purposes for creating a new object of
@@ -439,7 +451,7 @@ def _constructor(self):
return self.__class__
@HasMemoized.memoized_attribute
- def _cloned_set(self):
+ def _cloned_set(self) -> Set[ClauseElement]:
"""Return the set consisting all cloned ancestors of this
ClauseElement.
@@ -462,13 +474,13 @@ def _cloned_set(self):
return s
@property
- def entity_namespace(self):
+ def entity_namespace(self) -> Union[_EntityNamespace, NoReturn]:
raise AttributeError(
"This SQL expression has no entity namespace "
"with which to filter from."
)
- def __getstate__(self):
+ def __getstate__(self) -> Dict[str, Any]:
d = self.__dict__.copy()
d.pop("_is_clone_of", None)
d.pop("_generate_cache_key", None)
@@ -680,7 +692,7 @@ def _compile_w_cache(
return compiled_sql, extracted_params, cache_hit
- def __invert__(self):
+ def __invert__(self) -> Union[operators.Operators, ClauseElement]:
# undocumented element currently used by the ORM for
# relationship.contains()
if hasattr(self, "negation_clause"):
@@ -693,10 +705,10 @@ def _negate(self) -> ClauseElement:
assert isinstance(grouped, ColumnElement)
return UnaryExpression(grouped, operator=operators.inv)
- def __bool__(self):
+ def __bool__(self) -> bool:
raise TypeError("Boolean value of this clause is not defined")
- def __repr__(self):
+ def __repr__(self) -> str:
friendly = self.description
if friendly is None:
return object.__repr__(self)
@@ -1365,7 +1377,7 @@ def _non_anon_label(self) -> Optional[str]:
"""
return getattr(self, "name", None)
- _render_label_in_columns_clause = True
+ _render_label_in_columns_clause: bool = True
"""A flag used by select._columns_plus_names that helps to determine
we are actually going to render in terms of "SELECT
AS ".
This flag can be returned as False for some Column objects that want
@@ -1548,7 +1560,9 @@ def shares_lineage(self, othercolumn: ColumnElement[Any]) -> bool:
return bool(self.proxy_set.intersection(othercolumn.proxy_set))
- def _compare_name_for_result(self, other: ColumnElement[Any]) -> bool:
+ def _compare_name_for_result(
+ self, other: ColumnElement[Any]
+ ) -> Union[bool, FrozenSet[ColumnElement[Any]]]:
"""Return True if the given column element compares to this one
when targeting within a result row."""
@@ -1858,7 +1872,7 @@ def _dedupe_anon_label_idx(self, idx: int) -> str:
return self._dedupe_anon_tq_label_idx(idx)
@property
- def _proxy_key(self):
+ def _proxy_key(self) -> Optional[str]:
wce = self.wrapped_column_expression
if not wce._is_text_clause:
@@ -2008,9 +2022,15 @@ def __init__(
else:
self.type = type_
- def _with_value(self, value, maintain_key=False, required=NO_ARG):
+ def _with_value(
+ self,
+ value: Optional[_T],
+ maintain_key: bool = False,
+ required: Union[bool, _NoArg] = NO_ARG,
+ ) -> BindParameter[_T]:
"""Return a copy of this :class:`.BindParameter` with the given value
set.
+
"""
cloned = self._clone(maintain_key=maintain_key)
cloned.value = value
@@ -2064,7 +2084,9 @@ def render_literal_execute(self) -> BindParameter[_T]:
literal_execute=True,
)
- def _negate_in_binary(self, negated_op, original_op):
+ def _negate_in_binary(
+ self, negated_op: OperatorType, original_op: OperatorType
+ ) -> BindParameter[_T]:
if self.expand_op is original_op:
bind = self._clone()
bind.expand_op = negated_op
@@ -2072,7 +2094,7 @@ def _negate_in_binary(self, negated_op, original_op):
else:
return self
- def _with_binary_element_type(self, type_):
+ def _with_binary_element_type(self, type_: Any) -> ClauseElement:
c = ClauseElement._clone(self)
c.type = type_
return c
@@ -2094,7 +2116,11 @@ def _clone(self, maintain_key: bool = False, **kw: Any) -> Self:
)
return c
- def _gen_cache_key(self, anon_map, bindparams):
+ def _gen_cache_key(
+ self,
+ anon_map: anon_map,
+ bindparams: List[BindParameter[_T]],
+ ) -> Optional[typing_Tuple[Any, ...]]:
_gen_cache_ok = self.__class__.__dict__.get("inherit_cache", False)
if not _gen_cache_ok:
@@ -2117,14 +2143,14 @@ def _gen_cache_key(self, anon_map, bindparams):
self.literal_execute,
)
- def _convert_to_unique(self):
+ def _convert_to_unique(self) -> None:
if not self.unique:
self.unique = True
self.key = _anonymous_label.safe_construct(
id(self), self._orig_key or "param", sanitize_key=True
)
- def __getstate__(self):
+ def __getstate__(self) -> Dict[str, Any]:
"""execute a deferred value for serialization purposes."""
d = self.__dict__.copy()
@@ -2135,14 +2161,14 @@ def __getstate__(self):
d["value"] = v
return d
- def __setstate__(self, state):
+ def __setstate__(self, state: Dict[str, Any]) -> None:
if state.get("unique", False):
state["key"] = _anonymous_label.safe_construct(
id(self), state.get("_orig_key", "param"), sanitize_key=True
)
self.__dict__.update(state)
- def __repr__(self):
+ def __repr__(self) -> str:
return "%s(%r, %r, type_=%r)" % (
self.__class__.__name__,
self.key,
@@ -2164,7 +2190,7 @@ class TypeClause(DQLDMLClauseElement):
("type", InternalTraversal.dp_type)
]
- def __init__(self, type_):
+ def __init__(self, type_: Any) -> None:
self.type = type_
@@ -2223,7 +2249,9 @@ class TextClause(
def _hide_froms(self) -> Iterable[FromClause]:
return ()
- def __and__(self, other):
+ def __and__(
+ self, other: _ColumnExpressionArgument[bool]
+ ) -> ColumnElement[bool]:
# support use in select.where(), query.filter()
return and_(self, other)
@@ -2239,13 +2267,13 @@ def _select_iterable(self) -> _SelectIterable:
_allow_label_resolve = False
@property
- def _is_star(self):
+ def _is_star(self) -> bool:
return self.text == "*"
- def __init__(self, text: str):
+ def __init__(self, text: str) -> None:
self._bindparams: Dict[str, BindParameter[Any]] = {}
- def repl(m):
+ def repl(m: Match[Any]) -> str:
self._bindparams[m.group(1)] = BindParameter(m.group(1))
return ":%s" % m.group(1)
@@ -2539,7 +2567,9 @@ def comparator(self):
# be using this method.
return self.type.comparator_factory(self) # type: ignore
- def self_group(self, against=None):
+ def self_group(
+ self, against: Optional[OperatorType] = None
+ ) -> Union[Grouping[TextClause], TextClause]:
if against is operators.in_op:
return Grouping(self)
else:
@@ -2560,7 +2590,7 @@ class Null(SingletonConstant, roles.ConstExprRole[None], ColumnElement[None]):
_singleton: Null
@util.memoized_property
- def type(self):
+ def type(self) -> NullType:
return type_api.NULLTYPE
@classmethod
@@ -2588,7 +2618,7 @@ class False_(
_singleton: False_
@util.memoized_property
- def type(self):
+ def type(self) -> Boolean:
return type_api.BOOLEANTYPE
def _negate(self) -> True_:
@@ -2616,7 +2646,7 @@ class True_(SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]):
_singleton: True_
@util.memoized_property
- def type(self):
+ def type(self) -> Boolean:
return type_api.BOOLEANTYPE
def _negate(self) -> False_:
@@ -2672,7 +2702,7 @@ def __init__(
group: bool = True,
group_contents: bool = True,
_literal_as_text_role: Type[roles.SQLRole] = roles.WhereHavingRole,
- ):
+ ) -> None:
self.operator = operator
self.group = group
self.group_contents = group_contents
@@ -2722,7 +2752,7 @@ def _select_iterable(self) -> _SelectIterable:
[elem._select_iterable for elem in self.clauses]
)
- def append(self, clause):
+ def append(self, clause: Any) -> None:
if self.group_contents:
self.clauses.append(
coercions.expect(self._text_converter_role, clause).self_group(
@@ -2738,7 +2768,9 @@ def append(self, clause):
def _from_objects(self) -> List[FromClause]:
return list(itertools.chain(*[c._from_objects for c in self.clauses]))
- def self_group(self, against=None):
+ def self_group(
+ self, against: Optional[OperatorType] = None
+ ) -> ClauseElement:
if self.group and operators.is_precedent(self.operator, against):
return Grouping(self)
else:
@@ -2758,10 +2790,12 @@ class OperatorExpression(ColumnElement[_T]):
group: bool = True
@property
- def is_comparison(self):
+ def is_comparison(self) -> bool:
return operators.is_comparison(self.operator)
- def self_group(self, against=None):
+ def self_group(
+ self, against: Optional[OperatorType] = None
+ ) -> ColumnElement[Any]:
if (
self.group
and operators.is_precedent(self.operator, against)
@@ -2923,7 +2957,7 @@ class BooleanClauseList(ExpressionClauseList[bool]):
__visit_name__ = "expression_clauselist"
inherit_cache = True
- def __init__(self, *arg, **kw):
+ def __init__(self, *arg: Any, **kw: Any):
raise NotImplementedError(
"BooleanClauseList has a private constructor"
)
@@ -3116,7 +3150,9 @@ def or_(
def _select_iterable(self) -> _SelectIterable:
return (self,)
- def self_group(self, against=None):
+ def self_group(
+ self, against: Optional[OperatorType] = None
+ ) -> ColumnElement[Any]:
if not self.clauses:
return self
else:
@@ -3173,7 +3209,13 @@ def __init__(
def _select_iterable(self) -> _SelectIterable:
return (self,)
- def _bind_param(self, operator, obj, type_=None, expanding=False):
+ def _bind_param(
+ self,
+ operator: OperatorType,
+ obj: List[typing_Tuple[int, str]],
+ type_: Optional[_TypeEngineArgument[_T]] = None,
+ expanding: bool = False,
+ ) -> BindParameter[_T]:
if expanding:
return BindParameter(
None,
@@ -3199,7 +3241,7 @@ def _bind_param(self, operator, obj, type_=None, expanding=False):
]
)
- def self_group(self, against=None):
+ def self_group(self, against: Optional[OperatorType] = None) -> Tuple:
# Tuple is parenthesized by definition.
return self
@@ -3352,7 +3394,11 @@ def _from_objects(self) -> List[FromClause]:
return self.clause._from_objects
@property
- def wrapped_column_expression(self):
+ def wrapped_column_expression(
+ self,
+ ) -> Union[
+ BindParameter[Any], ColumnClause[Any], ColumnElement[Any], Column[Any]
+ ]:
return self.clause
@@ -3402,7 +3448,7 @@ def _from_objects(self) -> List[FromClause]:
return self.clause._from_objects
@HasMemoized.memoized_attribute
- def typed_expression(self):
+ def typed_expression(self) -> Any:
if isinstance(self.clause, BindParameter):
bp = self.clause._clone()
bp.type = self.type
@@ -3411,10 +3457,12 @@ def typed_expression(self):
return self.clause
@property
- def wrapped_column_expression(self):
+ def wrapped_column_expression(self) -> ColumnElement[Any]:
return self.clause
- def self_group(self, against=None):
+ def self_group(
+ self, against: Optional[OperatorType] = None
+ ) -> TypeCoerce[_T]:
grouped = self.clause.self_group(against=against)
if grouped is not self.clause:
return TypeCoerce(grouped, self.type)
@@ -3518,7 +3566,7 @@ def __init__(
element: ColumnElement[Any],
operator: Optional[OperatorType] = None,
modifier: Optional[OperatorType] = None,
- type_: Optional[_TypeEngineArgument[_T]] = None,
+ type_: Optional[Union[_TypeEngineArgument[_T], Boolean]] = None,
wraps_column_expression: bool = False,
):
self.operator = operator
@@ -3618,7 +3666,7 @@ def _order_by_label_element(self) -> Optional[Label[Any]]:
def _from_objects(self) -> List[FromClause]:
return self.element._from_objects
- def _negate(self):
+ def _negate(self) -> ClauseElement:
if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
return UnaryExpression(
self.self_group(against=operators.inv),
@@ -3629,7 +3677,9 @@ def _negate(self):
else:
return ClauseElement._negate(self)
- def self_group(self, against=None):
+ def self_group(
+ self, against: Optional[OperatorType] = None
+ ) -> ColumnElement[Any]:
if self.operator and operators.is_precedent(self.operator, against):
return Grouping(self)
else:
@@ -3683,7 +3733,9 @@ def _create_all(
# operate and reverse_operate are hardwired to
# dispatch onto the type comparator directly, so that we can
# ensure "reversed" behavior.
- def operate(self, op, *other, **kwargs):
+ def operate(
+ self, op: OperatorType, *other: Any, **kwargs: Any
+ ) -> ColumnElement[_T]:
if not operators.is_comparison(op):
raise exc.ArgumentError(
"Only comparison operators may be used with ANY/ALL"
@@ -3691,7 +3743,9 @@ def operate(self, op, *other, **kwargs):
kwargs["reverse"] = kwargs["_any_all_expr"] = True
return self.comparator.operate(operators.mirror(op), *other, **kwargs)
- def reverse_operate(self, op, other, **kwargs):
+ def reverse_operate(
+ self, op: OperatorType, other: Any, **kwargs: Any
+ ) -> NoReturn:
# comparison operators should never call reverse_operate
assert not operators.is_comparison(op)
raise exc.ArgumentError(
@@ -3702,7 +3756,9 @@ def reverse_operate(self, op, other, **kwargs):
class AsBoolean(WrapsColumnExpression[bool], UnaryExpression[bool]):
inherit_cache = True
- def __init__(self, element, operator, negate):
+ def __init__(
+ self, element: Any, operator: OperatorType, negate: OperatorType
+ ) -> None:
self.element = element
self.type = type_api.BOOLEANTYPE
self.operator = operator
@@ -3712,13 +3768,13 @@ def __init__(self, element, operator, negate):
self._is_implicitly_boolean = element._is_implicitly_boolean
@property
- def wrapped_column_expression(self):
+ def wrapped_column_expression(self) -> ColumnClause[Any]:
return self.element
- def self_group(self, against=None):
+ def self_group(self, against: Optional[OperatorType] = None) -> AsBoolean:
return self
- def _negate(self):
+ def _negate(self) -> Union[AsBoolean, False_, True_]:
if isinstance(self.element, (True_, False_)):
return self.element._negate()
else:
@@ -3816,7 +3872,7 @@ def _flattened_operator_clauses(
) -> typing_Tuple[ColumnElement[Any], ...]:
return (self.left, self.right)
- def __bool__(self):
+ def __bool__(self) -> bool:
"""Implement Python-side "bool" for BinaryExpression as a
simple "identity" check for the left and right attributes,
if the operator is "eq" or "ne". Otherwise the expression
@@ -3865,7 +3921,7 @@ def __invert__(
def _from_objects(self) -> List[FromClause]:
return self.left._from_objects + self.right._from_objects
- def _negate(self):
+ def _negate(self) -> Any:
if self.negate is not None:
return BinaryExpression(
self.left,
@@ -3895,7 +3951,13 @@ class Slice(ColumnElement[Any]):
("step", InternalTraversal.dp_clauseelement),
]
- def __init__(self, start, stop, step, _name=None):
+ def __init__(
+ self,
+ start: int,
+ stop: int,
+ step: Optional[int],
+ _name: Union[None, quoted_name, str] = None,
+ ) -> None:
self.start = coercions.expect(
roles.ExpressionElementRole,
start,
@@ -3916,7 +3978,7 @@ def __init__(self, start, stop, step, _name=None):
)
self.type = type_api.NULLTYPE
- def self_group(self, against=None):
+ def self_group(self, against: Optional[OperatorType] = None) -> Slice:
assert against is operator.getitem
return self
@@ -3935,10 +3997,12 @@ class GroupedElement(DQLDMLClauseElement):
element: ClauseElement
- def self_group(self, against=None):
+ def self_group(
+ self, against: Optional[OperatorType] = None
+ ) -> ClauseElement:
return self
- def _ungroup(self):
+ def _ungroup(self) -> ClauseElement:
return self.element._ungroup()
@@ -3964,11 +4028,13 @@ def __init__(
# nulltype assignment issue
self.type = getattr(element, "type", type_api.NULLTYPE) # type: ignore
- def _with_binary_element_type(self, type_):
+ def _with_binary_element_type(
+ self, type_: Union[String, TupleType]
+ ) -> Grouping[_T]:
return self.__class__(self.element._with_binary_element_type(type_))
@util.memoized_property
- def _is_implicitly_boolean(self):
+ def _is_implicitly_boolean(self) -> bool:
return self.element._is_implicitly_boolean
@util.non_memoized_property
@@ -3988,13 +4054,13 @@ def _proxies(self) -> List[ColumnElement[Any]]:
def _from_objects(self) -> List[FromClause]:
return self.element._from_objects
- def __getattr__(self, attr):
+ def __getattr__(self, attr: str) -> Any:
return getattr(self.element, attr)
- def __getstate__(self):
+ def __getstate__(self) -> dict[str, Any]:
return {"element": self.element, "type": self.type}
- def __setstate__(self, state):
+ def __setstate__(self, state: dict[str, Any]) -> None:
self.element = state["element"]
self.type = state["type"]
@@ -4080,7 +4146,18 @@ def __init__(
else:
self.rows = self.range_ = None
- def __reduce__(self):
+ def __reduce__(
+ self,
+ ) -> typing_Tuple[
+ Type[Over[_T]],
+ typing_Tuple[
+ ColumnElement[_T],
+ Optional[ClauseList],
+ Optional[ClauseList],
+ Optional[typing_Tuple[int, int]],
+ Optional[typing_Tuple[int, int]],
+ ],
+ ]:
return self.__class__, (
self.element,
self.partition_by,
@@ -4127,7 +4204,7 @@ def _interpret_range(
return lower, upper
@util.memoized_property
- def type(self):
+ def type(self) -> TypeEngine[_T]:
return self.element.type
@util.ro_non_memoized_property
@@ -4180,12 +4257,23 @@ def __init__(
*util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
)
- def __reduce__(self):
+ def __reduce__(
+ self,
+ ) -> typing_Tuple[
+ Type[WithinGroup[_T]],
+ typing_Tuple[Union[FunctionElement[_T], ColumnElement[Any]], ...],
+ ]:
return self.__class__, (self.element,) + (
tuple(self.order_by) if self.order_by is not None else ()
)
- def over(self, partition_by=None, order_by=None, range_=None, rows=None):
+ def over(
+ self,
+ partition_by: Optional[ColumnClause[_T]] = None,
+ order_by: Optional[ColumnClause[_T]] = None,
+ range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
+ rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
+ ) -> Over[_T]:
"""Produce an OVER clause against this :class:`.WithinGroup`
construct.
@@ -4202,7 +4290,7 @@ def over(self, partition_by=None, order_by=None, range_=None, rows=None):
)
@util.memoized_property
- def type(self):
+ def type(self) -> TypeEngine[_T]:
wgt = self.element.within_group_type(self)
if wgt is not None:
return wgt
@@ -4259,7 +4347,9 @@ def __init__(
self.func = func
self.filter(*criterion)
- def filter(self, *criterion):
+ def filter(
+ self, *criterion: _ColumnExpressionArgument[bool]
+ ) -> FunctionFilter[_T]:
"""Produce an additional FILTER against the function.
This method adds additional criteria to the initial criteria
@@ -4323,14 +4413,16 @@ def over(
rows=rows,
)
- def self_group(self, against=None):
+ def self_group(
+ self, against: Optional[OperatorType] = None
+ ) -> ColumnElement[_T]:
if operators.is_precedent(operators.filter_op, against):
return Grouping(self)
else:
return self
@util.memoized_property
- def type(self):
+ def type(self) -> TypeEngine[_T]:
return self.func.type
@util.ro_non_memoized_property
@@ -4352,7 +4444,9 @@ class NamedColumn(KeyedColumnElement[_T]):
name: str
key: str
- def _compare_name_for_result(self, other):
+ def _compare_name_for_result(
+ self, other: ColumnElement[Any]
+ ) -> Union[bool, FrozenSet[ColumnElement[Any]]]:
return (hasattr(other, "name") and self.name == other.name) or (
hasattr(other, "_label") and self._label == other._label
)
@@ -4362,7 +4456,7 @@ def description(self) -> str:
return self.name
@HasMemoized.memoized_attribute
- def _tq_key_label(self):
+ def _tq_key_label(self) -> Optional[str]:
"""table qualified label based on column key.
for table-bound columns this is _;
@@ -4387,11 +4481,11 @@ def _tq_label(self) -> Optional[str]:
return self._gen_tq_label(self.name)
@HasMemoized.memoized_attribute
- def _render_label_in_columns_clause(self):
+ def _render_label_in_columns_clause(self) -> bool:
return True
@HasMemoized.memoized_attribute
- def _non_anon_label(self):
+ def _non_anon_label(self) -> Optional[str]:
return self.name
def _gen_tq_label(
@@ -4512,14 +4606,24 @@ def __init__(
self._proxies = [element]
- def __reduce__(self):
+ def __reduce__(
+ self,
+ ) -> typing_Tuple[
+ Type[Any], typing_Tuple[str, ColumnElement[_T], TypeEngine[_T]]
+ ]:
return self.__class__, (self.name, self._element, self.type)
@HasMemoized.memoized_attribute
- def _render_label_in_columns_clause(self):
+ def _render_label_in_columns_clause(self) -> bool:
return True
- def _bind_param(self, operator, obj, type_=None, expanding=False):
+ def _bind_param(
+ self,
+ operator: OperatorType,
+ obj: Union[List[str], int, str],
+ type_: Optional[_TypeEngineArgument[_T]] = None,
+ expanding: bool = False,
+ ) -> BindParameter[_T]:
return BindParameter(
None,
obj,
@@ -4531,28 +4635,30 @@ def _bind_param(self, operator, obj, type_=None, expanding=False):
)
@util.memoized_property
- def _is_implicitly_boolean(self):
+ def _is_implicitly_boolean(self) -> bool:
return self.element._is_implicitly_boolean
@HasMemoized.memoized_attribute
- def _allow_label_resolve(self):
+ def _allow_label_resolve(self) -> bool:
return self.element._allow_label_resolve
@property
- def _order_by_label_element(self):
+ def _order_by_label_element(self) -> Label[_T]:
return self
@HasMemoized.memoized_attribute
def element(self) -> ColumnElement[_T]:
return self._element.self_group(against=operators.as_)
- def self_group(self, against=None):
+ def self_group(self, against: Optional[OperatorType] = None) -> Label[_T]:
return self._apply_to_inner(self._element.self_group, against=against)
- def _negate(self):
+ def _negate(self) -> ColumnElement[Any]:
return self._apply_to_inner(self._element._negate)
- def _apply_to_inner(self, fn, *arg, **kw):
+ def _apply_to_inner(
+ self, fn: Callable[..., Any], *arg: Any, **kw: Any
+ ) -> Label[_T]:
sub_element = fn(*arg, **kw)
if sub_element is not self._element:
return Label(self.name, sub_element, type_=self.type)
@@ -4560,11 +4666,11 @@ def _apply_to_inner(self, fn, *arg, **kw):
return self
@property
- def primary_key(self):
+ def primary_key(self) -> bool:
return self.element.primary_key
@property
- def foreign_keys(self):
+ def foreign_keys(self) -> AbstractSet[ForeignKey]:
return self.element.foreign_keys
def _copy_internals(
@@ -4691,7 +4797,7 @@ class is usable by itself in those cases where behavioral requirements
_is_multiparam_column = False
@property
- def _is_star(self):
+ def _is_star(self) -> bool:
return self.is_literal and self.name == "*"
def __init__(
@@ -4710,20 +4816,22 @@ def __init__(
self.is_literal = is_literal
- def get_children(self, *, column_tables=False, **kw):
+ def get_children(
+ self, *, column_tables: bool = False, **kw: Any
+ ) -> List[Any]:
# override base get_children() to not return the Table
# or selectable that is parent to this column. Traversals
# expect the columns of tables and subqueries to be leaf nodes.
return []
@property
- def entity_namespace(self):
+ def entity_namespace(self) -> Union[_EntityNamespace, NoReturn]:
if self.table is not None:
return self.table.entity_namespace
else:
return super().entity_namespace
- def _clone(self, detect_subquery_cols=False, **kw):
+ def _clone(self, detect_subquery_cols: bool = False, **kw: Any) -> Any:
if (
detect_subquery_cols
and self.table is not None
@@ -4745,14 +4853,16 @@ def _from_objects(self) -> List[FromClause]:
return []
@HasMemoized.memoized_attribute
- def _render_label_in_columns_clause(self):
+ def _render_label_in_columns_clause(self) -> bool:
return self.table is not None
@property
- def _ddl_label(self):
+ def _ddl_label(self) -> Optional[str]:
return self._gen_tq_label(self.name, dedupe_on_key=False)
- def _compare_name_for_result(self, other):
+ def _compare_name_for_result(
+ self, other: ColumnElement[Any]
+ ) -> Union[bool, FrozenSet[ColumnElement[Any]]]:
if (
self.is_literal
or self.table is None
@@ -4922,14 +5032,14 @@ def _create_collation_expression(
type_=expr.type,
)
- def __init__(self, collation):
+ def __init__(self, collation: str) -> None:
self.collation = collation
class _IdentifiedClause(Executable, ClauseElement):
__visit_name__ = "identified"
- def __init__(self, ident):
+ def __init__(self, ident: Any) -> None:
self.ident = ident
@@ -5033,16 +5143,18 @@ def __new__(cls, value: str, quote: Optional[bool]) -> quoted_name:
self.quote = quote
return self
- def __reduce__(self):
+ def __reduce__(
+ self,
+ ) -> typing_Tuple[Type[quoted_name], typing_Tuple[str, Optional[bool]]]:
return quoted_name, (str(self), self.quote)
- def _memoized_method_lower(self):
+ def _memoized_method_lower(self) -> str:
if self.quote:
return self
else:
return str(self).lower()
- def _memoized_method_upper(self):
+ def _memoized_method_upper(self) -> Union[quoted_name, str]:
if self.quote:
return self
else:
@@ -5057,7 +5169,7 @@ def _find_columns(clause: ClauseElement) -> Set[ColumnClause[Any]]:
return cols
-def _type_from_args(args):
+def _type_from_args(args: Iterable[Any]) -> Union[Any, NullType]:
for a in args:
if not a.type._isnull:
return a.type
@@ -5065,7 +5177,11 @@ def _type_from_args(args):
return type_api.NULLTYPE
-def _corresponding_column_or_error(fromclause, column, require_embedded=False):
+def _corresponding_column_or_error(
+ fromclause: FromClause,
+ column: KeyedColumnElement[_T],
+ require_embedded: bool = False,
+) -> Optional[KeyedColumnElement[_T]]:
c = fromclause.corresponding_column(
column, require_embedded=require_embedded
)
@@ -5081,7 +5197,7 @@ def _corresponding_column_or_error(fromclause, column, require_embedded=False):
class AnnotatedColumnElement(Annotated):
_Annotated__element: ColumnElement[Any]
- def __init__(self, element, values):
+ def __init__(self, element: Any, values: Mapping[str, Any]) -> None:
Annotated.__init__(self, element, values)
for attr in (
"comparator",
@@ -5095,23 +5211,25 @@ def __init__(self, element, values):
if self.__dict__.get(attr, False) is None:
self.__dict__.pop(attr)
- def _with_annotations(self, values):
+ def _with_annotations(
+ self, values: Mapping[str, Any]
+ ) -> SupportsAnnotations:
clone = super()._with_annotations(values)
clone.__dict__.pop("comparator", None)
return clone
@util.memoized_property
- def name(self):
+ def name(self) -> Any:
"""pull 'name' from parent, if not present"""
return self._Annotated__element.name
@util.memoized_property
- def table(self):
+ def table(self) -> Any:
"""pull 'table' from parent, if not present"""
return self._Annotated__element.table
@util.memoized_property
- def key(self):
+ def key(self) -> Union[str, None]:
"""pull 'key' from parent, if not present"""
return self._Annotated__element.key
@@ -5227,7 +5345,7 @@ def safe_construct(
return _anonymous_label(label)
- def __add__(self, other):
+ def __add__(self, other: Any) -> _anonymous_label:
if "%" in other and not isinstance(other, _anonymous_label):
other = str(other).replace("%", "%%")
else:
@@ -5240,7 +5358,7 @@ def __add__(self, other):
)
)
- def __radd__(self, other):
+ def __radd__(self, other: str) -> _anonymous_label:
if "%" in other and not isinstance(other, _anonymous_label):
other = str(other).replace("%", "%%")
else:
@@ -5253,7 +5371,7 @@ def __radd__(self, other):
)
)
- def apply_map(self, map_):
+ def apply_map(self, map_: Mapping[str, Any]) -> str:
if self.quote is not None:
# preserve quoting only if necessary
return quoted_name(self % map_, self.quote)