Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# sql/elements.py
2# Copyright (C) 2005-2025 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7# mypy: allow-untyped-defs, allow-untyped-calls
9"""Core SQL expression elements, including :class:`_expression.ClauseElement`,
10:class:`_expression.ColumnElement`, and derived classes.
12"""
14from __future__ import annotations
16from decimal import Decimal
17from enum import Enum
18import itertools
19import operator
20import re
21import typing
22from typing import AbstractSet
23from typing import Any
24from typing import Callable
25from typing import cast
26from typing import Dict
27from typing import FrozenSet
28from typing import Generic
29from typing import Iterable
30from typing import Iterator
31from typing import List
32from typing import Literal
33from typing import Mapping
34from typing import Optional
35from typing import overload
36from typing import ParamSpec
37from typing import Sequence
38from typing import Set
39from typing import Tuple as typing_Tuple
40from typing import Type
41from typing import TYPE_CHECKING
42from typing import TypeVar
43from typing import Union
45from . import coercions
46from . import operators
47from . import roles
48from . import traversals
49from . import type_api
50from ._typing import has_schema_attr
51from ._typing import is_named_from_clause
52from ._typing import is_quoted_name
53from ._typing import is_tuple_type
54from .annotation import Annotated
55from .annotation import SupportsWrappingAnnotations
56from .base import _clone
57from .base import _expand_cloned
58from .base import _generative
59from .base import _NoArg
60from .base import Executable
61from .base import Generative
62from .base import HasMemoized
63from .base import Immutable
64from .base import NO_ARG
65from .base import SingletonConstant
66from .cache_key import MemoizedHasCacheKey
67from .cache_key import NO_CACHE
68from .coercions import _document_text_coercion # noqa
69from .operators import ColumnOperators
70from .operators import OperatorClass
71from .traversals import HasCopyInternals
72from .visitors import cloned_traverse
73from .visitors import ExternallyTraversible
74from .visitors import InternalTraversal
75from .visitors import traverse
76from .visitors import Visitable
77from .. import exc
78from .. import inspection
79from .. import util
80from ..util import HasMemoized_ro_memoized_attribute
81from ..util import TypingOnly
82from ..util.typing import Self
83from ..util.typing import TupleAny
84from ..util.typing import Unpack
87if typing.TYPE_CHECKING:
88 from ._typing import _ByArgument
89 from ._typing import _ColumnExpressionArgument
90 from ._typing import _ColumnExpressionOrStrLabelArgument
91 from ._typing import _DMLOnlyColumnArgument
92 from ._typing import _HasDialect
93 from ._typing import _InfoType
94 from ._typing import _PropagateAttrsType
95 from ._typing import _TypeEngineArgument
96 from .base import _EntityNamespace
97 from .base import ColumnSet
98 from .cache_key import _CacheKeyTraversalType
99 from .cache_key import CacheKey
100 from .compiler import Compiled
101 from .compiler import SQLCompiler
102 from .functions import FunctionElement
103 from .operators import OperatorType
104 from .schema import Column
105 from .schema import DefaultGenerator
106 from .schema import FetchedValue
107 from .schema import ForeignKey
108 from .selectable import _SelectIterable
109 from .selectable import FromClause
110 from .selectable import NamedFromClause
111 from .selectable import TextualSelect
112 from .sqltypes import TupleType
113 from .type_api import TypeEngine
114 from .visitors import _CloneCallableType
115 from .visitors import _TraverseInternalsType
116 from .visitors import anon_map
117 from ..engine import Connection
118 from ..engine import Dialect
119 from ..engine.interfaces import _CoreMultiExecuteParams
120 from ..engine.interfaces import CacheStats
121 from ..engine.interfaces import CompiledCacheType
122 from ..engine.interfaces import CoreExecuteOptionsParameter
123 from ..engine.interfaces import SchemaTranslateMapType
124 from ..engine.result import Result
127_NUMERIC = Union[float, Decimal]
128_NUMBER = Union[float, int, Decimal]
130_T = TypeVar("_T", bound="Any")
131_T_co = TypeVar("_T_co", bound=Any, covariant=True)
132_OPT = TypeVar("_OPT", bound="Any")
133_NT = TypeVar("_NT", bound="_NUMERIC")
135_NMT = TypeVar("_NMT", bound="_NUMBER")
138@overload
139def literal(
140 value: Any,
141 type_: _TypeEngineArgument[_T],
142 literal_execute: bool = False,
143) -> BindParameter[_T]: ...
146@overload
147def literal(
148 value: _T,
149 type_: None = None,
150 literal_execute: bool = False,
151) -> BindParameter[_T]: ...
154@overload
155def literal(
156 value: Any,
157 type_: Optional[_TypeEngineArgument[Any]] = None,
158 literal_execute: bool = False,
159) -> BindParameter[Any]: ...
162def literal(
163 value: Any,
164 type_: Optional[_TypeEngineArgument[Any]] = None,
165 literal_execute: bool = False,
166) -> BindParameter[Any]:
167 r"""Return a literal clause, bound to a bind parameter.
169 Literal clauses are created automatically when non-
170 :class:`_expression.ClauseElement` objects (such as strings, ints, dates,
171 etc.) are
172 used in a comparison operation with a :class:`_expression.ColumnElement`
173 subclass,
174 such as a :class:`~sqlalchemy.schema.Column` object. Use this function
175 to force the generation of a literal clause, which will be created as a
176 :class:`BindParameter` with a bound value.
178 :param value: the value to be bound. Can be any Python object supported by
179 the underlying DB-API, or is translatable via the given type argument.
181 :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine` which will
182 provide bind-parameter translation for this literal.
184 :param literal_execute: optional bool, when True, the SQL engine will
185 attempt to render the bound value directly in the SQL statement at
186 execution time rather than providing as a parameter value.
188 .. versionadded:: 2.0
190 """
191 return coercions.expect(
192 roles.LiteralValueRole,
193 value,
194 type_=type_,
195 literal_execute=literal_execute,
196 )
199def literal_column(
200 text: str, type_: Optional[_TypeEngineArgument[_T]] = None
201) -> ColumnClause[_T]:
202 r"""Produce a :class:`.ColumnClause` object that has the
203 :paramref:`_expression.column.is_literal` flag set to True.
205 :func:`_expression.literal_column` is similar to
206 :func:`_expression.column`, except that
207 it is more often used as a "standalone" column expression that renders
208 exactly as stated; while :func:`_expression.column`
209 stores a string name that
210 will be assumed to be part of a table and may be quoted as such,
211 :func:`_expression.literal_column` can be that,
212 or any other arbitrary column-oriented
213 expression.
215 :param text: the text of the expression; can be any SQL expression.
216 Quoting rules will not be applied. To specify a column-name expression
217 which should be subject to quoting rules, use the :func:`column`
218 function.
220 :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine`
221 object which will
222 provide result-set translation and additional expression semantics for
223 this column. If left as ``None`` the type will be :class:`.NullType`.
225 .. seealso::
227 :func:`_expression.column`
229 :func:`_expression.text`
231 :ref:`tutorial_select_arbitrary_text`
233 """
234 return ColumnClause(text, type_=type_, is_literal=True)
237class CompilerElement(Visitable):
238 """base class for SQL elements that can be compiled to produce a
239 SQL string.
241 .. versionadded:: 2.0
243 """
245 __slots__ = ()
246 __visit_name__ = "compiler_element"
248 supports_execution = False
250 stringify_dialect = "default"
252 @util.preload_module("sqlalchemy.engine.default")
253 @util.preload_module("sqlalchemy.engine.url")
254 def compile(
255 self,
256 bind: Optional[_HasDialect] = None,
257 dialect: Optional[Dialect] = None,
258 **kw: Any,
259 ) -> Compiled:
260 """Compile this SQL expression.
262 The return value is a :class:`~.Compiled` object.
263 Calling ``str()`` or ``unicode()`` on the returned value will yield a
264 string representation of the result. The
265 :class:`~.Compiled` object also can return a
266 dictionary of bind parameter names and values
267 using the ``params`` accessor.
269 :param bind: An :class:`.Connection` or :class:`.Engine` which
270 can provide a :class:`.Dialect` in order to generate a
271 :class:`.Compiled` object. If the ``bind`` and
272 ``dialect`` parameters are both omitted, a default SQL compiler
273 is used.
275 :param column_keys: Used for INSERT and UPDATE statements, a list of
276 column names which should be present in the VALUES clause of the
277 compiled statement. If ``None``, all columns from the target table
278 object are rendered.
280 :param dialect: A :class:`.Dialect` instance which can generate
281 a :class:`.Compiled` object. This argument takes precedence over
282 the ``bind`` argument.
284 :param compile_kwargs: optional dictionary of additional parameters
285 that will be passed through to the compiler within all "visit"
286 methods. This allows any custom flag to be passed through to
287 a custom compilation construct, for example. It is also used
288 for the case of passing the ``literal_binds`` flag through::
290 from sqlalchemy.sql import table, column, select
292 t = table("t", column("x"))
294 s = select(t).where(t.c.x == 5)
296 print(s.compile(compile_kwargs={"literal_binds": True}))
298 .. seealso::
300 :ref:`faq_sql_expression_string`
302 """
304 if dialect is None:
305 if bind:
306 dialect = bind.dialect
307 elif self.stringify_dialect == "default":
308 dialect = self._default_dialect()
309 else:
310 url = util.preloaded.engine_url
311 dialect = url.URL.create(
312 self.stringify_dialect
313 ).get_dialect()()
315 return self._compiler(dialect, **kw)
317 def _default_dialect(self):
318 default = util.preloaded.engine_default
319 return default.StrCompileDialect()
321 def _compiler(self, dialect: Dialect, **kw: Any) -> Compiled:
322 """Return a compiler appropriate for this ClauseElement, given a
323 Dialect."""
325 if TYPE_CHECKING:
326 assert isinstance(self, ClauseElement)
327 return dialect.statement_compiler(dialect, self, **kw)
329 def __str__(self) -> str:
330 return str(self.compile())
333@inspection._self_inspects
334class ClauseElement(
335 SupportsWrappingAnnotations,
336 MemoizedHasCacheKey,
337 HasCopyInternals,
338 ExternallyTraversible,
339 CompilerElement,
340):
341 """Base class for elements of a programmatically constructed SQL
342 expression.
344 """
346 __visit_name__ = "clause"
348 if TYPE_CHECKING:
350 @util.memoized_property
351 def _propagate_attrs(self) -> _PropagateAttrsType:
352 """like annotations, however these propagate outwards liberally
353 as SQL constructs are built, and are set up at construction time.
355 """
356 ...
358 else:
359 _propagate_attrs = util.EMPTY_DICT
361 @util.ro_memoized_property
362 def description(self) -> Optional[str]:
363 return None
365 _is_clone_of: Optional[Self] = None
367 is_clause_element = True
368 is_selectable = False
369 is_dml = False
370 _is_column_element = False
371 _is_keyed_column_element = False
372 _is_table = False
373 _gen_static_annotations_cache_key = False
374 _is_textual = False
375 _is_from_clause = False
376 _is_returns_rows = False
377 _is_text_clause = False
378 _is_from_container = False
379 _is_select_container = False
380 _is_select_base = False
381 _is_select_statement = False
382 _is_bind_parameter = False
383 _is_clause_list = False
384 _is_lambda_element = False
385 _is_singleton_constant = False
386 _is_immutable = False
387 _is_star = False
389 @property
390 def _order_by_label_element(self) -> Optional[Label[Any]]:
391 return None
393 _cache_key_traversal: _CacheKeyTraversalType = None
395 negation_clause: ColumnElement[bool]
397 if typing.TYPE_CHECKING:
399 def get_children(
400 self, *, omit_attrs: typing_Tuple[str, ...] = ..., **kw: Any
401 ) -> Iterable[ClauseElement]: ...
403 @util.ro_non_memoized_property
404 def _from_objects(self) -> List[FromClause]:
405 return []
407 def _set_propagate_attrs(self, values: Mapping[str, Any]) -> Self:
408 # usually, self._propagate_attrs is empty here. one case where it's
409 # not is a subquery against ORM select, that is then pulled as a
410 # property of an aliased class. should all be good
412 # assert not self._propagate_attrs
414 self._propagate_attrs = util.immutabledict(values)
415 return self
417 def _default_compiler(self) -> SQLCompiler:
418 dialect = self._default_dialect()
419 return dialect.statement_compiler(dialect, self) # type: ignore
421 def _clone(self, **kw: Any) -> Self:
422 """Create a shallow copy of this ClauseElement.
424 This method may be used by a generative API. Its also used as
425 part of the "deep" copy afforded by a traversal that combines
426 the _copy_internals() method.
428 """
430 skip = self._memoized_keys
431 c = self.__class__.__new__(self.__class__)
433 if skip:
434 # ensure this iteration remains atomic
435 c.__dict__ = {
436 k: v for k, v in self.__dict__.copy().items() if k not in skip
437 }
438 else:
439 c.__dict__ = self.__dict__.copy()
441 # this is a marker that helps to "equate" clauses to each other
442 # when a Select returns its list of FROM clauses. the cloning
443 # process leaves around a lot of remnants of the previous clause
444 # typically in the form of column expressions still attached to the
445 # old table.
446 cc = self._is_clone_of
447 c._is_clone_of = cc if cc is not None else self
448 return c
450 def _negate_in_binary(self, negated_op, original_op):
451 """a hook to allow the right side of a binary expression to respond
452 to a negation of the binary expression.
454 Used for the special case of expanding bind parameter with IN.
456 """
457 return self
459 def _with_binary_element_type(self, type_):
460 """in the context of binary expression, convert the type of this
461 object to the one given.
463 applies only to :class:`_expression.ColumnElement` classes.
465 """
466 return self
468 @property
469 def _constructor(self): # type: ignore[override]
470 """return the 'constructor' for this ClauseElement.
472 This is for the purposes for creating a new object of
473 this type. Usually, its just the element's __class__.
474 However, the "Annotated" version of the object overrides
475 to return the class of its proxied element.
477 """
478 return self.__class__
480 @HasMemoized.memoized_attribute
481 def _cloned_set(self):
482 """Return the set consisting all cloned ancestors of this
483 ClauseElement.
485 Includes this ClauseElement. This accessor tends to be used for
486 FromClause objects to identify 'equivalent' FROM clauses, regardless
487 of transformative operations.
489 """
490 s = util.column_set()
491 f: Optional[ClauseElement] = self
493 # note this creates a cycle, asserted in test_memusage. however,
494 # turning this into a plain @property adds tends of thousands of method
495 # calls to Core / ORM performance tests, so the small overhead
496 # introduced by the relatively small amount of short term cycles
497 # produced here is preferable
498 while f is not None:
499 s.add(f)
500 f = f._is_clone_of
501 return s
503 def _de_clone(self):
504 while self._is_clone_of is not None:
505 self = self._is_clone_of
506 return self
508 @util.ro_non_memoized_property
509 def entity_namespace(self) -> _EntityNamespace:
510 raise AttributeError(
511 "This SQL expression has no entity namespace "
512 "with which to filter from."
513 )
515 def __getstate__(self):
516 d = self.__dict__.copy()
517 d.pop("_is_clone_of", None)
518 d.pop("_generate_cache_key", None)
519 return d
521 def _execute_on_connection(
522 self,
523 connection: Connection,
524 distilled_params: _CoreMultiExecuteParams,
525 execution_options: CoreExecuteOptionsParameter,
526 ) -> Result[Unpack[TupleAny]]:
527 if self.supports_execution:
528 if TYPE_CHECKING:
529 assert isinstance(self, Executable)
530 return connection._execute_clauseelement(
531 self, distilled_params, execution_options
532 )
533 else:
534 raise exc.ObjectNotExecutableError(self)
536 def _execute_on_scalar(
537 self,
538 connection: Connection,
539 distilled_params: _CoreMultiExecuteParams,
540 execution_options: CoreExecuteOptionsParameter,
541 ) -> Any:
542 """an additional hook for subclasses to provide a different
543 implementation for connection.scalar() vs. connection.execute().
545 .. versionadded:: 2.0
547 """
548 return self._execute_on_connection(
549 connection, distilled_params, execution_options
550 ).scalar()
552 def _get_embedded_bindparams(self) -> Sequence[BindParameter[Any]]:
553 """Return the list of :class:`.BindParameter` objects embedded in the
554 object.
556 This accomplishes the same purpose as ``visitors.traverse()`` or
557 similar would provide, however by making use of the cache key
558 it takes advantage of memoization of the key to result in fewer
559 net method calls, assuming the statement is also going to be
560 executed.
562 """
564 key = self._generate_cache_key()
565 if key is None:
566 bindparams: List[BindParameter[Any]] = []
568 traverse(self, {}, {"bindparam": bindparams.append})
569 return bindparams
571 else:
572 return key.bindparams
574 def unique_params(
575 self,
576 __optionaldict: Optional[Dict[str, Any]] = None,
577 /,
578 **kwargs: Any,
579 ) -> Self:
580 """Return a copy with :func:`_expression.bindparam` elements
581 replaced.
583 Same functionality as :meth:`_expression.ClauseElement.params`,
584 except adds `unique=True`
585 to affected bind parameters so that multiple statements can be
586 used.
588 """
589 return self._replace_params(True, __optionaldict, kwargs)
591 def params(
592 self,
593 __optionaldict: Optional[Mapping[str, Any]] = None,
594 /,
595 **kwargs: Any,
596 ) -> Self:
597 """Return a copy with :func:`_expression.bindparam` elements
598 replaced.
600 Returns a copy of this ClauseElement with
601 :func:`_expression.bindparam`
602 elements replaced with values taken from the given dictionary::
604 >>> clause = column("x") + bindparam("foo")
605 >>> print(clause.compile().params)
606 {'foo':None}
607 >>> print(clause.params({"foo": 7}).compile().params)
608 {'foo':7}
610 """
611 return self._replace_params(False, __optionaldict, kwargs)
613 def _replace_params(
614 self,
615 unique: bool,
616 optionaldict: Optional[Mapping[str, Any]],
617 kwargs: Dict[str, Any],
618 ) -> Self:
619 if optionaldict:
620 kwargs.update(optionaldict)
622 def visit_bindparam(bind: BindParameter[Any]) -> None:
623 if bind.key in kwargs:
624 bind.value = kwargs[bind.key]
625 bind.required = False
626 if unique:
627 bind._convert_to_unique()
629 return cloned_traverse(
630 self,
631 {"maintain_key": True, "detect_subquery_cols": True},
632 {"bindparam": visit_bindparam},
633 )
635 def compare(self, other: ClauseElement, **kw: Any) -> bool:
636 r"""Compare this :class:`_expression.ClauseElement` to
637 the given :class:`_expression.ClauseElement`.
639 Subclasses should override the default behavior, which is a
640 straight identity comparison.
642 \**kw are arguments consumed by subclass ``compare()`` methods and
643 may be used to modify the criteria for comparison
644 (see :class:`_expression.ColumnElement`).
646 """
647 return traversals.compare(self, other, **kw)
649 def self_group(
650 self, against: Optional[OperatorType] = None
651 ) -> ClauseElement:
652 """Apply a 'grouping' to this :class:`_expression.ClauseElement`.
654 This method is overridden by subclasses to return a "grouping"
655 construct, i.e. parenthesis. In particular it's used by "binary"
656 expressions to provide a grouping around themselves when placed into a
657 larger expression, as well as by :func:`_expression.select`
658 constructs when placed into the FROM clause of another
659 :func:`_expression.select`. (Note that subqueries should be
660 normally created using the :meth:`_expression.Select.alias` method,
661 as many
662 platforms require nested SELECT statements to be named).
664 As expressions are composed together, the application of
665 :meth:`self_group` is automatic - end-user code should never
666 need to use this method directly. Note that SQLAlchemy's
667 clause constructs take operator precedence into account -
668 so parenthesis might not be needed, for example, in
669 an expression like ``x OR (y AND z)`` - AND takes precedence
670 over OR.
672 The base :meth:`self_group` method of
673 :class:`_expression.ClauseElement`
674 just returns self.
675 """
676 return self
678 def _ungroup(self) -> ClauseElement:
679 """Return this :class:`_expression.ClauseElement`
680 without any groupings.
681 """
683 return self
685 def _compile_w_cache(
686 self,
687 dialect: Dialect,
688 *,
689 compiled_cache: Optional[CompiledCacheType],
690 column_keys: List[str],
691 for_executemany: bool = False,
692 schema_translate_map: Optional[SchemaTranslateMapType] = None,
693 **kw: Any,
694 ) -> typing_Tuple[
695 Compiled, Optional[Sequence[BindParameter[Any]]], CacheStats
696 ]:
697 elem_cache_key: Optional[CacheKey]
699 if compiled_cache is not None and dialect._supports_statement_cache:
700 elem_cache_key = self._generate_cache_key()
701 else:
702 elem_cache_key = None
704 extracted_params: Optional[Sequence[BindParameter[Any]]]
705 if elem_cache_key is not None:
706 if TYPE_CHECKING:
707 assert compiled_cache is not None
709 cache_key, extracted_params = elem_cache_key
710 key = (
711 dialect,
712 cache_key,
713 tuple(column_keys),
714 bool(schema_translate_map),
715 for_executemany,
716 )
717 compiled_sql = compiled_cache.get(key)
719 if compiled_sql is None:
720 cache_hit = dialect.CACHE_MISS
721 compiled_sql = self._compiler(
722 dialect,
723 cache_key=elem_cache_key,
724 column_keys=column_keys,
725 for_executemany=for_executemany,
726 schema_translate_map=schema_translate_map,
727 **kw,
728 )
729 compiled_cache[key] = compiled_sql
730 else:
731 cache_hit = dialect.CACHE_HIT
732 else:
733 extracted_params = None
734 compiled_sql = self._compiler(
735 dialect,
736 cache_key=elem_cache_key,
737 column_keys=column_keys,
738 for_executemany=for_executemany,
739 schema_translate_map=schema_translate_map,
740 **kw,
741 )
743 if not dialect._supports_statement_cache:
744 cache_hit = dialect.NO_DIALECT_SUPPORT
745 elif compiled_cache is None:
746 cache_hit = dialect.CACHING_DISABLED
747 else:
748 cache_hit = dialect.NO_CACHE_KEY
750 return compiled_sql, extracted_params, cache_hit
752 def __invert__(self):
753 # undocumented element currently used by the ORM for
754 # relationship.contains()
755 if hasattr(self, "negation_clause"):
756 return self.negation_clause
757 else:
758 return self._negate()
760 def _negate(self) -> ClauseElement:
761 # TODO: this code is uncovered and in all likelihood is not included
762 # in any codepath. So this should raise NotImplementedError in 2.1
763 grouped = self.self_group(against=operators.inv)
764 assert isinstance(grouped, ColumnElement)
765 return UnaryExpression(grouped, operator=operators.inv)
767 def __bool__(self):
768 raise TypeError("Boolean value of this clause is not defined")
770 def __repr__(self):
771 friendly = self.description
772 if friendly is None:
773 return object.__repr__(self)
774 else:
775 return "<%s.%s at 0x%x; %s>" % (
776 self.__module__,
777 self.__class__.__name__,
778 id(self),
779 friendly,
780 )
783class DQLDMLClauseElement(ClauseElement):
784 """represents a :class:`.ClauseElement` that compiles to a DQL or DML
785 expression, not DDL.
787 .. versionadded:: 2.0
789 """
791 if typing.TYPE_CHECKING:
793 def _compiler(self, dialect: Dialect, **kw: Any) -> SQLCompiler:
794 """Return a compiler appropriate for this ClauseElement, given a
795 Dialect."""
796 ...
798 def compile( # noqa: A001
799 self,
800 bind: Optional[_HasDialect] = None,
801 dialect: Optional[Dialect] = None,
802 **kw: Any,
803 ) -> SQLCompiler: ...
806class CompilerColumnElement(
807 roles.DMLColumnRole,
808 roles.DDLConstraintColumnRole,
809 roles.ColumnsClauseRole,
810 CompilerElement,
811):
812 """A compiler-only column element used for ad-hoc string compilations.
814 .. versionadded:: 2.0
816 """
818 __slots__ = ()
820 _propagate_attrs = util.EMPTY_DICT
821 _is_collection_aggregate = False
824# SQLCoreOperations should be suiting the ExpressionElementRole
825# and ColumnsClauseRole. however the MRO issues become too elaborate
826# at the moment.
827class SQLCoreOperations(Generic[_T_co], ColumnOperators, TypingOnly):
828 __slots__ = ()
830 # annotations for comparison methods
831 # these are from operators->Operators / ColumnOperators,
832 # redefined with the specific types returned by ColumnElement hierarchies
833 if typing.TYPE_CHECKING:
835 @util.non_memoized_property
836 def _propagate_attrs(self) -> _PropagateAttrsType: ...
838 def operate(
839 self, op: OperatorType, *other: Any, **kwargs: Any
840 ) -> ColumnElement[Any]: ...
842 def reverse_operate(
843 self, op: OperatorType, other: Any, **kwargs: Any
844 ) -> ColumnElement[Any]: ...
846 @overload
847 def op(
848 self,
849 opstring: str,
850 precedence: int = ...,
851 is_comparison: bool = ...,
852 *,
853 return_type: _TypeEngineArgument[_OPT],
854 python_impl: Optional[Callable[..., Any]] = None,
855 operator_class: OperatorClass = ...,
856 ) -> Callable[[Any], BinaryExpression[_OPT]]: ...
858 @overload
859 def op(
860 self,
861 opstring: str,
862 precedence: int = ...,
863 is_comparison: bool = ...,
864 return_type: Optional[_TypeEngineArgument[Any]] = ...,
865 python_impl: Optional[Callable[..., Any]] = ...,
866 operator_class: OperatorClass = ...,
867 ) -> Callable[[Any], BinaryExpression[Any]]: ...
869 def op(
870 self,
871 opstring: str,
872 precedence: int = 0,
873 is_comparison: bool = False,
874 return_type: Optional[_TypeEngineArgument[Any]] = None,
875 python_impl: Optional[Callable[..., Any]] = None,
876 operator_class: OperatorClass = OperatorClass.BASE,
877 ) -> Callable[[Any], BinaryExpression[Any]]: ...
879 def bool_op(
880 self,
881 opstring: str,
882 precedence: int = 0,
883 python_impl: Optional[Callable[..., Any]] = None,
884 ) -> Callable[[Any], BinaryExpression[bool]]: ...
886 def __and__(self, other: Any) -> BooleanClauseList: ...
888 def __or__(self, other: Any) -> BooleanClauseList: ...
890 def __invert__(self) -> ColumnElement[_T_co]: ...
892 def __lt__(self, other: Any) -> ColumnElement[bool]: ...
894 def __le__(self, other: Any) -> ColumnElement[bool]: ...
896 # declare also that this class has an hash method otherwise
897 # it may be assumed to be None by type checkers since the
898 # object defines __eq__ and python sets it to None in that case:
899 # https://docs.python.org/3/reference/datamodel.html#object.__hash__
900 def __hash__(self) -> int: ...
902 def __eq__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
903 ...
905 def __ne__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
906 ...
908 def is_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
910 def is_not_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
912 def __gt__(self, other: Any) -> ColumnElement[bool]: ...
914 def __ge__(self, other: Any) -> ColumnElement[bool]: ...
916 def __neg__(self) -> UnaryExpression[_T_co]: ...
918 def __contains__(self, other: Any) -> ColumnElement[bool]: ...
920 def __getitem__(self, index: Any) -> ColumnElement[Any]: ...
922 @overload
923 def __lshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
925 @overload
926 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
928 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
930 @overload
931 def __rlshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
933 @overload
934 def __rlshift__(self, other: Any) -> ColumnElement[Any]: ...
936 def __rlshift__(self, other: Any) -> ColumnElement[Any]: ...
938 @overload
939 def __rshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
941 @overload
942 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
944 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
946 @overload
947 def __rrshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
949 @overload
950 def __rrshift__(self, other: Any) -> ColumnElement[Any]: ...
952 def __rrshift__(self, other: Any) -> ColumnElement[Any]: ...
954 def __matmul__(self, other: Any) -> ColumnElement[Any]: ...
956 def __rmatmul__(self, other: Any) -> ColumnElement[Any]: ...
958 @overload
959 def concat(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
961 @overload
962 def concat(self, other: Any) -> ColumnElement[Any]: ...
964 def concat(self, other: Any) -> ColumnElement[Any]: ...
966 def like(
967 self, other: Any, escape: Optional[str] = None
968 ) -> BinaryExpression[bool]: ...
970 def ilike(
971 self, other: Any, escape: Optional[str] = None
972 ) -> BinaryExpression[bool]: ...
974 def bitwise_xor(self, other: Any) -> BinaryExpression[Any]: ...
976 def bitwise_or(self, other: Any) -> BinaryExpression[Any]: ...
978 def bitwise_and(self, other: Any) -> BinaryExpression[Any]: ...
980 def bitwise_not(self) -> UnaryExpression[_T_co]: ...
982 def bitwise_lshift(self, other: Any) -> BinaryExpression[Any]: ...
984 def bitwise_rshift(self, other: Any) -> BinaryExpression[Any]: ...
986 def in_(
987 self,
988 other: Union[
989 Iterable[Any], BindParameter[Any], roles.InElementRole
990 ],
991 ) -> BinaryExpression[bool]: ...
993 def not_in(
994 self,
995 other: Union[
996 Iterable[Any], BindParameter[Any], roles.InElementRole
997 ],
998 ) -> BinaryExpression[bool]: ...
1000 def notin_(
1001 self,
1002 other: Union[
1003 Iterable[Any], BindParameter[Any], roles.InElementRole
1004 ],
1005 ) -> BinaryExpression[bool]: ...
1007 def not_like(
1008 self, other: Any, escape: Optional[str] = None
1009 ) -> BinaryExpression[bool]: ...
1011 def notlike(
1012 self, other: Any, escape: Optional[str] = None
1013 ) -> BinaryExpression[bool]: ...
1015 def not_ilike(
1016 self, other: Any, escape: Optional[str] = None
1017 ) -> BinaryExpression[bool]: ...
1019 def notilike(
1020 self, other: Any, escape: Optional[str] = None
1021 ) -> BinaryExpression[bool]: ...
1023 def is_(self, other: Any) -> BinaryExpression[bool]: ...
1025 def is_not(self, other: Any) -> BinaryExpression[bool]: ...
1027 def isnot(self, other: Any) -> BinaryExpression[bool]: ...
1029 def startswith(
1030 self,
1031 other: Any,
1032 escape: Optional[str] = None,
1033 autoescape: bool = False,
1034 ) -> ColumnElement[bool]: ...
1036 def istartswith(
1037 self,
1038 other: Any,
1039 escape: Optional[str] = None,
1040 autoescape: bool = False,
1041 ) -> ColumnElement[bool]: ...
1043 def endswith(
1044 self,
1045 other: Any,
1046 escape: Optional[str] = None,
1047 autoescape: bool = False,
1048 ) -> ColumnElement[bool]: ...
1050 def iendswith(
1051 self,
1052 other: Any,
1053 escape: Optional[str] = None,
1054 autoescape: bool = False,
1055 ) -> ColumnElement[bool]: ...
1057 def contains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1059 def icontains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1061 def match(self, other: Any, **kwargs: Any) -> ColumnElement[bool]: ...
1063 def regexp_match(
1064 self, pattern: Any, flags: Optional[str] = None
1065 ) -> ColumnElement[bool]: ...
1067 def regexp_replace(
1068 self, pattern: Any, replacement: Any, flags: Optional[str] = None
1069 ) -> ColumnElement[str]: ...
1071 def desc(self) -> UnaryExpression[_T_co]: ...
1073 def asc(self) -> UnaryExpression[_T_co]: ...
1075 def nulls_first(self) -> UnaryExpression[_T_co]: ...
1077 def nullsfirst(self) -> UnaryExpression[_T_co]: ...
1079 def nulls_last(self) -> UnaryExpression[_T_co]: ...
1081 def nullslast(self) -> UnaryExpression[_T_co]: ...
1083 def collate(self, collation: str) -> CollationClause: ...
1085 def between(
1086 self, cleft: Any, cright: Any, symmetric: bool = False
1087 ) -> BinaryExpression[bool]: ...
1089 def distinct(self: _SQO[_T_co]) -> UnaryExpression[_T_co]: ...
1091 def any_(self) -> CollectionAggregate[Any]: ...
1093 def all_(self) -> CollectionAggregate[Any]: ...
1095 # numeric overloads. These need more tweaking
1096 # in particular they all need to have a variant for Optiona[_T]
1097 # because Optional only applies to the data side, not the expression
1098 # side
1100 @overload
1101 def __add__(
1102 self: _SQO[_NMT],
1103 other: Any,
1104 ) -> ColumnElement[_NMT]: ...
1106 @overload
1107 def __add__(
1108 self: _SQO[str],
1109 other: Any,
1110 ) -> ColumnElement[str]: ...
1112 @overload
1113 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1115 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1117 @overload
1118 def __radd__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1120 @overload
1121 def __radd__(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
1123 def __radd__(self, other: Any) -> ColumnElement[Any]: ...
1125 @overload
1126 def __sub__(
1127 self: _SQO[_NMT],
1128 other: Any,
1129 ) -> ColumnElement[_NMT]: ...
1131 @overload
1132 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1134 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1136 @overload
1137 def __rsub__(
1138 self: _SQO[_NMT],
1139 other: Any,
1140 ) -> ColumnElement[_NMT]: ...
1142 @overload
1143 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1145 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1147 @overload
1148 def __mul__(
1149 self: _SQO[_NMT],
1150 other: Any,
1151 ) -> ColumnElement[_NMT]: ...
1153 @overload
1154 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1156 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1158 @overload
1159 def __rmul__(
1160 self: _SQO[_NMT],
1161 other: Any,
1162 ) -> ColumnElement[_NMT]: ...
1164 @overload
1165 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1167 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1169 @overload
1170 def __mod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1172 @overload
1173 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1175 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1177 @overload
1178 def __rmod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1180 @overload
1181 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1183 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1185 @overload
1186 def __truediv__(
1187 self: _SQO[int], other: Any
1188 ) -> ColumnElement[_NUMERIC]: ...
1190 @overload
1191 def __truediv__(self: _SQO[_NT], other: Any) -> ColumnElement[_NT]: ...
1193 @overload
1194 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1196 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1198 @overload
1199 def __rtruediv__(
1200 self: _SQO[_NMT], other: Any
1201 ) -> ColumnElement[_NUMERIC]: ...
1203 @overload
1204 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1206 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1208 @overload
1209 def __floordiv__(
1210 self: _SQO[_NMT], other: Any
1211 ) -> ColumnElement[_NMT]: ...
1213 @overload
1214 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1216 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1218 @overload
1219 def __rfloordiv__(
1220 self: _SQO[_NMT], other: Any
1221 ) -> ColumnElement[_NMT]: ...
1223 @overload
1224 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1226 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1229class SQLColumnExpression(
1230 SQLCoreOperations[_T_co], roles.ExpressionElementRole[_T_co], TypingOnly
1231):
1232 """A type that may be used to indicate any SQL column element or object
1233 that acts in place of one.
1235 :class:`.SQLColumnExpression` is a base of
1236 :class:`.ColumnElement`, as well as within the bases of ORM elements
1237 such as :class:`.InstrumentedAttribute`, and may be used in :pep:`484`
1238 typing to indicate arguments or return values that should behave
1239 as column expressions.
1241 .. versionadded:: 2.0.0b4
1244 """
1246 __slots__ = ()
1249_SQO = SQLCoreOperations
1252class ColumnElement(
1253 roles.ColumnArgumentOrKeyRole,
1254 roles.StatementOptionRole,
1255 roles.WhereHavingRole,
1256 roles.BinaryElementRole[_T],
1257 roles.OrderByRole,
1258 roles.ColumnsClauseRole,
1259 roles.LimitOffsetRole,
1260 roles.DMLColumnRole,
1261 roles.DDLConstraintColumnRole,
1262 roles.DDLExpressionRole,
1263 SQLColumnExpression[_T],
1264 DQLDMLClauseElement,
1265):
1266 """Represent a column-oriented SQL expression suitable for usage in the
1267 "columns" clause, WHERE clause etc. of a statement.
1269 While the most familiar kind of :class:`_expression.ColumnElement` is the
1270 :class:`_schema.Column` object, :class:`_expression.ColumnElement`
1271 serves as the basis
1272 for any unit that may be present in a SQL expression, including
1273 the expressions themselves, SQL functions, bound parameters,
1274 literal expressions, keywords such as ``NULL``, etc.
1275 :class:`_expression.ColumnElement`
1276 is the ultimate base class for all such elements.
1278 A wide variety of SQLAlchemy Core functions work at the SQL expression
1279 level, and are intended to accept instances of
1280 :class:`_expression.ColumnElement` as
1281 arguments. These functions will typically document that they accept a
1282 "SQL expression" as an argument. What this means in terms of SQLAlchemy
1283 usually refers to an input which is either already in the form of a
1284 :class:`_expression.ColumnElement` object,
1285 or a value which can be **coerced** into
1286 one. The coercion rules followed by most, but not all, SQLAlchemy Core
1287 functions with regards to SQL expressions are as follows:
1289 * a literal Python value, such as a string, integer or floating
1290 point value, boolean, datetime, ``Decimal`` object, or virtually
1291 any other Python object, will be coerced into a "literal bound
1292 value". This generally means that a :func:`.bindparam` will be
1293 produced featuring the given value embedded into the construct; the
1294 resulting :class:`.BindParameter` object is an instance of
1295 :class:`_expression.ColumnElement`.
1296 The Python value will ultimately be sent
1297 to the DBAPI at execution time as a parameterized argument to the
1298 ``execute()`` or ``executemany()`` methods, after SQLAlchemy
1299 type-specific converters (e.g. those provided by any associated
1300 :class:`.TypeEngine` objects) are applied to the value.
1302 * any special object value, typically ORM-level constructs, which
1303 feature an accessor called ``__clause_element__()``. The Core
1304 expression system looks for this method when an object of otherwise
1305 unknown type is passed to a function that is looking to coerce the
1306 argument into a :class:`_expression.ColumnElement` and sometimes a
1307 :class:`_expression.SelectBase` expression.
1308 It is used within the ORM to
1309 convert from ORM-specific objects like mapped classes and
1310 mapped attributes into Core expression objects.
1312 * The Python ``None`` value is typically interpreted as ``NULL``,
1313 which in SQLAlchemy Core produces an instance of :func:`.null`.
1315 A :class:`_expression.ColumnElement` provides the ability to generate new
1316 :class:`_expression.ColumnElement`
1317 objects using Python expressions. This means that Python operators
1318 such as ``==``, ``!=`` and ``<`` are overloaded to mimic SQL operations,
1319 and allow the instantiation of further :class:`_expression.ColumnElement`
1320 instances
1321 which are composed from other, more fundamental
1322 :class:`_expression.ColumnElement`
1323 objects. For example, two :class:`.ColumnClause` objects can be added
1324 together with the addition operator ``+`` to produce
1325 a :class:`.BinaryExpression`.
1326 Both :class:`.ColumnClause` and :class:`.BinaryExpression` are subclasses
1327 of :class:`_expression.ColumnElement`:
1329 .. sourcecode:: pycon+sql
1331 >>> from sqlalchemy.sql import column
1332 >>> column("a") + column("b")
1333 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
1334 >>> print(column("a") + column("b"))
1335 {printsql}a + b
1337 .. seealso::
1339 :class:`_schema.Column`
1341 :func:`_expression.column`
1343 """
1345 __visit_name__ = "column_element"
1347 primary_key: bool = False
1348 _is_clone_of: Optional[ColumnElement[_T]]
1349 _is_column_element = True
1350 _insert_sentinel: bool = False
1351 _omit_from_statements = False
1352 _is_collection_aggregate = False
1354 foreign_keys: AbstractSet[ForeignKey] = frozenset()
1356 @util.memoized_property
1357 def _proxies(self) -> List[ColumnElement[Any]]:
1358 return []
1360 @util.non_memoized_property
1361 def _tq_label(self) -> Optional[str]:
1362 """The named label that can be used to target
1363 this column in a result set in a "table qualified" context.
1365 This label is almost always the label used when
1366 rendering <expr> AS <label> in a SELECT statement when using
1367 the LABEL_STYLE_TABLENAME_PLUS_COL label style, which is what the
1368 legacy ORM ``Query`` object uses as well.
1370 For a regular Column bound to a Table, this is typically the label
1371 <tablename>_<columnname>. For other constructs, different rules
1372 may apply, such as anonymized labels and others.
1374 .. versionchanged:: 1.4.21 renamed from ``._label``
1376 """
1377 return None
1379 key: Optional[str] = None
1380 """The 'key' that in some circumstances refers to this object in a
1381 Python namespace.
1383 This typically refers to the "key" of the column as present in the
1384 ``.c`` collection of a selectable, e.g. ``sometable.c["somekey"]`` would
1385 return a :class:`_schema.Column` with a ``.key`` of "somekey".
1387 """
1389 @HasMemoized.memoized_attribute
1390 def _tq_key_label(self) -> Optional[str]:
1391 """A label-based version of 'key' that in some circumstances refers
1392 to this object in a Python namespace.
1395 _tq_key_label comes into play when a select() statement is constructed
1396 with apply_labels(); in this case, all Column objects in the ``.c``
1397 collection are rendered as <tablename>_<columnname> in SQL; this is
1398 essentially the value of ._label. But to locate those columns in the
1399 ``.c`` collection, the name is along the lines of <tablename>_<key>;
1400 that's the typical value of .key_label.
1402 .. versionchanged:: 1.4.21 renamed from ``._key_label``
1404 """
1405 return self._proxy_key
1407 @property
1408 def _key_label(self) -> Optional[str]:
1409 """legacy; renamed to _tq_key_label"""
1410 return self._tq_key_label
1412 @property
1413 def _label(self) -> Optional[str]:
1414 """legacy; renamed to _tq_label"""
1415 return self._tq_label
1417 @property
1418 def _non_anon_label(self) -> Optional[str]:
1419 """the 'name' that naturally applies this element when rendered in
1420 SQL.
1422 Concretely, this is the "name" of a column or a label in a
1423 SELECT statement; ``<columnname>`` and ``<labelname>`` below:
1425 .. sourcecode:: sql
1427 SELECT <columnmame> FROM table
1429 SELECT column AS <labelname> FROM table
1431 Above, the two names noted will be what's present in the DBAPI
1432 ``cursor.description`` as the names.
1434 If this attribute returns ``None``, it means that the SQL element as
1435 written does not have a 100% fully predictable "name" that would appear
1436 in the ``cursor.description``. Examples include SQL functions, CAST
1437 functions, etc. While such things do return names in
1438 ``cursor.description``, they are only predictable on a
1439 database-specific basis; e.g. an expression like ``MAX(table.col)`` may
1440 appear as the string ``max`` on one database (like PostgreSQL) or may
1441 appear as the whole expression ``max(table.col)`` on SQLite.
1443 The default implementation looks for a ``.name`` attribute on the
1444 object, as has been the precedent established in SQLAlchemy for many
1445 years. An exception is made on the ``FunctionElement`` subclass
1446 so that the return value is always ``None``.
1448 .. versionadded:: 1.4.21
1452 """
1453 return getattr(self, "name", None)
1455 _render_label_in_columns_clause = True
1456 """A flag used by select._columns_plus_names that helps to determine
1457 we are actually going to render in terms of "SELECT <col> AS <label>".
1458 This flag can be returned as False for some Column objects that want
1459 to be rendered as simple "SELECT <col>"; typically columns that don't have
1460 any parent table and are named the same as what the label would be
1461 in any case.
1463 """
1465 _allow_label_resolve = True
1466 """A flag that can be flipped to prevent a column from being resolvable
1467 by string label name.
1469 The joined eager loader strategy in the ORM uses this, for example.
1471 """
1473 _is_implicitly_boolean = False
1475 _alt_names: Sequence[str] = ()
1477 if TYPE_CHECKING:
1479 def _ungroup(self) -> ColumnElement[_T]: ...
1481 @overload
1482 def self_group(self, against: None = None) -> ColumnElement[_T]: ...
1484 @overload
1485 def self_group(
1486 self, against: Optional[OperatorType] = None
1487 ) -> ColumnElement[Any]: ...
1489 def self_group(
1490 self, against: Optional[OperatorType] = None
1491 ) -> ColumnElement[Any]:
1492 if (
1493 against in (operators.and_, operators.or_, operators._asbool)
1494 and self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity
1495 ):
1496 return AsBoolean(self, operators.is_true, operators.is_false)
1497 elif against in (operators.any_op, operators.all_op):
1498 return Grouping(self)
1499 else:
1500 return self
1502 @overload
1503 def _negate(self: ColumnElement[bool]) -> ColumnElement[bool]: ...
1505 @overload
1506 def _negate(self: ColumnElement[_T]) -> ColumnElement[_T]: ...
1508 def _negate(self) -> ColumnElement[Any]:
1509 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
1510 return AsBoolean(self, operators.is_false, operators.is_true)
1511 else:
1512 grouped = self.self_group(against=operators.inv)
1513 assert isinstance(grouped, ColumnElement)
1514 return UnaryExpression(
1515 grouped,
1516 operator=operators.inv,
1517 )
1519 type: TypeEngine[_T]
1521 if not TYPE_CHECKING:
1523 @util.memoized_property
1524 def type(self) -> TypeEngine[_T]: # noqa: A001
1525 # used for delayed setup of
1526 # type_api
1527 return type_api.NULLTYPE
1529 @HasMemoized.memoized_attribute
1530 def comparator(self) -> TypeEngine.Comparator[_T]:
1531 try:
1532 comparator_factory = self.type.comparator_factory
1533 except AttributeError as err:
1534 raise TypeError(
1535 "Object %r associated with '.type' attribute "
1536 "is not a TypeEngine class or object" % self.type
1537 ) from err
1538 else:
1539 return comparator_factory(self)
1541 def __setstate__(self, state):
1542 self.__dict__.update(state)
1544 def __getattr__(self, key: str) -> Any:
1545 try:
1546 return getattr(self.comparator, key)
1547 except AttributeError as err:
1548 raise AttributeError(
1549 "Neither %r object nor %r object has an attribute %r"
1550 % (
1551 type(self).__name__,
1552 type(self.comparator).__name__,
1553 key,
1554 )
1555 ) from err
1557 def operate(
1558 self,
1559 op: operators.OperatorType,
1560 *other: Any,
1561 **kwargs: Any,
1562 ) -> ColumnElement[Any]:
1563 return op(self.comparator, *other, **kwargs) # type: ignore[no-any-return] # noqa: E501
1565 def reverse_operate(
1566 self, op: operators.OperatorType, other: Any, **kwargs: Any
1567 ) -> ColumnElement[Any]:
1568 return op(other, self.comparator, **kwargs) # type: ignore[no-any-return] # noqa: E501
1570 def _bind_param(
1571 self,
1572 operator: operators.OperatorType,
1573 obj: Any,
1574 type_: Optional[TypeEngine[_T]] = None,
1575 expanding: bool = False,
1576 ) -> BindParameter[_T]:
1577 return BindParameter(
1578 None,
1579 obj,
1580 _compared_to_operator=operator,
1581 type_=type_,
1582 _compared_to_type=self.type,
1583 unique=True,
1584 expanding=expanding,
1585 )
1587 @property
1588 def expression(self) -> ColumnElement[Any]:
1589 """Return a column expression.
1591 Part of the inspection interface; returns self.
1593 """
1594 return self
1596 @property
1597 def _select_iterable(self) -> _SelectIterable:
1598 return (self,)
1600 @util.memoized_property
1601 def base_columns(self) -> FrozenSet[ColumnElement[Any]]:
1602 return frozenset(c for c in self.proxy_set if not c._proxies)
1604 @util.memoized_property
1605 def proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1606 """set of all columns we are proxying
1608 as of 2.0 this is explicitly deannotated columns. previously it was
1609 effectively deannotated columns but wasn't enforced. annotated
1610 columns should basically not go into sets if at all possible because
1611 their hashing behavior is very non-performant.
1613 """
1614 return frozenset([self._deannotate()]).union(
1615 itertools.chain(*[c.proxy_set for c in self._proxies])
1616 )
1618 @util.memoized_property
1619 def _expanded_proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1620 return frozenset(_expand_cloned(self.proxy_set))
1622 def _uncached_proxy_list(self) -> List[ColumnElement[Any]]:
1623 """An 'uncached' version of proxy set.
1625 This list includes annotated columns which perform very poorly in
1626 set operations.
1628 """
1630 return [self] + list(
1631 itertools.chain(*[c._uncached_proxy_list() for c in self._proxies])
1632 )
1634 def shares_lineage(self, othercolumn: ColumnElement[Any]) -> bool:
1635 """Return True if the given :class:`_expression.ColumnElement`
1636 has a common ancestor to this :class:`_expression.ColumnElement`."""
1638 return bool(self.proxy_set.intersection(othercolumn.proxy_set))
1640 def _compare_name_for_result(self, other: ColumnElement[Any]) -> bool:
1641 """Return True if the given column element compares to this one
1642 when targeting within a result row."""
1644 return (
1645 hasattr(other, "name")
1646 and hasattr(self, "name")
1647 and other.name == self.name
1648 )
1650 @HasMemoized.memoized_attribute
1651 def _proxy_key(self) -> Optional[str]:
1652 if self._annotations and "proxy_key" in self._annotations:
1653 return cast(str, self._annotations["proxy_key"])
1655 name = self.key
1656 if not name:
1657 # there's a bit of a seeming contradiction which is that the
1658 # "_non_anon_label" of a column can in fact be an
1659 # "_anonymous_label"; this is when it's on a column that is
1660 # proxying for an anonymous expression in a subquery.
1661 name = self._non_anon_label
1663 if isinstance(name, _anonymous_label):
1664 return None
1665 else:
1666 return name
1668 @HasMemoized.memoized_attribute
1669 def _expression_label(self) -> Optional[str]:
1670 """a suggested label to use in the case that the column has no name,
1671 which should be used if possible as the explicit 'AS <label>'
1672 where this expression would normally have an anon label.
1674 this is essentially mostly what _proxy_key does except it returns
1675 None if the column has a normal name that can be used.
1677 """
1679 if getattr(self, "name", None) is not None:
1680 return None
1681 elif self._annotations and "proxy_key" in self._annotations:
1682 return cast(str, self._annotations["proxy_key"])
1683 else:
1684 return None
1686 def _make_proxy(
1687 self,
1688 selectable: FromClause,
1689 *,
1690 primary_key: ColumnSet,
1691 foreign_keys: Set[KeyedColumnElement[Any]],
1692 name: Optional[str] = None,
1693 key: Optional[str] = None,
1694 name_is_truncatable: bool = False,
1695 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
1696 **kw: Any,
1697 ) -> typing_Tuple[str, ColumnClause[_T]]:
1698 """Create a new :class:`_expression.ColumnElement` representing this
1699 :class:`_expression.ColumnElement` as it appears in the select list of
1700 a descending selectable.
1702 """
1703 if name is None:
1704 name = self._anon_name_label
1705 if key is None:
1706 key = self._proxy_key
1707 else:
1708 key = name
1710 assert key is not None
1712 co: ColumnClause[_T] = ColumnClause(
1713 (
1714 coercions.expect(roles.TruncatedLabelRole, name)
1715 if name_is_truncatable
1716 else name
1717 ),
1718 type_=getattr(self, "type", None),
1719 _selectable=selectable,
1720 )
1722 co._propagate_attrs = selectable._propagate_attrs
1723 if compound_select_cols:
1724 co._proxies = list(compound_select_cols)
1725 else:
1726 co._proxies = [self]
1727 if selectable._is_clone_of is not None:
1728 co._is_clone_of = selectable._is_clone_of.columns.get(key)
1729 return key, co
1731 def cast(self, type_: _TypeEngineArgument[_OPT]) -> Cast[_OPT]:
1732 """Produce a type cast, i.e. ``CAST(<expression> AS <type>)``.
1734 This is a shortcut to the :func:`_expression.cast` function.
1736 .. seealso::
1738 :ref:`tutorial_casts`
1740 :func:`_expression.cast`
1742 :func:`_expression.type_coerce`
1744 """
1745 return Cast(self, type_)
1747 def label(self, name: Optional[str]) -> Label[_T]:
1748 """Produce a column label, i.e. ``<columnname> AS <name>``.
1750 This is a shortcut to the :func:`_expression.label` function.
1752 If 'name' is ``None``, an anonymous label name will be generated.
1754 """
1755 return Label(name, self, self.type)
1757 def _anon_label(
1758 self, seed: Optional[str], add_hash: Optional[int] = None
1759 ) -> _anonymous_label:
1760 while self._is_clone_of is not None:
1761 self = self._is_clone_of
1763 # as of 1.4 anonymous label for ColumnElement uses hash(), not id(),
1764 # as the identifier, because a column and its annotated version are
1765 # the same thing in a SQL statement
1766 hash_value = hash(self)
1768 if add_hash:
1769 # this path is used for disambiguating anon labels that would
1770 # otherwise be the same name for the same element repeated.
1771 # an additional numeric value is factored in for each label.
1773 # shift hash(self) (which is id(self), typically 8 byte integer)
1774 # 16 bits leftward. fill extra add_hash on right
1775 assert add_hash < (2 << 15)
1776 assert seed
1777 hash_value = (hash_value << 16) | add_hash
1779 # extra underscore is added for labels with extra hash
1780 # values, to isolate the "deduped anon" namespace from the
1781 # regular namespace. eliminates chance of these
1782 # manufactured hash values overlapping with regular ones for some
1783 # undefined python interpreter
1784 seed = seed + "_"
1786 if isinstance(seed, _anonymous_label):
1787 # NOTE: the space after the hash is required
1788 return _anonymous_label(f"{seed}%({hash_value} )s")
1790 return _anonymous_label.safe_construct(hash_value, seed or "anon")
1792 @util.memoized_property
1793 def _anon_name_label(self) -> str:
1794 """Provides a constant 'anonymous label' for this ColumnElement.
1796 This is a label() expression which will be named at compile time.
1797 The same label() is returned each time ``anon_label`` is called so
1798 that expressions can reference ``anon_label`` multiple times,
1799 producing the same label name at compile time.
1801 The compiler uses this function automatically at compile time
1802 for expressions that are known to be 'unnamed' like binary
1803 expressions and function calls.
1805 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1806 public and is renamed to _anon_name_label. anon_name exists
1807 for backwards compat
1809 """
1810 name = getattr(self, "name", None)
1811 return self._anon_label(name)
1813 @util.memoized_property
1814 def _anon_key_label(self) -> _anonymous_label:
1815 """Provides a constant 'anonymous key label' for this ColumnElement.
1817 Compare to ``anon_label``, except that the "key" of the column,
1818 if available, is used to generate the label.
1820 This is used when a deduplicating key is placed into the columns
1821 collection of a selectable.
1823 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1824 public and is renamed to _anon_key_label. anon_key_label exists
1825 for backwards compat
1827 """
1828 return self._anon_label(self._proxy_key)
1830 @property
1831 @util.deprecated(
1832 "1.4",
1833 "The :attr:`_expression.ColumnElement.anon_label` attribute is now "
1834 "private, and the public accessor is deprecated.",
1835 )
1836 def anon_label(self) -> str:
1837 return self._anon_name_label
1839 @property
1840 @util.deprecated(
1841 "1.4",
1842 "The :attr:`_expression.ColumnElement.anon_key_label` attribute is "
1843 "now private, and the public accessor is deprecated.",
1844 )
1845 def anon_key_label(self) -> str:
1846 return self._anon_key_label
1848 def _dedupe_anon_label_idx(self, idx: int) -> str:
1849 """label to apply to a column that is anon labeled, but repeated
1850 in the SELECT, so that we have to make an "extra anon" label that
1851 disambiguates it from the previous appearance.
1853 these labels come out like "foo_bar_id__1" and have double underscores
1854 in them.
1856 """
1857 label = getattr(self, "name", None)
1859 # current convention is that if the element doesn't have a
1860 # ".name" (usually because it is not NamedColumn), we try to
1861 # use a "table qualified" form for the "dedupe anon" label,
1862 # based on the notion that a label like
1863 # "CAST(casttest.v1 AS DECIMAL) AS casttest_v1__1" looks better than
1864 # "CAST(casttest.v1 AS DECIMAL) AS anon__1"
1866 if label is None:
1867 return self._dedupe_anon_tq_label_idx(idx)
1868 else:
1869 return self._anon_label(label, add_hash=idx)
1871 @util.memoized_property
1872 def _anon_tq_label(self) -> _anonymous_label:
1873 return self._anon_label(getattr(self, "_tq_label", None))
1875 @util.memoized_property
1876 def _anon_tq_key_label(self) -> _anonymous_label:
1877 return self._anon_label(getattr(self, "_tq_key_label", None))
1879 def _dedupe_anon_tq_label_idx(self, idx: int) -> _anonymous_label:
1880 label = getattr(self, "_tq_label", None) or "anon"
1882 return self._anon_label(label, add_hash=idx)
1885class KeyedColumnElement(ColumnElement[_T]):
1886 """ColumnElement where ``.key`` is non-None."""
1888 _is_keyed_column_element = True
1890 key: str
1893class WrapsColumnExpression(ColumnElement[_T]):
1894 """Mixin that defines a :class:`_expression.ColumnElement`
1895 as a wrapper with special
1896 labeling behavior for an expression that already has a name.
1898 .. versionadded:: 1.4
1900 .. seealso::
1902 :ref:`change_4449`
1905 """
1907 @property
1908 def wrapped_column_expression(self) -> ColumnElement[_T]:
1909 raise NotImplementedError()
1911 @util.non_memoized_property
1912 def _tq_label(self) -> Optional[str]:
1913 wce = self.wrapped_column_expression
1914 if hasattr(wce, "_tq_label"):
1915 return wce._tq_label
1916 else:
1917 return None
1919 @property
1920 def _label(self) -> Optional[str]:
1921 return self._tq_label
1923 @property
1924 def _non_anon_label(self) -> Optional[str]:
1925 return None
1927 @util.non_memoized_property
1928 def _anon_name_label(self) -> str:
1929 wce = self.wrapped_column_expression
1931 # this logic tries to get the WrappedColumnExpression to render
1932 # with "<expr> AS <name>", where "<name>" is the natural name
1933 # within the expression itself. e.g. "CAST(table.foo) AS foo".
1934 if not wce._is_text_clause:
1935 nal = wce._non_anon_label
1936 if nal:
1937 return nal
1938 elif hasattr(wce, "_anon_name_label"):
1939 return wce._anon_name_label
1940 return super()._anon_name_label
1942 def _dedupe_anon_label_idx(self, idx: int) -> str:
1943 wce = self.wrapped_column_expression
1944 nal = wce._non_anon_label
1945 if nal:
1946 return self._anon_label(nal + "_")
1947 else:
1948 return self._dedupe_anon_tq_label_idx(idx)
1950 @property
1951 def _proxy_key(self):
1952 wce = self.wrapped_column_expression
1954 if not wce._is_text_clause:
1955 return wce._proxy_key
1956 return super()._proxy_key
1959class DMLTargetCopy(roles.InElementRole, KeyedColumnElement[_T]):
1960 """Refer to another column's VALUES or SET expression in an INSERT or
1961 UPDATE statement.
1963 See the public-facing :func:`_sql.from_dml_column` constructor for
1964 background.
1966 .. versionadded:: 2.1
1969 """
1971 def __init__(self, column: _DMLOnlyColumnArgument[_T]):
1972 self.column = coercions.expect(roles.ColumnArgumentRole, column)
1973 self.type = self.column.type
1975 __visit_name__ = "dmltargetcopy"
1977 _traverse_internals: _TraverseInternalsType = [
1978 ("column", InternalTraversal.dp_clauseelement),
1979 ]
1982class BindParameter(roles.InElementRole, KeyedColumnElement[_T]):
1983 r"""Represent a "bound expression".
1985 :class:`.BindParameter` is invoked explicitly using the
1986 :func:`.bindparam` function, as in::
1988 from sqlalchemy import bindparam
1990 stmt = select(users_table).where(
1991 users_table.c.name == bindparam("username")
1992 )
1994 Detailed discussion of how :class:`.BindParameter` is used is
1995 at :func:`.bindparam`.
1997 .. seealso::
1999 :func:`.bindparam`
2001 """
2003 __visit_name__ = "bindparam"
2005 _traverse_internals: _TraverseInternalsType = [
2006 ("key", InternalTraversal.dp_anon_name),
2007 ("type", InternalTraversal.dp_type),
2008 ("callable", InternalTraversal.dp_plain_dict),
2009 ("value", InternalTraversal.dp_plain_obj),
2010 ("literal_execute", InternalTraversal.dp_boolean),
2011 ]
2013 key: str
2014 _anon_map_key: Optional[str] = None
2015 type: TypeEngine[_T]
2016 value: Optional[_T]
2018 _is_crud = False
2019 _is_bind_parameter = True
2021 # bindparam implements its own _gen_cache_key() method however
2022 # we check subclasses for this flag, else no cache key is generated
2023 inherit_cache = True
2025 def __init__(
2026 self,
2027 key: Optional[str],
2028 value: Any = _NoArg.NO_ARG,
2029 type_: Optional[_TypeEngineArgument[_T]] = None,
2030 unique: bool = False,
2031 required: Union[bool, Literal[_NoArg.NO_ARG]] = _NoArg.NO_ARG,
2032 quote: Optional[bool] = None,
2033 callable_: Optional[Callable[[], Any]] = None,
2034 expanding: bool = False,
2035 isoutparam: bool = False,
2036 literal_execute: bool = False,
2037 _compared_to_operator: Optional[OperatorType] = None,
2038 _compared_to_type: Optional[TypeEngine[Any]] = None,
2039 _is_crud: bool = False,
2040 ):
2041 if required is _NoArg.NO_ARG:
2042 required = value is _NoArg.NO_ARG and callable_ is None
2043 if value is _NoArg.NO_ARG:
2044 value = None
2046 if quote is not None:
2047 key = quoted_name.construct(key, quote)
2049 if unique:
2050 self.key, self._anon_map_key = (
2051 _anonymous_label.safe_construct_with_key(
2052 id(self),
2053 (
2054 key
2055 if key is not None
2056 and not isinstance(key, _anonymous_label)
2057 else "param"
2058 ),
2059 sanitize_key=True,
2060 )
2061 )
2062 elif key:
2063 self.key = key
2064 else:
2065 self.key, self._anon_map_key = (
2066 _anonymous_label.safe_construct_with_key(id(self), "param")
2067 )
2069 # identifying key that won't change across
2070 # clones, used to identify the bind's logical
2071 # identity
2072 self._identifying_key = self.key
2074 # key that was passed in the first place, used to
2075 # generate new keys
2076 self._orig_key = key or "param"
2078 self.unique = unique
2079 self.value = value
2080 self.callable = callable_
2081 self.isoutparam = isoutparam
2082 self.required = required
2084 # indicate an "expanding" parameter; the compiler sets this
2085 # automatically in the compiler _render_in_expr_w_bindparam method
2086 # for an IN expression
2087 self.expanding = expanding
2089 # this is another hint to help w/ expanding and is typically
2090 # set in the compiler _render_in_expr_w_bindparam method for an
2091 # IN expression
2092 self.expand_op = None
2094 self.literal_execute = literal_execute
2095 if _is_crud:
2096 self._is_crud = True
2098 if type_ is None:
2099 if expanding:
2100 if value:
2101 check_value = value[0]
2102 else:
2103 check_value = type_api._NO_VALUE_IN_LIST
2104 else:
2105 check_value = value
2106 if _compared_to_type is not None:
2107 self.type = _compared_to_type.coerce_compared_value(
2108 _compared_to_operator, check_value
2109 )
2110 else:
2111 self.type = type_api._resolve_value_to_type(check_value)
2112 elif isinstance(type_, type):
2113 self.type = type_()
2114 elif is_tuple_type(type_):
2115 if value:
2116 if expanding:
2117 check_value = value[0]
2118 else:
2119 check_value = value
2120 cast("BindParameter[TupleAny]", self).type = (
2121 type_._resolve_values_to_types(check_value)
2122 )
2123 else:
2124 cast("BindParameter[TupleAny]", self).type = type_
2125 else:
2126 self.type = type_
2128 def _with_value(self, value, maintain_key=False, required=NO_ARG):
2129 """Return a copy of this :class:`.BindParameter` with the given value
2130 set.
2131 """
2132 cloned = self._clone(maintain_key=maintain_key)
2133 cloned.value = value
2134 cloned.callable = None
2135 cloned.required = required if required is not NO_ARG else self.required
2136 if cloned.type is type_api.NULLTYPE:
2137 cloned.type = type_api._resolve_value_to_type(value)
2138 return cloned
2140 @property
2141 def effective_value(self) -> Optional[_T]:
2142 """Return the value of this bound parameter,
2143 taking into account if the ``callable`` parameter
2144 was set.
2146 The ``callable`` value will be evaluated
2147 and returned if present, else ``value``.
2149 """
2150 if self.callable:
2151 # TODO: set up protocol for bind parameter callable
2152 return self.callable() # type: ignore
2153 else:
2154 return self.value
2156 def render_literal_execute(self) -> Self:
2157 """Produce a copy of this bound parameter that will enable the
2158 :paramref:`_sql.BindParameter.literal_execute` flag.
2160 The :paramref:`_sql.BindParameter.literal_execute` flag will
2161 have the effect of the parameter rendered in the compiled SQL
2162 string using ``[POSTCOMPILE]`` form, which is a special form that
2163 is converted to be a rendering of the literal value of the parameter
2164 at SQL execution time. The rationale is to support caching
2165 of SQL statement strings that can embed per-statement literal values,
2166 such as LIMIT and OFFSET parameters, in the final SQL string that
2167 is passed to the DBAPI. Dialects in particular may want to use
2168 this method within custom compilation schemes.
2170 .. versionadded:: 1.4.5
2172 .. seealso::
2174 :ref:`engine_thirdparty_caching`
2176 """
2177 c: Self = ClauseElement._clone(self)
2178 c.literal_execute = True
2179 return c
2181 def _negate_in_binary(self, negated_op, original_op):
2182 if self.expand_op is original_op:
2183 bind = self._clone()
2184 bind.expand_op = negated_op
2185 return bind
2186 else:
2187 return self
2189 def _with_binary_element_type(self, type_: TypeEngine[Any]) -> Self:
2190 c: Self = ClauseElement._clone(self)
2191 c.type = type_
2192 return c
2194 def _clone(self, maintain_key: bool = False, **kw: Any) -> Self:
2195 c: Self = ClauseElement._clone(self, **kw)
2196 # ensure all the BindParameter objects stay in cloned set.
2197 # in #7823, we changed "clone" so that a clone only keeps a reference
2198 # to the "original" element, since for column correspondence, that's
2199 # all we need. However, for BindParam, _cloned_set is used by
2200 # the "cache key bind match" lookup, which means if any of those
2201 # interim BindParameter objects became part of a cache key in the
2202 # cache, we need it. So here, make sure all clones keep carrying
2203 # forward.
2204 c._cloned_set.update(self._cloned_set)
2205 if not maintain_key and self.unique:
2206 c.key, c._anon_map_key = _anonymous_label.safe_construct_with_key(
2207 id(c), c._orig_key or "param", sanitize_key=True
2208 )
2209 return c
2211 def _gen_cache_key(self, anon_map, bindparams):
2212 _gen_cache_ok = self.__class__.__dict__.get("inherit_cache", False)
2214 if not _gen_cache_ok:
2215 if anon_map is not None:
2216 anon_map[NO_CACHE] = True
2217 return None
2219 id_, found = anon_map.get_anon(self)
2220 if found:
2221 return (id_, self.__class__)
2223 if bindparams is not None:
2224 bindparams.append(self)
2226 return (
2227 id_,
2228 self.__class__,
2229 self.type._static_cache_key,
2230 (
2231 anon_map[self._anon_map_key]
2232 if self._anon_map_key is not None
2233 else self.key
2234 ),
2235 self.literal_execute,
2236 )
2238 def _convert_to_unique(self):
2239 if not self.unique:
2240 self.unique = True
2241 self.key, self._anon_map_key = (
2242 _anonymous_label.safe_construct_with_key(
2243 id(self), self._orig_key or "param", sanitize_key=True
2244 )
2245 )
2247 def __getstate__(self):
2248 """execute a deferred value for serialization purposes."""
2250 d = self.__dict__.copy()
2251 v = self.value
2252 if self.callable:
2253 v = self.callable()
2254 d["callable"] = None
2255 d["value"] = v
2256 return d
2258 def __setstate__(self, state):
2259 if state.get("unique", False):
2260 anon_and_key = _anonymous_label.safe_construct_with_key(
2261 id(self), state.get("_orig_key", "param"), sanitize_key=True
2262 )
2263 state["key"], state["_anon_map_key"] = anon_and_key
2264 self.__dict__.update(state)
2266 def __repr__(self):
2267 return "%s(%r, %r, type_=%r)" % (
2268 self.__class__.__name__,
2269 self.key,
2270 self.value,
2271 self.type,
2272 )
2275class TypeClause(DQLDMLClauseElement):
2276 """Handle a type keyword in a SQL statement.
2278 Used by the ``Case`` statement.
2280 """
2282 __visit_name__ = "typeclause"
2284 _traverse_internals: _TraverseInternalsType = [
2285 ("type", InternalTraversal.dp_type)
2286 ]
2287 type: TypeEngine[Any]
2289 def __init__(self, type_: TypeEngine[Any]):
2290 self.type = type_
2293class TextClause(
2294 roles.DDLConstraintColumnRole,
2295 roles.DDLExpressionRole,
2296 roles.StatementOptionRole,
2297 roles.WhereHavingRole,
2298 roles.OrderByRole,
2299 roles.FromClauseRole,
2300 roles.SelectStatementRole,
2301 roles.InElementRole,
2302 Generative,
2303 Executable,
2304 DQLDMLClauseElement,
2305 roles.BinaryElementRole[Any],
2306 inspection.Inspectable["TextClause"],
2307):
2308 """Represent a literal SQL text fragment.
2310 E.g.::
2312 from sqlalchemy import text
2314 t = text("SELECT * FROM users")
2315 result = connection.execute(t)
2317 The :class:`_expression.TextClause` construct is produced using the
2318 :func:`_expression.text`
2319 function; see that function for full documentation.
2321 .. seealso::
2323 :func:`_expression.text`
2325 """
2327 __visit_name__ = "textclause"
2329 _traverse_internals: _TraverseInternalsType = [
2330 ("_bindparams", InternalTraversal.dp_string_clauseelement_dict),
2331 ("text", InternalTraversal.dp_string),
2332 ]
2334 _is_text_clause = True
2336 _is_textual = True
2338 _bind_params_regex = re.compile(r"(?<![:\w\x5c]):(\w+)(?!:)", re.UNICODE)
2339 _is_implicitly_boolean = False
2341 _render_label_in_columns_clause = False
2343 _omit_from_statements = False
2345 _is_collection_aggregate = False
2347 @property
2348 def _hide_froms(self) -> Iterable[FromClause]:
2349 return ()
2351 def __and__(self, other):
2352 # support use in select.where(), query.filter()
2353 return and_(self, other)
2355 @property
2356 def _select_iterable(self) -> _SelectIterable:
2357 return (self,)
2359 # help in those cases where text() is
2360 # interpreted in a column expression situation
2361 key: Optional[str] = None
2362 _label: Optional[str] = None
2364 _allow_label_resolve = False
2366 @property
2367 def _is_star(self): # type: ignore[override]
2368 return self.text == "*"
2370 def __init__(self, text: str):
2371 self._bindparams: Dict[str, BindParameter[Any]] = {}
2373 def repl(m):
2374 self._bindparams[m.group(1)] = BindParameter(m.group(1))
2375 return ":%s" % m.group(1)
2377 # scan the string and search for bind parameter names, add them
2378 # to the list of bindparams
2379 self.text = self._bind_params_regex.sub(repl, text)
2381 @_generative
2382 def bindparams(
2383 self,
2384 *binds: BindParameter[Any],
2385 **names_to_values: Any,
2386 ) -> Self:
2387 """Establish the values and/or types of bound parameters within
2388 this :class:`_expression.TextClause` construct.
2390 Given a text construct such as::
2392 from sqlalchemy import text
2394 stmt = text(
2395 "SELECT id, name FROM user WHERE name=:name AND timestamp=:timestamp"
2396 )
2398 the :meth:`_expression.TextClause.bindparams`
2399 method can be used to establish
2400 the initial value of ``:name`` and ``:timestamp``,
2401 using simple keyword arguments::
2403 stmt = stmt.bindparams(
2404 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2405 )
2407 Where above, new :class:`.BindParameter` objects
2408 will be generated with the names ``name`` and ``timestamp``, and
2409 values of ``jack`` and ``datetime.datetime(2012, 10, 8, 15, 12, 5)``,
2410 respectively. The types will be
2411 inferred from the values given, in this case :class:`.String` and
2412 :class:`.DateTime`.
2414 When specific typing behavior is needed, the positional ``*binds``
2415 argument can be used in which to specify :func:`.bindparam` constructs
2416 directly. These constructs must include at least the ``key``
2417 argument, then an optional value and type::
2419 from sqlalchemy import bindparam
2421 stmt = stmt.bindparams(
2422 bindparam("name", value="jack", type_=String),
2423 bindparam("timestamp", type_=DateTime),
2424 )
2426 Above, we specified the type of :class:`.DateTime` for the
2427 ``timestamp`` bind, and the type of :class:`.String` for the ``name``
2428 bind. In the case of ``name`` we also set the default value of
2429 ``"jack"``.
2431 Additional bound parameters can be supplied at statement execution
2432 time, e.g.::
2434 result = connection.execute(
2435 stmt, timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2436 )
2438 The :meth:`_expression.TextClause.bindparams`
2439 method can be called repeatedly,
2440 where it will re-use existing :class:`.BindParameter` objects to add
2441 new information. For example, we can call
2442 :meth:`_expression.TextClause.bindparams`
2443 first with typing information, and a
2444 second time with value information, and it will be combined::
2446 stmt = text(
2447 "SELECT id, name FROM user WHERE name=:name "
2448 "AND timestamp=:timestamp"
2449 )
2450 stmt = stmt.bindparams(
2451 bindparam("name", type_=String), bindparam("timestamp", type_=DateTime)
2452 )
2453 stmt = stmt.bindparams(
2454 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2455 )
2457 The :meth:`_expression.TextClause.bindparams`
2458 method also supports the concept of
2459 **unique** bound parameters. These are parameters that are
2460 "uniquified" on name at statement compilation time, so that multiple
2461 :func:`_expression.text`
2462 constructs may be combined together without the names
2463 conflicting. To use this feature, specify the
2464 :paramref:`.BindParameter.unique` flag on each :func:`.bindparam`
2465 object::
2467 stmt1 = text("select id from table where name=:name").bindparams(
2468 bindparam("name", value="name1", unique=True)
2469 )
2470 stmt2 = text("select id from table where name=:name").bindparams(
2471 bindparam("name", value="name2", unique=True)
2472 )
2474 union = union_all(stmt1.columns(column("id")), stmt2.columns(column("id")))
2476 The above statement will render as:
2478 .. sourcecode:: sql
2480 select id from table where name=:name_1
2481 UNION ALL select id from table where name=:name_2
2483 """ # noqa: E501
2484 self._bindparams = new_params = self._bindparams.copy()
2486 for bind in binds:
2487 try:
2488 # the regex used for text() currently will not match
2489 # a unique/anonymous key in any case, so use the _orig_key
2490 # so that a text() construct can support unique parameters
2491 existing = new_params[bind._orig_key]
2492 except KeyError as err:
2493 raise exc.ArgumentError(
2494 "This text() construct doesn't define a "
2495 "bound parameter named %r" % bind._orig_key
2496 ) from err
2497 else:
2498 new_params[existing._orig_key] = bind
2500 for key, value in names_to_values.items():
2501 try:
2502 existing = new_params[key]
2503 except KeyError as err:
2504 raise exc.ArgumentError(
2505 "This text() construct doesn't define a "
2506 "bound parameter named %r" % key
2507 ) from err
2508 else:
2509 new_params[key] = existing._with_value(value, required=False)
2510 return self
2512 @util.preload_module("sqlalchemy.sql.selectable")
2513 def columns(
2514 self,
2515 *cols: _ColumnExpressionArgument[Any],
2516 **types: _TypeEngineArgument[Any],
2517 ) -> TextualSelect:
2518 r"""Turn this :class:`_expression.TextClause` object into a
2519 :class:`_expression.TextualSelect`
2520 object that serves the same role as a SELECT
2521 statement.
2523 The :class:`_expression.TextualSelect` is part of the
2524 :class:`_expression.SelectBase`
2525 hierarchy and can be embedded into another statement by using the
2526 :meth:`_expression.TextualSelect.subquery` method to produce a
2527 :class:`.Subquery`
2528 object, which can then be SELECTed from.
2530 This function essentially bridges the gap between an entirely
2531 textual SELECT statement and the SQL expression language concept
2532 of a "selectable"::
2534 from sqlalchemy.sql import column, text
2536 stmt = text("SELECT id, name FROM some_table")
2537 stmt = stmt.columns(column("id"), column("name")).subquery("st")
2539 stmt = (
2540 select(mytable)
2541 .select_from(mytable.join(stmt, mytable.c.name == stmt.c.name))
2542 .where(stmt.c.id > 5)
2543 )
2545 Above, we pass a series of :func:`_expression.column` elements to the
2546 :meth:`_expression.TextClause.columns` method positionally. These
2547 :func:`_expression.column`
2548 elements now become first class elements upon the
2549 :attr:`_expression.TextualSelect.selected_columns` column collection,
2550 which then
2551 become part of the :attr:`.Subquery.c` collection after
2552 :meth:`_expression.TextualSelect.subquery` is invoked.
2554 The column expressions we pass to
2555 :meth:`_expression.TextClause.columns` may
2556 also be typed; when we do so, these :class:`.TypeEngine` objects become
2557 the effective return type of the column, so that SQLAlchemy's
2558 result-set-processing systems may be used on the return values.
2559 This is often needed for types such as date or boolean types, as well
2560 as for unicode processing on some dialect configurations::
2562 stmt = text("SELECT id, name, timestamp FROM some_table")
2563 stmt = stmt.columns(
2564 column("id", Integer),
2565 column("name", Unicode),
2566 column("timestamp", DateTime),
2567 )
2569 for id, name, timestamp in connection.execute(stmt):
2570 print(id, name, timestamp)
2572 As a shortcut to the above syntax, keyword arguments referring to
2573 types alone may be used, if only type conversion is needed::
2575 stmt = text("SELECT id, name, timestamp FROM some_table")
2576 stmt = stmt.columns(id=Integer, name=Unicode, timestamp=DateTime)
2578 for id, name, timestamp in connection.execute(stmt):
2579 print(id, name, timestamp)
2581 The positional form of :meth:`_expression.TextClause.columns`
2582 also provides the
2583 unique feature of **positional column targeting**, which is
2584 particularly useful when using the ORM with complex textual queries. If
2585 we specify the columns from our model to
2586 :meth:`_expression.TextClause.columns`,
2587 the result set will match to those columns positionally, meaning the
2588 name or origin of the column in the textual SQL doesn't matter::
2590 stmt = text(
2591 "SELECT users.id, addresses.id, users.id, "
2592 "users.name, addresses.email_address AS email "
2593 "FROM users JOIN addresses ON users.id=addresses.user_id "
2594 "WHERE users.id = 1"
2595 ).columns(
2596 User.id,
2597 Address.id,
2598 Address.user_id,
2599 User.name,
2600 Address.email_address,
2601 )
2603 query = (
2604 session.query(User)
2605 .from_statement(stmt)
2606 .options(contains_eager(User.addresses))
2607 )
2609 The :meth:`_expression.TextClause.columns` method provides a direct
2610 route to calling :meth:`_expression.FromClause.subquery` as well as
2611 :meth:`_expression.SelectBase.cte`
2612 against a textual SELECT statement::
2614 stmt = stmt.columns(id=Integer, name=String).cte("st")
2616 stmt = select(sometable).where(sometable.c.id == stmt.c.id)
2618 :param \*cols: A series of :class:`_expression.ColumnElement` objects,
2619 typically
2620 :class:`_schema.Column` objects from a :class:`_schema.Table`
2621 or ORM level
2622 column-mapped attributes, representing a set of columns that this
2623 textual string will SELECT from.
2625 :param \**types: A mapping of string names to :class:`.TypeEngine`
2626 type objects indicating the datatypes to use for names that are
2627 SELECTed from the textual string. Prefer to use the ``*cols``
2628 argument as it also indicates positional ordering.
2630 """
2631 selectable = util.preloaded.sql_selectable
2633 input_cols: List[NamedColumn[Any]] = [
2634 coercions.expect(roles.LabeledColumnExprRole, col) for col in cols
2635 ]
2637 positional_input_cols = [
2638 (
2639 ColumnClause(col.key, types.pop(col.key))
2640 if col.key in types
2641 else col
2642 )
2643 for col in input_cols
2644 ]
2645 keyed_input_cols: List[NamedColumn[Any]] = [
2646 ColumnClause(key, type_) for key, type_ in types.items()
2647 ]
2649 elem = selectable.TextualSelect.__new__(selectable.TextualSelect)
2650 elem._init(
2651 self,
2652 positional_input_cols + keyed_input_cols,
2653 positional=bool(positional_input_cols) and not keyed_input_cols,
2654 )
2655 return elem
2657 @property
2658 def type(self) -> TypeEngine[Any]:
2659 return type_api.NULLTYPE
2661 @property
2662 def comparator(self):
2663 # TODO: this seems wrong, it seems like we might not
2664 # be using this method.
2665 return self.type.comparator_factory(self) # type: ignore
2667 def self_group(
2668 self, against: Optional[OperatorType] = None
2669 ) -> Union[Self, Grouping[Any]]:
2670 if against is operators.in_op:
2671 return Grouping(self)
2672 else:
2673 return self
2676class Null(SingletonConstant, roles.ConstExprRole[None], ColumnElement[None]):
2677 """Represent the NULL keyword in a SQL statement.
2679 :class:`.Null` is accessed as a constant via the
2680 :func:`.null` function.
2682 """
2684 __visit_name__ = "null"
2686 _traverse_internals: _TraverseInternalsType = []
2687 _singleton: Null
2689 if not TYPE_CHECKING:
2691 @util.memoized_property
2692 def type(self) -> TypeEngine[_T]: # noqa: A001
2693 return type_api.NULLTYPE
2695 @classmethod
2696 def _instance(cls) -> Null:
2697 """Return a constant :class:`.Null` construct."""
2699 return Null._singleton
2702Null._create_singleton()
2705class False_(
2706 SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]
2707):
2708 """Represent the ``false`` keyword, or equivalent, in a SQL statement.
2710 :class:`.False_` is accessed as a constant via the
2711 :func:`.false` function.
2713 """
2715 __visit_name__ = "false"
2716 _traverse_internals: _TraverseInternalsType = []
2717 _singleton: False_
2719 if not TYPE_CHECKING:
2721 @util.memoized_property
2722 def type(self) -> TypeEngine[_T]: # noqa: A001
2723 return type_api.BOOLEANTYPE
2725 def _negate(self) -> True_:
2726 return True_._singleton
2728 @classmethod
2729 def _instance(cls) -> False_:
2730 return False_._singleton
2733False_._create_singleton()
2736class True_(SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]):
2737 """Represent the ``true`` keyword, or equivalent, in a SQL statement.
2739 :class:`.True_` is accessed as a constant via the
2740 :func:`.true` function.
2742 """
2744 __visit_name__ = "true"
2746 _traverse_internals: _TraverseInternalsType = []
2747 _singleton: True_
2749 if not TYPE_CHECKING:
2751 @util.memoized_property
2752 def type(self) -> TypeEngine[_T]: # noqa: A001
2753 return type_api.BOOLEANTYPE
2755 def _negate(self) -> False_:
2756 return False_._singleton
2758 @classmethod
2759 def _ifnone(
2760 cls, other: Optional[ColumnElement[Any]]
2761 ) -> ColumnElement[Any]:
2762 if other is None:
2763 return cls._instance()
2764 else:
2765 return other
2767 @classmethod
2768 def _instance(cls) -> True_:
2769 return True_._singleton
2772True_._create_singleton()
2775class ElementList(DQLDMLClauseElement):
2776 """Describe a list of clauses that will be space separated.
2778 This is a minimal version of :class:`.ClauseList` which is used by
2779 the :class:`.HasSyntaxExtension` class. It does not do any coercions
2780 so should be used internally only.
2782 .. versionadded:: 2.1
2784 """
2786 __visit_name__ = "element_list"
2788 _traverse_internals: _TraverseInternalsType = [
2789 ("clauses", InternalTraversal.dp_clauseelement_tuple),
2790 ]
2792 clauses: typing_Tuple[ClauseElement, ...]
2794 def __init__(self, clauses: Sequence[ClauseElement]):
2795 self.clauses = tuple(clauses)
2798class OrderByList(
2799 roles.OrderByRole,
2800 operators.OrderingOperators,
2801 DQLDMLClauseElement,
2802):
2803 """Describe a list of clauses that will be comma separated to nest
2804 within an ORDER BY.
2806 .. versionadded:: 2.1
2808 """
2810 __visit_name__ = "order_by_list"
2812 _traverse_internals: _TraverseInternalsType = [
2813 ("clauses", InternalTraversal.dp_clauseelement_tuple),
2814 ]
2816 clauses: List[ColumnElement[Any]]
2818 def __init__(
2819 self,
2820 clauses: Iterable[Union[OrderByList, _ColumnExpressionArgument[Any]]],
2821 ):
2822 text_converter_role: Type[roles.SQLRole] = roles.ByOfRole
2823 self._text_converter_role = text_converter_role
2825 self.clauses = [
2826 coercions.expect(
2827 text_converter_role, clause, apply_propagate_attrs=self
2828 )
2829 for clause in clauses
2830 ]
2832 def __iter__(self) -> Iterator[ColumnElement[Any]]:
2833 return iter(self.clauses)
2835 def __len__(self) -> int:
2836 return len(self.clauses)
2838 @property
2839 def _select_iterable(self) -> _SelectIterable:
2840 return itertools.chain.from_iterable(
2841 [elem._select_iterable for elem in self.clauses]
2842 )
2844 @util.ro_non_memoized_property
2845 def _from_objects(self) -> List[FromClause]:
2846 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
2848 def self_group(
2849 self, against: Optional[OperatorType] = None
2850 ) -> Union[Self, Grouping[Any]]:
2851 return self
2853 def desc(self) -> OrderByList:
2854 return OrderByList([e.desc() for e in self.clauses])
2856 def asc(self) -> OrderByList:
2857 return OrderByList([e.asc() for e in self.clauses])
2859 def nulls_first(self) -> OrderByList:
2860 return OrderByList([e.nulls_first() for e in self.clauses])
2862 def nulls_last(self) -> OrderByList:
2863 return OrderByList([e.nulls_last() for e in self.clauses])
2866class ClauseList(
2867 roles.InElementRole,
2868 roles.OrderByRole,
2869 roles.ColumnsClauseRole,
2870 roles.DMLColumnRole,
2871 DQLDMLClauseElement,
2872):
2873 """Describe a list of clauses, separated by an operator.
2875 By default, is comma-separated, such as a column listing.
2877 """
2879 __visit_name__ = "clauselist"
2881 # Used by ORM context.py to identify ClauseList objects in legacy
2882 # composite attribute queries (see test_query_cols_legacy test)
2883 _is_clause_list = True
2885 _traverse_internals: _TraverseInternalsType = [
2886 ("clauses", InternalTraversal.dp_clauseelement_list),
2887 ("operator", InternalTraversal.dp_operator),
2888 ]
2890 clauses: List[ColumnElement[Any]]
2892 def __init__(
2893 self,
2894 *clauses: _ColumnExpressionArgument[Any],
2895 operator: OperatorType = operators.comma_op,
2896 group: bool = True,
2897 group_contents: bool = True,
2898 _literal_as_text_role: Type[roles.SQLRole] = roles.WhereHavingRole,
2899 ):
2900 self.operator = operator
2901 self.group = group
2902 self.group_contents = group_contents
2903 clauses_iterator: Iterable[_ColumnExpressionArgument[Any]] = clauses
2904 text_converter_role: Type[roles.SQLRole] = _literal_as_text_role
2905 self._text_converter_role = text_converter_role
2907 if self.group_contents:
2908 self.clauses = [
2909 coercions.expect(
2910 text_converter_role, clause, apply_propagate_attrs=self
2911 ).self_group(against=self.operator)
2912 for clause in clauses_iterator
2913 ]
2914 else:
2915 self.clauses = [
2916 coercions.expect(
2917 text_converter_role, clause, apply_propagate_attrs=self
2918 )
2919 for clause in clauses_iterator
2920 ]
2921 self._is_implicitly_boolean = operators.is_boolean(self.operator)
2923 @classmethod
2924 def _construct_raw(
2925 cls,
2926 operator: OperatorType,
2927 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
2928 ) -> ClauseList:
2929 self = cls.__new__(cls)
2930 self.clauses = list(clauses) if clauses else []
2931 self.group = True
2932 self.operator = operator
2933 self.group_contents = True
2934 self._is_implicitly_boolean = False
2935 return self
2937 def __iter__(self) -> Iterator[ColumnElement[Any]]:
2938 return iter(self.clauses)
2940 def __len__(self) -> int:
2941 return len(self.clauses)
2943 @property
2944 def _select_iterable(self) -> _SelectIterable:
2945 return itertools.chain.from_iterable(
2946 [elem._select_iterable for elem in self.clauses]
2947 )
2949 def append(self, clause):
2950 if self.group_contents:
2951 self.clauses.append(
2952 coercions.expect(self._text_converter_role, clause).self_group(
2953 against=self.operator
2954 )
2955 )
2956 else:
2957 self.clauses.append(
2958 coercions.expect(self._text_converter_role, clause)
2959 )
2961 @util.ro_non_memoized_property
2962 def _from_objects(self) -> List[FromClause]:
2963 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
2965 def self_group(
2966 self, against: Optional[OperatorType] = None
2967 ) -> Union[Self, Grouping[Any]]:
2968 if self.group and operators.is_precedent(self.operator, against):
2969 return Grouping(self)
2970 else:
2971 return self
2974class OperatorExpression(ColumnElement[_T]):
2975 """base for expressions that contain an operator and operands
2977 .. versionadded:: 2.0
2979 """
2981 operator: OperatorType
2982 type: TypeEngine[_T]
2984 group: bool = True
2986 @property
2987 def is_comparison(self):
2988 return operators.is_comparison(self.operator)
2990 def self_group(
2991 self, against: Optional[OperatorType] = None
2992 ) -> Union[Self, Grouping[_T]]:
2993 if (
2994 self.group
2995 and operators.is_precedent(self.operator, against)
2996 or (
2997 # a negate against a non-boolean operator
2998 # doesn't make too much sense but we should
2999 # group for that
3000 against is operators.inv
3001 and not operators.is_boolean(self.operator)
3002 )
3003 ):
3004 return Grouping(self)
3005 else:
3006 return self
3008 @property
3009 def _flattened_operator_clauses(
3010 self,
3011 ) -> typing_Tuple[ColumnElement[Any], ...]:
3012 raise NotImplementedError()
3014 @classmethod
3015 def _construct_for_op(
3016 cls,
3017 left: ColumnElement[Any],
3018 right: ColumnElement[Any],
3019 op: OperatorType,
3020 *,
3021 type_: TypeEngine[_T],
3022 negate: Optional[OperatorType] = None,
3023 modifiers: Optional[Mapping[str, Any]] = None,
3024 ) -> OperatorExpression[_T]:
3025 if operators.is_associative(op):
3026 assert (
3027 negate is None
3028 ), f"negate not supported for associative operator {op}"
3030 multi = False
3031 if getattr(
3032 left, "operator", None
3033 ) is op and type_._compare_type_affinity(left.type):
3034 multi = True
3035 left_flattened = left._flattened_operator_clauses
3036 else:
3037 left_flattened = (left,)
3039 if getattr(
3040 right, "operator", None
3041 ) is op and type_._compare_type_affinity(right.type):
3042 multi = True
3043 right_flattened = right._flattened_operator_clauses
3044 else:
3045 right_flattened = (right,)
3047 if multi:
3048 return ExpressionClauseList._construct_for_list(
3049 op,
3050 type_,
3051 *(left_flattened + right_flattened),
3052 )
3054 if right._is_collection_aggregate:
3055 negate = None
3057 return BinaryExpression(
3058 left, right, op, type_=type_, negate=negate, modifiers=modifiers
3059 )
3062class ExpressionClauseList(OperatorExpression[_T]):
3063 """Describe a list of clauses, separated by an operator,
3064 in a column expression context.
3066 :class:`.ExpressionClauseList` differs from :class:`.ClauseList` in that
3067 it represents a column-oriented DQL expression only, not an open ended
3068 list of anything comma separated.
3070 .. versionadded:: 2.0
3072 """
3074 __visit_name__ = "expression_clauselist"
3076 _traverse_internals: _TraverseInternalsType = [
3077 ("clauses", InternalTraversal.dp_clauseelement_tuple),
3078 ("operator", InternalTraversal.dp_operator),
3079 ]
3081 clauses: typing_Tuple[ColumnElement[Any], ...]
3083 group: bool
3085 def __init__(
3086 self,
3087 operator: OperatorType,
3088 *clauses: _ColumnExpressionArgument[Any],
3089 type_: Optional[_TypeEngineArgument[_T]] = None,
3090 ):
3091 self.operator = operator
3093 self.clauses = tuple(
3094 coercions.expect(
3095 roles.ExpressionElementRole, clause, apply_propagate_attrs=self
3096 )
3097 for clause in clauses
3098 )
3099 self._is_implicitly_boolean = operators.is_boolean(self.operator)
3100 self.type = type_api.to_instance(type_) # type: ignore
3102 @property
3103 def _flattened_operator_clauses(
3104 self,
3105 ) -> typing_Tuple[ColumnElement[Any], ...]:
3106 return self.clauses
3108 def __iter__(self) -> Iterator[ColumnElement[Any]]:
3109 return iter(self.clauses)
3111 def __len__(self) -> int:
3112 return len(self.clauses)
3114 @property
3115 def _select_iterable(self) -> _SelectIterable:
3116 return (self,)
3118 @util.ro_non_memoized_property
3119 def _from_objects(self) -> List[FromClause]:
3120 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
3122 def _append_inplace(self, clause: ColumnElement[Any]) -> None:
3123 self.clauses += (clause,)
3125 @classmethod
3126 def _construct_for_list(
3127 cls,
3128 operator: OperatorType,
3129 type_: TypeEngine[_T],
3130 *clauses: ColumnElement[Any],
3131 group: bool = True,
3132 ) -> ExpressionClauseList[_T]:
3133 self = cls.__new__(cls)
3134 self.group = group
3135 if group:
3136 self.clauses = tuple(
3137 c.self_group(against=operator) for c in clauses
3138 )
3139 else:
3140 self.clauses = clauses
3141 self.operator = operator
3142 self.type = type_
3143 for c in clauses:
3144 if c._propagate_attrs:
3145 self._propagate_attrs = c._propagate_attrs
3146 break
3147 return self
3149 def _negate(self) -> Any:
3150 grouped = self.self_group(against=operators.inv)
3151 assert isinstance(grouped, ColumnElement)
3152 return UnaryExpression(grouped, operator=operators.inv)
3155class BooleanClauseList(ExpressionClauseList[bool]):
3156 __visit_name__ = "expression_clauselist"
3157 inherit_cache = True
3159 def __init__(self, *arg, **kw):
3160 raise NotImplementedError(
3161 "BooleanClauseList has a private constructor"
3162 )
3164 @classmethod
3165 def _process_clauses_for_boolean(
3166 cls,
3167 operator: OperatorType,
3168 continue_on: Any,
3169 skip_on: Any,
3170 clauses: Iterable[ColumnElement[Any]],
3171 ) -> typing_Tuple[int, List[ColumnElement[Any]]]:
3172 has_continue_on = None
3174 convert_clauses = []
3176 against = operators._asbool
3177 lcc = 0
3179 for clause in clauses:
3180 if clause is continue_on:
3181 # instance of continue_on, like and_(x, y, True, z), store it
3182 # if we didn't find one already, we will use it if there
3183 # are no other expressions here.
3184 has_continue_on = clause
3185 elif clause is skip_on:
3186 # instance of skip_on, e.g. and_(x, y, False, z), cancels
3187 # the rest out
3188 convert_clauses = [clause]
3189 lcc = 1
3190 break
3191 else:
3192 if not lcc:
3193 lcc = 1
3194 else:
3195 against = operator
3196 # technically this would be len(convert_clauses) + 1
3197 # however this only needs to indicate "greater than one"
3198 lcc = 2
3199 convert_clauses.append(clause)
3201 if not convert_clauses and has_continue_on is not None:
3202 convert_clauses = [has_continue_on]
3203 lcc = 1
3205 return lcc, [c.self_group(against=against) for c in convert_clauses]
3207 @classmethod
3208 def _construct(
3209 cls,
3210 operator: OperatorType,
3211 continue_on: Any,
3212 skip_on: Any,
3213 initial_clause: Any = _NoArg.NO_ARG,
3214 *clauses: Any,
3215 **kw: Any,
3216 ) -> ColumnElement[Any]:
3217 if initial_clause is _NoArg.NO_ARG:
3218 # no elements period. deprecated use case. return an empty
3219 # ClauseList construct that generates nothing unless it has
3220 # elements added to it.
3221 name = operator.__name__
3223 util.warn_deprecated(
3224 f"Invoking {name}() without arguments is deprecated, and "
3225 f"will be disallowed in a future release. For an empty "
3226 f"""{name}() construct, use '{name}({
3227 'true()' if continue_on is True_._singleton else 'false()'
3228 }, *args)' """
3229 f"""or '{name}({
3230 'True' if continue_on is True_._singleton else 'False'
3231 }, *args)'.""",
3232 version="1.4",
3233 )
3234 return cls._construct_raw(operator)
3236 lcc, convert_clauses = cls._process_clauses_for_boolean(
3237 operator,
3238 continue_on,
3239 skip_on,
3240 [
3241 coercions.expect(roles.WhereHavingRole, clause)
3242 for clause in util.coerce_generator_arg(
3243 (initial_clause,) + clauses
3244 )
3245 ],
3246 )
3248 if lcc > 1:
3249 # multiple elements. Return regular BooleanClauseList
3250 # which will link elements against the operator.
3252 flattened_clauses = itertools.chain.from_iterable(
3253 (
3254 (c for c in to_flat._flattened_operator_clauses)
3255 if getattr(to_flat, "operator", None) is operator
3256 else (to_flat,)
3257 )
3258 for to_flat in convert_clauses
3259 )
3261 return cls._construct_raw(operator, flattened_clauses) # type: ignore # noqa: E501
3262 else:
3263 assert lcc
3264 # just one element. return it as a single boolean element,
3265 # not a list and discard the operator.
3266 return convert_clauses[0]
3268 @classmethod
3269 def _construct_for_whereclause(
3270 cls, clauses: Iterable[ColumnElement[Any]]
3271 ) -> Optional[ColumnElement[bool]]:
3272 operator, continue_on, skip_on = (
3273 operators.and_,
3274 True_._singleton,
3275 False_._singleton,
3276 )
3278 lcc, convert_clauses = cls._process_clauses_for_boolean(
3279 operator,
3280 continue_on,
3281 skip_on,
3282 clauses, # these are assumed to be coerced already
3283 )
3285 if lcc > 1:
3286 # multiple elements. Return regular BooleanClauseList
3287 # which will link elements against the operator.
3288 return cls._construct_raw(operator, convert_clauses)
3289 elif lcc == 1:
3290 # just one element. return it as a single boolean element,
3291 # not a list and discard the operator.
3292 return convert_clauses[0]
3293 else:
3294 return None
3296 @classmethod
3297 def _construct_raw(
3298 cls,
3299 operator: OperatorType,
3300 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
3301 ) -> BooleanClauseList:
3302 self = cls.__new__(cls)
3303 self.clauses = tuple(clauses) if clauses else ()
3304 self.group = True
3305 self.operator = operator
3306 self.type = type_api.BOOLEANTYPE
3307 self._is_implicitly_boolean = True
3308 return self
3310 @classmethod
3311 def and_(
3312 cls,
3313 initial_clause: Union[
3314 Literal[True], _ColumnExpressionArgument[bool], _NoArg
3315 ] = _NoArg.NO_ARG,
3316 *clauses: _ColumnExpressionArgument[bool],
3317 ) -> ColumnElement[bool]:
3318 r"""Produce a conjunction of expressions joined by ``AND``.
3320 See :func:`_sql.and_` for full documentation.
3321 """
3322 return cls._construct(
3323 operators.and_,
3324 True_._singleton,
3325 False_._singleton,
3326 initial_clause,
3327 *clauses,
3328 )
3330 @classmethod
3331 def or_(
3332 cls,
3333 initial_clause: Union[
3334 Literal[False], _ColumnExpressionArgument[bool], _NoArg
3335 ] = _NoArg.NO_ARG,
3336 *clauses: _ColumnExpressionArgument[bool],
3337 ) -> ColumnElement[bool]:
3338 """Produce a conjunction of expressions joined by ``OR``.
3340 See :func:`_sql.or_` for full documentation.
3341 """
3342 return cls._construct(
3343 operators.or_,
3344 False_._singleton,
3345 True_._singleton,
3346 initial_clause,
3347 *clauses,
3348 )
3350 @property
3351 def _select_iterable(self) -> _SelectIterable:
3352 return (self,)
3354 def self_group(
3355 self, against: Optional[OperatorType] = None
3356 ) -> Union[Self, Grouping[bool]]:
3357 if not self.clauses:
3358 return self
3359 else:
3360 return super().self_group(against=against)
3363and_ = BooleanClauseList.and_
3364or_ = BooleanClauseList.or_
3367class Tuple(ClauseList, ColumnElement[TupleAny]):
3368 """Represent a SQL tuple."""
3370 __visit_name__ = "tuple"
3372 _traverse_internals: _TraverseInternalsType = (
3373 ClauseList._traverse_internals + []
3374 )
3376 type: TupleType
3378 @util.preload_module("sqlalchemy.sql.sqltypes")
3379 def __init__(
3380 self,
3381 *clauses: _ColumnExpressionArgument[Any],
3382 types: Optional[Sequence[_TypeEngineArgument[Any]]] = None,
3383 ):
3384 sqltypes = util.preloaded.sql_sqltypes
3386 if types is None:
3387 init_clauses: List[ColumnElement[Any]] = [
3388 coercions.expect(roles.ExpressionElementRole, c)
3389 for c in clauses
3390 ]
3391 else:
3392 if len(types) != len(clauses):
3393 raise exc.ArgumentError(
3394 "Wrong number of elements for %d-tuple: %r "
3395 % (len(types), clauses)
3396 )
3397 init_clauses = [
3398 coercions.expect(
3399 roles.ExpressionElementRole,
3400 c,
3401 type_=typ if not typ._isnull else None,
3402 )
3403 for typ, c in zip(types, clauses)
3404 ]
3406 self.type = sqltypes.TupleType(*[arg.type for arg in init_clauses])
3407 super().__init__(*init_clauses)
3409 @property
3410 def _select_iterable(self) -> _SelectIterable:
3411 return (self,)
3413 def _bind_param(self, operator, obj, type_=None, expanding=False):
3414 if expanding:
3415 return BindParameter(
3416 None,
3417 value=obj,
3418 _compared_to_operator=operator,
3419 unique=True,
3420 expanding=True,
3421 type_=type_,
3422 _compared_to_type=self.type,
3423 )
3424 else:
3425 return Tuple(
3426 *[
3427 BindParameter(
3428 None,
3429 o,
3430 _compared_to_operator=operator,
3431 _compared_to_type=compared_to_type,
3432 unique=True,
3433 type_=type_,
3434 )
3435 for o, compared_to_type in zip(obj, self.type.types)
3436 ]
3437 )
3439 def self_group(self, against: Optional[OperatorType] = None) -> Self:
3440 # Tuple is parenthesized by definition.
3441 return self
3444class Case(ColumnElement[_T]):
3445 """Represent a ``CASE`` expression.
3447 :class:`.Case` is produced using the :func:`.case` factory function,
3448 as in::
3450 from sqlalchemy import case
3452 stmt = select(users_table).where(
3453 case(
3454 (users_table.c.name == "wendy", "W"),
3455 (users_table.c.name == "jack", "J"),
3456 else_="E",
3457 )
3458 )
3460 Details on :class:`.Case` usage is at :func:`.case`.
3462 .. seealso::
3464 :func:`.case`
3466 """
3468 __visit_name__ = "case"
3470 _traverse_internals: _TraverseInternalsType = [
3471 ("value", InternalTraversal.dp_clauseelement),
3472 ("whens", InternalTraversal.dp_clauseelement_tuples),
3473 ("else_", InternalTraversal.dp_clauseelement),
3474 ]
3476 # for case(), the type is derived from the whens. so for the moment
3477 # users would have to cast() the case to get a specific type
3479 whens: List[typing_Tuple[ColumnElement[bool], ColumnElement[_T]]]
3480 else_: Optional[ColumnElement[_T]]
3481 value: Optional[ColumnElement[Any]]
3483 def __init__(
3484 self,
3485 *whens: Union[
3486 typing_Tuple[_ColumnExpressionArgument[bool], Any],
3487 Mapping[Any, Any],
3488 ],
3489 value: Optional[Any] = None,
3490 else_: Optional[Any] = None,
3491 ):
3492 new_whens: Iterable[Any] = coercions._expression_collection_was_a_list(
3493 "whens", "case", whens
3494 )
3495 try:
3496 new_whens = util.dictlike_iteritems(new_whens)
3497 except TypeError:
3498 pass
3500 self.whens = [
3501 (
3502 coercions.expect(
3503 roles.ExpressionElementRole,
3504 c,
3505 apply_propagate_attrs=self,
3506 ).self_group(),
3507 coercions.expect(roles.ExpressionElementRole, r),
3508 )
3509 for (c, r) in new_whens
3510 ]
3512 if value is None:
3513 self.value = None
3514 else:
3515 self.value = coercions.expect(roles.ExpressionElementRole, value)
3517 if else_ is not None:
3518 self.else_ = coercions.expect(roles.ExpressionElementRole, else_)
3519 else:
3520 self.else_ = None
3522 type_ = next(
3523 (
3524 then.type
3525 # Iterate `whens` in reverse to match previous behaviour
3526 # where type of final element took priority
3527 for *_, then in reversed(self.whens)
3528 if not then.type._isnull
3529 ),
3530 self.else_.type if self.else_ is not None else type_api.NULLTYPE,
3531 )
3532 self.type = cast(_T, type_)
3534 @util.ro_non_memoized_property
3535 def _from_objects(self) -> List[FromClause]:
3536 return list(
3537 itertools.chain(*[x._from_objects for x in self.get_children()])
3538 )
3541class Cast(WrapsColumnExpression[_T]):
3542 """Represent a ``CAST`` expression.
3544 :class:`.Cast` is produced using the :func:`.cast` factory function,
3545 as in::
3547 from sqlalchemy import cast, Numeric
3549 stmt = select(cast(product_table.c.unit_price, Numeric(10, 4)))
3551 Details on :class:`.Cast` usage is at :func:`.cast`.
3553 .. seealso::
3555 :ref:`tutorial_casts`
3557 :func:`.cast`
3559 :func:`.try_cast`
3561 :func:`.type_coerce` - an alternative to CAST that coerces the type
3562 on the Python side only, which is often sufficient to generate the
3563 correct SQL and data coercion.
3565 """
3567 __visit_name__ = "cast"
3569 _traverse_internals: _TraverseInternalsType = [
3570 ("clause", InternalTraversal.dp_clauseelement),
3571 ("type", InternalTraversal.dp_type),
3572 ]
3574 clause: ColumnElement[Any]
3575 type: TypeEngine[_T]
3576 typeclause: TypeClause
3578 def __init__(
3579 self,
3580 expression: _ColumnExpressionArgument[Any],
3581 type_: _TypeEngineArgument[_T],
3582 ):
3583 self.type = type_api.to_instance(type_)
3584 self.clause = coercions.expect(
3585 roles.ExpressionElementRole,
3586 expression,
3587 type_=self.type,
3588 apply_propagate_attrs=self,
3589 )
3590 self.typeclause = TypeClause(self.type)
3592 @util.ro_non_memoized_property
3593 def _from_objects(self) -> List[FromClause]:
3594 return self.clause._from_objects
3596 @property
3597 def wrapped_column_expression(self):
3598 return self.clause
3601class TryCast(Cast[_T]):
3602 """Represent a TRY_CAST expression.
3604 Details on :class:`.TryCast` usage is at :func:`.try_cast`.
3606 .. seealso::
3608 :func:`.try_cast`
3610 :ref:`tutorial_casts`
3611 """
3613 __visit_name__ = "try_cast"
3614 inherit_cache = True
3617class TypeCoerce(WrapsColumnExpression[_T]):
3618 """Represent a Python-side type-coercion wrapper.
3620 :class:`.TypeCoerce` supplies the :func:`_expression.type_coerce`
3621 function; see that function for usage details.
3623 .. seealso::
3625 :func:`_expression.type_coerce`
3627 :func:`.cast`
3629 """
3631 __visit_name__ = "type_coerce"
3633 _traverse_internals: _TraverseInternalsType = [
3634 ("clause", InternalTraversal.dp_clauseelement),
3635 ("type", InternalTraversal.dp_type),
3636 ]
3638 clause: ColumnElement[Any]
3639 type: TypeEngine[_T]
3641 def __init__(
3642 self,
3643 expression: _ColumnExpressionArgument[Any],
3644 type_: _TypeEngineArgument[_T],
3645 ):
3646 self.type = type_api.to_instance(type_)
3647 self.clause = coercions.expect(
3648 roles.ExpressionElementRole,
3649 expression,
3650 type_=self.type,
3651 apply_propagate_attrs=self,
3652 )
3654 @util.ro_non_memoized_property
3655 def _from_objects(self) -> List[FromClause]:
3656 return self.clause._from_objects
3658 @HasMemoized.memoized_attribute
3659 def typed_expression(self):
3660 if isinstance(self.clause, BindParameter):
3661 bp = self.clause._clone()
3662 bp.type = self.type
3663 return bp
3664 else:
3665 return self.clause
3667 @property
3668 def wrapped_column_expression(self):
3669 return self.clause
3671 def self_group(
3672 self, against: Optional[OperatorType] = None
3673 ) -> TypeCoerce[_T]:
3674 grouped = self.clause.self_group(against=against)
3675 if grouped is not self.clause:
3676 return TypeCoerce(grouped, self.type)
3677 else:
3678 return self
3681class Extract(ColumnElement[int]):
3682 """Represent a SQL EXTRACT clause, ``extract(field FROM expr)``."""
3684 __visit_name__ = "extract"
3686 _traverse_internals: _TraverseInternalsType = [
3687 ("expr", InternalTraversal.dp_clauseelement),
3688 ("field", InternalTraversal.dp_string),
3689 ]
3691 expr: ColumnElement[Any]
3692 field: str
3694 def __init__(self, field: str, expr: _ColumnExpressionArgument[Any]):
3695 self.type = type_api.INTEGERTYPE
3696 self.field = field
3697 self.expr = coercions.expect(roles.ExpressionElementRole, expr)
3699 @util.ro_non_memoized_property
3700 def _from_objects(self) -> List[FromClause]:
3701 return self.expr._from_objects
3704class _label_reference(ColumnElement[_T]):
3705 """Wrap a column expression as it appears in a 'reference' context.
3707 This expression is any that includes an _order_by_label_element,
3708 which is a Label, or a DESC / ASC construct wrapping a Label.
3710 The production of _label_reference() should occur when an expression
3711 is added to this context; this includes the ORDER BY or GROUP BY of a
3712 SELECT statement, as well as a few other places, such as the ORDER BY
3713 within an OVER clause.
3715 """
3717 __visit_name__ = "label_reference"
3719 _traverse_internals: _TraverseInternalsType = [
3720 ("element", InternalTraversal.dp_clauseelement)
3721 ]
3723 element: ColumnElement[_T]
3725 def __init__(self, element: ColumnElement[_T]):
3726 self.element = element
3727 self._propagate_attrs = element._propagate_attrs
3729 @util.ro_non_memoized_property
3730 def _from_objects(self) -> List[FromClause]:
3731 return []
3734class _textual_label_reference(ColumnElement[Any]):
3735 __visit_name__ = "textual_label_reference"
3737 _traverse_internals: _TraverseInternalsType = [
3738 ("element", InternalTraversal.dp_string)
3739 ]
3741 def __init__(self, element: str):
3742 self.element = element
3744 @util.memoized_property
3745 def _text_clause(self) -> TextClause:
3746 return TextClause(self.element)
3749class UnaryExpression(ColumnElement[_T]):
3750 """Define a 'unary' expression.
3752 A unary expression has a single column expression
3753 and an operator. The operator can be placed on the left
3754 (where it is called the 'operator') or right (where it is called the
3755 'modifier') of the column expression.
3757 :class:`.UnaryExpression` is the basis for several unary operators
3758 including those used by :func:`.desc`, :func:`.asc`, :func:`.distinct`,
3759 :func:`.nulls_first` and :func:`.nulls_last`.
3761 """
3763 __visit_name__ = "unary"
3765 _traverse_internals: _TraverseInternalsType = [
3766 ("element", InternalTraversal.dp_clauseelement),
3767 ("operator", InternalTraversal.dp_operator),
3768 ("modifier", InternalTraversal.dp_operator),
3769 ]
3771 element: ColumnElement[Any]
3772 operator: Optional[OperatorType]
3773 modifier: Optional[OperatorType]
3775 def __init__(
3776 self,
3777 element: ColumnElement[Any],
3778 *,
3779 operator: Optional[OperatorType] = None,
3780 modifier: Optional[OperatorType] = None,
3781 type_: Optional[_TypeEngineArgument[_T]] = None,
3782 wraps_column_expression: bool = False, # legacy, not used as of 2.0.42
3783 ):
3784 self.operator = operator
3785 self.modifier = modifier
3786 self._propagate_attrs = element._propagate_attrs
3787 self.element = element.self_group(
3788 against=self.operator or self.modifier
3789 )
3791 # if type is None, we get NULLTYPE, which is our _T. But I don't
3792 # know how to get the overloads to express that correctly
3793 self.type = type_api.to_instance(type_) # type: ignore
3795 def _wraps_unnamed_column(self):
3796 ungrouped = self.element._ungroup()
3797 return (
3798 not isinstance(ungrouped, NamedColumn)
3799 or ungrouped._non_anon_label is None
3800 )
3802 @classmethod
3803 def _create_nulls_first(
3804 cls,
3805 column: _ColumnExpressionArgument[_T],
3806 ) -> UnaryExpression[_T]:
3807 return UnaryExpression(
3808 coercions.expect(roles.ByOfRole, column),
3809 modifier=operators.nulls_first_op,
3810 )
3812 @classmethod
3813 def _create_nulls_last(
3814 cls,
3815 column: _ColumnExpressionArgument[_T],
3816 ) -> UnaryExpression[_T]:
3817 return UnaryExpression(
3818 coercions.expect(roles.ByOfRole, column),
3819 modifier=operators.nulls_last_op,
3820 )
3822 @classmethod
3823 def _create_desc(
3824 cls, column: _ColumnExpressionOrStrLabelArgument[_T]
3825 ) -> UnaryExpression[_T]:
3827 return UnaryExpression(
3828 coercions.expect(roles.ByOfRole, column),
3829 modifier=operators.desc_op,
3830 )
3832 @classmethod
3833 def _create_asc(
3834 cls,
3835 column: _ColumnExpressionOrStrLabelArgument[_T],
3836 ) -> UnaryExpression[_T]:
3837 return UnaryExpression(
3838 coercions.expect(roles.ByOfRole, column),
3839 modifier=operators.asc_op,
3840 )
3842 @classmethod
3843 def _create_distinct(
3844 cls,
3845 expr: _ColumnExpressionArgument[_T],
3846 ) -> UnaryExpression[_T]:
3847 col_expr: ColumnElement[_T] = coercions.expect(
3848 roles.ExpressionElementRole, expr
3849 )
3850 return UnaryExpression(
3851 col_expr,
3852 operator=operators.distinct_op,
3853 type_=col_expr.type,
3854 )
3856 @classmethod
3857 def _create_bitwise_not(
3858 cls,
3859 expr: _ColumnExpressionArgument[_T],
3860 ) -> UnaryExpression[_T]:
3861 col_expr: ColumnElement[_T] = coercions.expect(
3862 roles.ExpressionElementRole, expr
3863 )
3864 return UnaryExpression(
3865 col_expr,
3866 operator=operators.bitwise_not_op,
3867 type_=col_expr.type,
3868 )
3870 @property
3871 def _order_by_label_element(self) -> Optional[Label[Any]]:
3872 if operators.is_order_by_modifier(self.modifier):
3873 return self.element._order_by_label_element
3874 else:
3875 return None
3877 @util.ro_non_memoized_property
3878 def _from_objects(self) -> List[FromClause]:
3879 return self.element._from_objects
3881 def _negate(self) -> ColumnElement[Any]:
3882 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
3883 return UnaryExpression(
3884 self.self_group(against=operators.inv),
3885 operator=operators.inv,
3886 type_=type_api.BOOLEANTYPE,
3887 )
3888 else:
3889 return ColumnElement._negate(self)
3891 def self_group(
3892 self, against: Optional[OperatorType] = None
3893 ) -> Union[Self, Grouping[_T]]:
3894 if self.operator and operators.is_precedent(self.operator, against):
3895 return Grouping(self)
3896 else:
3897 return self
3900class CollectionAggregate(UnaryExpression[_T]):
3901 """Forms the basis for right-hand collection operator modifiers
3902 ANY and ALL.
3904 The ANY and ALL keywords are available in different ways on different
3905 backends. On PostgreSQL, they only work for an ARRAY type. On
3906 MySQL, they only work for subqueries.
3908 """
3910 inherit_cache = True
3911 _is_collection_aggregate = True
3913 @classmethod
3914 def _create_any(
3915 cls, expr: _ColumnExpressionArgument[_T]
3916 ) -> CollectionAggregate[bool]:
3917 col_expr: ColumnElement[_T] = coercions.expect(
3918 roles.ExpressionElementRole,
3919 expr,
3920 )
3921 col_expr = col_expr.self_group()
3922 return CollectionAggregate(
3923 col_expr,
3924 operator=operators.any_op,
3925 type_=type_api.BOOLEANTYPE,
3926 )
3928 @classmethod
3929 def _create_all(
3930 cls, expr: _ColumnExpressionArgument[_T]
3931 ) -> CollectionAggregate[bool]:
3932 col_expr: ColumnElement[_T] = coercions.expect(
3933 roles.ExpressionElementRole,
3934 expr,
3935 )
3936 col_expr = col_expr.self_group()
3937 return CollectionAggregate(
3938 col_expr,
3939 operator=operators.all_op,
3940 type_=type_api.BOOLEANTYPE,
3941 )
3943 # operate and reverse_operate are hardwired to
3944 # dispatch onto the type comparator directly, so that we can
3945 # ensure "reversed" behavior.
3946 def operate(
3947 self, op: OperatorType, *other: Any, **kwargs: Any
3948 ) -> ColumnElement[_T]:
3949 if not operators.is_comparison(op):
3950 raise exc.ArgumentError(
3951 "Only comparison operators may be used with ANY/ALL"
3952 )
3953 kwargs["reverse"] = True
3954 return self.comparator.operate(operators.mirror(op), *other, **kwargs)
3956 def reverse_operate(
3957 self, op: OperatorType, other: Any, **kwargs: Any
3958 ) -> ColumnElement[_T]:
3959 # comparison operators should never call reverse_operate
3960 assert not operators.is_comparison(op)
3961 raise exc.ArgumentError(
3962 "Only comparison operators may be used with ANY/ALL"
3963 )
3966class AsBoolean(WrapsColumnExpression[bool], UnaryExpression[bool]):
3967 inherit_cache = True
3969 def __init__(self, element, operator, negate):
3970 self.element = element
3971 self.type = type_api.BOOLEANTYPE
3972 self.operator = operator
3973 self.negate = negate
3974 self.modifier = None
3975 self._is_implicitly_boolean = element._is_implicitly_boolean
3977 @property
3978 def wrapped_column_expression(self):
3979 return self.element
3981 def self_group(self, against: Optional[OperatorType] = None) -> Self:
3982 return self
3984 def _negate(self):
3985 if isinstance(self.element, (True_, False_)):
3986 return self.element._negate()
3987 else:
3988 return AsBoolean(self.element, self.negate, self.operator)
3991class BinaryExpression(OperatorExpression[_T]):
3992 """Represent an expression that is ``LEFT <operator> RIGHT``.
3994 A :class:`.BinaryExpression` is generated automatically
3995 whenever two column expressions are used in a Python binary expression:
3997 .. sourcecode:: pycon+sql
3999 >>> from sqlalchemy.sql import column
4000 >>> column("a") + column("b")
4001 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
4002 >>> print(column("a") + column("b"))
4003 {printsql}a + b
4005 """
4007 __visit_name__ = "binary"
4009 _traverse_internals: _TraverseInternalsType = [
4010 ("left", InternalTraversal.dp_clauseelement),
4011 ("right", InternalTraversal.dp_clauseelement),
4012 ("operator", InternalTraversal.dp_operator),
4013 ("negate", InternalTraversal.dp_operator),
4014 ("modifiers", InternalTraversal.dp_plain_dict),
4015 (
4016 "type",
4017 InternalTraversal.dp_type,
4018 ),
4019 ]
4021 _cache_key_traversal = [
4022 ("left", InternalTraversal.dp_clauseelement),
4023 ("right", InternalTraversal.dp_clauseelement),
4024 ("operator", InternalTraversal.dp_operator),
4025 ("modifiers", InternalTraversal.dp_plain_dict),
4026 # "type" affects JSON CAST operators, so while redundant in most cases,
4027 # is needed for that one
4028 (
4029 "type",
4030 InternalTraversal.dp_type,
4031 ),
4032 ]
4034 _is_implicitly_boolean = True
4035 """Indicates that any database will know this is a boolean expression
4036 even if the database does not have an explicit boolean datatype.
4038 """
4040 left: ColumnElement[Any]
4041 right: ColumnElement[Any]
4042 modifiers: Mapping[str, Any]
4044 def __init__(
4045 self,
4046 left: ColumnElement[Any],
4047 right: ColumnElement[Any],
4048 operator: OperatorType,
4049 type_: Optional[_TypeEngineArgument[_T]] = None,
4050 negate: Optional[OperatorType] = None,
4051 modifiers: Optional[Mapping[str, Any]] = None,
4052 ):
4053 # allow compatibility with libraries that
4054 # refer to BinaryExpression directly and pass strings
4055 if isinstance(operator, str):
4056 operator = operators.custom_op(operator)
4057 self._orig = (left.__hash__(), right.__hash__())
4058 self._propagate_attrs = left._propagate_attrs or right._propagate_attrs
4059 self.left = left.self_group(against=operator)
4060 self.right = right.self_group(against=operator)
4061 self.operator = operator
4063 # if type is None, we get NULLTYPE, which is our _T. But I don't
4064 # know how to get the overloads to express that correctly
4065 self.type = type_api.to_instance(type_) # type: ignore
4067 self.negate = negate
4068 self._is_implicitly_boolean = operators.is_boolean(operator)
4070 if modifiers is None:
4071 self.modifiers = {}
4072 else:
4073 self.modifiers = modifiers
4075 @property
4076 def _flattened_operator_clauses(
4077 self,
4078 ) -> typing_Tuple[ColumnElement[Any], ...]:
4079 return (self.left, self.right)
4081 def __bool__(self):
4082 """Implement Python-side "bool" for BinaryExpression as a
4083 simple "identity" check for the left and right attributes,
4084 if the operator is "eq" or "ne". Otherwise the expression
4085 continues to not support "bool" like all other column expressions.
4087 The rationale here is so that ColumnElement objects can be hashable.
4088 What? Well, suppose you do this::
4090 c1, c2 = column("x"), column("y")
4091 s1 = set([c1, c2])
4093 We do that **a lot**, columns inside of sets is an extremely basic
4094 thing all over the ORM for example.
4096 So what happens if we do this? ::
4098 c1 in s1
4100 Hashing means it will normally use ``__hash__()`` of the object,
4101 but in case of hash collision, it's going to also do ``c1 == c1``
4102 and/or ``c1 == c2`` inside. Those operations need to return a
4103 True/False value. But because we override ``==`` and ``!=``, they're
4104 going to get a BinaryExpression. Hence we implement ``__bool__`` here
4105 so that these comparisons behave in this particular context mostly
4106 like regular object comparisons. Thankfully Python is OK with
4107 that! Otherwise we'd have to use special set classes for columns
4108 (which we used to do, decades ago).
4110 """
4111 if self.operator in (operators.eq, operators.ne):
4112 # this is using the eq/ne operator given int hash values,
4113 # rather than Operator, so that "bool" can be based on
4114 # identity
4115 return self.operator(*self._orig) # type: ignore
4116 else:
4117 raise TypeError("Boolean value of this clause is not defined")
4119 if typing.TYPE_CHECKING:
4121 def __invert__(
4122 self: BinaryExpression[_T],
4123 ) -> BinaryExpression[_T]: ...
4125 @util.ro_non_memoized_property
4126 def _from_objects(self) -> List[FromClause]:
4127 return self.left._from_objects + self.right._from_objects
4129 def _negate(self):
4130 if self.negate is not None:
4131 return BinaryExpression(
4132 self.left,
4133 self.right._negate_in_binary(self.negate, self.operator),
4134 self.negate,
4135 negate=self.operator,
4136 type_=self.type,
4137 modifiers=self.modifiers,
4138 )
4139 else:
4140 return self.self_group()._negate()
4143class Slice(ColumnElement[Any]):
4144 """Represent SQL for a Python array-slice object.
4146 This is not a specific SQL construct at this level, but
4147 may be interpreted by specific dialects, e.g. PostgreSQL.
4149 """
4151 __visit_name__ = "slice"
4153 _traverse_internals: _TraverseInternalsType = [
4154 ("start", InternalTraversal.dp_clauseelement),
4155 ("stop", InternalTraversal.dp_clauseelement),
4156 ("step", InternalTraversal.dp_clauseelement),
4157 ]
4159 def __init__(self, start, stop, step, _name=None):
4160 self.start = coercions.expect(
4161 roles.ExpressionElementRole,
4162 start,
4163 name=_name,
4164 type_=type_api.INTEGERTYPE,
4165 )
4166 self.stop = coercions.expect(
4167 roles.ExpressionElementRole,
4168 stop,
4169 name=_name,
4170 type_=type_api.INTEGERTYPE,
4171 )
4172 self.step = coercions.expect(
4173 roles.ExpressionElementRole,
4174 step,
4175 name=_name,
4176 type_=type_api.INTEGERTYPE,
4177 )
4178 self.type = type_api.NULLTYPE
4180 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4181 assert against is operator.getitem
4182 return self
4185class IndexExpression(BinaryExpression[Any]):
4186 """Represent the class of expressions that are like an "index"
4187 operation."""
4189 inherit_cache = True
4192class GroupedElement(DQLDMLClauseElement):
4193 """Represent any parenthesized expression"""
4195 __visit_name__ = "grouping"
4197 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4198 return self
4200 def _ungroup(self) -> ClauseElement:
4201 raise NotImplementedError()
4204class Grouping(GroupedElement, ColumnElement[_T]):
4205 """Represent a grouping within a column expression"""
4207 _traverse_internals: _TraverseInternalsType = [
4208 ("element", InternalTraversal.dp_clauseelement),
4209 ("type", InternalTraversal.dp_type),
4210 ]
4212 _cache_key_traversal = [
4213 ("element", InternalTraversal.dp_clauseelement),
4214 ]
4216 element: Union[TextClause, ClauseList, ColumnElement[_T]]
4218 def __init__(
4219 self, element: Union[TextClause, ClauseList, ColumnElement[_T]]
4220 ):
4221 self.element = element
4223 # nulltype assignment issue
4224 self.type = getattr(element, "type", type_api.NULLTYPE) # type: ignore
4225 self._propagate_attrs = element._propagate_attrs
4227 def _with_binary_element_type(self, type_):
4228 return self.__class__(self.element._with_binary_element_type(type_))
4230 def _ungroup(self) -> ColumnElement[_T]:
4231 assert isinstance(self.element, ColumnElement)
4232 return self.element._ungroup()
4234 @util.memoized_property
4235 def _is_implicitly_boolean(self):
4236 return self.element._is_implicitly_boolean
4238 @util.non_memoized_property
4239 def _tq_label(self) -> Optional[str]:
4240 return (
4241 getattr(self.element, "_tq_label", None) or self._anon_name_label
4242 )
4244 @util.non_memoized_property
4245 def _proxies(self) -> List[ColumnElement[Any]]:
4246 if isinstance(self.element, ColumnElement):
4247 return [self.element]
4248 else:
4249 return []
4251 @util.ro_non_memoized_property
4252 def _from_objects(self) -> List[FromClause]:
4253 return self.element._from_objects
4255 def __getattr__(self, attr):
4256 return getattr(self.element, attr)
4258 def __getstate__(self):
4259 return {"element": self.element, "type": self.type}
4261 def __setstate__(self, state):
4262 self.element = state["element"]
4263 self.type = state["type"]
4265 if TYPE_CHECKING:
4267 def self_group(
4268 self, against: Optional[OperatorType] = None
4269 ) -> Self: ...
4272class _OverrideBinds(Grouping[_T]):
4273 """used by cache_key->_apply_params_to_element to allow compilation /
4274 execution of a SQL element that's been cached, using an alternate set of
4275 bound parameter values.
4277 This is used by the ORM to swap new parameter values into expressions
4278 that are embedded into loader options like with_expression(),
4279 selectinload(). Previously, this task was accomplished using the
4280 .params() method which would perform a deep-copy instead. This deep
4281 copy proved to be too expensive for more complex expressions.
4283 See #11085
4285 """
4287 __visit_name__ = "override_binds"
4289 def __init__(
4290 self,
4291 element: ColumnElement[_T],
4292 bindparams: Sequence[BindParameter[Any]],
4293 replaces_params: Sequence[BindParameter[Any]],
4294 ):
4295 self.element = element
4296 self.translate = {
4297 k.key: v.value for k, v in zip(replaces_params, bindparams)
4298 }
4300 def _gen_cache_key(
4301 self, anon_map: anon_map, bindparams: List[BindParameter[Any]]
4302 ) -> Optional[typing_Tuple[Any, ...]]:
4303 """generate a cache key for the given element, substituting its bind
4304 values for the translation values present."""
4306 existing_bps: List[BindParameter[Any]] = []
4307 ck = self.element._gen_cache_key(anon_map, existing_bps)
4309 bindparams.extend(
4310 (
4311 bp._with_value(
4312 self.translate[bp.key], maintain_key=True, required=False
4313 )
4314 if bp.key in self.translate
4315 else bp
4316 )
4317 for bp in existing_bps
4318 )
4320 return ck
4323class Over(ColumnElement[_T]):
4324 """Represent an OVER clause.
4326 This is a special operator against a so-called
4327 "window" function, as well as any aggregate function,
4328 which produces results relative to the result set
4329 itself. Most modern SQL backends now support window functions.
4331 """
4333 __visit_name__ = "over"
4335 _traverse_internals: _TraverseInternalsType = [
4336 ("element", InternalTraversal.dp_clauseelement),
4337 ("order_by", InternalTraversal.dp_clauseelement),
4338 ("partition_by", InternalTraversal.dp_clauseelement),
4339 ("range_", InternalTraversal.dp_clauseelement),
4340 ("rows", InternalTraversal.dp_clauseelement),
4341 ("groups", InternalTraversal.dp_clauseelement),
4342 ]
4344 order_by: Optional[ClauseList] = None
4345 partition_by: Optional[ClauseList] = None
4347 element: ColumnElement[_T]
4348 """The underlying expression object to which this :class:`.Over`
4349 object refers."""
4351 range_: Optional[_FrameClause]
4352 rows: Optional[_FrameClause]
4353 groups: Optional[_FrameClause]
4355 def __init__(
4356 self,
4357 element: ColumnElement[_T],
4358 partition_by: Optional[_ByArgument] = None,
4359 order_by: Optional[_ByArgument] = None,
4360 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4361 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4362 groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4363 ):
4364 self.element = element
4365 if order_by is not None:
4366 self.order_by = ClauseList(
4367 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4368 )
4369 if partition_by is not None:
4370 self.partition_by = ClauseList(
4371 *util.to_list(partition_by),
4372 _literal_as_text_role=roles.ByOfRole,
4373 )
4375 if sum(bool(item) for item in (range_, rows, groups)) > 1:
4376 raise exc.ArgumentError(
4377 "only one of 'rows', 'range_', or 'groups' may be provided"
4378 )
4379 else:
4380 self.range_ = _FrameClause(range_) if range_ else None
4381 self.rows = _FrameClause(rows) if rows else None
4382 self.groups = _FrameClause(groups) if groups else None
4384 if not TYPE_CHECKING:
4386 @util.memoized_property
4387 def type(self) -> TypeEngine[_T]: # noqa: A001
4388 return self.element.type
4390 @util.ro_non_memoized_property
4391 def _from_objects(self) -> List[FromClause]:
4392 return list(
4393 itertools.chain(
4394 *[
4395 c._from_objects
4396 for c in (self.element, self.partition_by, self.order_by)
4397 if c is not None
4398 ]
4399 )
4400 )
4403class _FrameClauseType(Enum):
4404 RANGE_UNBOUNDED = 0
4405 RANGE_CURRENT = 1
4406 RANGE_PRECEDING = 2
4407 RANGE_FOLLOWING = 3
4410class _FrameClause(ClauseElement):
4411 """indicate the 'rows' or 'range' field of a window function, e.g. using
4412 :class:`.Over`.
4414 .. versionadded:: 2.1
4416 """
4418 __visit_name__ = "frame_clause"
4420 _traverse_internals: _TraverseInternalsType = [
4421 ("lower_integer_bind", InternalTraversal.dp_clauseelement),
4422 ("upper_integer_bind", InternalTraversal.dp_clauseelement),
4423 ("lower_type", InternalTraversal.dp_plain_obj),
4424 ("upper_type", InternalTraversal.dp_plain_obj),
4425 ]
4427 def __init__(
4428 self,
4429 range_: typing_Tuple[Optional[int], Optional[int]],
4430 ):
4431 try:
4432 r0, r1 = range_
4433 except (ValueError, TypeError) as ve:
4434 raise exc.ArgumentError("2-tuple expected for range/rows") from ve
4436 if r0 is None:
4437 self.lower_type = _FrameClauseType.RANGE_UNBOUNDED
4438 self.lower_integer_bind = None
4439 else:
4440 try:
4441 lower_integer = int(r0)
4442 except ValueError as err:
4443 raise exc.ArgumentError(
4444 "Integer or None expected for range value"
4445 ) from err
4446 else:
4447 if lower_integer == 0:
4448 self.lower_type = _FrameClauseType.RANGE_CURRENT
4449 self.lower_integer_bind = None
4450 elif lower_integer < 0:
4451 self.lower_type = _FrameClauseType.RANGE_PRECEDING
4452 self.lower_integer_bind = literal(
4453 abs(lower_integer), type_api.INTEGERTYPE
4454 )
4455 else:
4456 self.lower_type = _FrameClauseType.RANGE_FOLLOWING
4457 self.lower_integer_bind = literal(
4458 lower_integer, type_api.INTEGERTYPE
4459 )
4461 if r1 is None:
4462 self.upper_type = _FrameClauseType.RANGE_UNBOUNDED
4463 self.upper_integer_bind = None
4464 else:
4465 try:
4466 upper_integer = int(r1)
4467 except ValueError as err:
4468 raise exc.ArgumentError(
4469 "Integer or None expected for range value"
4470 ) from err
4471 else:
4472 if upper_integer == 0:
4473 self.upper_type = _FrameClauseType.RANGE_CURRENT
4474 self.upper_integer_bind = None
4475 elif upper_integer < 0:
4476 self.upper_type = _FrameClauseType.RANGE_PRECEDING
4477 self.upper_integer_bind = literal(
4478 abs(upper_integer), type_api.INTEGERTYPE
4479 )
4480 else:
4481 self.upper_type = _FrameClauseType.RANGE_FOLLOWING
4482 self.upper_integer_bind = literal(
4483 upper_integer, type_api.INTEGERTYPE
4484 )
4487class WithinGroup(ColumnElement[_T]):
4488 """Represent a WITHIN GROUP (ORDER BY) clause.
4490 This is a special operator against so-called
4491 "ordered set aggregate" and "hypothetical
4492 set aggregate" functions, including ``percentile_cont()``,
4493 ``rank()``, ``dense_rank()``, etc.
4495 It's supported only by certain database backends, such as PostgreSQL,
4496 Oracle Database and MS SQL Server.
4498 The :class:`.WithinGroup` construct extracts its type from the
4499 method :meth:`.FunctionElement.within_group_type`. If this returns
4500 ``None``, the function's ``.type`` is used.
4502 """
4504 __visit_name__ = "withingroup"
4506 _traverse_internals: _TraverseInternalsType = [
4507 ("element", InternalTraversal.dp_clauseelement),
4508 ("order_by", InternalTraversal.dp_clauseelement),
4509 ]
4511 order_by: Optional[ClauseList] = None
4513 def __init__(
4514 self,
4515 element: Union[FunctionElement[_T], FunctionFilter[_T]],
4516 *order_by: _ColumnExpressionArgument[Any],
4517 ):
4518 self.element = element
4519 if order_by is not None:
4520 self.order_by = ClauseList(
4521 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4522 )
4524 def __reduce__(self):
4525 return self.__class__, (self.element,) + (
4526 tuple(self.order_by) if self.order_by is not None else ()
4527 )
4529 def over(
4530 self,
4531 *,
4532 partition_by: Optional[_ByArgument] = None,
4533 order_by: Optional[_ByArgument] = None,
4534 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4535 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4536 groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4537 ) -> Over[_T]:
4538 """Produce an OVER clause against this :class:`.WithinGroup`
4539 construct.
4541 This function has the same signature as that of
4542 :meth:`.FunctionElement.over`.
4544 """
4545 return Over(
4546 self,
4547 partition_by=partition_by,
4548 order_by=order_by,
4549 range_=range_,
4550 rows=rows,
4551 groups=groups,
4552 )
4554 @overload
4555 def filter(self) -> Self: ...
4557 @overload
4558 def filter(
4559 self,
4560 __criterion0: _ColumnExpressionArgument[bool],
4561 *criterion: _ColumnExpressionArgument[bool],
4562 ) -> FunctionFilter[_T]: ...
4564 def filter(
4565 self, *criterion: _ColumnExpressionArgument[bool]
4566 ) -> Union[Self, FunctionFilter[_T]]:
4567 """Produce a FILTER clause against this function."""
4568 if not criterion:
4569 return self
4570 return FunctionFilter(self, *criterion)
4572 if not TYPE_CHECKING:
4574 @util.memoized_property
4575 def type(self) -> TypeEngine[_T]: # noqa: A001
4576 wgt = self.element.within_group_type(self)
4577 if wgt is not None:
4578 return wgt
4579 else:
4580 return self.element.type
4582 @util.ro_non_memoized_property
4583 def _from_objects(self) -> List[FromClause]:
4584 return list(
4585 itertools.chain(
4586 *[
4587 c._from_objects
4588 for c in (self.element, self.order_by)
4589 if c is not None
4590 ]
4591 )
4592 )
4595class FunctionFilter(Generative, ColumnElement[_T]):
4596 """Represent a function FILTER clause.
4598 This is a special operator against aggregate and window functions,
4599 which controls which rows are passed to it.
4600 It's supported only by certain database backends.
4602 Invocation of :class:`.FunctionFilter` is via
4603 :meth:`.FunctionElement.filter`::
4605 func.count(1).filter(True)
4607 .. seealso::
4609 :meth:`.FunctionElement.filter`
4611 """
4613 __visit_name__ = "funcfilter"
4615 _traverse_internals: _TraverseInternalsType = [
4616 ("func", InternalTraversal.dp_clauseelement),
4617 ("criterion", InternalTraversal.dp_clauseelement),
4618 ]
4620 criterion: Optional[ColumnElement[bool]] = None
4622 def __init__(
4623 self,
4624 func: Union[FunctionElement[_T], WithinGroup[_T]],
4625 *criterion: _ColumnExpressionArgument[bool],
4626 ):
4627 self.func = func
4628 self.filter.non_generative(self, *criterion) # type: ignore
4630 @_generative
4631 def filter(self, *criterion: _ColumnExpressionArgument[bool]) -> Self:
4632 """Produce an additional FILTER against the function.
4634 This method adds additional criteria to the initial criteria
4635 set up by :meth:`.FunctionElement.filter`.
4637 Multiple criteria are joined together at SQL render time
4638 via ``AND``.
4641 """
4643 for crit in list(criterion):
4644 crit = coercions.expect(roles.WhereHavingRole, crit)
4646 if self.criterion is not None:
4647 self.criterion = self.criterion & crit
4648 else:
4649 self.criterion = crit
4651 return self
4653 def over(
4654 self,
4655 partition_by: Optional[
4656 Union[
4657 Iterable[_ColumnExpressionArgument[Any]],
4658 _ColumnExpressionArgument[Any],
4659 ]
4660 ] = None,
4661 order_by: Optional[
4662 Union[
4663 Iterable[_ColumnExpressionArgument[Any]],
4664 _ColumnExpressionArgument[Any],
4665 ]
4666 ] = None,
4667 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4668 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4669 groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4670 ) -> Over[_T]:
4671 """Produce an OVER clause against this filtered function.
4673 Used against aggregate or so-called "window" functions,
4674 for database backends that support window functions.
4676 The expression::
4678 func.rank().filter(MyClass.y > 5).over(order_by="x")
4680 is shorthand for::
4682 from sqlalchemy import over, funcfilter
4684 over(funcfilter(func.rank(), MyClass.y > 5), order_by="x")
4686 See :func:`_expression.over` for a full description.
4688 """
4689 return Over(
4690 self,
4691 partition_by=partition_by,
4692 order_by=order_by,
4693 range_=range_,
4694 rows=rows,
4695 groups=groups,
4696 )
4698 def within_group(
4699 self, *order_by: _ColumnExpressionArgument[Any]
4700 ) -> WithinGroup[_T]:
4701 """Produce a WITHIN GROUP (ORDER BY expr) clause against
4702 this function.
4703 """
4704 return WithinGroup(self, *order_by)
4706 def within_group_type(
4707 self, within_group: WithinGroup[_T]
4708 ) -> Optional[TypeEngine[_T]]:
4709 return None
4711 def self_group(
4712 self, against: Optional[OperatorType] = None
4713 ) -> Union[Self, Grouping[_T]]:
4714 if operators.is_precedent(operators.filter_op, against):
4715 return Grouping(self)
4716 else:
4717 return self
4719 if not TYPE_CHECKING:
4721 @util.memoized_property
4722 def type(self) -> TypeEngine[_T]: # noqa: A001
4723 return self.func.type
4725 @util.ro_non_memoized_property
4726 def _from_objects(self) -> List[FromClause]:
4727 return list(
4728 itertools.chain(
4729 *[
4730 c._from_objects
4731 for c in (self.func, self.criterion)
4732 if c is not None
4733 ]
4734 )
4735 )
4738class NamedColumn(KeyedColumnElement[_T]):
4739 is_literal = False
4740 table: Optional[FromClause] = None
4741 name: str
4742 key: str
4744 def _compare_name_for_result(self, other):
4745 return (hasattr(other, "name") and self.name == other.name) or (
4746 hasattr(other, "_label") and self._label == other._label
4747 )
4749 @util.ro_memoized_property
4750 def description(self) -> str:
4751 return self.name
4753 @HasMemoized.memoized_attribute
4754 def _tq_key_label(self) -> Optional[str]:
4755 """table qualified label based on column key.
4757 for table-bound columns this is <tablename>_<column key/proxy key>;
4759 all other expressions it resolves to key/proxy key.
4761 """
4762 proxy_key = self._proxy_key
4763 if proxy_key and proxy_key != self.name:
4764 return self._gen_tq_label(proxy_key)
4765 else:
4766 return self._tq_label
4768 @HasMemoized.memoized_attribute
4769 def _tq_label(self) -> Optional[str]:
4770 """table qualified label based on column name.
4772 for table-bound columns this is <tablename>_<columnname>; all other
4773 expressions it resolves to .name.
4775 """
4776 return self._gen_tq_label(self.name)
4778 @HasMemoized.memoized_attribute
4779 def _render_label_in_columns_clause(self):
4780 return True
4782 @HasMemoized.memoized_attribute
4783 def _non_anon_label(self):
4784 return self.name
4786 def _gen_tq_label(
4787 self, name: str, dedupe_on_key: bool = True
4788 ) -> Optional[str]:
4789 return name
4791 def _bind_param(
4792 self,
4793 operator: OperatorType,
4794 obj: Any,
4795 type_: Optional[TypeEngine[_T]] = None,
4796 expanding: bool = False,
4797 ) -> BindParameter[_T]:
4798 return BindParameter(
4799 self.key,
4800 obj,
4801 _compared_to_operator=operator,
4802 _compared_to_type=self.type,
4803 type_=type_,
4804 unique=True,
4805 expanding=expanding,
4806 )
4808 def _make_proxy(
4809 self,
4810 selectable: FromClause,
4811 *,
4812 primary_key: ColumnSet,
4813 foreign_keys: Set[KeyedColumnElement[Any]],
4814 name: Optional[str] = None,
4815 key: Optional[str] = None,
4816 name_is_truncatable: bool = False,
4817 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
4818 disallow_is_literal: bool = False,
4819 **kw: Any,
4820 ) -> typing_Tuple[str, ColumnClause[_T]]:
4821 c = ColumnClause(
4822 (
4823 coercions.expect(roles.TruncatedLabelRole, name or self.name)
4824 if name_is_truncatable
4825 else (name or self.name)
4826 ),
4827 type_=self.type,
4828 _selectable=selectable,
4829 is_literal=False,
4830 )
4832 c._propagate_attrs = selectable._propagate_attrs
4833 if name is None:
4834 c.key = self.key
4835 if compound_select_cols:
4836 c._proxies = list(compound_select_cols)
4837 else:
4838 c._proxies = [self]
4840 if selectable._is_clone_of is not None:
4841 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
4842 return c.key, c
4845_PS = ParamSpec("_PS")
4848class Label(roles.LabeledColumnExprRole[_T], NamedColumn[_T]):
4849 """Represents a column label (AS).
4851 Represent a label, as typically applied to any column-level
4852 element using the ``AS`` sql keyword.
4854 """
4856 __visit_name__ = "label"
4858 _traverse_internals: _TraverseInternalsType = [
4859 ("name", InternalTraversal.dp_anon_name),
4860 ("type", InternalTraversal.dp_type),
4861 ("_element", InternalTraversal.dp_clauseelement),
4862 ]
4864 _cache_key_traversal = [
4865 ("name", InternalTraversal.dp_anon_name),
4866 ("_element", InternalTraversal.dp_clauseelement),
4867 ]
4869 _element: ColumnElement[_T]
4870 name: str
4872 def __init__(
4873 self,
4874 name: Optional[str],
4875 element: _ColumnExpressionArgument[_T],
4876 type_: Optional[_TypeEngineArgument[_T]] = None,
4877 ):
4878 orig_element = element
4879 element = coercions.expect(
4880 roles.ExpressionElementRole,
4881 element,
4882 apply_propagate_attrs=self,
4883 )
4884 while isinstance(element, Label):
4885 # TODO: this is only covered in test_text.py, but nothing
4886 # fails if it's removed. determine rationale
4887 element = element.element
4889 if name:
4890 self.name = name
4891 else:
4892 self.name = _anonymous_label.safe_construct(
4893 id(self), getattr(element, "name", "anon")
4894 )
4895 if isinstance(orig_element, Label):
4896 # TODO: no coverage for this block, again would be in
4897 # test_text.py where the resolve_label concept is important
4898 self._resolve_label = orig_element._label
4900 self.key = self._tq_label = self._tq_key_label = self.name
4901 self._element = element
4903 self.type = (
4904 type_api.to_instance(type_)
4905 if type_ is not None
4906 else self._element.type
4907 )
4909 self._proxies = [element]
4911 def __reduce__(self):
4912 return self.__class__, (self.name, self._element, self.type)
4914 @HasMemoized.memoized_attribute
4915 def _render_label_in_columns_clause(self):
4916 return True
4918 def _bind_param(self, operator, obj, type_=None, expanding=False):
4919 return BindParameter(
4920 None,
4921 obj,
4922 _compared_to_operator=operator,
4923 type_=type_,
4924 _compared_to_type=self.type,
4925 unique=True,
4926 expanding=expanding,
4927 )
4929 @util.memoized_property
4930 def _is_implicitly_boolean(self):
4931 return self.element._is_implicitly_boolean
4933 @HasMemoized.memoized_attribute
4934 def _allow_label_resolve(self):
4935 return self.element._allow_label_resolve
4937 @property
4938 def _order_by_label_element(self):
4939 return self
4941 def as_reference(self) -> _label_reference[_T]:
4942 """refer to this labeled expression in a clause such as GROUP BY,
4943 ORDER BY etc. as the label name itself, without expanding
4944 into the full expression.
4946 .. versionadded:: 2.1
4948 """
4949 return _label_reference(self)
4951 @HasMemoized.memoized_attribute
4952 def element(self) -> ColumnElement[_T]:
4953 return self._element.self_group(against=operators.as_)
4955 def self_group(self, against: Optional[OperatorType] = None) -> Label[_T]:
4956 return self._apply_to_inner(self._element.self_group, against=against)
4958 def _negate(self):
4959 return self._apply_to_inner(self._element._negate)
4961 def _apply_to_inner(
4962 self,
4963 fn: Callable[_PS, ColumnElement[_T]],
4964 *arg: _PS.args,
4965 **kw: _PS.kwargs,
4966 ) -> Label[_T]:
4967 sub_element = fn(*arg, **kw)
4968 if sub_element is not self._element:
4969 return Label(self.name, sub_element, type_=self.type)
4970 else:
4971 return self
4973 @property
4974 def primary_key(self): # type: ignore[override]
4975 return self.element.primary_key
4977 @property
4978 def foreign_keys(self): # type: ignore[override]
4979 return self.element.foreign_keys
4981 def _copy_internals(
4982 self,
4983 *,
4984 clone: _CloneCallableType = _clone,
4985 anonymize_labels: bool = False,
4986 **kw: Any,
4987 ) -> None:
4988 self._reset_memoizations()
4989 self._element = clone(self._element, **kw)
4990 if anonymize_labels:
4991 self.name = _anonymous_label.safe_construct(
4992 id(self), getattr(self.element, "name", "anon")
4993 )
4994 self.key = self._tq_label = self._tq_key_label = self.name
4996 @util.ro_non_memoized_property
4997 def _from_objects(self) -> List[FromClause]:
4998 return self.element._from_objects
5000 def _make_proxy(
5001 self,
5002 selectable: FromClause,
5003 *,
5004 primary_key: ColumnSet,
5005 foreign_keys: Set[KeyedColumnElement[Any]],
5006 name: Optional[str] = None,
5007 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5008 **kw: Any,
5009 ) -> typing_Tuple[str, ColumnClause[_T]]:
5010 name = self.name if not name else name
5012 key, e = self.element._make_proxy(
5013 selectable,
5014 name=name,
5015 disallow_is_literal=True,
5016 name_is_truncatable=isinstance(name, _truncated_label),
5017 compound_select_cols=compound_select_cols,
5018 primary_key=primary_key,
5019 foreign_keys=foreign_keys,
5020 )
5022 # there was a note here to remove this assertion, which was here
5023 # to determine if we later could support a use case where
5024 # the key and name of a label are separate. But I don't know what
5025 # that case was. For now, this is an unexpected case that occurs
5026 # when a label name conflicts with other columns and select()
5027 # is attempting to disambiguate an explicit label, which is not what
5028 # the user would want. See issue #6090.
5029 if key != self.name and not isinstance(self.name, _anonymous_label):
5030 raise exc.InvalidRequestError(
5031 "Label name %s is being renamed to an anonymous label due "
5032 "to disambiguation "
5033 "which is not supported right now. Please use unique names "
5034 "for explicit labels." % (self.name)
5035 )
5037 e._propagate_attrs = selectable._propagate_attrs
5038 e._proxies.append(self)
5039 if self.type is not None:
5040 e.type = self.type
5042 return self.key, e
5045class ColumnClause(
5046 roles.DDLReferredColumnRole,
5047 roles.LabeledColumnExprRole[_T],
5048 roles.StrAsPlainColumnRole,
5049 Immutable,
5050 NamedColumn[_T],
5051):
5052 """Represents a column expression from any textual string.
5054 The :class:`.ColumnClause`, a lightweight analogue to the
5055 :class:`_schema.Column` class, is typically invoked using the
5056 :func:`_expression.column` function, as in::
5058 from sqlalchemy import column
5060 id, name = column("id"), column("name")
5061 stmt = select(id, name).select_from("user")
5063 The above statement would produce SQL like:
5065 .. sourcecode:: sql
5067 SELECT id, name FROM user
5069 :class:`.ColumnClause` is the immediate superclass of the schema-specific
5070 :class:`_schema.Column` object. While the :class:`_schema.Column`
5071 class has all the
5072 same capabilities as :class:`.ColumnClause`, the :class:`.ColumnClause`
5073 class is usable by itself in those cases where behavioral requirements
5074 are limited to simple SQL expression generation. The object has none of
5075 the associations with schema-level metadata or with execution-time
5076 behavior that :class:`_schema.Column` does,
5077 so in that sense is a "lightweight"
5078 version of :class:`_schema.Column`.
5080 Full details on :class:`.ColumnClause` usage is at
5081 :func:`_expression.column`.
5083 .. seealso::
5085 :func:`_expression.column`
5087 :class:`_schema.Column`
5089 """
5091 table: Optional[FromClause]
5092 is_literal: bool
5094 __visit_name__ = "column"
5096 _traverse_internals: _TraverseInternalsType = [
5097 ("name", InternalTraversal.dp_anon_name),
5098 ("type", InternalTraversal.dp_type),
5099 ("table", InternalTraversal.dp_clauseelement),
5100 ("is_literal", InternalTraversal.dp_boolean),
5101 ]
5103 onupdate: Optional[DefaultGenerator] = None
5104 default: Optional[DefaultGenerator] = None
5105 server_default: Optional[FetchedValue] = None
5106 server_onupdate: Optional[FetchedValue] = None
5108 _is_multiparam_column = False
5110 @property
5111 def _is_star(self): # type: ignore[override]
5112 return self.is_literal and self.name == "*"
5114 def __init__(
5115 self,
5116 text: str,
5117 type_: Optional[_TypeEngineArgument[_T]] = None,
5118 is_literal: bool = False,
5119 _selectable: Optional[FromClause] = None,
5120 ):
5121 self.key = self.name = text
5122 self.table = _selectable
5124 # if type is None, we get NULLTYPE, which is our _T. But I don't
5125 # know how to get the overloads to express that correctly
5126 self.type = type_api.to_instance(type_) # type: ignore
5128 self.is_literal = is_literal
5130 def get_children(self, *, column_tables=False, **kw):
5131 # override base get_children() to not return the Table
5132 # or selectable that is parent to this column. Traversals
5133 # expect the columns of tables and subqueries to be leaf nodes.
5134 return []
5136 @property
5137 def entity_namespace(self):
5138 if self.table is not None:
5139 return self.table.entity_namespace
5140 else:
5141 return super().entity_namespace
5143 def _clone(self, detect_subquery_cols=False, **kw):
5144 if (
5145 detect_subquery_cols
5146 and self.table is not None
5147 and self.table._is_subquery
5148 ):
5149 clone = kw.pop("clone")
5150 table = clone(self.table, **kw)
5151 new = table.c.corresponding_column(self)
5152 return new
5154 return super()._clone(**kw)
5156 @HasMemoized_ro_memoized_attribute
5157 def _from_objects(self) -> List[FromClause]:
5158 t = self.table
5159 if t is not None:
5160 return [t]
5161 else:
5162 return []
5164 @HasMemoized.memoized_attribute
5165 def _render_label_in_columns_clause(self):
5166 return self.table is not None
5168 @property
5169 def _ddl_label(self):
5170 return self._gen_tq_label(self.name, dedupe_on_key=False)
5172 def _compare_name_for_result(self, other):
5173 if (
5174 self.is_literal
5175 or self.table is None
5176 or self.table._is_textual
5177 or not hasattr(other, "proxy_set")
5178 or (
5179 isinstance(other, ColumnClause)
5180 and (
5181 other.is_literal
5182 or other.table is None
5183 or other.table._is_textual
5184 )
5185 )
5186 ):
5187 return (hasattr(other, "name") and self.name == other.name) or (
5188 hasattr(other, "_tq_label")
5189 and self._tq_label == other._tq_label
5190 )
5191 else:
5192 return other.proxy_set.intersection(self.proxy_set)
5194 def _gen_tq_label(
5195 self, name: str, dedupe_on_key: bool = True
5196 ) -> Optional[str]:
5197 """generate table-qualified label
5199 for a table-bound column this is <tablename>_<columnname>.
5201 used primarily for LABEL_STYLE_TABLENAME_PLUS_COL
5202 as well as the .columns collection on a Join object.
5204 """
5205 label: str
5206 t = self.table
5207 if self.is_literal:
5208 return None
5209 elif t is not None and is_named_from_clause(t):
5210 if has_schema_attr(t) and t.schema:
5211 label = (
5212 t.schema.replace(".", "_") + "_" + t.name + ("_" + name)
5213 )
5214 else:
5215 assert not TYPE_CHECKING or isinstance(t, NamedFromClause)
5216 label = t.name + ("_" + name)
5218 # propagate name quoting rules for labels.
5219 if is_quoted_name(name) and name.quote is not None:
5220 if is_quoted_name(label):
5221 label.quote = name.quote
5222 else:
5223 label = quoted_name(label, name.quote)
5224 elif is_quoted_name(t.name) and t.name.quote is not None:
5225 # can't get this situation to occur, so let's
5226 # assert false on it for now
5227 assert not isinstance(label, quoted_name)
5228 label = quoted_name(label, t.name.quote)
5230 if dedupe_on_key:
5231 # ensure the label name doesn't conflict with that of an
5232 # existing column. note that this implies that any Column
5233 # must **not** set up its _label before its parent table has
5234 # all of its other Column objects set up. There are several
5235 # tables in the test suite which will fail otherwise; example:
5236 # table "owner" has columns "name" and "owner_name". Therefore
5237 # column owner.name cannot use the label "owner_name", it has
5238 # to be "owner_name_1".
5239 if label in t.c:
5240 _label = label
5241 counter = 1
5242 while _label in t.c:
5243 _label = label + f"_{counter}"
5244 counter += 1
5245 label = _label
5247 return coercions.expect(roles.TruncatedLabelRole, label)
5249 else:
5250 return name
5252 def _make_proxy(
5253 self,
5254 selectable: FromClause,
5255 *,
5256 primary_key: ColumnSet,
5257 foreign_keys: Set[KeyedColumnElement[Any]],
5258 name: Optional[str] = None,
5259 key: Optional[str] = None,
5260 name_is_truncatable: bool = False,
5261 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5262 disallow_is_literal: bool = False,
5263 **kw: Any,
5264 ) -> typing_Tuple[str, ColumnClause[_T]]:
5265 # the "is_literal" flag normally should never be propagated; a proxied
5266 # column is always a SQL identifier and never the actual expression
5267 # being evaluated. however, there is a case where the "is_literal" flag
5268 # might be used to allow the given identifier to have a fixed quoting
5269 # pattern already, so maintain the flag for the proxy unless a
5270 # :class:`.Label` object is creating the proxy. See [ticket:4730].
5271 is_literal = (
5272 not disallow_is_literal
5273 and self.is_literal
5274 and (
5275 # note this does not accommodate for quoted_name differences
5276 # right now
5277 name is None
5278 or name == self.name
5279 )
5280 )
5281 c = self._constructor(
5282 (
5283 coercions.expect(roles.TruncatedLabelRole, name or self.name)
5284 if name_is_truncatable
5285 else (name or self.name)
5286 ),
5287 type_=self.type,
5288 _selectable=selectable,
5289 is_literal=is_literal,
5290 )
5291 c._propagate_attrs = selectable._propagate_attrs
5292 if name is None:
5293 c.key = self.key
5294 if compound_select_cols:
5295 c._proxies = list(compound_select_cols)
5296 else:
5297 c._proxies = [self]
5299 if selectable._is_clone_of is not None:
5300 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
5301 return c.key, c
5304class TableValuedColumn(NamedColumn[_T]):
5305 __visit_name__ = "table_valued_column"
5307 _traverse_internals: _TraverseInternalsType = [
5308 ("name", InternalTraversal.dp_anon_name),
5309 ("type", InternalTraversal.dp_type),
5310 ("scalar_alias", InternalTraversal.dp_clauseelement),
5311 ]
5313 def __init__(self, scalar_alias: NamedFromClause, type_: TypeEngine[_T]):
5314 self.scalar_alias = scalar_alias
5315 self.key = self.name = scalar_alias.name
5316 self.type = type_
5318 def _copy_internals(
5319 self, clone: _CloneCallableType = _clone, **kw: Any
5320 ) -> None:
5321 self.scalar_alias = clone(self.scalar_alias, **kw)
5322 self.key = self.name = self.scalar_alias.name
5324 @util.ro_non_memoized_property
5325 def _from_objects(self) -> List[FromClause]:
5326 return [self.scalar_alias]
5329class CollationClause(ColumnElement[str]):
5330 __visit_name__ = "collation"
5332 _traverse_internals: _TraverseInternalsType = [
5333 ("collation", InternalTraversal.dp_string)
5334 ]
5336 @classmethod
5337 @util.preload_module("sqlalchemy.sql.sqltypes")
5338 def _create_collation_expression(
5339 cls, expression: _ColumnExpressionArgument[str], collation: str
5340 ) -> BinaryExpression[str]:
5342 sqltypes = util.preloaded.sql_sqltypes
5344 expr = coercions.expect(roles.ExpressionElementRole[str], expression)
5346 if expr.type._type_affinity is sqltypes.String:
5347 collate_type = expr.type._with_collation(collation)
5348 else:
5349 collate_type = expr.type
5351 return BinaryExpression(
5352 expr,
5353 CollationClause(collation),
5354 operators.collate,
5355 type_=collate_type,
5356 )
5358 def __init__(self, collation):
5359 self.collation = collation
5362class _IdentifiedClause(Executable, ClauseElement):
5363 __visit_name__ = "identified"
5365 def __init__(self, ident):
5366 self.ident = ident
5369class SavepointClause(_IdentifiedClause):
5370 __visit_name__ = "savepoint"
5371 inherit_cache = False
5374class RollbackToSavepointClause(_IdentifiedClause):
5375 __visit_name__ = "rollback_to_savepoint"
5376 inherit_cache = False
5379class ReleaseSavepointClause(_IdentifiedClause):
5380 __visit_name__ = "release_savepoint"
5381 inherit_cache = False
5384class quoted_name(util.MemoizedSlots, str):
5385 """Represent a SQL identifier combined with quoting preferences.
5387 :class:`.quoted_name` is a Python unicode/str subclass which
5388 represents a particular identifier name along with a
5389 ``quote`` flag. This ``quote`` flag, when set to
5390 ``True`` or ``False``, overrides automatic quoting behavior
5391 for this identifier in order to either unconditionally quote
5392 or to not quote the name. If left at its default of ``None``,
5393 quoting behavior is applied to the identifier on a per-backend basis
5394 based on an examination of the token itself.
5396 A :class:`.quoted_name` object with ``quote=True`` is also
5397 prevented from being modified in the case of a so-called
5398 "name normalize" option. Certain database backends, such as
5399 Oracle Database, Firebird, and DB2 "normalize" case-insensitive names
5400 as uppercase. The SQLAlchemy dialects for these backends
5401 convert from SQLAlchemy's lower-case-means-insensitive convention
5402 to the upper-case-means-insensitive conventions of those backends.
5403 The ``quote=True`` flag here will prevent this conversion from occurring
5404 to support an identifier that's quoted as all lower case against
5405 such a backend.
5407 The :class:`.quoted_name` object is normally created automatically
5408 when specifying the name for key schema constructs such as
5409 :class:`_schema.Table`, :class:`_schema.Column`, and others.
5410 The class can also be
5411 passed explicitly as the name to any function that receives a name which
5412 can be quoted. Such as to use the :meth:`_engine.Engine.has_table`
5413 method with
5414 an unconditionally quoted name::
5416 from sqlalchemy import create_engine
5417 from sqlalchemy import inspect
5418 from sqlalchemy.sql import quoted_name
5420 engine = create_engine("oracle+oracledb://some_dsn")
5421 print(inspect(engine).has_table(quoted_name("some_table", True)))
5423 The above logic will run the "has table" logic against the Oracle Database
5424 backend, passing the name exactly as ``"some_table"`` without converting to
5425 upper case.
5427 """
5429 __slots__ = "quote", "lower", "upper"
5431 quote: Optional[bool]
5433 @overload
5434 @classmethod
5435 def construct(cls, value: str, quote: Optional[bool]) -> quoted_name: ...
5437 @overload
5438 @classmethod
5439 def construct(cls, value: None, quote: Optional[bool]) -> None: ...
5441 @classmethod
5442 def construct(
5443 cls, value: Optional[str], quote: Optional[bool]
5444 ) -> Optional[quoted_name]:
5445 if value is None:
5446 return None
5447 else:
5448 return quoted_name(value, quote)
5450 def __new__(cls, value: str, quote: Optional[bool]) -> quoted_name:
5451 assert (
5452 value is not None
5453 ), "use quoted_name.construct() for None passthrough"
5454 if isinstance(value, cls) and (quote is None or value.quote == quote):
5455 return value
5456 self = super().__new__(cls, value)
5458 self.quote = quote
5459 return self
5461 def __reduce__(self):
5462 return quoted_name, (str(self), self.quote)
5464 def _memoized_method_lower(self):
5465 if self.quote:
5466 return self
5467 else:
5468 return str(self).lower()
5470 def _memoized_method_upper(self):
5471 if self.quote:
5472 return self
5473 else:
5474 return str(self).upper()
5477def _find_columns(clause: ClauseElement) -> Set[ColumnClause[Any]]:
5478 """locate Column objects within the given expression."""
5480 cols: Set[ColumnClause[Any]] = set()
5481 traverse(clause, {}, {"column": cols.add})
5482 return cols
5485def _type_from_args(args: Sequence[ColumnElement[_T]]) -> TypeEngine[_T]:
5486 for a in args:
5487 if not a.type._isnull:
5488 return a.type
5489 else:
5490 return type_api.NULLTYPE # type: ignore
5493def _corresponding_column_or_error(fromclause, column, require_embedded=False):
5494 c = fromclause.corresponding_column(
5495 column, require_embedded=require_embedded
5496 )
5497 if c is None:
5498 raise exc.InvalidRequestError(
5499 "Given column '%s', attached to table '%s', "
5500 "failed to locate a corresponding column from table '%s'"
5501 % (column, getattr(column, "table", None), fromclause.description)
5502 )
5503 return c
5506class _memoized_property_but_not_nulltype(
5507 util.memoized_property["TypeEngine[_T]"]
5508):
5509 """memoized property, but dont memoize NullType"""
5511 def __get__(self, obj, cls):
5512 if obj is None:
5513 return self
5514 result = self.fget(obj)
5515 if not result._isnull:
5516 obj.__dict__[self.__name__] = result
5517 return result
5520class AnnotatedColumnElement(Annotated):
5521 _Annotated__element: ColumnElement[Any]
5523 def __init__(self, element, values):
5524 Annotated.__init__(self, element, values)
5525 for attr in (
5526 "comparator",
5527 "_proxy_key",
5528 "_tq_key_label",
5529 "_tq_label",
5530 "_non_anon_label",
5531 "type",
5532 ):
5533 self.__dict__.pop(attr, None)
5534 for attr in ("name", "key", "table"):
5535 if self.__dict__.get(attr, False) is None:
5536 self.__dict__.pop(attr)
5538 def _with_annotations(self, values):
5539 clone = super()._with_annotations(values)
5540 for attr in (
5541 "comparator",
5542 "_proxy_key",
5543 "_tq_key_label",
5544 "_tq_label",
5545 "_non_anon_label",
5546 ):
5547 clone.__dict__.pop(attr, None)
5548 return clone
5550 @util.memoized_property
5551 def name(self):
5552 """pull 'name' from parent, if not present"""
5553 return self._Annotated__element.name
5555 @_memoized_property_but_not_nulltype
5556 def type(self):
5557 """pull 'type' from parent and don't cache if null.
5559 type is routinely changed on existing columns within the
5560 mapped_column() initialization process, and "type" is also consulted
5561 during the creation of SQL expressions. Therefore it can change after
5562 it was already retrieved. At the same time we don't want annotated
5563 objects having overhead when expressions are produced, so continue
5564 to memoize, but only when we have a non-null type.
5566 """
5567 return self._Annotated__element.type
5569 @util.memoized_property
5570 def table(self):
5571 """pull 'table' from parent, if not present"""
5572 return self._Annotated__element.table
5574 @util.memoized_property
5575 def key(self):
5576 """pull 'key' from parent, if not present"""
5577 return self._Annotated__element.key
5579 @util.memoized_property
5580 def info(self) -> _InfoType:
5581 if TYPE_CHECKING:
5582 assert isinstance(self._Annotated__element, Column)
5583 return self._Annotated__element.info
5585 @util.memoized_property
5586 def _anon_name_label(self) -> str:
5587 return self._Annotated__element._anon_name_label
5590class _truncated_label(quoted_name):
5591 """A unicode subclass used to identify symbolic "
5592 "names that may require truncation."""
5594 __slots__ = ()
5596 def __new__(cls, value: str, quote: Optional[bool] = None) -> Any:
5597 quote = getattr(value, "quote", quote)
5598 # return super(_truncated_label, cls).__new__(cls, value, quote, True)
5599 return super().__new__(cls, value, quote)
5601 def __reduce__(self) -> Any:
5602 return self.__class__, (str(self), self.quote)
5604 def apply_map(self, map_: Mapping[str, Any]) -> str:
5605 return self
5608class conv(_truncated_label):
5609 """Mark a string indicating that a name has already been converted
5610 by a naming convention.
5612 This is a string subclass that indicates a name that should not be
5613 subject to any further naming conventions.
5615 E.g. when we create a :class:`.Constraint` using a naming convention
5616 as follows::
5618 m = MetaData(
5619 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5620 )
5621 t = Table(
5622 "t", m, Column("x", Integer), CheckConstraint("x > 5", name="x5")
5623 )
5625 The name of the above constraint will be rendered as ``"ck_t_x5"``.
5626 That is, the existing name ``x5`` is used in the naming convention as the
5627 ``constraint_name`` token.
5629 In some situations, such as in migration scripts, we may be rendering
5630 the above :class:`.CheckConstraint` with a name that's already been
5631 converted. In order to make sure the name isn't double-modified, the
5632 new name is applied using the :func:`_schema.conv` marker. We can
5633 use this explicitly as follows::
5636 m = MetaData(
5637 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5638 )
5639 t = Table(
5640 "t",
5641 m,
5642 Column("x", Integer),
5643 CheckConstraint("x > 5", name=conv("ck_t_x5")),
5644 )
5646 Where above, the :func:`_schema.conv` marker indicates that the constraint
5647 name here is final, and the name will render as ``"ck_t_x5"`` and not
5648 ``"ck_t_ck_t_x5"``
5650 .. seealso::
5652 :ref:`constraint_naming_conventions`
5654 """
5656 __slots__ = ()
5659# for backwards compatibility in case
5660# someone is re-implementing the
5661# _truncated_identifier() sequence in a custom
5662# compiler
5663_generated_label = _truncated_label
5664_anonymous_label_escape = re.compile(r"[%\(\) \$]+")
5667class _anonymous_label(_truncated_label):
5668 """A unicode subclass used to identify anonymously
5669 generated names."""
5671 __slots__ = ()
5673 @classmethod
5674 def safe_construct_with_key(
5675 cls, seed: int, body: str, sanitize_key: bool = False
5676 ) -> typing_Tuple[_anonymous_label, str]:
5677 # need to escape chars that interfere with format
5678 # strings in any case, issue #8724
5679 body = _anonymous_label_escape.sub("_", body)
5681 if sanitize_key:
5682 # sanitize_key is then an extra step used by BindParameter
5683 body = body.strip("_")
5685 key = f"{seed} {body.replace('%', '%%')}"
5686 label = _anonymous_label(f"%({key})s")
5687 return label, key
5689 @classmethod
5690 def safe_construct(
5691 cls, seed: int, body: str, sanitize_key: bool = False
5692 ) -> _anonymous_label:
5693 # need to escape chars that interfere with format
5694 # strings in any case, issue #8724
5695 body = _anonymous_label_escape.sub("_", body)
5697 if sanitize_key:
5698 # sanitize_key is then an extra step used by BindParameter
5699 body = body.strip("_")
5701 return _anonymous_label(f"%({seed} {body.replace('%', '%%')})s")
5703 def __add__(self, other: str) -> _anonymous_label:
5704 if "%" in other and not isinstance(other, _anonymous_label):
5705 other = str(other).replace("%", "%%")
5706 else:
5707 other = str(other)
5709 return _anonymous_label(
5710 quoted_name(
5711 str.__add__(self, other),
5712 self.quote,
5713 )
5714 )
5716 def __radd__(self, other: str) -> _anonymous_label:
5717 if "%" in other and not isinstance(other, _anonymous_label):
5718 other = str(other).replace("%", "%%")
5719 else:
5720 other = str(other)
5722 return _anonymous_label(
5723 quoted_name(
5724 str.__add__(other, self),
5725 self.quote,
5726 )
5727 )
5729 def apply_map(self, map_: Mapping[str, Any]) -> str:
5730 if self.quote is not None:
5731 # preserve quoting only if necessary
5732 return quoted_name(self % map_, self.quote)
5733 else:
5734 # else skip the constructor call
5735 return self % map_