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 .traversals import HasCopyInternals
71from .visitors import cloned_traverse
72from .visitors import ExternallyTraversible
73from .visitors import InternalTraversal
74from .visitors import traverse
75from .visitors import Visitable
76from .. import exc
77from .. import inspection
78from .. import util
79from ..util import HasMemoized_ro_memoized_attribute
80from ..util import TypingOnly
81from ..util.typing import Self
82from ..util.typing import TupleAny
83from ..util.typing import Unpack
86if typing.TYPE_CHECKING:
87 from ._typing import _ByArgument
88 from ._typing import _ColumnExpressionArgument
89 from ._typing import _ColumnExpressionOrStrLabelArgument
90 from ._typing import _DMLOnlyColumnArgument
91 from ._typing import _HasDialect
92 from ._typing import _InfoType
93 from ._typing import _PropagateAttrsType
94 from ._typing import _TypeEngineArgument
95 from .base import _EntityNamespace
96 from .base import ColumnSet
97 from .cache_key import _CacheKeyTraversalType
98 from .cache_key import CacheKey
99 from .compiler import Compiled
100 from .compiler import SQLCompiler
101 from .functions import FunctionElement
102 from .operators import OperatorType
103 from .schema import Column
104 from .schema import DefaultGenerator
105 from .schema import FetchedValue
106 from .schema import ForeignKey
107 from .selectable import _SelectIterable
108 from .selectable import FromClause
109 from .selectable import NamedFromClause
110 from .selectable import TextualSelect
111 from .sqltypes import TupleType
112 from .type_api import TypeEngine
113 from .visitors import _CloneCallableType
114 from .visitors import _TraverseInternalsType
115 from .visitors import anon_map
116 from ..engine import Connection
117 from ..engine import Dialect
118 from ..engine.interfaces import _CoreMultiExecuteParams
119 from ..engine.interfaces import CacheStats
120 from ..engine.interfaces import CompiledCacheType
121 from ..engine.interfaces import CoreExecuteOptionsParameter
122 from ..engine.interfaces import SchemaTranslateMapType
123 from ..engine.result import Result
126_NUMERIC = Union[float, Decimal]
127_NUMBER = Union[float, int, Decimal]
129_T = TypeVar("_T", bound="Any")
130_T_co = TypeVar("_T_co", bound=Any, covariant=True)
131_OPT = TypeVar("_OPT", bound="Any")
132_NT = TypeVar("_NT", bound="_NUMERIC")
134_NMT = TypeVar("_NMT", bound="_NUMBER")
137@overload
138def literal(
139 value: Any,
140 type_: _TypeEngineArgument[_T],
141 literal_execute: bool = False,
142) -> BindParameter[_T]: ...
145@overload
146def literal(
147 value: _T,
148 type_: None = None,
149 literal_execute: bool = False,
150) -> BindParameter[_T]: ...
153@overload
154def literal(
155 value: Any,
156 type_: Optional[_TypeEngineArgument[Any]] = None,
157 literal_execute: bool = False,
158) -> BindParameter[Any]: ...
161def literal(
162 value: Any,
163 type_: Optional[_TypeEngineArgument[Any]] = None,
164 literal_execute: bool = False,
165) -> BindParameter[Any]:
166 r"""Return a literal clause, bound to a bind parameter.
168 Literal clauses are created automatically when non-
169 :class:`_expression.ClauseElement` objects (such as strings, ints, dates,
170 etc.) are
171 used in a comparison operation with a :class:`_expression.ColumnElement`
172 subclass,
173 such as a :class:`~sqlalchemy.schema.Column` object. Use this function
174 to force the generation of a literal clause, which will be created as a
175 :class:`BindParameter` with a bound value.
177 :param value: the value to be bound. Can be any Python object supported by
178 the underlying DB-API, or is translatable via the given type argument.
180 :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine` which will
181 provide bind-parameter translation for this literal.
183 :param literal_execute: optional bool, when True, the SQL engine will
184 attempt to render the bound value directly in the SQL statement at
185 execution time rather than providing as a parameter value.
187 .. versionadded:: 2.0
189 """
190 return coercions.expect(
191 roles.LiteralValueRole,
192 value,
193 type_=type_,
194 literal_execute=literal_execute,
195 )
198def literal_column(
199 text: str, type_: Optional[_TypeEngineArgument[_T]] = None
200) -> ColumnClause[_T]:
201 r"""Produce a :class:`.ColumnClause` object that has the
202 :paramref:`_expression.column.is_literal` flag set to True.
204 :func:`_expression.literal_column` is similar to
205 :func:`_expression.column`, except that
206 it is more often used as a "standalone" column expression that renders
207 exactly as stated; while :func:`_expression.column`
208 stores a string name that
209 will be assumed to be part of a table and may be quoted as such,
210 :func:`_expression.literal_column` can be that,
211 or any other arbitrary column-oriented
212 expression.
214 :param text: the text of the expression; can be any SQL expression.
215 Quoting rules will not be applied. To specify a column-name expression
216 which should be subject to quoting rules, use the :func:`column`
217 function.
219 :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine`
220 object which will
221 provide result-set translation and additional expression semantics for
222 this column. If left as ``None`` the type will be :class:`.NullType`.
224 .. seealso::
226 :func:`_expression.column`
228 :func:`_expression.text`
230 :ref:`tutorial_select_arbitrary_text`
232 """
233 return ColumnClause(text, type_=type_, is_literal=True)
236class CompilerElement(Visitable):
237 """base class for SQL elements that can be compiled to produce a
238 SQL string.
240 .. versionadded:: 2.0
242 """
244 __slots__ = ()
245 __visit_name__ = "compiler_element"
247 supports_execution = False
249 stringify_dialect = "default"
251 @util.preload_module("sqlalchemy.engine.default")
252 @util.preload_module("sqlalchemy.engine.url")
253 def compile(
254 self,
255 bind: Optional[_HasDialect] = None,
256 dialect: Optional[Dialect] = None,
257 **kw: Any,
258 ) -> Compiled:
259 """Compile this SQL expression.
261 The return value is a :class:`~.Compiled` object.
262 Calling ``str()`` or ``unicode()`` on the returned value will yield a
263 string representation of the result. The
264 :class:`~.Compiled` object also can return a
265 dictionary of bind parameter names and values
266 using the ``params`` accessor.
268 :param bind: An :class:`.Connection` or :class:`.Engine` which
269 can provide a :class:`.Dialect` in order to generate a
270 :class:`.Compiled` object. If the ``bind`` and
271 ``dialect`` parameters are both omitted, a default SQL compiler
272 is used.
274 :param column_keys: Used for INSERT and UPDATE statements, a list of
275 column names which should be present in the VALUES clause of the
276 compiled statement. If ``None``, all columns from the target table
277 object are rendered.
279 :param dialect: A :class:`.Dialect` instance which can generate
280 a :class:`.Compiled` object. This argument takes precedence over
281 the ``bind`` argument.
283 :param compile_kwargs: optional dictionary of additional parameters
284 that will be passed through to the compiler within all "visit"
285 methods. This allows any custom flag to be passed through to
286 a custom compilation construct, for example. It is also used
287 for the case of passing the ``literal_binds`` flag through::
289 from sqlalchemy.sql import table, column, select
291 t = table("t", column("x"))
293 s = select(t).where(t.c.x == 5)
295 print(s.compile(compile_kwargs={"literal_binds": True}))
297 .. seealso::
299 :ref:`faq_sql_expression_string`
301 """
303 if dialect is None:
304 if bind:
305 dialect = bind.dialect
306 elif self.stringify_dialect == "default":
307 dialect = self._default_dialect()
308 else:
309 url = util.preloaded.engine_url
310 dialect = url.URL.create(
311 self.stringify_dialect
312 ).get_dialect()()
314 return self._compiler(dialect, **kw)
316 def _default_dialect(self):
317 default = util.preloaded.engine_default
318 return default.StrCompileDialect()
320 def _compiler(self, dialect: Dialect, **kw: Any) -> Compiled:
321 """Return a compiler appropriate for this ClauseElement, given a
322 Dialect."""
324 if TYPE_CHECKING:
325 assert isinstance(self, ClauseElement)
326 return dialect.statement_compiler(dialect, self, **kw)
328 def __str__(self) -> str:
329 return str(self.compile())
332@inspection._self_inspects
333class ClauseElement(
334 SupportsWrappingAnnotations,
335 MemoizedHasCacheKey,
336 HasCopyInternals,
337 ExternallyTraversible,
338 CompilerElement,
339):
340 """Base class for elements of a programmatically constructed SQL
341 expression.
343 """
345 __visit_name__ = "clause"
347 if TYPE_CHECKING:
349 @util.memoized_property
350 def _propagate_attrs(self) -> _PropagateAttrsType:
351 """like annotations, however these propagate outwards liberally
352 as SQL constructs are built, and are set up at construction time.
354 """
355 ...
357 else:
358 _propagate_attrs = util.EMPTY_DICT
360 @util.ro_memoized_property
361 def description(self) -> Optional[str]:
362 return None
364 _is_clone_of: Optional[Self] = None
366 is_clause_element = True
367 is_selectable = False
368 is_dml = False
369 _is_column_element = False
370 _is_keyed_column_element = False
371 _is_table = False
372 _gen_static_annotations_cache_key = False
373 _is_textual = False
374 _is_from_clause = False
375 _is_returns_rows = False
376 _is_text_clause = False
377 _is_from_container = False
378 _is_select_container = False
379 _is_select_base = False
380 _is_select_statement = False
381 _is_bind_parameter = False
382 _is_clause_list = False
383 _is_lambda_element = False
384 _is_singleton_constant = False
385 _is_immutable = False
386 _is_star = False
388 @property
389 def _order_by_label_element(self) -> Optional[Label[Any]]:
390 return None
392 _cache_key_traversal: _CacheKeyTraversalType = None
394 negation_clause: ColumnElement[bool]
396 if typing.TYPE_CHECKING:
398 def get_children(
399 self, *, omit_attrs: typing_Tuple[str, ...] = ..., **kw: Any
400 ) -> Iterable[ClauseElement]: ...
402 @util.ro_non_memoized_property
403 def _from_objects(self) -> List[FromClause]:
404 return []
406 def _set_propagate_attrs(self, values: Mapping[str, Any]) -> Self:
407 # usually, self._propagate_attrs is empty here. one case where it's
408 # not is a subquery against ORM select, that is then pulled as a
409 # property of an aliased class. should all be good
411 # assert not self._propagate_attrs
413 self._propagate_attrs = util.immutabledict(values)
414 return self
416 def _default_compiler(self) -> SQLCompiler:
417 dialect = self._default_dialect()
418 return dialect.statement_compiler(dialect, self) # type: ignore
420 def _clone(self, **kw: Any) -> Self:
421 """Create a shallow copy of this ClauseElement.
423 This method may be used by a generative API. Its also used as
424 part of the "deep" copy afforded by a traversal that combines
425 the _copy_internals() method.
427 """
429 skip = self._memoized_keys
430 c = self.__class__.__new__(self.__class__)
432 if skip:
433 # ensure this iteration remains atomic
434 c.__dict__ = {
435 k: v for k, v in self.__dict__.copy().items() if k not in skip
436 }
437 else:
438 c.__dict__ = self.__dict__.copy()
440 # this is a marker that helps to "equate" clauses to each other
441 # when a Select returns its list of FROM clauses. the cloning
442 # process leaves around a lot of remnants of the previous clause
443 # typically in the form of column expressions still attached to the
444 # old table.
445 cc = self._is_clone_of
446 c._is_clone_of = cc if cc is not None else self
447 return c
449 def _negate_in_binary(self, negated_op, original_op):
450 """a hook to allow the right side of a binary expression to respond
451 to a negation of the binary expression.
453 Used for the special case of expanding bind parameter with IN.
455 """
456 return self
458 def _with_binary_element_type(self, type_):
459 """in the context of binary expression, convert the type of this
460 object to the one given.
462 applies only to :class:`_expression.ColumnElement` classes.
464 """
465 return self
467 @property
468 def _constructor(self): # type: ignore[override]
469 """return the 'constructor' for this ClauseElement.
471 This is for the purposes for creating a new object of
472 this type. Usually, its just the element's __class__.
473 However, the "Annotated" version of the object overrides
474 to return the class of its proxied element.
476 """
477 return self.__class__
479 @HasMemoized.memoized_attribute
480 def _cloned_set(self):
481 """Return the set consisting all cloned ancestors of this
482 ClauseElement.
484 Includes this ClauseElement. This accessor tends to be used for
485 FromClause objects to identify 'equivalent' FROM clauses, regardless
486 of transformative operations.
488 """
489 s = util.column_set()
490 f: Optional[ClauseElement] = self
492 # note this creates a cycle, asserted in test_memusage. however,
493 # turning this into a plain @property adds tends of thousands of method
494 # calls to Core / ORM performance tests, so the small overhead
495 # introduced by the relatively small amount of short term cycles
496 # produced here is preferable
497 while f is not None:
498 s.add(f)
499 f = f._is_clone_of
500 return s
502 def _de_clone(self):
503 while self._is_clone_of is not None:
504 self = self._is_clone_of
505 return self
507 @util.ro_non_memoized_property
508 def entity_namespace(self) -> _EntityNamespace:
509 raise AttributeError(
510 "This SQL expression has no entity namespace "
511 "with which to filter from."
512 )
514 def __getstate__(self):
515 d = self.__dict__.copy()
516 d.pop("_is_clone_of", None)
517 d.pop("_generate_cache_key", None)
518 return d
520 def _execute_on_connection(
521 self,
522 connection: Connection,
523 distilled_params: _CoreMultiExecuteParams,
524 execution_options: CoreExecuteOptionsParameter,
525 ) -> Result[Unpack[TupleAny]]:
526 if self.supports_execution:
527 if TYPE_CHECKING:
528 assert isinstance(self, Executable)
529 return connection._execute_clauseelement(
530 self, distilled_params, execution_options
531 )
532 else:
533 raise exc.ObjectNotExecutableError(self)
535 def _execute_on_scalar(
536 self,
537 connection: Connection,
538 distilled_params: _CoreMultiExecuteParams,
539 execution_options: CoreExecuteOptionsParameter,
540 ) -> Any:
541 """an additional hook for subclasses to provide a different
542 implementation for connection.scalar() vs. connection.execute().
544 .. versionadded:: 2.0
546 """
547 return self._execute_on_connection(
548 connection, distilled_params, execution_options
549 ).scalar()
551 def _get_embedded_bindparams(self) -> Sequence[BindParameter[Any]]:
552 """Return the list of :class:`.BindParameter` objects embedded in the
553 object.
555 This accomplishes the same purpose as ``visitors.traverse()`` or
556 similar would provide, however by making use of the cache key
557 it takes advantage of memoization of the key to result in fewer
558 net method calls, assuming the statement is also going to be
559 executed.
561 """
563 key = self._generate_cache_key()
564 if key is None:
565 bindparams: List[BindParameter[Any]] = []
567 traverse(self, {}, {"bindparam": bindparams.append})
568 return bindparams
570 else:
571 return key.bindparams
573 def unique_params(
574 self,
575 __optionaldict: Optional[Dict[str, Any]] = None,
576 /,
577 **kwargs: Any,
578 ) -> Self:
579 """Return a copy with :func:`_expression.bindparam` elements
580 replaced.
582 Same functionality as :meth:`_expression.ClauseElement.params`,
583 except adds `unique=True`
584 to affected bind parameters so that multiple statements can be
585 used.
587 """
588 return self._replace_params(True, __optionaldict, kwargs)
590 def params(
591 self,
592 __optionaldict: Optional[Mapping[str, Any]] = None,
593 /,
594 **kwargs: Any,
595 ) -> Self:
596 """Return a copy with :func:`_expression.bindparam` elements
597 replaced.
599 Returns a copy of this ClauseElement with
600 :func:`_expression.bindparam`
601 elements replaced with values taken from the given dictionary::
603 >>> clause = column("x") + bindparam("foo")
604 >>> print(clause.compile().params)
605 {'foo':None}
606 >>> print(clause.params({"foo": 7}).compile().params)
607 {'foo':7}
609 """
610 return self._replace_params(False, __optionaldict, kwargs)
612 def _replace_params(
613 self,
614 unique: bool,
615 optionaldict: Optional[Mapping[str, Any]],
616 kwargs: Dict[str, Any],
617 ) -> Self:
618 if optionaldict:
619 kwargs.update(optionaldict)
621 def visit_bindparam(bind: BindParameter[Any]) -> None:
622 if bind.key in kwargs:
623 bind.value = kwargs[bind.key]
624 bind.required = False
625 if unique:
626 bind._convert_to_unique()
628 return cloned_traverse(
629 self,
630 {"maintain_key": True, "detect_subquery_cols": True},
631 {"bindparam": visit_bindparam},
632 )
634 def compare(self, other: ClauseElement, **kw: Any) -> bool:
635 r"""Compare this :class:`_expression.ClauseElement` to
636 the given :class:`_expression.ClauseElement`.
638 Subclasses should override the default behavior, which is a
639 straight identity comparison.
641 \**kw are arguments consumed by subclass ``compare()`` methods and
642 may be used to modify the criteria for comparison
643 (see :class:`_expression.ColumnElement`).
645 """
646 return traversals.compare(self, other, **kw)
648 def self_group(
649 self, against: Optional[OperatorType] = None
650 ) -> ClauseElement:
651 """Apply a 'grouping' to this :class:`_expression.ClauseElement`.
653 This method is overridden by subclasses to return a "grouping"
654 construct, i.e. parenthesis. In particular it's used by "binary"
655 expressions to provide a grouping around themselves when placed into a
656 larger expression, as well as by :func:`_expression.select`
657 constructs when placed into the FROM clause of another
658 :func:`_expression.select`. (Note that subqueries should be
659 normally created using the :meth:`_expression.Select.alias` method,
660 as many
661 platforms require nested SELECT statements to be named).
663 As expressions are composed together, the application of
664 :meth:`self_group` is automatic - end-user code should never
665 need to use this method directly. Note that SQLAlchemy's
666 clause constructs take operator precedence into account -
667 so parenthesis might not be needed, for example, in
668 an expression like ``x OR (y AND z)`` - AND takes precedence
669 over OR.
671 The base :meth:`self_group` method of
672 :class:`_expression.ClauseElement`
673 just returns self.
674 """
675 return self
677 def _ungroup(self) -> ClauseElement:
678 """Return this :class:`_expression.ClauseElement`
679 without any groupings.
680 """
682 return self
684 def _compile_w_cache(
685 self,
686 dialect: Dialect,
687 *,
688 compiled_cache: Optional[CompiledCacheType],
689 column_keys: List[str],
690 for_executemany: bool = False,
691 schema_translate_map: Optional[SchemaTranslateMapType] = None,
692 **kw: Any,
693 ) -> typing_Tuple[
694 Compiled, Optional[Sequence[BindParameter[Any]]], CacheStats
695 ]:
696 elem_cache_key: Optional[CacheKey]
698 if compiled_cache is not None and dialect._supports_statement_cache:
699 elem_cache_key = self._generate_cache_key()
700 else:
701 elem_cache_key = None
703 extracted_params: Optional[Sequence[BindParameter[Any]]]
704 if elem_cache_key is not None:
705 if TYPE_CHECKING:
706 assert compiled_cache is not None
708 cache_key, extracted_params = elem_cache_key
709 key = (
710 dialect,
711 cache_key,
712 tuple(column_keys),
713 bool(schema_translate_map),
714 for_executemany,
715 )
716 compiled_sql = compiled_cache.get(key)
718 if compiled_sql is None:
719 cache_hit = dialect.CACHE_MISS
720 compiled_sql = self._compiler(
721 dialect,
722 cache_key=elem_cache_key,
723 column_keys=column_keys,
724 for_executemany=for_executemany,
725 schema_translate_map=schema_translate_map,
726 **kw,
727 )
728 compiled_cache[key] = compiled_sql
729 else:
730 cache_hit = dialect.CACHE_HIT
731 else:
732 extracted_params = None
733 compiled_sql = self._compiler(
734 dialect,
735 cache_key=elem_cache_key,
736 column_keys=column_keys,
737 for_executemany=for_executemany,
738 schema_translate_map=schema_translate_map,
739 **kw,
740 )
742 if not dialect._supports_statement_cache:
743 cache_hit = dialect.NO_DIALECT_SUPPORT
744 elif compiled_cache is None:
745 cache_hit = dialect.CACHING_DISABLED
746 else:
747 cache_hit = dialect.NO_CACHE_KEY
749 return compiled_sql, extracted_params, cache_hit
751 def __invert__(self):
752 # undocumented element currently used by the ORM for
753 # relationship.contains()
754 if hasattr(self, "negation_clause"):
755 return self.negation_clause
756 else:
757 return self._negate()
759 def _negate(self) -> ClauseElement:
760 # TODO: this code is uncovered and in all likelihood is not included
761 # in any codepath. So this should raise NotImplementedError in 2.1
762 grouped = self.self_group(against=operators.inv)
763 assert isinstance(grouped, ColumnElement)
764 return UnaryExpression(grouped, operator=operators.inv)
766 def __bool__(self):
767 raise TypeError("Boolean value of this clause is not defined")
769 def __repr__(self):
770 friendly = self.description
771 if friendly is None:
772 return object.__repr__(self)
773 else:
774 return "<%s.%s at 0x%x; %s>" % (
775 self.__module__,
776 self.__class__.__name__,
777 id(self),
778 friendly,
779 )
782class DQLDMLClauseElement(ClauseElement):
783 """represents a :class:`.ClauseElement` that compiles to a DQL or DML
784 expression, not DDL.
786 .. versionadded:: 2.0
788 """
790 if typing.TYPE_CHECKING:
792 def _compiler(self, dialect: Dialect, **kw: Any) -> SQLCompiler:
793 """Return a compiler appropriate for this ClauseElement, given a
794 Dialect."""
795 ...
797 def compile( # noqa: A001
798 self,
799 bind: Optional[_HasDialect] = None,
800 dialect: Optional[Dialect] = None,
801 **kw: Any,
802 ) -> SQLCompiler: ...
805class CompilerColumnElement(
806 roles.DMLColumnRole,
807 roles.DDLConstraintColumnRole,
808 roles.ColumnsClauseRole,
809 CompilerElement,
810):
811 """A compiler-only column element used for ad-hoc string compilations.
813 .. versionadded:: 2.0
815 """
817 __slots__ = ()
819 _propagate_attrs = util.EMPTY_DICT
820 _is_collection_aggregate = False
823# SQLCoreOperations should be suiting the ExpressionElementRole
824# and ColumnsClauseRole. however the MRO issues become too elaborate
825# at the moment.
826class SQLCoreOperations(Generic[_T_co], ColumnOperators, TypingOnly):
827 __slots__ = ()
829 # annotations for comparison methods
830 # these are from operators->Operators / ColumnOperators,
831 # redefined with the specific types returned by ColumnElement hierarchies
832 if typing.TYPE_CHECKING:
834 @util.non_memoized_property
835 def _propagate_attrs(self) -> _PropagateAttrsType: ...
837 def operate(
838 self, op: OperatorType, *other: Any, **kwargs: Any
839 ) -> ColumnElement[Any]: ...
841 def reverse_operate(
842 self, op: OperatorType, other: Any, **kwargs: Any
843 ) -> ColumnElement[Any]: ...
845 @overload
846 def op(
847 self,
848 opstring: str,
849 precedence: int = ...,
850 is_comparison: bool = ...,
851 *,
852 return_type: _TypeEngineArgument[_OPT],
853 python_impl: Optional[Callable[..., Any]] = None,
854 ) -> Callable[[Any], BinaryExpression[_OPT]]: ...
856 @overload
857 def op(
858 self,
859 opstring: str,
860 precedence: int = ...,
861 is_comparison: bool = ...,
862 return_type: Optional[_TypeEngineArgument[Any]] = ...,
863 python_impl: Optional[Callable[..., Any]] = ...,
864 ) -> Callable[[Any], BinaryExpression[Any]]: ...
866 def op(
867 self,
868 opstring: str,
869 precedence: int = 0,
870 is_comparison: bool = False,
871 return_type: Optional[_TypeEngineArgument[Any]] = None,
872 python_impl: Optional[Callable[..., Any]] = None,
873 ) -> Callable[[Any], BinaryExpression[Any]]: ...
875 def bool_op(
876 self,
877 opstring: str,
878 precedence: int = 0,
879 python_impl: Optional[Callable[..., Any]] = None,
880 ) -> Callable[[Any], BinaryExpression[bool]]: ...
882 def __and__(self, other: Any) -> BooleanClauseList: ...
884 def __or__(self, other: Any) -> BooleanClauseList: ...
886 def __invert__(self) -> ColumnElement[_T_co]: ...
888 def __lt__(self, other: Any) -> ColumnElement[bool]: ...
890 def __le__(self, other: Any) -> ColumnElement[bool]: ...
892 # declare also that this class has an hash method otherwise
893 # it may be assumed to be None by type checkers since the
894 # object defines __eq__ and python sets it to None in that case:
895 # https://docs.python.org/3/reference/datamodel.html#object.__hash__
896 def __hash__(self) -> int: ...
898 def __eq__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
899 ...
901 def __ne__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
902 ...
904 def is_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
906 def is_not_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
908 def __gt__(self, other: Any) -> ColumnElement[bool]: ...
910 def __ge__(self, other: Any) -> ColumnElement[bool]: ...
912 def __neg__(self) -> UnaryExpression[_T_co]: ...
914 def __contains__(self, other: Any) -> ColumnElement[bool]: ...
916 def __getitem__(self, index: Any) -> ColumnElement[Any]: ...
918 @overload
919 def __lshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
921 @overload
922 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
924 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
926 @overload
927 def __rlshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
929 @overload
930 def __rlshift__(self, other: Any) -> ColumnElement[Any]: ...
932 def __rlshift__(self, other: Any) -> ColumnElement[Any]: ...
934 @overload
935 def __rshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
937 @overload
938 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
940 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
942 @overload
943 def __rrshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
945 @overload
946 def __rrshift__(self, other: Any) -> ColumnElement[Any]: ...
948 def __rrshift__(self, other: Any) -> ColumnElement[Any]: ...
950 def __matmul__(self, other: Any) -> ColumnElement[Any]: ...
952 def __rmatmul__(self, other: Any) -> ColumnElement[Any]: ...
954 @overload
955 def concat(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
957 @overload
958 def concat(self, other: Any) -> ColumnElement[Any]: ...
960 def concat(self, other: Any) -> ColumnElement[Any]: ...
962 def like(
963 self, other: Any, escape: Optional[str] = None
964 ) -> BinaryExpression[bool]: ...
966 def ilike(
967 self, other: Any, escape: Optional[str] = None
968 ) -> BinaryExpression[bool]: ...
970 def bitwise_xor(self, other: Any) -> BinaryExpression[Any]: ...
972 def bitwise_or(self, other: Any) -> BinaryExpression[Any]: ...
974 def bitwise_and(self, other: Any) -> BinaryExpression[Any]: ...
976 def bitwise_not(self) -> UnaryExpression[_T_co]: ...
978 def bitwise_lshift(self, other: Any) -> BinaryExpression[Any]: ...
980 def bitwise_rshift(self, other: Any) -> BinaryExpression[Any]: ...
982 def in_(
983 self,
984 other: Union[
985 Iterable[Any], BindParameter[Any], roles.InElementRole
986 ],
987 ) -> BinaryExpression[bool]: ...
989 def not_in(
990 self,
991 other: Union[
992 Iterable[Any], BindParameter[Any], roles.InElementRole
993 ],
994 ) -> BinaryExpression[bool]: ...
996 def notin_(
997 self,
998 other: Union[
999 Iterable[Any], BindParameter[Any], roles.InElementRole
1000 ],
1001 ) -> BinaryExpression[bool]: ...
1003 def not_like(
1004 self, other: Any, escape: Optional[str] = None
1005 ) -> BinaryExpression[bool]: ...
1007 def notlike(
1008 self, other: Any, escape: Optional[str] = None
1009 ) -> BinaryExpression[bool]: ...
1011 def not_ilike(
1012 self, other: Any, escape: Optional[str] = None
1013 ) -> BinaryExpression[bool]: ...
1015 def notilike(
1016 self, other: Any, escape: Optional[str] = None
1017 ) -> BinaryExpression[bool]: ...
1019 def is_(self, other: Any) -> BinaryExpression[bool]: ...
1021 def is_not(self, other: Any) -> BinaryExpression[bool]: ...
1023 def isnot(self, other: Any) -> BinaryExpression[bool]: ...
1025 def startswith(
1026 self,
1027 other: Any,
1028 escape: Optional[str] = None,
1029 autoescape: bool = False,
1030 ) -> ColumnElement[bool]: ...
1032 def istartswith(
1033 self,
1034 other: Any,
1035 escape: Optional[str] = None,
1036 autoescape: bool = False,
1037 ) -> ColumnElement[bool]: ...
1039 def endswith(
1040 self,
1041 other: Any,
1042 escape: Optional[str] = None,
1043 autoescape: bool = False,
1044 ) -> ColumnElement[bool]: ...
1046 def iendswith(
1047 self,
1048 other: Any,
1049 escape: Optional[str] = None,
1050 autoescape: bool = False,
1051 ) -> ColumnElement[bool]: ...
1053 def contains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1055 def icontains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1057 def match(self, other: Any, **kwargs: Any) -> ColumnElement[bool]: ...
1059 def regexp_match(
1060 self, pattern: Any, flags: Optional[str] = None
1061 ) -> ColumnElement[bool]: ...
1063 def regexp_replace(
1064 self, pattern: Any, replacement: Any, flags: Optional[str] = None
1065 ) -> ColumnElement[str]: ...
1067 def desc(self) -> UnaryExpression[_T_co]: ...
1069 def asc(self) -> UnaryExpression[_T_co]: ...
1071 def nulls_first(self) -> UnaryExpression[_T_co]: ...
1073 def nullsfirst(self) -> UnaryExpression[_T_co]: ...
1075 def nulls_last(self) -> UnaryExpression[_T_co]: ...
1077 def nullslast(self) -> UnaryExpression[_T_co]: ...
1079 def collate(self, collation: str) -> CollationClause: ...
1081 def between(
1082 self, cleft: Any, cright: Any, symmetric: bool = False
1083 ) -> BinaryExpression[bool]: ...
1085 def distinct(self: _SQO[_T_co]) -> UnaryExpression[_T_co]: ...
1087 def any_(self) -> CollectionAggregate[Any]: ...
1089 def all_(self) -> CollectionAggregate[Any]: ...
1091 # numeric overloads. These need more tweaking
1092 # in particular they all need to have a variant for Optiona[_T]
1093 # because Optional only applies to the data side, not the expression
1094 # side
1096 @overload
1097 def __add__(
1098 self: _SQO[_NMT],
1099 other: Any,
1100 ) -> ColumnElement[_NMT]: ...
1102 @overload
1103 def __add__(
1104 self: _SQO[str],
1105 other: Any,
1106 ) -> ColumnElement[str]: ...
1108 @overload
1109 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1111 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1113 @overload
1114 def __radd__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1116 @overload
1117 def __radd__(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
1119 def __radd__(self, other: Any) -> ColumnElement[Any]: ...
1121 @overload
1122 def __sub__(
1123 self: _SQO[_NMT],
1124 other: Any,
1125 ) -> ColumnElement[_NMT]: ...
1127 @overload
1128 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1130 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1132 @overload
1133 def __rsub__(
1134 self: _SQO[_NMT],
1135 other: Any,
1136 ) -> ColumnElement[_NMT]: ...
1138 @overload
1139 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1141 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1143 @overload
1144 def __mul__(
1145 self: _SQO[_NMT],
1146 other: Any,
1147 ) -> ColumnElement[_NMT]: ...
1149 @overload
1150 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1152 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1154 @overload
1155 def __rmul__(
1156 self: _SQO[_NMT],
1157 other: Any,
1158 ) -> ColumnElement[_NMT]: ...
1160 @overload
1161 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1163 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1165 @overload
1166 def __mod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1168 @overload
1169 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1171 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1173 @overload
1174 def __rmod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1176 @overload
1177 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1179 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1181 @overload
1182 def __truediv__(
1183 self: _SQO[int], other: Any
1184 ) -> ColumnElement[_NUMERIC]: ...
1186 @overload
1187 def __truediv__(self: _SQO[_NT], other: Any) -> ColumnElement[_NT]: ...
1189 @overload
1190 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1192 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1194 @overload
1195 def __rtruediv__(
1196 self: _SQO[_NMT], other: Any
1197 ) -> ColumnElement[_NUMERIC]: ...
1199 @overload
1200 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1202 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1204 @overload
1205 def __floordiv__(
1206 self: _SQO[_NMT], other: Any
1207 ) -> ColumnElement[_NMT]: ...
1209 @overload
1210 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1212 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1214 @overload
1215 def __rfloordiv__(
1216 self: _SQO[_NMT], other: Any
1217 ) -> ColumnElement[_NMT]: ...
1219 @overload
1220 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1222 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1225class SQLColumnExpression(
1226 SQLCoreOperations[_T_co], roles.ExpressionElementRole[_T_co], TypingOnly
1227):
1228 """A type that may be used to indicate any SQL column element or object
1229 that acts in place of one.
1231 :class:`.SQLColumnExpression` is a base of
1232 :class:`.ColumnElement`, as well as within the bases of ORM elements
1233 such as :class:`.InstrumentedAttribute`, and may be used in :pep:`484`
1234 typing to indicate arguments or return values that should behave
1235 as column expressions.
1237 .. versionadded:: 2.0.0b4
1240 """
1242 __slots__ = ()
1245_SQO = SQLCoreOperations
1248class ColumnElement(
1249 roles.ColumnArgumentOrKeyRole,
1250 roles.StatementOptionRole,
1251 roles.WhereHavingRole,
1252 roles.BinaryElementRole[_T],
1253 roles.OrderByRole,
1254 roles.ColumnsClauseRole,
1255 roles.LimitOffsetRole,
1256 roles.DMLColumnRole,
1257 roles.DDLConstraintColumnRole,
1258 roles.DDLExpressionRole,
1259 SQLColumnExpression[_T],
1260 DQLDMLClauseElement,
1261):
1262 """Represent a column-oriented SQL expression suitable for usage in the
1263 "columns" clause, WHERE clause etc. of a statement.
1265 While the most familiar kind of :class:`_expression.ColumnElement` is the
1266 :class:`_schema.Column` object, :class:`_expression.ColumnElement`
1267 serves as the basis
1268 for any unit that may be present in a SQL expression, including
1269 the expressions themselves, SQL functions, bound parameters,
1270 literal expressions, keywords such as ``NULL``, etc.
1271 :class:`_expression.ColumnElement`
1272 is the ultimate base class for all such elements.
1274 A wide variety of SQLAlchemy Core functions work at the SQL expression
1275 level, and are intended to accept instances of
1276 :class:`_expression.ColumnElement` as
1277 arguments. These functions will typically document that they accept a
1278 "SQL expression" as an argument. What this means in terms of SQLAlchemy
1279 usually refers to an input which is either already in the form of a
1280 :class:`_expression.ColumnElement` object,
1281 or a value which can be **coerced** into
1282 one. The coercion rules followed by most, but not all, SQLAlchemy Core
1283 functions with regards to SQL expressions are as follows:
1285 * a literal Python value, such as a string, integer or floating
1286 point value, boolean, datetime, ``Decimal`` object, or virtually
1287 any other Python object, will be coerced into a "literal bound
1288 value". This generally means that a :func:`.bindparam` will be
1289 produced featuring the given value embedded into the construct; the
1290 resulting :class:`.BindParameter` object is an instance of
1291 :class:`_expression.ColumnElement`.
1292 The Python value will ultimately be sent
1293 to the DBAPI at execution time as a parameterized argument to the
1294 ``execute()`` or ``executemany()`` methods, after SQLAlchemy
1295 type-specific converters (e.g. those provided by any associated
1296 :class:`.TypeEngine` objects) are applied to the value.
1298 * any special object value, typically ORM-level constructs, which
1299 feature an accessor called ``__clause_element__()``. The Core
1300 expression system looks for this method when an object of otherwise
1301 unknown type is passed to a function that is looking to coerce the
1302 argument into a :class:`_expression.ColumnElement` and sometimes a
1303 :class:`_expression.SelectBase` expression.
1304 It is used within the ORM to
1305 convert from ORM-specific objects like mapped classes and
1306 mapped attributes into Core expression objects.
1308 * The Python ``None`` value is typically interpreted as ``NULL``,
1309 which in SQLAlchemy Core produces an instance of :func:`.null`.
1311 A :class:`_expression.ColumnElement` provides the ability to generate new
1312 :class:`_expression.ColumnElement`
1313 objects using Python expressions. This means that Python operators
1314 such as ``==``, ``!=`` and ``<`` are overloaded to mimic SQL operations,
1315 and allow the instantiation of further :class:`_expression.ColumnElement`
1316 instances
1317 which are composed from other, more fundamental
1318 :class:`_expression.ColumnElement`
1319 objects. For example, two :class:`.ColumnClause` objects can be added
1320 together with the addition operator ``+`` to produce
1321 a :class:`.BinaryExpression`.
1322 Both :class:`.ColumnClause` and :class:`.BinaryExpression` are subclasses
1323 of :class:`_expression.ColumnElement`:
1325 .. sourcecode:: pycon+sql
1327 >>> from sqlalchemy.sql import column
1328 >>> column("a") + column("b")
1329 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
1330 >>> print(column("a") + column("b"))
1331 {printsql}a + b
1333 .. seealso::
1335 :class:`_schema.Column`
1337 :func:`_expression.column`
1339 """
1341 __visit_name__ = "column_element"
1343 primary_key: bool = False
1344 _is_clone_of: Optional[ColumnElement[_T]]
1345 _is_column_element = True
1346 _insert_sentinel: bool = False
1347 _omit_from_statements = False
1348 _is_collection_aggregate = False
1350 foreign_keys: AbstractSet[ForeignKey] = frozenset()
1352 @util.memoized_property
1353 def _proxies(self) -> List[ColumnElement[Any]]:
1354 return []
1356 @util.non_memoized_property
1357 def _tq_label(self) -> Optional[str]:
1358 """The named label that can be used to target
1359 this column in a result set in a "table qualified" context.
1361 This label is almost always the label used when
1362 rendering <expr> AS <label> in a SELECT statement when using
1363 the LABEL_STYLE_TABLENAME_PLUS_COL label style, which is what the
1364 legacy ORM ``Query`` object uses as well.
1366 For a regular Column bound to a Table, this is typically the label
1367 <tablename>_<columnname>. For other constructs, different rules
1368 may apply, such as anonymized labels and others.
1370 .. versionchanged:: 1.4.21 renamed from ``._label``
1372 """
1373 return None
1375 key: Optional[str] = None
1376 """The 'key' that in some circumstances refers to this object in a
1377 Python namespace.
1379 This typically refers to the "key" of the column as present in the
1380 ``.c`` collection of a selectable, e.g. ``sometable.c["somekey"]`` would
1381 return a :class:`_schema.Column` with a ``.key`` of "somekey".
1383 """
1385 @HasMemoized.memoized_attribute
1386 def _tq_key_label(self) -> Optional[str]:
1387 """A label-based version of 'key' that in some circumstances refers
1388 to this object in a Python namespace.
1391 _tq_key_label comes into play when a select() statement is constructed
1392 with apply_labels(); in this case, all Column objects in the ``.c``
1393 collection are rendered as <tablename>_<columnname> in SQL; this is
1394 essentially the value of ._label. But to locate those columns in the
1395 ``.c`` collection, the name is along the lines of <tablename>_<key>;
1396 that's the typical value of .key_label.
1398 .. versionchanged:: 1.4.21 renamed from ``._key_label``
1400 """
1401 return self._proxy_key
1403 @property
1404 def _key_label(self) -> Optional[str]:
1405 """legacy; renamed to _tq_key_label"""
1406 return self._tq_key_label
1408 @property
1409 def _label(self) -> Optional[str]:
1410 """legacy; renamed to _tq_label"""
1411 return self._tq_label
1413 @property
1414 def _non_anon_label(self) -> Optional[str]:
1415 """the 'name' that naturally applies this element when rendered in
1416 SQL.
1418 Concretely, this is the "name" of a column or a label in a
1419 SELECT statement; ``<columnname>`` and ``<labelname>`` below:
1421 .. sourcecode:: sql
1423 SELECT <columnmame> FROM table
1425 SELECT column AS <labelname> FROM table
1427 Above, the two names noted will be what's present in the DBAPI
1428 ``cursor.description`` as the names.
1430 If this attribute returns ``None``, it means that the SQL element as
1431 written does not have a 100% fully predictable "name" that would appear
1432 in the ``cursor.description``. Examples include SQL functions, CAST
1433 functions, etc. While such things do return names in
1434 ``cursor.description``, they are only predictable on a
1435 database-specific basis; e.g. an expression like ``MAX(table.col)`` may
1436 appear as the string ``max`` on one database (like PostgreSQL) or may
1437 appear as the whole expression ``max(table.col)`` on SQLite.
1439 The default implementation looks for a ``.name`` attribute on the
1440 object, as has been the precedent established in SQLAlchemy for many
1441 years. An exception is made on the ``FunctionElement`` subclass
1442 so that the return value is always ``None``.
1444 .. versionadded:: 1.4.21
1448 """
1449 return getattr(self, "name", None)
1451 _render_label_in_columns_clause = True
1452 """A flag used by select._columns_plus_names that helps to determine
1453 we are actually going to render in terms of "SELECT <col> AS <label>".
1454 This flag can be returned as False for some Column objects that want
1455 to be rendered as simple "SELECT <col>"; typically columns that don't have
1456 any parent table and are named the same as what the label would be
1457 in any case.
1459 """
1461 _allow_label_resolve = True
1462 """A flag that can be flipped to prevent a column from being resolvable
1463 by string label name.
1465 The joined eager loader strategy in the ORM uses this, for example.
1467 """
1469 _is_implicitly_boolean = False
1471 _alt_names: Sequence[str] = ()
1473 if TYPE_CHECKING:
1475 def _ungroup(self) -> ColumnElement[_T]: ...
1477 @overload
1478 def self_group(self, against: None = None) -> ColumnElement[_T]: ...
1480 @overload
1481 def self_group(
1482 self, against: Optional[OperatorType] = None
1483 ) -> ColumnElement[Any]: ...
1485 def self_group(
1486 self, against: Optional[OperatorType] = None
1487 ) -> ColumnElement[Any]:
1488 if (
1489 against in (operators.and_, operators.or_, operators._asbool)
1490 and self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity
1491 ):
1492 return AsBoolean(self, operators.is_true, operators.is_false)
1493 elif against in (operators.any_op, operators.all_op):
1494 return Grouping(self)
1495 else:
1496 return self
1498 @overload
1499 def _negate(self: ColumnElement[bool]) -> ColumnElement[bool]: ...
1501 @overload
1502 def _negate(self: ColumnElement[_T]) -> ColumnElement[_T]: ...
1504 def _negate(self) -> ColumnElement[Any]:
1505 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
1506 return AsBoolean(self, operators.is_false, operators.is_true)
1507 else:
1508 grouped = self.self_group(against=operators.inv)
1509 assert isinstance(grouped, ColumnElement)
1510 return UnaryExpression(
1511 grouped,
1512 operator=operators.inv,
1513 )
1515 type: TypeEngine[_T]
1517 if not TYPE_CHECKING:
1519 @util.memoized_property
1520 def type(self) -> TypeEngine[_T]: # noqa: A001
1521 # used for delayed setup of
1522 # type_api
1523 return type_api.NULLTYPE
1525 @HasMemoized.memoized_attribute
1526 def comparator(self) -> TypeEngine.Comparator[_T]:
1527 try:
1528 comparator_factory = self.type.comparator_factory
1529 except AttributeError as err:
1530 raise TypeError(
1531 "Object %r associated with '.type' attribute "
1532 "is not a TypeEngine class or object" % self.type
1533 ) from err
1534 else:
1535 return comparator_factory(self)
1537 def __setstate__(self, state):
1538 self.__dict__.update(state)
1540 def __getattr__(self, key: str) -> Any:
1541 try:
1542 return getattr(self.comparator, key)
1543 except AttributeError as err:
1544 raise AttributeError(
1545 "Neither %r object nor %r object has an attribute %r"
1546 % (
1547 type(self).__name__,
1548 type(self.comparator).__name__,
1549 key,
1550 )
1551 ) from err
1553 def operate(
1554 self,
1555 op: operators.OperatorType,
1556 *other: Any,
1557 **kwargs: Any,
1558 ) -> ColumnElement[Any]:
1559 return op(self.comparator, *other, **kwargs) # type: ignore[no-any-return] # noqa: E501
1561 def reverse_operate(
1562 self, op: operators.OperatorType, other: Any, **kwargs: Any
1563 ) -> ColumnElement[Any]:
1564 return op(other, self.comparator, **kwargs) # type: ignore[no-any-return] # noqa: E501
1566 def _bind_param(
1567 self,
1568 operator: operators.OperatorType,
1569 obj: Any,
1570 type_: Optional[TypeEngine[_T]] = None,
1571 expanding: bool = False,
1572 ) -> BindParameter[_T]:
1573 return BindParameter(
1574 None,
1575 obj,
1576 _compared_to_operator=operator,
1577 type_=type_,
1578 _compared_to_type=self.type,
1579 unique=True,
1580 expanding=expanding,
1581 )
1583 @property
1584 def expression(self) -> ColumnElement[Any]:
1585 """Return a column expression.
1587 Part of the inspection interface; returns self.
1589 """
1590 return self
1592 @property
1593 def _select_iterable(self) -> _SelectIterable:
1594 return (self,)
1596 @util.memoized_property
1597 def base_columns(self) -> FrozenSet[ColumnElement[Any]]:
1598 return frozenset(c for c in self.proxy_set if not c._proxies)
1600 @util.memoized_property
1601 def proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1602 """set of all columns we are proxying
1604 as of 2.0 this is explicitly deannotated columns. previously it was
1605 effectively deannotated columns but wasn't enforced. annotated
1606 columns should basically not go into sets if at all possible because
1607 their hashing behavior is very non-performant.
1609 """
1610 return frozenset([self._deannotate()]).union(
1611 itertools.chain(*[c.proxy_set for c in self._proxies])
1612 )
1614 @util.memoized_property
1615 def _expanded_proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1616 return frozenset(_expand_cloned(self.proxy_set))
1618 def _uncached_proxy_list(self) -> List[ColumnElement[Any]]:
1619 """An 'uncached' version of proxy set.
1621 This list includes annotated columns which perform very poorly in
1622 set operations.
1624 """
1626 return [self] + list(
1627 itertools.chain(*[c._uncached_proxy_list() for c in self._proxies])
1628 )
1630 def shares_lineage(self, othercolumn: ColumnElement[Any]) -> bool:
1631 """Return True if the given :class:`_expression.ColumnElement`
1632 has a common ancestor to this :class:`_expression.ColumnElement`."""
1634 return bool(self.proxy_set.intersection(othercolumn.proxy_set))
1636 def _compare_name_for_result(self, other: ColumnElement[Any]) -> bool:
1637 """Return True if the given column element compares to this one
1638 when targeting within a result row."""
1640 return (
1641 hasattr(other, "name")
1642 and hasattr(self, "name")
1643 and other.name == self.name
1644 )
1646 @HasMemoized.memoized_attribute
1647 def _proxy_key(self) -> Optional[str]:
1648 if self._annotations and "proxy_key" in self._annotations:
1649 return cast(str, self._annotations["proxy_key"])
1651 name = self.key
1652 if not name:
1653 # there's a bit of a seeming contradiction which is that the
1654 # "_non_anon_label" of a column can in fact be an
1655 # "_anonymous_label"; this is when it's on a column that is
1656 # proxying for an anonymous expression in a subquery.
1657 name = self._non_anon_label
1659 if isinstance(name, _anonymous_label):
1660 return None
1661 else:
1662 return name
1664 @HasMemoized.memoized_attribute
1665 def _expression_label(self) -> Optional[str]:
1666 """a suggested label to use in the case that the column has no name,
1667 which should be used if possible as the explicit 'AS <label>'
1668 where this expression would normally have an anon label.
1670 this is essentially mostly what _proxy_key does except it returns
1671 None if the column has a normal name that can be used.
1673 """
1675 if getattr(self, "name", None) is not None:
1676 return None
1677 elif self._annotations and "proxy_key" in self._annotations:
1678 return cast(str, self._annotations["proxy_key"])
1679 else:
1680 return None
1682 def _make_proxy(
1683 self,
1684 selectable: FromClause,
1685 *,
1686 primary_key: ColumnSet,
1687 foreign_keys: Set[KeyedColumnElement[Any]],
1688 name: Optional[str] = None,
1689 key: Optional[str] = None,
1690 name_is_truncatable: bool = False,
1691 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
1692 **kw: Any,
1693 ) -> typing_Tuple[str, ColumnClause[_T]]:
1694 """Create a new :class:`_expression.ColumnElement` representing this
1695 :class:`_expression.ColumnElement` as it appears in the select list of
1696 a descending selectable.
1698 """
1699 if name is None:
1700 name = self._anon_name_label
1701 if key is None:
1702 key = self._proxy_key
1703 else:
1704 key = name
1706 assert key is not None
1708 co: ColumnClause[_T] = ColumnClause(
1709 (
1710 coercions.expect(roles.TruncatedLabelRole, name)
1711 if name_is_truncatable
1712 else name
1713 ),
1714 type_=getattr(self, "type", None),
1715 _selectable=selectable,
1716 )
1718 co._propagate_attrs = selectable._propagate_attrs
1719 if compound_select_cols:
1720 co._proxies = list(compound_select_cols)
1721 else:
1722 co._proxies = [self]
1723 if selectable._is_clone_of is not None:
1724 co._is_clone_of = selectable._is_clone_of.columns.get(key)
1725 return key, co
1727 def cast(self, type_: _TypeEngineArgument[_OPT]) -> Cast[_OPT]:
1728 """Produce a type cast, i.e. ``CAST(<expression> AS <type>)``.
1730 This is a shortcut to the :func:`_expression.cast` function.
1732 .. seealso::
1734 :ref:`tutorial_casts`
1736 :func:`_expression.cast`
1738 :func:`_expression.type_coerce`
1740 """
1741 return Cast(self, type_)
1743 def label(self, name: Optional[str]) -> Label[_T]:
1744 """Produce a column label, i.e. ``<columnname> AS <name>``.
1746 This is a shortcut to the :func:`_expression.label` function.
1748 If 'name' is ``None``, an anonymous label name will be generated.
1750 """
1751 return Label(name, self, self.type)
1753 def _anon_label(
1754 self, seed: Optional[str], add_hash: Optional[int] = None
1755 ) -> _anonymous_label:
1756 while self._is_clone_of is not None:
1757 self = self._is_clone_of
1759 # as of 1.4 anonymous label for ColumnElement uses hash(), not id(),
1760 # as the identifier, because a column and its annotated version are
1761 # the same thing in a SQL statement
1762 hash_value = hash(self)
1764 if add_hash:
1765 # this path is used for disambiguating anon labels that would
1766 # otherwise be the same name for the same element repeated.
1767 # an additional numeric value is factored in for each label.
1769 # shift hash(self) (which is id(self), typically 8 byte integer)
1770 # 16 bits leftward. fill extra add_hash on right
1771 assert add_hash < (2 << 15)
1772 assert seed
1773 hash_value = (hash_value << 16) | add_hash
1775 # extra underscore is added for labels with extra hash
1776 # values, to isolate the "deduped anon" namespace from the
1777 # regular namespace. eliminates chance of these
1778 # manufactured hash values overlapping with regular ones for some
1779 # undefined python interpreter
1780 seed = seed + "_"
1782 if isinstance(seed, _anonymous_label):
1783 # NOTE: the space after the hash is required
1784 return _anonymous_label(f"{seed}%({hash_value} )s")
1786 return _anonymous_label.safe_construct(hash_value, seed or "anon")
1788 @util.memoized_property
1789 def _anon_name_label(self) -> str:
1790 """Provides a constant 'anonymous label' for this ColumnElement.
1792 This is a label() expression which will be named at compile time.
1793 The same label() is returned each time ``anon_label`` is called so
1794 that expressions can reference ``anon_label`` multiple times,
1795 producing the same label name at compile time.
1797 The compiler uses this function automatically at compile time
1798 for expressions that are known to be 'unnamed' like binary
1799 expressions and function calls.
1801 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1802 public and is renamed to _anon_name_label. anon_name exists
1803 for backwards compat
1805 """
1806 name = getattr(self, "name", None)
1807 return self._anon_label(name)
1809 @util.memoized_property
1810 def _anon_key_label(self) -> _anonymous_label:
1811 """Provides a constant 'anonymous key label' for this ColumnElement.
1813 Compare to ``anon_label``, except that the "key" of the column,
1814 if available, is used to generate the label.
1816 This is used when a deduplicating key is placed into the columns
1817 collection of a selectable.
1819 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1820 public and is renamed to _anon_key_label. anon_key_label exists
1821 for backwards compat
1823 """
1824 return self._anon_label(self._proxy_key)
1826 @property
1827 @util.deprecated(
1828 "1.4",
1829 "The :attr:`_expression.ColumnElement.anon_label` attribute is now "
1830 "private, and the public accessor is deprecated.",
1831 )
1832 def anon_label(self) -> str:
1833 return self._anon_name_label
1835 @property
1836 @util.deprecated(
1837 "1.4",
1838 "The :attr:`_expression.ColumnElement.anon_key_label` attribute is "
1839 "now private, and the public accessor is deprecated.",
1840 )
1841 def anon_key_label(self) -> str:
1842 return self._anon_key_label
1844 def _dedupe_anon_label_idx(self, idx: int) -> str:
1845 """label to apply to a column that is anon labeled, but repeated
1846 in the SELECT, so that we have to make an "extra anon" label that
1847 disambiguates it from the previous appearance.
1849 these labels come out like "foo_bar_id__1" and have double underscores
1850 in them.
1852 """
1853 label = getattr(self, "name", None)
1855 # current convention is that if the element doesn't have a
1856 # ".name" (usually because it is not NamedColumn), we try to
1857 # use a "table qualified" form for the "dedupe anon" label,
1858 # based on the notion that a label like
1859 # "CAST(casttest.v1 AS DECIMAL) AS casttest_v1__1" looks better than
1860 # "CAST(casttest.v1 AS DECIMAL) AS anon__1"
1862 if label is None:
1863 return self._dedupe_anon_tq_label_idx(idx)
1864 else:
1865 return self._anon_label(label, add_hash=idx)
1867 @util.memoized_property
1868 def _anon_tq_label(self) -> _anonymous_label:
1869 return self._anon_label(getattr(self, "_tq_label", None))
1871 @util.memoized_property
1872 def _anon_tq_key_label(self) -> _anonymous_label:
1873 return self._anon_label(getattr(self, "_tq_key_label", None))
1875 def _dedupe_anon_tq_label_idx(self, idx: int) -> _anonymous_label:
1876 label = getattr(self, "_tq_label", None) or "anon"
1878 return self._anon_label(label, add_hash=idx)
1881class KeyedColumnElement(ColumnElement[_T]):
1882 """ColumnElement where ``.key`` is non-None."""
1884 _is_keyed_column_element = True
1886 key: str
1889class WrapsColumnExpression(ColumnElement[_T]):
1890 """Mixin that defines a :class:`_expression.ColumnElement`
1891 as a wrapper with special
1892 labeling behavior for an expression that already has a name.
1894 .. versionadded:: 1.4
1896 .. seealso::
1898 :ref:`change_4449`
1901 """
1903 @property
1904 def wrapped_column_expression(self) -> ColumnElement[_T]:
1905 raise NotImplementedError()
1907 @util.non_memoized_property
1908 def _tq_label(self) -> Optional[str]:
1909 wce = self.wrapped_column_expression
1910 if hasattr(wce, "_tq_label"):
1911 return wce._tq_label
1912 else:
1913 return None
1915 @property
1916 def _label(self) -> Optional[str]:
1917 return self._tq_label
1919 @property
1920 def _non_anon_label(self) -> Optional[str]:
1921 return None
1923 @util.non_memoized_property
1924 def _anon_name_label(self) -> str:
1925 wce = self.wrapped_column_expression
1927 # this logic tries to get the WrappedColumnExpression to render
1928 # with "<expr> AS <name>", where "<name>" is the natural name
1929 # within the expression itself. e.g. "CAST(table.foo) AS foo".
1930 if not wce._is_text_clause:
1931 nal = wce._non_anon_label
1932 if nal:
1933 return nal
1934 elif hasattr(wce, "_anon_name_label"):
1935 return wce._anon_name_label
1936 return super()._anon_name_label
1938 def _dedupe_anon_label_idx(self, idx: int) -> str:
1939 wce = self.wrapped_column_expression
1940 nal = wce._non_anon_label
1941 if nal:
1942 return self._anon_label(nal + "_")
1943 else:
1944 return self._dedupe_anon_tq_label_idx(idx)
1946 @property
1947 def _proxy_key(self):
1948 wce = self.wrapped_column_expression
1950 if not wce._is_text_clause:
1951 return wce._proxy_key
1952 return super()._proxy_key
1955class DMLTargetCopy(roles.InElementRole, KeyedColumnElement[_T]):
1956 """Refer to another column's VALUES or SET expression in an INSERT or
1957 UPDATE statement.
1959 See the public-facing :func:`_sql.from_dml_column` constructor for
1960 background.
1962 .. versionadded:: 2.1
1965 """
1967 def __init__(self, column: _DMLOnlyColumnArgument[_T]):
1968 self.column = coercions.expect(roles.ColumnArgumentRole, column)
1969 self.type = self.column.type
1971 __visit_name__ = "dmltargetcopy"
1973 _traverse_internals: _TraverseInternalsType = [
1974 ("column", InternalTraversal.dp_clauseelement),
1975 ]
1978class BindParameter(roles.InElementRole, KeyedColumnElement[_T]):
1979 r"""Represent a "bound expression".
1981 :class:`.BindParameter` is invoked explicitly using the
1982 :func:`.bindparam` function, as in::
1984 from sqlalchemy import bindparam
1986 stmt = select(users_table).where(
1987 users_table.c.name == bindparam("username")
1988 )
1990 Detailed discussion of how :class:`.BindParameter` is used is
1991 at :func:`.bindparam`.
1993 .. seealso::
1995 :func:`.bindparam`
1997 """
1999 __visit_name__ = "bindparam"
2001 _traverse_internals: _TraverseInternalsType = [
2002 ("key", InternalTraversal.dp_anon_name),
2003 ("type", InternalTraversal.dp_type),
2004 ("callable", InternalTraversal.dp_plain_dict),
2005 ("value", InternalTraversal.dp_plain_obj),
2006 ("literal_execute", InternalTraversal.dp_boolean),
2007 ]
2009 key: str
2010 _anon_map_key: Optional[str] = None
2011 type: TypeEngine[_T]
2012 value: Optional[_T]
2014 _is_crud = False
2015 _is_bind_parameter = True
2017 # bindparam implements its own _gen_cache_key() method however
2018 # we check subclasses for this flag, else no cache key is generated
2019 inherit_cache = True
2021 def __init__(
2022 self,
2023 key: Optional[str],
2024 value: Any = _NoArg.NO_ARG,
2025 type_: Optional[_TypeEngineArgument[_T]] = None,
2026 unique: bool = False,
2027 required: Union[bool, Literal[_NoArg.NO_ARG]] = _NoArg.NO_ARG,
2028 quote: Optional[bool] = None,
2029 callable_: Optional[Callable[[], Any]] = None,
2030 expanding: bool = False,
2031 isoutparam: bool = False,
2032 literal_execute: bool = False,
2033 _compared_to_operator: Optional[OperatorType] = None,
2034 _compared_to_type: Optional[TypeEngine[Any]] = None,
2035 _is_crud: bool = False,
2036 ):
2037 if required is _NoArg.NO_ARG:
2038 required = value is _NoArg.NO_ARG and callable_ is None
2039 if value is _NoArg.NO_ARG:
2040 value = None
2042 if quote is not None:
2043 key = quoted_name.construct(key, quote)
2045 if unique:
2046 self.key, self._anon_map_key = (
2047 _anonymous_label.safe_construct_with_key(
2048 id(self),
2049 (
2050 key
2051 if key is not None
2052 and not isinstance(key, _anonymous_label)
2053 else "param"
2054 ),
2055 sanitize_key=True,
2056 )
2057 )
2058 elif key:
2059 self.key = key
2060 else:
2061 self.key, self._anon_map_key = (
2062 _anonymous_label.safe_construct_with_key(id(self), "param")
2063 )
2065 # identifying key that won't change across
2066 # clones, used to identify the bind's logical
2067 # identity
2068 self._identifying_key = self.key
2070 # key that was passed in the first place, used to
2071 # generate new keys
2072 self._orig_key = key or "param"
2074 self.unique = unique
2075 self.value = value
2076 self.callable = callable_
2077 self.isoutparam = isoutparam
2078 self.required = required
2080 # indicate an "expanding" parameter; the compiler sets this
2081 # automatically in the compiler _render_in_expr_w_bindparam method
2082 # for an IN expression
2083 self.expanding = expanding
2085 # this is another hint to help w/ expanding and is typically
2086 # set in the compiler _render_in_expr_w_bindparam method for an
2087 # IN expression
2088 self.expand_op = None
2090 self.literal_execute = literal_execute
2091 if _is_crud:
2092 self._is_crud = True
2094 if type_ is None:
2095 if expanding:
2096 if value:
2097 check_value = value[0]
2098 else:
2099 check_value = type_api._NO_VALUE_IN_LIST
2100 else:
2101 check_value = value
2102 if _compared_to_type is not None:
2103 self.type = _compared_to_type.coerce_compared_value(
2104 _compared_to_operator, check_value
2105 )
2106 else:
2107 self.type = type_api._resolve_value_to_type(check_value)
2108 elif isinstance(type_, type):
2109 self.type = type_()
2110 elif is_tuple_type(type_):
2111 if value:
2112 if expanding:
2113 check_value = value[0]
2114 else:
2115 check_value = value
2116 cast("BindParameter[TupleAny]", self).type = (
2117 type_._resolve_values_to_types(check_value)
2118 )
2119 else:
2120 cast("BindParameter[TupleAny]", self).type = type_
2121 else:
2122 self.type = type_
2124 def _with_value(self, value, maintain_key=False, required=NO_ARG):
2125 """Return a copy of this :class:`.BindParameter` with the given value
2126 set.
2127 """
2128 cloned = self._clone(maintain_key=maintain_key)
2129 cloned.value = value
2130 cloned.callable = None
2131 cloned.required = required if required is not NO_ARG else self.required
2132 if cloned.type is type_api.NULLTYPE:
2133 cloned.type = type_api._resolve_value_to_type(value)
2134 return cloned
2136 @property
2137 def effective_value(self) -> Optional[_T]:
2138 """Return the value of this bound parameter,
2139 taking into account if the ``callable`` parameter
2140 was set.
2142 The ``callable`` value will be evaluated
2143 and returned if present, else ``value``.
2145 """
2146 if self.callable:
2147 # TODO: set up protocol for bind parameter callable
2148 return self.callable() # type: ignore
2149 else:
2150 return self.value
2152 def render_literal_execute(self) -> Self:
2153 """Produce a copy of this bound parameter that will enable the
2154 :paramref:`_sql.BindParameter.literal_execute` flag.
2156 The :paramref:`_sql.BindParameter.literal_execute` flag will
2157 have the effect of the parameter rendered in the compiled SQL
2158 string using ``[POSTCOMPILE]`` form, which is a special form that
2159 is converted to be a rendering of the literal value of the parameter
2160 at SQL execution time. The rationale is to support caching
2161 of SQL statement strings that can embed per-statement literal values,
2162 such as LIMIT and OFFSET parameters, in the final SQL string that
2163 is passed to the DBAPI. Dialects in particular may want to use
2164 this method within custom compilation schemes.
2166 .. versionadded:: 1.4.5
2168 .. seealso::
2170 :ref:`engine_thirdparty_caching`
2172 """
2173 c: Self = ClauseElement._clone(self)
2174 c.literal_execute = True
2175 return c
2177 def _negate_in_binary(self, negated_op, original_op):
2178 if self.expand_op is original_op:
2179 bind = self._clone()
2180 bind.expand_op = negated_op
2181 return bind
2182 else:
2183 return self
2185 def _with_binary_element_type(self, type_: TypeEngine[Any]) -> Self:
2186 c: Self = ClauseElement._clone(self)
2187 c.type = type_
2188 return c
2190 def _clone(self, maintain_key: bool = False, **kw: Any) -> Self:
2191 c: Self = ClauseElement._clone(self, **kw)
2192 # ensure all the BindParameter objects stay in cloned set.
2193 # in #7823, we changed "clone" so that a clone only keeps a reference
2194 # to the "original" element, since for column correspondence, that's
2195 # all we need. However, for BindParam, _cloned_set is used by
2196 # the "cache key bind match" lookup, which means if any of those
2197 # interim BindParameter objects became part of a cache key in the
2198 # cache, we need it. So here, make sure all clones keep carrying
2199 # forward.
2200 c._cloned_set.update(self._cloned_set)
2201 if not maintain_key and self.unique:
2202 c.key, c._anon_map_key = _anonymous_label.safe_construct_with_key(
2203 id(c), c._orig_key or "param", sanitize_key=True
2204 )
2205 return c
2207 def _gen_cache_key(self, anon_map, bindparams):
2208 _gen_cache_ok = self.__class__.__dict__.get("inherit_cache", False)
2210 if not _gen_cache_ok:
2211 if anon_map is not None:
2212 anon_map[NO_CACHE] = True
2213 return None
2215 id_, found = anon_map.get_anon(self)
2216 if found:
2217 return (id_, self.__class__)
2219 if bindparams is not None:
2220 bindparams.append(self)
2222 return (
2223 id_,
2224 self.__class__,
2225 self.type._static_cache_key,
2226 (
2227 anon_map[self._anon_map_key]
2228 if self._anon_map_key is not None
2229 else self.key
2230 ),
2231 self.literal_execute,
2232 )
2234 def _convert_to_unique(self):
2235 if not self.unique:
2236 self.unique = True
2237 self.key, self._anon_map_key = (
2238 _anonymous_label.safe_construct_with_key(
2239 id(self), self._orig_key or "param", sanitize_key=True
2240 )
2241 )
2243 def __getstate__(self):
2244 """execute a deferred value for serialization purposes."""
2246 d = self.__dict__.copy()
2247 v = self.value
2248 if self.callable:
2249 v = self.callable()
2250 d["callable"] = None
2251 d["value"] = v
2252 return d
2254 def __setstate__(self, state):
2255 if state.get("unique", False):
2256 anon_and_key = _anonymous_label.safe_construct_with_key(
2257 id(self), state.get("_orig_key", "param"), sanitize_key=True
2258 )
2259 state["key"], state["_anon_map_key"] = anon_and_key
2260 self.__dict__.update(state)
2262 def __repr__(self):
2263 return "%s(%r, %r, type_=%r)" % (
2264 self.__class__.__name__,
2265 self.key,
2266 self.value,
2267 self.type,
2268 )
2271class TypeClause(DQLDMLClauseElement):
2272 """Handle a type keyword in a SQL statement.
2274 Used by the ``Case`` statement.
2276 """
2278 __visit_name__ = "typeclause"
2280 _traverse_internals: _TraverseInternalsType = [
2281 ("type", InternalTraversal.dp_type)
2282 ]
2283 type: TypeEngine[Any]
2285 def __init__(self, type_: TypeEngine[Any]):
2286 self.type = type_
2289class TextClause(
2290 roles.DDLConstraintColumnRole,
2291 roles.DDLExpressionRole,
2292 roles.StatementOptionRole,
2293 roles.WhereHavingRole,
2294 roles.OrderByRole,
2295 roles.FromClauseRole,
2296 roles.SelectStatementRole,
2297 roles.InElementRole,
2298 Generative,
2299 Executable,
2300 DQLDMLClauseElement,
2301 roles.BinaryElementRole[Any],
2302 inspection.Inspectable["TextClause"],
2303):
2304 """Represent a literal SQL text fragment.
2306 E.g.::
2308 from sqlalchemy import text
2310 t = text("SELECT * FROM users")
2311 result = connection.execute(t)
2313 The :class:`_expression.TextClause` construct is produced using the
2314 :func:`_expression.text`
2315 function; see that function for full documentation.
2317 .. seealso::
2319 :func:`_expression.text`
2321 """
2323 __visit_name__ = "textclause"
2325 _traverse_internals: _TraverseInternalsType = [
2326 ("_bindparams", InternalTraversal.dp_string_clauseelement_dict),
2327 ("text", InternalTraversal.dp_string),
2328 ]
2330 _is_text_clause = True
2332 _is_textual = True
2334 _bind_params_regex = re.compile(r"(?<![:\w\x5c]):(\w+)(?!:)", re.UNICODE)
2335 _is_implicitly_boolean = False
2337 _render_label_in_columns_clause = False
2339 _omit_from_statements = False
2341 _is_collection_aggregate = False
2343 @property
2344 def _hide_froms(self) -> Iterable[FromClause]:
2345 return ()
2347 def __and__(self, other):
2348 # support use in select.where(), query.filter()
2349 return and_(self, other)
2351 @property
2352 def _select_iterable(self) -> _SelectIterable:
2353 return (self,)
2355 # help in those cases where text() is
2356 # interpreted in a column expression situation
2357 key: Optional[str] = None
2358 _label: Optional[str] = None
2360 _allow_label_resolve = False
2362 @property
2363 def _is_star(self): # type: ignore[override]
2364 return self.text == "*"
2366 def __init__(self, text: str):
2367 self._bindparams: Dict[str, BindParameter[Any]] = {}
2369 def repl(m):
2370 self._bindparams[m.group(1)] = BindParameter(m.group(1))
2371 return ":%s" % m.group(1)
2373 # scan the string and search for bind parameter names, add them
2374 # to the list of bindparams
2375 self.text = self._bind_params_regex.sub(repl, text)
2377 @_generative
2378 def bindparams(
2379 self,
2380 *binds: BindParameter[Any],
2381 **names_to_values: Any,
2382 ) -> Self:
2383 """Establish the values and/or types of bound parameters within
2384 this :class:`_expression.TextClause` construct.
2386 Given a text construct such as::
2388 from sqlalchemy import text
2390 stmt = text(
2391 "SELECT id, name FROM user WHERE name=:name AND timestamp=:timestamp"
2392 )
2394 the :meth:`_expression.TextClause.bindparams`
2395 method can be used to establish
2396 the initial value of ``:name`` and ``:timestamp``,
2397 using simple keyword arguments::
2399 stmt = stmt.bindparams(
2400 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2401 )
2403 Where above, new :class:`.BindParameter` objects
2404 will be generated with the names ``name`` and ``timestamp``, and
2405 values of ``jack`` and ``datetime.datetime(2012, 10, 8, 15, 12, 5)``,
2406 respectively. The types will be
2407 inferred from the values given, in this case :class:`.String` and
2408 :class:`.DateTime`.
2410 When specific typing behavior is needed, the positional ``*binds``
2411 argument can be used in which to specify :func:`.bindparam` constructs
2412 directly. These constructs must include at least the ``key``
2413 argument, then an optional value and type::
2415 from sqlalchemy import bindparam
2417 stmt = stmt.bindparams(
2418 bindparam("name", value="jack", type_=String),
2419 bindparam("timestamp", type_=DateTime),
2420 )
2422 Above, we specified the type of :class:`.DateTime` for the
2423 ``timestamp`` bind, and the type of :class:`.String` for the ``name``
2424 bind. In the case of ``name`` we also set the default value of
2425 ``"jack"``.
2427 Additional bound parameters can be supplied at statement execution
2428 time, e.g.::
2430 result = connection.execute(
2431 stmt, timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2432 )
2434 The :meth:`_expression.TextClause.bindparams`
2435 method can be called repeatedly,
2436 where it will re-use existing :class:`.BindParameter` objects to add
2437 new information. For example, we can call
2438 :meth:`_expression.TextClause.bindparams`
2439 first with typing information, and a
2440 second time with value information, and it will be combined::
2442 stmt = text(
2443 "SELECT id, name FROM user WHERE name=:name "
2444 "AND timestamp=:timestamp"
2445 )
2446 stmt = stmt.bindparams(
2447 bindparam("name", type_=String), bindparam("timestamp", type_=DateTime)
2448 )
2449 stmt = stmt.bindparams(
2450 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2451 )
2453 The :meth:`_expression.TextClause.bindparams`
2454 method also supports the concept of
2455 **unique** bound parameters. These are parameters that are
2456 "uniquified" on name at statement compilation time, so that multiple
2457 :func:`_expression.text`
2458 constructs may be combined together without the names
2459 conflicting. To use this feature, specify the
2460 :paramref:`.BindParameter.unique` flag on each :func:`.bindparam`
2461 object::
2463 stmt1 = text("select id from table where name=:name").bindparams(
2464 bindparam("name", value="name1", unique=True)
2465 )
2466 stmt2 = text("select id from table where name=:name").bindparams(
2467 bindparam("name", value="name2", unique=True)
2468 )
2470 union = union_all(stmt1.columns(column("id")), stmt2.columns(column("id")))
2472 The above statement will render as:
2474 .. sourcecode:: sql
2476 select id from table where name=:name_1
2477 UNION ALL select id from table where name=:name_2
2479 """ # noqa: E501
2480 self._bindparams = new_params = self._bindparams.copy()
2482 for bind in binds:
2483 try:
2484 # the regex used for text() currently will not match
2485 # a unique/anonymous key in any case, so use the _orig_key
2486 # so that a text() construct can support unique parameters
2487 existing = new_params[bind._orig_key]
2488 except KeyError as err:
2489 raise exc.ArgumentError(
2490 "This text() construct doesn't define a "
2491 "bound parameter named %r" % bind._orig_key
2492 ) from err
2493 else:
2494 new_params[existing._orig_key] = bind
2496 for key, value in names_to_values.items():
2497 try:
2498 existing = new_params[key]
2499 except KeyError as err:
2500 raise exc.ArgumentError(
2501 "This text() construct doesn't define a "
2502 "bound parameter named %r" % key
2503 ) from err
2504 else:
2505 new_params[key] = existing._with_value(value, required=False)
2506 return self
2508 @util.preload_module("sqlalchemy.sql.selectable")
2509 def columns(
2510 self,
2511 *cols: _ColumnExpressionArgument[Any],
2512 **types: _TypeEngineArgument[Any],
2513 ) -> TextualSelect:
2514 r"""Turn this :class:`_expression.TextClause` object into a
2515 :class:`_expression.TextualSelect`
2516 object that serves the same role as a SELECT
2517 statement.
2519 The :class:`_expression.TextualSelect` is part of the
2520 :class:`_expression.SelectBase`
2521 hierarchy and can be embedded into another statement by using the
2522 :meth:`_expression.TextualSelect.subquery` method to produce a
2523 :class:`.Subquery`
2524 object, which can then be SELECTed from.
2526 This function essentially bridges the gap between an entirely
2527 textual SELECT statement and the SQL expression language concept
2528 of a "selectable"::
2530 from sqlalchemy.sql import column, text
2532 stmt = text("SELECT id, name FROM some_table")
2533 stmt = stmt.columns(column("id"), column("name")).subquery("st")
2535 stmt = (
2536 select(mytable)
2537 .select_from(mytable.join(stmt, mytable.c.name == stmt.c.name))
2538 .where(stmt.c.id > 5)
2539 )
2541 Above, we pass a series of :func:`_expression.column` elements to the
2542 :meth:`_expression.TextClause.columns` method positionally. These
2543 :func:`_expression.column`
2544 elements now become first class elements upon the
2545 :attr:`_expression.TextualSelect.selected_columns` column collection,
2546 which then
2547 become part of the :attr:`.Subquery.c` collection after
2548 :meth:`_expression.TextualSelect.subquery` is invoked.
2550 The column expressions we pass to
2551 :meth:`_expression.TextClause.columns` may
2552 also be typed; when we do so, these :class:`.TypeEngine` objects become
2553 the effective return type of the column, so that SQLAlchemy's
2554 result-set-processing systems may be used on the return values.
2555 This is often needed for types such as date or boolean types, as well
2556 as for unicode processing on some dialect configurations::
2558 stmt = text("SELECT id, name, timestamp FROM some_table")
2559 stmt = stmt.columns(
2560 column("id", Integer),
2561 column("name", Unicode),
2562 column("timestamp", DateTime),
2563 )
2565 for id, name, timestamp in connection.execute(stmt):
2566 print(id, name, timestamp)
2568 As a shortcut to the above syntax, keyword arguments referring to
2569 types alone may be used, if only type conversion is needed::
2571 stmt = text("SELECT id, name, timestamp FROM some_table")
2572 stmt = stmt.columns(id=Integer, name=Unicode, timestamp=DateTime)
2574 for id, name, timestamp in connection.execute(stmt):
2575 print(id, name, timestamp)
2577 The positional form of :meth:`_expression.TextClause.columns`
2578 also provides the
2579 unique feature of **positional column targeting**, which is
2580 particularly useful when using the ORM with complex textual queries. If
2581 we specify the columns from our model to
2582 :meth:`_expression.TextClause.columns`,
2583 the result set will match to those columns positionally, meaning the
2584 name or origin of the column in the textual SQL doesn't matter::
2586 stmt = text(
2587 "SELECT users.id, addresses.id, users.id, "
2588 "users.name, addresses.email_address AS email "
2589 "FROM users JOIN addresses ON users.id=addresses.user_id "
2590 "WHERE users.id = 1"
2591 ).columns(
2592 User.id,
2593 Address.id,
2594 Address.user_id,
2595 User.name,
2596 Address.email_address,
2597 )
2599 query = (
2600 session.query(User)
2601 .from_statement(stmt)
2602 .options(contains_eager(User.addresses))
2603 )
2605 The :meth:`_expression.TextClause.columns` method provides a direct
2606 route to calling :meth:`_expression.FromClause.subquery` as well as
2607 :meth:`_expression.SelectBase.cte`
2608 against a textual SELECT statement::
2610 stmt = stmt.columns(id=Integer, name=String).cte("st")
2612 stmt = select(sometable).where(sometable.c.id == stmt.c.id)
2614 :param \*cols: A series of :class:`_expression.ColumnElement` objects,
2615 typically
2616 :class:`_schema.Column` objects from a :class:`_schema.Table`
2617 or ORM level
2618 column-mapped attributes, representing a set of columns that this
2619 textual string will SELECT from.
2621 :param \**types: A mapping of string names to :class:`.TypeEngine`
2622 type objects indicating the datatypes to use for names that are
2623 SELECTed from the textual string. Prefer to use the ``*cols``
2624 argument as it also indicates positional ordering.
2626 """
2627 selectable = util.preloaded.sql_selectable
2629 input_cols: List[NamedColumn[Any]] = [
2630 coercions.expect(roles.LabeledColumnExprRole, col) for col in cols
2631 ]
2633 positional_input_cols = [
2634 (
2635 ColumnClause(col.key, types.pop(col.key))
2636 if col.key in types
2637 else col
2638 )
2639 for col in input_cols
2640 ]
2641 keyed_input_cols: List[NamedColumn[Any]] = [
2642 ColumnClause(key, type_) for key, type_ in types.items()
2643 ]
2645 elem = selectable.TextualSelect.__new__(selectable.TextualSelect)
2646 elem._init(
2647 self,
2648 positional_input_cols + keyed_input_cols,
2649 positional=bool(positional_input_cols) and not keyed_input_cols,
2650 )
2651 return elem
2653 @property
2654 def type(self) -> TypeEngine[Any]:
2655 return type_api.NULLTYPE
2657 @property
2658 def comparator(self):
2659 # TODO: this seems wrong, it seems like we might not
2660 # be using this method.
2661 return self.type.comparator_factory(self) # type: ignore
2663 def self_group(
2664 self, against: Optional[OperatorType] = None
2665 ) -> Union[Self, Grouping[Any]]:
2666 if against is operators.in_op:
2667 return Grouping(self)
2668 else:
2669 return self
2672class Null(SingletonConstant, roles.ConstExprRole[None], ColumnElement[None]):
2673 """Represent the NULL keyword in a SQL statement.
2675 :class:`.Null` is accessed as a constant via the
2676 :func:`.null` function.
2678 """
2680 __visit_name__ = "null"
2682 _traverse_internals: _TraverseInternalsType = []
2683 _singleton: Null
2685 if not TYPE_CHECKING:
2687 @util.memoized_property
2688 def type(self) -> TypeEngine[_T]: # noqa: A001
2689 return type_api.NULLTYPE
2691 @classmethod
2692 def _instance(cls) -> Null:
2693 """Return a constant :class:`.Null` construct."""
2695 return Null._singleton
2698Null._create_singleton()
2701class False_(
2702 SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]
2703):
2704 """Represent the ``false`` keyword, or equivalent, in a SQL statement.
2706 :class:`.False_` is accessed as a constant via the
2707 :func:`.false` function.
2709 """
2711 __visit_name__ = "false"
2712 _traverse_internals: _TraverseInternalsType = []
2713 _singleton: False_
2715 if not TYPE_CHECKING:
2717 @util.memoized_property
2718 def type(self) -> TypeEngine[_T]: # noqa: A001
2719 return type_api.BOOLEANTYPE
2721 def _negate(self) -> True_:
2722 return True_._singleton
2724 @classmethod
2725 def _instance(cls) -> False_:
2726 return False_._singleton
2729False_._create_singleton()
2732class True_(SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]):
2733 """Represent the ``true`` keyword, or equivalent, in a SQL statement.
2735 :class:`.True_` is accessed as a constant via the
2736 :func:`.true` function.
2738 """
2740 __visit_name__ = "true"
2742 _traverse_internals: _TraverseInternalsType = []
2743 _singleton: True_
2745 if not TYPE_CHECKING:
2747 @util.memoized_property
2748 def type(self) -> TypeEngine[_T]: # noqa: A001
2749 return type_api.BOOLEANTYPE
2751 def _negate(self) -> False_:
2752 return False_._singleton
2754 @classmethod
2755 def _ifnone(
2756 cls, other: Optional[ColumnElement[Any]]
2757 ) -> ColumnElement[Any]:
2758 if other is None:
2759 return cls._instance()
2760 else:
2761 return other
2763 @classmethod
2764 def _instance(cls) -> True_:
2765 return True_._singleton
2768True_._create_singleton()
2771class ElementList(DQLDMLClauseElement):
2772 """Describe a list of clauses that will be space separated.
2774 This is a minimal version of :class:`.ClauseList` which is used by
2775 the :class:`.HasSyntaxExtension` class. It does not do any coercions
2776 so should be used internally only.
2778 .. versionadded:: 2.1
2780 """
2782 __visit_name__ = "element_list"
2784 _traverse_internals: _TraverseInternalsType = [
2785 ("clauses", InternalTraversal.dp_clauseelement_tuple),
2786 ]
2788 clauses: typing_Tuple[ClauseElement, ...]
2790 def __init__(self, clauses: Sequence[ClauseElement]):
2791 self.clauses = tuple(clauses)
2794class OrderByList(
2795 roles.OrderByRole,
2796 operators.OrderingOperators,
2797 DQLDMLClauseElement,
2798):
2799 """Describe a list of clauses that will be comma separated to nest
2800 within an ORDER BY.
2802 .. versionadded:: 2.1
2804 """
2806 __visit_name__ = "order_by_list"
2808 _traverse_internals: _TraverseInternalsType = [
2809 ("clauses", InternalTraversal.dp_clauseelement_tuple),
2810 ]
2812 clauses: List[ColumnElement[Any]]
2814 def __init__(
2815 self,
2816 clauses: Iterable[Union[OrderByList, _ColumnExpressionArgument[Any]]],
2817 ):
2818 text_converter_role: Type[roles.SQLRole] = roles.ByOfRole
2819 self._text_converter_role = text_converter_role
2821 self.clauses = [
2822 coercions.expect(
2823 text_converter_role, clause, apply_propagate_attrs=self
2824 )
2825 for clause in clauses
2826 ]
2828 def __iter__(self) -> Iterator[ColumnElement[Any]]:
2829 return iter(self.clauses)
2831 def __len__(self) -> int:
2832 return len(self.clauses)
2834 @property
2835 def _select_iterable(self) -> _SelectIterable:
2836 return itertools.chain.from_iterable(
2837 [elem._select_iterable for elem in self.clauses]
2838 )
2840 @util.ro_non_memoized_property
2841 def _from_objects(self) -> List[FromClause]:
2842 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
2844 def self_group(
2845 self, against: Optional[OperatorType] = None
2846 ) -> Union[Self, Grouping[Any]]:
2847 return self
2849 def desc(self) -> OrderByList:
2850 return OrderByList([e.desc() for e in self.clauses])
2852 def asc(self) -> OrderByList:
2853 return OrderByList([e.asc() for e in self.clauses])
2855 def nulls_first(self) -> OrderByList:
2856 return OrderByList([e.nulls_first() for e in self.clauses])
2858 def nulls_last(self) -> OrderByList:
2859 return OrderByList([e.nulls_last() for e in self.clauses])
2862class ClauseList(
2863 roles.InElementRole,
2864 roles.OrderByRole,
2865 roles.ColumnsClauseRole,
2866 roles.DMLColumnRole,
2867 DQLDMLClauseElement,
2868):
2869 """Describe a list of clauses, separated by an operator.
2871 By default, is comma-separated, such as a column listing.
2873 """
2875 __visit_name__ = "clauselist"
2877 # Used by ORM context.py to identify ClauseList objects in legacy
2878 # composite attribute queries (see test_query_cols_legacy test)
2879 _is_clause_list = True
2881 _traverse_internals: _TraverseInternalsType = [
2882 ("clauses", InternalTraversal.dp_clauseelement_list),
2883 ("operator", InternalTraversal.dp_operator),
2884 ]
2886 clauses: List[ColumnElement[Any]]
2888 def __init__(
2889 self,
2890 *clauses: _ColumnExpressionArgument[Any],
2891 operator: OperatorType = operators.comma_op,
2892 group: bool = True,
2893 group_contents: bool = True,
2894 _literal_as_text_role: Type[roles.SQLRole] = roles.WhereHavingRole,
2895 ):
2896 self.operator = operator
2897 self.group = group
2898 self.group_contents = group_contents
2899 clauses_iterator: Iterable[_ColumnExpressionArgument[Any]] = clauses
2900 text_converter_role: Type[roles.SQLRole] = _literal_as_text_role
2901 self._text_converter_role = text_converter_role
2903 if self.group_contents:
2904 self.clauses = [
2905 coercions.expect(
2906 text_converter_role, clause, apply_propagate_attrs=self
2907 ).self_group(against=self.operator)
2908 for clause in clauses_iterator
2909 ]
2910 else:
2911 self.clauses = [
2912 coercions.expect(
2913 text_converter_role, clause, apply_propagate_attrs=self
2914 )
2915 for clause in clauses_iterator
2916 ]
2917 self._is_implicitly_boolean = operators.is_boolean(self.operator)
2919 @classmethod
2920 def _construct_raw(
2921 cls,
2922 operator: OperatorType,
2923 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
2924 ) -> ClauseList:
2925 self = cls.__new__(cls)
2926 self.clauses = list(clauses) if clauses else []
2927 self.group = True
2928 self.operator = operator
2929 self.group_contents = True
2930 self._is_implicitly_boolean = False
2931 return self
2933 def __iter__(self) -> Iterator[ColumnElement[Any]]:
2934 return iter(self.clauses)
2936 def __len__(self) -> int:
2937 return len(self.clauses)
2939 @property
2940 def _select_iterable(self) -> _SelectIterable:
2941 return itertools.chain.from_iterable(
2942 [elem._select_iterable for elem in self.clauses]
2943 )
2945 def append(self, clause):
2946 if self.group_contents:
2947 self.clauses.append(
2948 coercions.expect(self._text_converter_role, clause).self_group(
2949 against=self.operator
2950 )
2951 )
2952 else:
2953 self.clauses.append(
2954 coercions.expect(self._text_converter_role, clause)
2955 )
2957 @util.ro_non_memoized_property
2958 def _from_objects(self) -> List[FromClause]:
2959 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
2961 def self_group(
2962 self, against: Optional[OperatorType] = None
2963 ) -> Union[Self, Grouping[Any]]:
2964 if self.group and operators.is_precedent(self.operator, against):
2965 return Grouping(self)
2966 else:
2967 return self
2970class OperatorExpression(ColumnElement[_T]):
2971 """base for expressions that contain an operator and operands
2973 .. versionadded:: 2.0
2975 """
2977 operator: OperatorType
2978 type: TypeEngine[_T]
2980 group: bool = True
2982 @property
2983 def is_comparison(self):
2984 return operators.is_comparison(self.operator)
2986 def self_group(
2987 self, against: Optional[OperatorType] = None
2988 ) -> Union[Self, Grouping[_T]]:
2989 if (
2990 self.group
2991 and operators.is_precedent(self.operator, against)
2992 or (
2993 # a negate against a non-boolean operator
2994 # doesn't make too much sense but we should
2995 # group for that
2996 against is operators.inv
2997 and not operators.is_boolean(self.operator)
2998 )
2999 ):
3000 return Grouping(self)
3001 else:
3002 return self
3004 @property
3005 def _flattened_operator_clauses(
3006 self,
3007 ) -> typing_Tuple[ColumnElement[Any], ...]:
3008 raise NotImplementedError()
3010 @classmethod
3011 def _construct_for_op(
3012 cls,
3013 left: ColumnElement[Any],
3014 right: ColumnElement[Any],
3015 op: OperatorType,
3016 *,
3017 type_: TypeEngine[_T],
3018 negate: Optional[OperatorType] = None,
3019 modifiers: Optional[Mapping[str, Any]] = None,
3020 ) -> OperatorExpression[_T]:
3021 if operators.is_associative(op):
3022 assert (
3023 negate is None
3024 ), f"negate not supported for associative operator {op}"
3026 multi = False
3027 if getattr(
3028 left, "operator", None
3029 ) is op and type_._compare_type_affinity(left.type):
3030 multi = True
3031 left_flattened = left._flattened_operator_clauses
3032 else:
3033 left_flattened = (left,)
3035 if getattr(
3036 right, "operator", None
3037 ) is op and type_._compare_type_affinity(right.type):
3038 multi = True
3039 right_flattened = right._flattened_operator_clauses
3040 else:
3041 right_flattened = (right,)
3043 if multi:
3044 return ExpressionClauseList._construct_for_list(
3045 op,
3046 type_,
3047 *(left_flattened + right_flattened),
3048 )
3050 if right._is_collection_aggregate:
3051 negate = None
3053 return BinaryExpression(
3054 left, right, op, type_=type_, negate=negate, modifiers=modifiers
3055 )
3058class ExpressionClauseList(OperatorExpression[_T]):
3059 """Describe a list of clauses, separated by an operator,
3060 in a column expression context.
3062 :class:`.ExpressionClauseList` differs from :class:`.ClauseList` in that
3063 it represents a column-oriented DQL expression only, not an open ended
3064 list of anything comma separated.
3066 .. versionadded:: 2.0
3068 """
3070 __visit_name__ = "expression_clauselist"
3072 _traverse_internals: _TraverseInternalsType = [
3073 ("clauses", InternalTraversal.dp_clauseelement_tuple),
3074 ("operator", InternalTraversal.dp_operator),
3075 ]
3077 clauses: typing_Tuple[ColumnElement[Any], ...]
3079 group: bool
3081 def __init__(
3082 self,
3083 operator: OperatorType,
3084 *clauses: _ColumnExpressionArgument[Any],
3085 type_: Optional[_TypeEngineArgument[_T]] = None,
3086 ):
3087 self.operator = operator
3089 self.clauses = tuple(
3090 coercions.expect(
3091 roles.ExpressionElementRole, clause, apply_propagate_attrs=self
3092 )
3093 for clause in clauses
3094 )
3095 self._is_implicitly_boolean = operators.is_boolean(self.operator)
3096 self.type = type_api.to_instance(type_) # type: ignore
3098 @property
3099 def _flattened_operator_clauses(
3100 self,
3101 ) -> typing_Tuple[ColumnElement[Any], ...]:
3102 return self.clauses
3104 def __iter__(self) -> Iterator[ColumnElement[Any]]:
3105 return iter(self.clauses)
3107 def __len__(self) -> int:
3108 return len(self.clauses)
3110 @property
3111 def _select_iterable(self) -> _SelectIterable:
3112 return (self,)
3114 @util.ro_non_memoized_property
3115 def _from_objects(self) -> List[FromClause]:
3116 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
3118 def _append_inplace(self, clause: ColumnElement[Any]) -> None:
3119 self.clauses += (clause,)
3121 @classmethod
3122 def _construct_for_list(
3123 cls,
3124 operator: OperatorType,
3125 type_: TypeEngine[_T],
3126 *clauses: ColumnElement[Any],
3127 group: bool = True,
3128 ) -> ExpressionClauseList[_T]:
3129 self = cls.__new__(cls)
3130 self.group = group
3131 if group:
3132 self.clauses = tuple(
3133 c.self_group(against=operator) for c in clauses
3134 )
3135 else:
3136 self.clauses = clauses
3137 self.operator = operator
3138 self.type = type_
3139 for c in clauses:
3140 if c._propagate_attrs:
3141 self._propagate_attrs = c._propagate_attrs
3142 break
3143 return self
3145 def _negate(self) -> Any:
3146 grouped = self.self_group(against=operators.inv)
3147 assert isinstance(grouped, ColumnElement)
3148 return UnaryExpression(grouped, operator=operators.inv)
3151class BooleanClauseList(ExpressionClauseList[bool]):
3152 __visit_name__ = "expression_clauselist"
3153 inherit_cache = True
3155 def __init__(self, *arg, **kw):
3156 raise NotImplementedError(
3157 "BooleanClauseList has a private constructor"
3158 )
3160 @classmethod
3161 def _process_clauses_for_boolean(
3162 cls,
3163 operator: OperatorType,
3164 continue_on: Any,
3165 skip_on: Any,
3166 clauses: Iterable[ColumnElement[Any]],
3167 ) -> typing_Tuple[int, List[ColumnElement[Any]]]:
3168 has_continue_on = None
3170 convert_clauses = []
3172 against = operators._asbool
3173 lcc = 0
3175 for clause in clauses:
3176 if clause is continue_on:
3177 # instance of continue_on, like and_(x, y, True, z), store it
3178 # if we didn't find one already, we will use it if there
3179 # are no other expressions here.
3180 has_continue_on = clause
3181 elif clause is skip_on:
3182 # instance of skip_on, e.g. and_(x, y, False, z), cancels
3183 # the rest out
3184 convert_clauses = [clause]
3185 lcc = 1
3186 break
3187 else:
3188 if not lcc:
3189 lcc = 1
3190 else:
3191 against = operator
3192 # technically this would be len(convert_clauses) + 1
3193 # however this only needs to indicate "greater than one"
3194 lcc = 2
3195 convert_clauses.append(clause)
3197 if not convert_clauses and has_continue_on is not None:
3198 convert_clauses = [has_continue_on]
3199 lcc = 1
3201 return lcc, [c.self_group(against=against) for c in convert_clauses]
3203 @classmethod
3204 def _construct(
3205 cls,
3206 operator: OperatorType,
3207 continue_on: Any,
3208 skip_on: Any,
3209 initial_clause: Any = _NoArg.NO_ARG,
3210 *clauses: Any,
3211 **kw: Any,
3212 ) -> ColumnElement[Any]:
3213 if initial_clause is _NoArg.NO_ARG:
3214 # no elements period. deprecated use case. return an empty
3215 # ClauseList construct that generates nothing unless it has
3216 # elements added to it.
3217 name = operator.__name__
3219 util.warn_deprecated(
3220 f"Invoking {name}() without arguments is deprecated, and "
3221 f"will be disallowed in a future release. For an empty "
3222 f"""{name}() construct, use '{name}({
3223 'true()' if continue_on is True_._singleton else 'false()'
3224 }, *args)' """
3225 f"""or '{name}({
3226 'True' if continue_on is True_._singleton else 'False'
3227 }, *args)'.""",
3228 version="1.4",
3229 )
3230 return cls._construct_raw(operator)
3232 lcc, convert_clauses = cls._process_clauses_for_boolean(
3233 operator,
3234 continue_on,
3235 skip_on,
3236 [
3237 coercions.expect(roles.WhereHavingRole, clause)
3238 for clause in util.coerce_generator_arg(
3239 (initial_clause,) + clauses
3240 )
3241 ],
3242 )
3244 if lcc > 1:
3245 # multiple elements. Return regular BooleanClauseList
3246 # which will link elements against the operator.
3248 flattened_clauses = itertools.chain.from_iterable(
3249 (
3250 (c for c in to_flat._flattened_operator_clauses)
3251 if getattr(to_flat, "operator", None) is operator
3252 else (to_flat,)
3253 )
3254 for to_flat in convert_clauses
3255 )
3257 return cls._construct_raw(operator, flattened_clauses) # type: ignore # noqa: E501
3258 else:
3259 assert lcc
3260 # just one element. return it as a single boolean element,
3261 # not a list and discard the operator.
3262 return convert_clauses[0]
3264 @classmethod
3265 def _construct_for_whereclause(
3266 cls, clauses: Iterable[ColumnElement[Any]]
3267 ) -> Optional[ColumnElement[bool]]:
3268 operator, continue_on, skip_on = (
3269 operators.and_,
3270 True_._singleton,
3271 False_._singleton,
3272 )
3274 lcc, convert_clauses = cls._process_clauses_for_boolean(
3275 operator,
3276 continue_on,
3277 skip_on,
3278 clauses, # these are assumed to be coerced already
3279 )
3281 if lcc > 1:
3282 # multiple elements. Return regular BooleanClauseList
3283 # which will link elements against the operator.
3284 return cls._construct_raw(operator, convert_clauses)
3285 elif lcc == 1:
3286 # just one element. return it as a single boolean element,
3287 # not a list and discard the operator.
3288 return convert_clauses[0]
3289 else:
3290 return None
3292 @classmethod
3293 def _construct_raw(
3294 cls,
3295 operator: OperatorType,
3296 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
3297 ) -> BooleanClauseList:
3298 self = cls.__new__(cls)
3299 self.clauses = tuple(clauses) if clauses else ()
3300 self.group = True
3301 self.operator = operator
3302 self.type = type_api.BOOLEANTYPE
3303 self._is_implicitly_boolean = True
3304 return self
3306 @classmethod
3307 def and_(
3308 cls,
3309 initial_clause: Union[
3310 Literal[True], _ColumnExpressionArgument[bool], _NoArg
3311 ] = _NoArg.NO_ARG,
3312 *clauses: _ColumnExpressionArgument[bool],
3313 ) -> ColumnElement[bool]:
3314 r"""Produce a conjunction of expressions joined by ``AND``.
3316 See :func:`_sql.and_` for full documentation.
3317 """
3318 return cls._construct(
3319 operators.and_,
3320 True_._singleton,
3321 False_._singleton,
3322 initial_clause,
3323 *clauses,
3324 )
3326 @classmethod
3327 def or_(
3328 cls,
3329 initial_clause: Union[
3330 Literal[False], _ColumnExpressionArgument[bool], _NoArg
3331 ] = _NoArg.NO_ARG,
3332 *clauses: _ColumnExpressionArgument[bool],
3333 ) -> ColumnElement[bool]:
3334 """Produce a conjunction of expressions joined by ``OR``.
3336 See :func:`_sql.or_` for full documentation.
3337 """
3338 return cls._construct(
3339 operators.or_,
3340 False_._singleton,
3341 True_._singleton,
3342 initial_clause,
3343 *clauses,
3344 )
3346 @property
3347 def _select_iterable(self) -> _SelectIterable:
3348 return (self,)
3350 def self_group(
3351 self, against: Optional[OperatorType] = None
3352 ) -> Union[Self, Grouping[bool]]:
3353 if not self.clauses:
3354 return self
3355 else:
3356 return super().self_group(against=against)
3359and_ = BooleanClauseList.and_
3360or_ = BooleanClauseList.or_
3363class Tuple(ClauseList, ColumnElement[TupleAny]):
3364 """Represent a SQL tuple."""
3366 __visit_name__ = "tuple"
3368 _traverse_internals: _TraverseInternalsType = (
3369 ClauseList._traverse_internals + []
3370 )
3372 type: TupleType
3374 @util.preload_module("sqlalchemy.sql.sqltypes")
3375 def __init__(
3376 self,
3377 *clauses: _ColumnExpressionArgument[Any],
3378 types: Optional[Sequence[_TypeEngineArgument[Any]]] = None,
3379 ):
3380 sqltypes = util.preloaded.sql_sqltypes
3382 if types is None:
3383 init_clauses: List[ColumnElement[Any]] = [
3384 coercions.expect(roles.ExpressionElementRole, c)
3385 for c in clauses
3386 ]
3387 else:
3388 if len(types) != len(clauses):
3389 raise exc.ArgumentError(
3390 "Wrong number of elements for %d-tuple: %r "
3391 % (len(types), clauses)
3392 )
3393 init_clauses = [
3394 coercions.expect(
3395 roles.ExpressionElementRole,
3396 c,
3397 type_=typ if not typ._isnull else None,
3398 )
3399 for typ, c in zip(types, clauses)
3400 ]
3402 self.type = sqltypes.TupleType(*[arg.type for arg in init_clauses])
3403 super().__init__(*init_clauses)
3405 @property
3406 def _select_iterable(self) -> _SelectIterable:
3407 return (self,)
3409 def _bind_param(self, operator, obj, type_=None, expanding=False):
3410 if expanding:
3411 return BindParameter(
3412 None,
3413 value=obj,
3414 _compared_to_operator=operator,
3415 unique=True,
3416 expanding=True,
3417 type_=type_,
3418 _compared_to_type=self.type,
3419 )
3420 else:
3421 return Tuple(
3422 *[
3423 BindParameter(
3424 None,
3425 o,
3426 _compared_to_operator=operator,
3427 _compared_to_type=compared_to_type,
3428 unique=True,
3429 type_=type_,
3430 )
3431 for o, compared_to_type in zip(obj, self.type.types)
3432 ]
3433 )
3435 def self_group(self, against: Optional[OperatorType] = None) -> Self:
3436 # Tuple is parenthesized by definition.
3437 return self
3440class Case(ColumnElement[_T]):
3441 """Represent a ``CASE`` expression.
3443 :class:`.Case` is produced using the :func:`.case` factory function,
3444 as in::
3446 from sqlalchemy import case
3448 stmt = select(users_table).where(
3449 case(
3450 (users_table.c.name == "wendy", "W"),
3451 (users_table.c.name == "jack", "J"),
3452 else_="E",
3453 )
3454 )
3456 Details on :class:`.Case` usage is at :func:`.case`.
3458 .. seealso::
3460 :func:`.case`
3462 """
3464 __visit_name__ = "case"
3466 _traverse_internals: _TraverseInternalsType = [
3467 ("value", InternalTraversal.dp_clauseelement),
3468 ("whens", InternalTraversal.dp_clauseelement_tuples),
3469 ("else_", InternalTraversal.dp_clauseelement),
3470 ]
3472 # for case(), the type is derived from the whens. so for the moment
3473 # users would have to cast() the case to get a specific type
3475 whens: List[typing_Tuple[ColumnElement[bool], ColumnElement[_T]]]
3476 else_: Optional[ColumnElement[_T]]
3477 value: Optional[ColumnElement[Any]]
3479 def __init__(
3480 self,
3481 *whens: Union[
3482 typing_Tuple[_ColumnExpressionArgument[bool], Any],
3483 Mapping[Any, Any],
3484 ],
3485 value: Optional[Any] = None,
3486 else_: Optional[Any] = None,
3487 ):
3488 new_whens: Iterable[Any] = coercions._expression_collection_was_a_list(
3489 "whens", "case", whens
3490 )
3491 try:
3492 new_whens = util.dictlike_iteritems(new_whens)
3493 except TypeError:
3494 pass
3496 self.whens = [
3497 (
3498 coercions.expect(
3499 roles.ExpressionElementRole,
3500 c,
3501 apply_propagate_attrs=self,
3502 ).self_group(),
3503 coercions.expect(roles.ExpressionElementRole, r),
3504 )
3505 for (c, r) in new_whens
3506 ]
3508 if value is None:
3509 self.value = None
3510 else:
3511 self.value = coercions.expect(roles.ExpressionElementRole, value)
3513 if else_ is not None:
3514 self.else_ = coercions.expect(roles.ExpressionElementRole, else_)
3515 else:
3516 self.else_ = None
3518 type_ = next(
3519 (
3520 then.type
3521 # Iterate `whens` in reverse to match previous behaviour
3522 # where type of final element took priority
3523 for *_, then in reversed(self.whens)
3524 if not then.type._isnull
3525 ),
3526 self.else_.type if self.else_ is not None else type_api.NULLTYPE,
3527 )
3528 self.type = cast(_T, type_)
3530 @util.ro_non_memoized_property
3531 def _from_objects(self) -> List[FromClause]:
3532 return list(
3533 itertools.chain(*[x._from_objects for x in self.get_children()])
3534 )
3537class Cast(WrapsColumnExpression[_T]):
3538 """Represent a ``CAST`` expression.
3540 :class:`.Cast` is produced using the :func:`.cast` factory function,
3541 as in::
3543 from sqlalchemy import cast, Numeric
3545 stmt = select(cast(product_table.c.unit_price, Numeric(10, 4)))
3547 Details on :class:`.Cast` usage is at :func:`.cast`.
3549 .. seealso::
3551 :ref:`tutorial_casts`
3553 :func:`.cast`
3555 :func:`.try_cast`
3557 :func:`.type_coerce` - an alternative to CAST that coerces the type
3558 on the Python side only, which is often sufficient to generate the
3559 correct SQL and data coercion.
3561 """
3563 __visit_name__ = "cast"
3565 _traverse_internals: _TraverseInternalsType = [
3566 ("clause", InternalTraversal.dp_clauseelement),
3567 ("type", InternalTraversal.dp_type),
3568 ]
3570 clause: ColumnElement[Any]
3571 type: TypeEngine[_T]
3572 typeclause: TypeClause
3574 def __init__(
3575 self,
3576 expression: _ColumnExpressionArgument[Any],
3577 type_: _TypeEngineArgument[_T],
3578 ):
3579 self.type = type_api.to_instance(type_)
3580 self.clause = coercions.expect(
3581 roles.ExpressionElementRole,
3582 expression,
3583 type_=self.type,
3584 apply_propagate_attrs=self,
3585 )
3586 self.typeclause = TypeClause(self.type)
3588 @util.ro_non_memoized_property
3589 def _from_objects(self) -> List[FromClause]:
3590 return self.clause._from_objects
3592 @property
3593 def wrapped_column_expression(self):
3594 return self.clause
3597class TryCast(Cast[_T]):
3598 """Represent a TRY_CAST expression.
3600 Details on :class:`.TryCast` usage is at :func:`.try_cast`.
3602 .. seealso::
3604 :func:`.try_cast`
3606 :ref:`tutorial_casts`
3607 """
3609 __visit_name__ = "try_cast"
3610 inherit_cache = True
3613class TypeCoerce(WrapsColumnExpression[_T]):
3614 """Represent a Python-side type-coercion wrapper.
3616 :class:`.TypeCoerce` supplies the :func:`_expression.type_coerce`
3617 function; see that function for usage details.
3619 .. seealso::
3621 :func:`_expression.type_coerce`
3623 :func:`.cast`
3625 """
3627 __visit_name__ = "type_coerce"
3629 _traverse_internals: _TraverseInternalsType = [
3630 ("clause", InternalTraversal.dp_clauseelement),
3631 ("type", InternalTraversal.dp_type),
3632 ]
3634 clause: ColumnElement[Any]
3635 type: TypeEngine[_T]
3637 def __init__(
3638 self,
3639 expression: _ColumnExpressionArgument[Any],
3640 type_: _TypeEngineArgument[_T],
3641 ):
3642 self.type = type_api.to_instance(type_)
3643 self.clause = coercions.expect(
3644 roles.ExpressionElementRole,
3645 expression,
3646 type_=self.type,
3647 apply_propagate_attrs=self,
3648 )
3650 @util.ro_non_memoized_property
3651 def _from_objects(self) -> List[FromClause]:
3652 return self.clause._from_objects
3654 @HasMemoized.memoized_attribute
3655 def typed_expression(self):
3656 if isinstance(self.clause, BindParameter):
3657 bp = self.clause._clone()
3658 bp.type = self.type
3659 return bp
3660 else:
3661 return self.clause
3663 @property
3664 def wrapped_column_expression(self):
3665 return self.clause
3667 def self_group(
3668 self, against: Optional[OperatorType] = None
3669 ) -> TypeCoerce[_T]:
3670 grouped = self.clause.self_group(against=against)
3671 if grouped is not self.clause:
3672 return TypeCoerce(grouped, self.type)
3673 else:
3674 return self
3677class Extract(ColumnElement[int]):
3678 """Represent a SQL EXTRACT clause, ``extract(field FROM expr)``."""
3680 __visit_name__ = "extract"
3682 _traverse_internals: _TraverseInternalsType = [
3683 ("expr", InternalTraversal.dp_clauseelement),
3684 ("field", InternalTraversal.dp_string),
3685 ]
3687 expr: ColumnElement[Any]
3688 field: str
3690 def __init__(self, field: str, expr: _ColumnExpressionArgument[Any]):
3691 self.type = type_api.INTEGERTYPE
3692 self.field = field
3693 self.expr = coercions.expect(roles.ExpressionElementRole, expr)
3695 @util.ro_non_memoized_property
3696 def _from_objects(self) -> List[FromClause]:
3697 return self.expr._from_objects
3700class _label_reference(ColumnElement[_T]):
3701 """Wrap a column expression as it appears in a 'reference' context.
3703 This expression is any that includes an _order_by_label_element,
3704 which is a Label, or a DESC / ASC construct wrapping a Label.
3706 The production of _label_reference() should occur when an expression
3707 is added to this context; this includes the ORDER BY or GROUP BY of a
3708 SELECT statement, as well as a few other places, such as the ORDER BY
3709 within an OVER clause.
3711 """
3713 __visit_name__ = "label_reference"
3715 _traverse_internals: _TraverseInternalsType = [
3716 ("element", InternalTraversal.dp_clauseelement)
3717 ]
3719 element: ColumnElement[_T]
3721 def __init__(self, element: ColumnElement[_T]):
3722 self.element = element
3723 self._propagate_attrs = element._propagate_attrs
3725 @util.ro_non_memoized_property
3726 def _from_objects(self) -> List[FromClause]:
3727 return []
3730class _textual_label_reference(ColumnElement[Any]):
3731 __visit_name__ = "textual_label_reference"
3733 _traverse_internals: _TraverseInternalsType = [
3734 ("element", InternalTraversal.dp_string)
3735 ]
3737 def __init__(self, element: str):
3738 self.element = element
3740 @util.memoized_property
3741 def _text_clause(self) -> TextClause:
3742 return TextClause(self.element)
3745class UnaryExpression(ColumnElement[_T]):
3746 """Define a 'unary' expression.
3748 A unary expression has a single column expression
3749 and an operator. The operator can be placed on the left
3750 (where it is called the 'operator') or right (where it is called the
3751 'modifier') of the column expression.
3753 :class:`.UnaryExpression` is the basis for several unary operators
3754 including those used by :func:`.desc`, :func:`.asc`, :func:`.distinct`,
3755 :func:`.nulls_first` and :func:`.nulls_last`.
3757 """
3759 __visit_name__ = "unary"
3761 _traverse_internals: _TraverseInternalsType = [
3762 ("element", InternalTraversal.dp_clauseelement),
3763 ("operator", InternalTraversal.dp_operator),
3764 ("modifier", InternalTraversal.dp_operator),
3765 ]
3767 element: ColumnElement[Any]
3768 operator: Optional[OperatorType]
3769 modifier: Optional[OperatorType]
3771 def __init__(
3772 self,
3773 element: ColumnElement[Any],
3774 *,
3775 operator: Optional[OperatorType] = None,
3776 modifier: Optional[OperatorType] = None,
3777 type_: Optional[_TypeEngineArgument[_T]] = None,
3778 wraps_column_expression: bool = False, # legacy, not used as of 2.0.42
3779 ):
3780 self.operator = operator
3781 self.modifier = modifier
3782 self._propagate_attrs = element._propagate_attrs
3783 self.element = element.self_group(
3784 against=self.operator or self.modifier
3785 )
3787 # if type is None, we get NULLTYPE, which is our _T. But I don't
3788 # know how to get the overloads to express that correctly
3789 self.type = type_api.to_instance(type_) # type: ignore
3791 def _wraps_unnamed_column(self):
3792 ungrouped = self.element._ungroup()
3793 return (
3794 not isinstance(ungrouped, NamedColumn)
3795 or ungrouped._non_anon_label is None
3796 )
3798 @classmethod
3799 def _create_nulls_first(
3800 cls,
3801 column: _ColumnExpressionArgument[_T],
3802 ) -> UnaryExpression[_T]:
3803 return UnaryExpression(
3804 coercions.expect(roles.ByOfRole, column),
3805 modifier=operators.nulls_first_op,
3806 )
3808 @classmethod
3809 def _create_nulls_last(
3810 cls,
3811 column: _ColumnExpressionArgument[_T],
3812 ) -> UnaryExpression[_T]:
3813 return UnaryExpression(
3814 coercions.expect(roles.ByOfRole, column),
3815 modifier=operators.nulls_last_op,
3816 )
3818 @classmethod
3819 def _create_desc(
3820 cls, column: _ColumnExpressionOrStrLabelArgument[_T]
3821 ) -> UnaryExpression[_T]:
3823 return UnaryExpression(
3824 coercions.expect(roles.ByOfRole, column),
3825 modifier=operators.desc_op,
3826 )
3828 @classmethod
3829 def _create_asc(
3830 cls,
3831 column: _ColumnExpressionOrStrLabelArgument[_T],
3832 ) -> UnaryExpression[_T]:
3833 return UnaryExpression(
3834 coercions.expect(roles.ByOfRole, column),
3835 modifier=operators.asc_op,
3836 )
3838 @classmethod
3839 def _create_distinct(
3840 cls,
3841 expr: _ColumnExpressionArgument[_T],
3842 ) -> UnaryExpression[_T]:
3843 col_expr: ColumnElement[_T] = coercions.expect(
3844 roles.ExpressionElementRole, expr
3845 )
3846 return UnaryExpression(
3847 col_expr,
3848 operator=operators.distinct_op,
3849 type_=col_expr.type,
3850 )
3852 @classmethod
3853 def _create_bitwise_not(
3854 cls,
3855 expr: _ColumnExpressionArgument[_T],
3856 ) -> UnaryExpression[_T]:
3857 col_expr: ColumnElement[_T] = coercions.expect(
3858 roles.ExpressionElementRole, expr
3859 )
3860 return UnaryExpression(
3861 col_expr,
3862 operator=operators.bitwise_not_op,
3863 type_=col_expr.type,
3864 )
3866 @property
3867 def _order_by_label_element(self) -> Optional[Label[Any]]:
3868 if operators.is_order_by_modifier(self.modifier):
3869 return self.element._order_by_label_element
3870 else:
3871 return None
3873 @util.ro_non_memoized_property
3874 def _from_objects(self) -> List[FromClause]:
3875 return self.element._from_objects
3877 def _negate(self) -> ColumnElement[Any]:
3878 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
3879 return UnaryExpression(
3880 self.self_group(against=operators.inv),
3881 operator=operators.inv,
3882 type_=type_api.BOOLEANTYPE,
3883 )
3884 else:
3885 return ColumnElement._negate(self)
3887 def self_group(
3888 self, against: Optional[OperatorType] = None
3889 ) -> Union[Self, Grouping[_T]]:
3890 if self.operator and operators.is_precedent(self.operator, against):
3891 return Grouping(self)
3892 else:
3893 return self
3896class CollectionAggregate(UnaryExpression[_T]):
3897 """Forms the basis for right-hand collection operator modifiers
3898 ANY and ALL.
3900 The ANY and ALL keywords are available in different ways on different
3901 backends. On PostgreSQL, they only work for an ARRAY type. On
3902 MySQL, they only work for subqueries.
3904 """
3906 inherit_cache = True
3907 _is_collection_aggregate = True
3909 @classmethod
3910 def _create_any(
3911 cls, expr: _ColumnExpressionArgument[_T]
3912 ) -> CollectionAggregate[bool]:
3913 col_expr: ColumnElement[_T] = coercions.expect(
3914 roles.ExpressionElementRole,
3915 expr,
3916 )
3917 col_expr = col_expr.self_group()
3918 return CollectionAggregate(
3919 col_expr,
3920 operator=operators.any_op,
3921 type_=type_api.BOOLEANTYPE,
3922 )
3924 @classmethod
3925 def _create_all(
3926 cls, expr: _ColumnExpressionArgument[_T]
3927 ) -> CollectionAggregate[bool]:
3928 col_expr: ColumnElement[_T] = coercions.expect(
3929 roles.ExpressionElementRole,
3930 expr,
3931 )
3932 col_expr = col_expr.self_group()
3933 return CollectionAggregate(
3934 col_expr,
3935 operator=operators.all_op,
3936 type_=type_api.BOOLEANTYPE,
3937 )
3939 # operate and reverse_operate are hardwired to
3940 # dispatch onto the type comparator directly, so that we can
3941 # ensure "reversed" behavior.
3942 def operate(
3943 self, op: OperatorType, *other: Any, **kwargs: Any
3944 ) -> ColumnElement[_T]:
3945 if not operators.is_comparison(op):
3946 raise exc.ArgumentError(
3947 "Only comparison operators may be used with ANY/ALL"
3948 )
3949 kwargs["reverse"] = True
3950 return self.comparator.operate(operators.mirror(op), *other, **kwargs)
3952 def reverse_operate(
3953 self, op: OperatorType, other: Any, **kwargs: Any
3954 ) -> ColumnElement[_T]:
3955 # comparison operators should never call reverse_operate
3956 assert not operators.is_comparison(op)
3957 raise exc.ArgumentError(
3958 "Only comparison operators may be used with ANY/ALL"
3959 )
3962class AsBoolean(WrapsColumnExpression[bool], UnaryExpression[bool]):
3963 inherit_cache = True
3965 def __init__(self, element, operator, negate):
3966 self.element = element
3967 self.type = type_api.BOOLEANTYPE
3968 self.operator = operator
3969 self.negate = negate
3970 self.modifier = None
3971 self._is_implicitly_boolean = element._is_implicitly_boolean
3973 @property
3974 def wrapped_column_expression(self):
3975 return self.element
3977 def self_group(self, against: Optional[OperatorType] = None) -> Self:
3978 return self
3980 def _negate(self):
3981 if isinstance(self.element, (True_, False_)):
3982 return self.element._negate()
3983 else:
3984 return AsBoolean(self.element, self.negate, self.operator)
3987class BinaryExpression(OperatorExpression[_T]):
3988 """Represent an expression that is ``LEFT <operator> RIGHT``.
3990 A :class:`.BinaryExpression` is generated automatically
3991 whenever two column expressions are used in a Python binary expression:
3993 .. sourcecode:: pycon+sql
3995 >>> from sqlalchemy.sql import column
3996 >>> column("a") + column("b")
3997 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
3998 >>> print(column("a") + column("b"))
3999 {printsql}a + b
4001 """
4003 __visit_name__ = "binary"
4005 _traverse_internals: _TraverseInternalsType = [
4006 ("left", InternalTraversal.dp_clauseelement),
4007 ("right", InternalTraversal.dp_clauseelement),
4008 ("operator", InternalTraversal.dp_operator),
4009 ("negate", InternalTraversal.dp_operator),
4010 ("modifiers", InternalTraversal.dp_plain_dict),
4011 (
4012 "type",
4013 InternalTraversal.dp_type,
4014 ),
4015 ]
4017 _cache_key_traversal = [
4018 ("left", InternalTraversal.dp_clauseelement),
4019 ("right", InternalTraversal.dp_clauseelement),
4020 ("operator", InternalTraversal.dp_operator),
4021 ("modifiers", InternalTraversal.dp_plain_dict),
4022 # "type" affects JSON CAST operators, so while redundant in most cases,
4023 # is needed for that one
4024 (
4025 "type",
4026 InternalTraversal.dp_type,
4027 ),
4028 ]
4030 _is_implicitly_boolean = True
4031 """Indicates that any database will know this is a boolean expression
4032 even if the database does not have an explicit boolean datatype.
4034 """
4036 left: ColumnElement[Any]
4037 right: ColumnElement[Any]
4038 modifiers: Mapping[str, Any]
4040 def __init__(
4041 self,
4042 left: ColumnElement[Any],
4043 right: ColumnElement[Any],
4044 operator: OperatorType,
4045 type_: Optional[_TypeEngineArgument[_T]] = None,
4046 negate: Optional[OperatorType] = None,
4047 modifiers: Optional[Mapping[str, Any]] = None,
4048 ):
4049 # allow compatibility with libraries that
4050 # refer to BinaryExpression directly and pass strings
4051 if isinstance(operator, str):
4052 operator = operators.custom_op(operator)
4053 self._orig = (left.__hash__(), right.__hash__())
4054 self._propagate_attrs = left._propagate_attrs or right._propagate_attrs
4055 self.left = left.self_group(against=operator)
4056 self.right = right.self_group(against=operator)
4057 self.operator = operator
4059 # if type is None, we get NULLTYPE, which is our _T. But I don't
4060 # know how to get the overloads to express that correctly
4061 self.type = type_api.to_instance(type_) # type: ignore
4063 self.negate = negate
4064 self._is_implicitly_boolean = operators.is_boolean(operator)
4066 if modifiers is None:
4067 self.modifiers = {}
4068 else:
4069 self.modifiers = modifiers
4071 @property
4072 def _flattened_operator_clauses(
4073 self,
4074 ) -> typing_Tuple[ColumnElement[Any], ...]:
4075 return (self.left, self.right)
4077 def __bool__(self):
4078 """Implement Python-side "bool" for BinaryExpression as a
4079 simple "identity" check for the left and right attributes,
4080 if the operator is "eq" or "ne". Otherwise the expression
4081 continues to not support "bool" like all other column expressions.
4083 The rationale here is so that ColumnElement objects can be hashable.
4084 What? Well, suppose you do this::
4086 c1, c2 = column("x"), column("y")
4087 s1 = set([c1, c2])
4089 We do that **a lot**, columns inside of sets is an extremely basic
4090 thing all over the ORM for example.
4092 So what happens if we do this? ::
4094 c1 in s1
4096 Hashing means it will normally use ``__hash__()`` of the object,
4097 but in case of hash collision, it's going to also do ``c1 == c1``
4098 and/or ``c1 == c2`` inside. Those operations need to return a
4099 True/False value. But because we override ``==`` and ``!=``, they're
4100 going to get a BinaryExpression. Hence we implement ``__bool__`` here
4101 so that these comparisons behave in this particular context mostly
4102 like regular object comparisons. Thankfully Python is OK with
4103 that! Otherwise we'd have to use special set classes for columns
4104 (which we used to do, decades ago).
4106 """
4107 if self.operator in (operators.eq, operators.ne):
4108 # this is using the eq/ne operator given int hash values,
4109 # rather than Operator, so that "bool" can be based on
4110 # identity
4111 return self.operator(*self._orig) # type: ignore
4112 else:
4113 raise TypeError("Boolean value of this clause is not defined")
4115 if typing.TYPE_CHECKING:
4117 def __invert__(
4118 self: BinaryExpression[_T],
4119 ) -> BinaryExpression[_T]: ...
4121 @util.ro_non_memoized_property
4122 def _from_objects(self) -> List[FromClause]:
4123 return self.left._from_objects + self.right._from_objects
4125 def _negate(self):
4126 if self.negate is not None:
4127 return BinaryExpression(
4128 self.left,
4129 self.right._negate_in_binary(self.negate, self.operator),
4130 self.negate,
4131 negate=self.operator,
4132 type_=self.type,
4133 modifiers=self.modifiers,
4134 )
4135 else:
4136 return self.self_group()._negate()
4139class Slice(ColumnElement[Any]):
4140 """Represent SQL for a Python array-slice object.
4142 This is not a specific SQL construct at this level, but
4143 may be interpreted by specific dialects, e.g. PostgreSQL.
4145 """
4147 __visit_name__ = "slice"
4149 _traverse_internals: _TraverseInternalsType = [
4150 ("start", InternalTraversal.dp_clauseelement),
4151 ("stop", InternalTraversal.dp_clauseelement),
4152 ("step", InternalTraversal.dp_clauseelement),
4153 ]
4155 def __init__(self, start, stop, step, _name=None):
4156 self.start = coercions.expect(
4157 roles.ExpressionElementRole,
4158 start,
4159 name=_name,
4160 type_=type_api.INTEGERTYPE,
4161 )
4162 self.stop = coercions.expect(
4163 roles.ExpressionElementRole,
4164 stop,
4165 name=_name,
4166 type_=type_api.INTEGERTYPE,
4167 )
4168 self.step = coercions.expect(
4169 roles.ExpressionElementRole,
4170 step,
4171 name=_name,
4172 type_=type_api.INTEGERTYPE,
4173 )
4174 self.type = type_api.NULLTYPE
4176 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4177 assert against is operator.getitem
4178 return self
4181class IndexExpression(BinaryExpression[Any]):
4182 """Represent the class of expressions that are like an "index"
4183 operation."""
4185 inherit_cache = True
4188class GroupedElement(DQLDMLClauseElement):
4189 """Represent any parenthesized expression"""
4191 __visit_name__ = "grouping"
4193 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4194 return self
4196 def _ungroup(self) -> ClauseElement:
4197 raise NotImplementedError()
4200class Grouping(GroupedElement, ColumnElement[_T]):
4201 """Represent a grouping within a column expression"""
4203 _traverse_internals: _TraverseInternalsType = [
4204 ("element", InternalTraversal.dp_clauseelement),
4205 ("type", InternalTraversal.dp_type),
4206 ]
4208 _cache_key_traversal = [
4209 ("element", InternalTraversal.dp_clauseelement),
4210 ]
4212 element: Union[TextClause, ClauseList, ColumnElement[_T]]
4214 def __init__(
4215 self, element: Union[TextClause, ClauseList, ColumnElement[_T]]
4216 ):
4217 self.element = element
4219 # nulltype assignment issue
4220 self.type = getattr(element, "type", type_api.NULLTYPE) # type: ignore
4221 self._propagate_attrs = element._propagate_attrs
4223 def _with_binary_element_type(self, type_):
4224 return self.__class__(self.element._with_binary_element_type(type_))
4226 def _ungroup(self) -> ColumnElement[_T]:
4227 assert isinstance(self.element, ColumnElement)
4228 return self.element._ungroup()
4230 @util.memoized_property
4231 def _is_implicitly_boolean(self):
4232 return self.element._is_implicitly_boolean
4234 @util.non_memoized_property
4235 def _tq_label(self) -> Optional[str]:
4236 return (
4237 getattr(self.element, "_tq_label", None) or self._anon_name_label
4238 )
4240 @util.non_memoized_property
4241 def _proxies(self) -> List[ColumnElement[Any]]:
4242 if isinstance(self.element, ColumnElement):
4243 return [self.element]
4244 else:
4245 return []
4247 @util.ro_non_memoized_property
4248 def _from_objects(self) -> List[FromClause]:
4249 return self.element._from_objects
4251 def __getattr__(self, attr):
4252 return getattr(self.element, attr)
4254 def __getstate__(self):
4255 return {"element": self.element, "type": self.type}
4257 def __setstate__(self, state):
4258 self.element = state["element"]
4259 self.type = state["type"]
4261 if TYPE_CHECKING:
4263 def self_group(
4264 self, against: Optional[OperatorType] = None
4265 ) -> Self: ...
4268class _OverrideBinds(Grouping[_T]):
4269 """used by cache_key->_apply_params_to_element to allow compilation /
4270 execution of a SQL element that's been cached, using an alternate set of
4271 bound parameter values.
4273 This is used by the ORM to swap new parameter values into expressions
4274 that are embedded into loader options like with_expression(),
4275 selectinload(). Previously, this task was accomplished using the
4276 .params() method which would perform a deep-copy instead. This deep
4277 copy proved to be too expensive for more complex expressions.
4279 See #11085
4281 """
4283 __visit_name__ = "override_binds"
4285 def __init__(
4286 self,
4287 element: ColumnElement[_T],
4288 bindparams: Sequence[BindParameter[Any]],
4289 replaces_params: Sequence[BindParameter[Any]],
4290 ):
4291 self.element = element
4292 self.translate = {
4293 k.key: v.value for k, v in zip(replaces_params, bindparams)
4294 }
4296 def _gen_cache_key(
4297 self, anon_map: anon_map, bindparams: List[BindParameter[Any]]
4298 ) -> Optional[typing_Tuple[Any, ...]]:
4299 """generate a cache key for the given element, substituting its bind
4300 values for the translation values present."""
4302 existing_bps: List[BindParameter[Any]] = []
4303 ck = self.element._gen_cache_key(anon_map, existing_bps)
4305 bindparams.extend(
4306 (
4307 bp._with_value(
4308 self.translate[bp.key], maintain_key=True, required=False
4309 )
4310 if bp.key in self.translate
4311 else bp
4312 )
4313 for bp in existing_bps
4314 )
4316 return ck
4319class Over(ColumnElement[_T]):
4320 """Represent an OVER clause.
4322 This is a special operator against a so-called
4323 "window" function, as well as any aggregate function,
4324 which produces results relative to the result set
4325 itself. Most modern SQL backends now support window functions.
4327 """
4329 __visit_name__ = "over"
4331 _traverse_internals: _TraverseInternalsType = [
4332 ("element", InternalTraversal.dp_clauseelement),
4333 ("order_by", InternalTraversal.dp_clauseelement),
4334 ("partition_by", InternalTraversal.dp_clauseelement),
4335 ("range_", InternalTraversal.dp_clauseelement),
4336 ("rows", InternalTraversal.dp_clauseelement),
4337 ("groups", InternalTraversal.dp_clauseelement),
4338 ]
4340 order_by: Optional[ClauseList] = None
4341 partition_by: Optional[ClauseList] = None
4343 element: ColumnElement[_T]
4344 """The underlying expression object to which this :class:`.Over`
4345 object refers."""
4347 range_: Optional[_FrameClause]
4348 rows: Optional[_FrameClause]
4349 groups: Optional[_FrameClause]
4351 def __init__(
4352 self,
4353 element: ColumnElement[_T],
4354 partition_by: Optional[_ByArgument] = None,
4355 order_by: Optional[_ByArgument] = None,
4356 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4357 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4358 groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4359 ):
4360 self.element = element
4361 if order_by is not None:
4362 self.order_by = ClauseList(
4363 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4364 )
4365 if partition_by is not None:
4366 self.partition_by = ClauseList(
4367 *util.to_list(partition_by),
4368 _literal_as_text_role=roles.ByOfRole,
4369 )
4371 if sum(bool(item) for item in (range_, rows, groups)) > 1:
4372 raise exc.ArgumentError(
4373 "only one of 'rows', 'range_', or 'groups' may be provided"
4374 )
4375 else:
4376 self.range_ = _FrameClause(range_) if range_ else None
4377 self.rows = _FrameClause(rows) if rows else None
4378 self.groups = _FrameClause(groups) if groups else None
4380 if not TYPE_CHECKING:
4382 @util.memoized_property
4383 def type(self) -> TypeEngine[_T]: # noqa: A001
4384 return self.element.type
4386 @util.ro_non_memoized_property
4387 def _from_objects(self) -> List[FromClause]:
4388 return list(
4389 itertools.chain(
4390 *[
4391 c._from_objects
4392 for c in (self.element, self.partition_by, self.order_by)
4393 if c is not None
4394 ]
4395 )
4396 )
4399class _FrameClauseType(Enum):
4400 RANGE_UNBOUNDED = 0
4401 RANGE_CURRENT = 1
4402 RANGE_PRECEDING = 2
4403 RANGE_FOLLOWING = 3
4406class _FrameClause(ClauseElement):
4407 """indicate the 'rows' or 'range' field of a window function, e.g. using
4408 :class:`.Over`.
4410 .. versionadded:: 2.1
4412 """
4414 __visit_name__ = "frame_clause"
4416 _traverse_internals: _TraverseInternalsType = [
4417 ("lower_integer_bind", InternalTraversal.dp_clauseelement),
4418 ("upper_integer_bind", InternalTraversal.dp_clauseelement),
4419 ("lower_type", InternalTraversal.dp_plain_obj),
4420 ("upper_type", InternalTraversal.dp_plain_obj),
4421 ]
4423 def __init__(
4424 self,
4425 range_: typing_Tuple[Optional[int], Optional[int]],
4426 ):
4427 try:
4428 r0, r1 = range_
4429 except (ValueError, TypeError) as ve:
4430 raise exc.ArgumentError("2-tuple expected for range/rows") from ve
4432 if r0 is None:
4433 self.lower_type = _FrameClauseType.RANGE_UNBOUNDED
4434 self.lower_integer_bind = None
4435 else:
4436 try:
4437 lower_integer = int(r0)
4438 except ValueError as err:
4439 raise exc.ArgumentError(
4440 "Integer or None expected for range value"
4441 ) from err
4442 else:
4443 if lower_integer == 0:
4444 self.lower_type = _FrameClauseType.RANGE_CURRENT
4445 self.lower_integer_bind = None
4446 elif lower_integer < 0:
4447 self.lower_type = _FrameClauseType.RANGE_PRECEDING
4448 self.lower_integer_bind = literal(
4449 abs(lower_integer), type_api.INTEGERTYPE
4450 )
4451 else:
4452 self.lower_type = _FrameClauseType.RANGE_FOLLOWING
4453 self.lower_integer_bind = literal(
4454 lower_integer, type_api.INTEGERTYPE
4455 )
4457 if r1 is None:
4458 self.upper_type = _FrameClauseType.RANGE_UNBOUNDED
4459 self.upper_integer_bind = None
4460 else:
4461 try:
4462 upper_integer = int(r1)
4463 except ValueError as err:
4464 raise exc.ArgumentError(
4465 "Integer or None expected for range value"
4466 ) from err
4467 else:
4468 if upper_integer == 0:
4469 self.upper_type = _FrameClauseType.RANGE_CURRENT
4470 self.upper_integer_bind = None
4471 elif upper_integer < 0:
4472 self.upper_type = _FrameClauseType.RANGE_PRECEDING
4473 self.upper_integer_bind = literal(
4474 abs(upper_integer), type_api.INTEGERTYPE
4475 )
4476 else:
4477 self.upper_type = _FrameClauseType.RANGE_FOLLOWING
4478 self.upper_integer_bind = literal(
4479 upper_integer, type_api.INTEGERTYPE
4480 )
4483class WithinGroup(ColumnElement[_T]):
4484 """Represent a WITHIN GROUP (ORDER BY) clause.
4486 This is a special operator against so-called
4487 "ordered set aggregate" and "hypothetical
4488 set aggregate" functions, including ``percentile_cont()``,
4489 ``rank()``, ``dense_rank()``, etc.
4491 It's supported only by certain database backends, such as PostgreSQL,
4492 Oracle Database and MS SQL Server.
4494 The :class:`.WithinGroup` construct extracts its type from the
4495 method :meth:`.FunctionElement.within_group_type`. If this returns
4496 ``None``, the function's ``.type`` is used.
4498 """
4500 __visit_name__ = "withingroup"
4502 _traverse_internals: _TraverseInternalsType = [
4503 ("element", InternalTraversal.dp_clauseelement),
4504 ("order_by", InternalTraversal.dp_clauseelement),
4505 ]
4507 order_by: Optional[ClauseList] = None
4509 def __init__(
4510 self,
4511 element: Union[FunctionElement[_T], FunctionFilter[_T]],
4512 *order_by: _ColumnExpressionArgument[Any],
4513 ):
4514 self.element = element
4515 if order_by is not None:
4516 self.order_by = ClauseList(
4517 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4518 )
4520 def __reduce__(self):
4521 return self.__class__, (self.element,) + (
4522 tuple(self.order_by) if self.order_by is not None else ()
4523 )
4525 def over(
4526 self,
4527 *,
4528 partition_by: Optional[_ByArgument] = None,
4529 order_by: Optional[_ByArgument] = None,
4530 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4531 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4532 groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4533 ) -> Over[_T]:
4534 """Produce an OVER clause against this :class:`.WithinGroup`
4535 construct.
4537 This function has the same signature as that of
4538 :meth:`.FunctionElement.over`.
4540 """
4541 return Over(
4542 self,
4543 partition_by=partition_by,
4544 order_by=order_by,
4545 range_=range_,
4546 rows=rows,
4547 groups=groups,
4548 )
4550 @overload
4551 def filter(self) -> Self: ...
4553 @overload
4554 def filter(
4555 self,
4556 __criterion0: _ColumnExpressionArgument[bool],
4557 *criterion: _ColumnExpressionArgument[bool],
4558 ) -> FunctionFilter[_T]: ...
4560 def filter(
4561 self, *criterion: _ColumnExpressionArgument[bool]
4562 ) -> Union[Self, FunctionFilter[_T]]:
4563 """Produce a FILTER clause against this function."""
4564 if not criterion:
4565 return self
4566 return FunctionFilter(self, *criterion)
4568 if not TYPE_CHECKING:
4570 @util.memoized_property
4571 def type(self) -> TypeEngine[_T]: # noqa: A001
4572 wgt = self.element.within_group_type(self)
4573 if wgt is not None:
4574 return wgt
4575 else:
4576 return self.element.type
4578 @util.ro_non_memoized_property
4579 def _from_objects(self) -> List[FromClause]:
4580 return list(
4581 itertools.chain(
4582 *[
4583 c._from_objects
4584 for c in (self.element, self.order_by)
4585 if c is not None
4586 ]
4587 )
4588 )
4591class FunctionFilter(Generative, ColumnElement[_T]):
4592 """Represent a function FILTER clause.
4594 This is a special operator against aggregate and window functions,
4595 which controls which rows are passed to it.
4596 It's supported only by certain database backends.
4598 Invocation of :class:`.FunctionFilter` is via
4599 :meth:`.FunctionElement.filter`::
4601 func.count(1).filter(True)
4603 .. seealso::
4605 :meth:`.FunctionElement.filter`
4607 """
4609 __visit_name__ = "funcfilter"
4611 _traverse_internals: _TraverseInternalsType = [
4612 ("func", InternalTraversal.dp_clauseelement),
4613 ("criterion", InternalTraversal.dp_clauseelement),
4614 ]
4616 criterion: Optional[ColumnElement[bool]] = None
4618 def __init__(
4619 self,
4620 func: Union[FunctionElement[_T], WithinGroup[_T]],
4621 *criterion: _ColumnExpressionArgument[bool],
4622 ):
4623 self.func = func
4624 self.filter.non_generative(self, *criterion) # type: ignore
4626 @_generative
4627 def filter(self, *criterion: _ColumnExpressionArgument[bool]) -> Self:
4628 """Produce an additional FILTER against the function.
4630 This method adds additional criteria to the initial criteria
4631 set up by :meth:`.FunctionElement.filter`.
4633 Multiple criteria are joined together at SQL render time
4634 via ``AND``.
4637 """
4639 for crit in list(criterion):
4640 crit = coercions.expect(roles.WhereHavingRole, crit)
4642 if self.criterion is not None:
4643 self.criterion = self.criterion & crit
4644 else:
4645 self.criterion = crit
4647 return self
4649 def over(
4650 self,
4651 partition_by: Optional[
4652 Union[
4653 Iterable[_ColumnExpressionArgument[Any]],
4654 _ColumnExpressionArgument[Any],
4655 ]
4656 ] = None,
4657 order_by: Optional[
4658 Union[
4659 Iterable[_ColumnExpressionArgument[Any]],
4660 _ColumnExpressionArgument[Any],
4661 ]
4662 ] = None,
4663 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4664 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4665 groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4666 ) -> Over[_T]:
4667 """Produce an OVER clause against this filtered function.
4669 Used against aggregate or so-called "window" functions,
4670 for database backends that support window functions.
4672 The expression::
4674 func.rank().filter(MyClass.y > 5).over(order_by="x")
4676 is shorthand for::
4678 from sqlalchemy import over, funcfilter
4680 over(funcfilter(func.rank(), MyClass.y > 5), order_by="x")
4682 See :func:`_expression.over` for a full description.
4684 """
4685 return Over(
4686 self,
4687 partition_by=partition_by,
4688 order_by=order_by,
4689 range_=range_,
4690 rows=rows,
4691 groups=groups,
4692 )
4694 def within_group(
4695 self, *order_by: _ColumnExpressionArgument[Any]
4696 ) -> WithinGroup[_T]:
4697 """Produce a WITHIN GROUP (ORDER BY expr) clause against
4698 this function.
4699 """
4700 return WithinGroup(self, *order_by)
4702 def within_group_type(
4703 self, within_group: WithinGroup[_T]
4704 ) -> Optional[TypeEngine[_T]]:
4705 return None
4707 def self_group(
4708 self, against: Optional[OperatorType] = None
4709 ) -> Union[Self, Grouping[_T]]:
4710 if operators.is_precedent(operators.filter_op, against):
4711 return Grouping(self)
4712 else:
4713 return self
4715 if not TYPE_CHECKING:
4717 @util.memoized_property
4718 def type(self) -> TypeEngine[_T]: # noqa: A001
4719 return self.func.type
4721 @util.ro_non_memoized_property
4722 def _from_objects(self) -> List[FromClause]:
4723 return list(
4724 itertools.chain(
4725 *[
4726 c._from_objects
4727 for c in (self.func, self.criterion)
4728 if c is not None
4729 ]
4730 )
4731 )
4734class NamedColumn(KeyedColumnElement[_T]):
4735 is_literal = False
4736 table: Optional[FromClause] = None
4737 name: str
4738 key: str
4740 def _compare_name_for_result(self, other):
4741 return (hasattr(other, "name") and self.name == other.name) or (
4742 hasattr(other, "_label") and self._label == other._label
4743 )
4745 @util.ro_memoized_property
4746 def description(self) -> str:
4747 return self.name
4749 @HasMemoized.memoized_attribute
4750 def _tq_key_label(self) -> Optional[str]:
4751 """table qualified label based on column key.
4753 for table-bound columns this is <tablename>_<column key/proxy key>;
4755 all other expressions it resolves to key/proxy key.
4757 """
4758 proxy_key = self._proxy_key
4759 if proxy_key and proxy_key != self.name:
4760 return self._gen_tq_label(proxy_key)
4761 else:
4762 return self._tq_label
4764 @HasMemoized.memoized_attribute
4765 def _tq_label(self) -> Optional[str]:
4766 """table qualified label based on column name.
4768 for table-bound columns this is <tablename>_<columnname>; all other
4769 expressions it resolves to .name.
4771 """
4772 return self._gen_tq_label(self.name)
4774 @HasMemoized.memoized_attribute
4775 def _render_label_in_columns_clause(self):
4776 return True
4778 @HasMemoized.memoized_attribute
4779 def _non_anon_label(self):
4780 return self.name
4782 def _gen_tq_label(
4783 self, name: str, dedupe_on_key: bool = True
4784 ) -> Optional[str]:
4785 return name
4787 def _bind_param(
4788 self,
4789 operator: OperatorType,
4790 obj: Any,
4791 type_: Optional[TypeEngine[_T]] = None,
4792 expanding: bool = False,
4793 ) -> BindParameter[_T]:
4794 return BindParameter(
4795 self.key,
4796 obj,
4797 _compared_to_operator=operator,
4798 _compared_to_type=self.type,
4799 type_=type_,
4800 unique=True,
4801 expanding=expanding,
4802 )
4804 def _make_proxy(
4805 self,
4806 selectable: FromClause,
4807 *,
4808 primary_key: ColumnSet,
4809 foreign_keys: Set[KeyedColumnElement[Any]],
4810 name: Optional[str] = None,
4811 key: Optional[str] = None,
4812 name_is_truncatable: bool = False,
4813 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
4814 disallow_is_literal: bool = False,
4815 **kw: Any,
4816 ) -> typing_Tuple[str, ColumnClause[_T]]:
4817 c = ColumnClause(
4818 (
4819 coercions.expect(roles.TruncatedLabelRole, name or self.name)
4820 if name_is_truncatable
4821 else (name or self.name)
4822 ),
4823 type_=self.type,
4824 _selectable=selectable,
4825 is_literal=False,
4826 )
4828 c._propagate_attrs = selectable._propagate_attrs
4829 if name is None:
4830 c.key = self.key
4831 if compound_select_cols:
4832 c._proxies = list(compound_select_cols)
4833 else:
4834 c._proxies = [self]
4836 if selectable._is_clone_of is not None:
4837 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
4838 return c.key, c
4841_PS = ParamSpec("_PS")
4844class Label(roles.LabeledColumnExprRole[_T], NamedColumn[_T]):
4845 """Represents a column label (AS).
4847 Represent a label, as typically applied to any column-level
4848 element using the ``AS`` sql keyword.
4850 """
4852 __visit_name__ = "label"
4854 _traverse_internals: _TraverseInternalsType = [
4855 ("name", InternalTraversal.dp_anon_name),
4856 ("type", InternalTraversal.dp_type),
4857 ("_element", InternalTraversal.dp_clauseelement),
4858 ]
4860 _cache_key_traversal = [
4861 ("name", InternalTraversal.dp_anon_name),
4862 ("_element", InternalTraversal.dp_clauseelement),
4863 ]
4865 _element: ColumnElement[_T]
4866 name: str
4868 def __init__(
4869 self,
4870 name: Optional[str],
4871 element: _ColumnExpressionArgument[_T],
4872 type_: Optional[_TypeEngineArgument[_T]] = None,
4873 ):
4874 orig_element = element
4875 element = coercions.expect(
4876 roles.ExpressionElementRole,
4877 element,
4878 apply_propagate_attrs=self,
4879 )
4880 while isinstance(element, Label):
4881 # TODO: this is only covered in test_text.py, but nothing
4882 # fails if it's removed. determine rationale
4883 element = element.element
4885 if name:
4886 self.name = name
4887 else:
4888 self.name = _anonymous_label.safe_construct(
4889 id(self), getattr(element, "name", "anon")
4890 )
4891 if isinstance(orig_element, Label):
4892 # TODO: no coverage for this block, again would be in
4893 # test_text.py where the resolve_label concept is important
4894 self._resolve_label = orig_element._label
4896 self.key = self._tq_label = self._tq_key_label = self.name
4897 self._element = element
4899 self.type = (
4900 type_api.to_instance(type_)
4901 if type_ is not None
4902 else self._element.type
4903 )
4905 self._proxies = [element]
4907 def __reduce__(self):
4908 return self.__class__, (self.name, self._element, self.type)
4910 @HasMemoized.memoized_attribute
4911 def _render_label_in_columns_clause(self):
4912 return True
4914 def _bind_param(self, operator, obj, type_=None, expanding=False):
4915 return BindParameter(
4916 None,
4917 obj,
4918 _compared_to_operator=operator,
4919 type_=type_,
4920 _compared_to_type=self.type,
4921 unique=True,
4922 expanding=expanding,
4923 )
4925 @util.memoized_property
4926 def _is_implicitly_boolean(self):
4927 return self.element._is_implicitly_boolean
4929 @HasMemoized.memoized_attribute
4930 def _allow_label_resolve(self):
4931 return self.element._allow_label_resolve
4933 @property
4934 def _order_by_label_element(self):
4935 return self
4937 def as_reference(self) -> _label_reference[_T]:
4938 """refer to this labeled expression in a clause such as GROUP BY,
4939 ORDER BY etc. as the label name itself, without expanding
4940 into the full expression.
4942 .. versionadded:: 2.1
4944 """
4945 return _label_reference(self)
4947 @HasMemoized.memoized_attribute
4948 def element(self) -> ColumnElement[_T]:
4949 return self._element.self_group(against=operators.as_)
4951 def self_group(self, against: Optional[OperatorType] = None) -> Label[_T]:
4952 return self._apply_to_inner(self._element.self_group, against=against)
4954 def _negate(self):
4955 return self._apply_to_inner(self._element._negate)
4957 def _apply_to_inner(
4958 self,
4959 fn: Callable[_PS, ColumnElement[_T]],
4960 *arg: _PS.args,
4961 **kw: _PS.kwargs,
4962 ) -> Label[_T]:
4963 sub_element = fn(*arg, **kw)
4964 if sub_element is not self._element:
4965 return Label(self.name, sub_element, type_=self.type)
4966 else:
4967 return self
4969 @property
4970 def primary_key(self): # type: ignore[override]
4971 return self.element.primary_key
4973 @property
4974 def foreign_keys(self): # type: ignore[override]
4975 return self.element.foreign_keys
4977 def _copy_internals(
4978 self,
4979 *,
4980 clone: _CloneCallableType = _clone,
4981 anonymize_labels: bool = False,
4982 **kw: Any,
4983 ) -> None:
4984 self._reset_memoizations()
4985 self._element = clone(self._element, **kw)
4986 if anonymize_labels:
4987 self.name = _anonymous_label.safe_construct(
4988 id(self), getattr(self.element, "name", "anon")
4989 )
4990 self.key = self._tq_label = self._tq_key_label = self.name
4992 @util.ro_non_memoized_property
4993 def _from_objects(self) -> List[FromClause]:
4994 return self.element._from_objects
4996 def _make_proxy(
4997 self,
4998 selectable: FromClause,
4999 *,
5000 primary_key: ColumnSet,
5001 foreign_keys: Set[KeyedColumnElement[Any]],
5002 name: Optional[str] = None,
5003 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5004 **kw: Any,
5005 ) -> typing_Tuple[str, ColumnClause[_T]]:
5006 name = self.name if not name else name
5008 key, e = self.element._make_proxy(
5009 selectable,
5010 name=name,
5011 disallow_is_literal=True,
5012 name_is_truncatable=isinstance(name, _truncated_label),
5013 compound_select_cols=compound_select_cols,
5014 primary_key=primary_key,
5015 foreign_keys=foreign_keys,
5016 )
5018 # there was a note here to remove this assertion, which was here
5019 # to determine if we later could support a use case where
5020 # the key and name of a label are separate. But I don't know what
5021 # that case was. For now, this is an unexpected case that occurs
5022 # when a label name conflicts with other columns and select()
5023 # is attempting to disambiguate an explicit label, which is not what
5024 # the user would want. See issue #6090.
5025 if key != self.name and not isinstance(self.name, _anonymous_label):
5026 raise exc.InvalidRequestError(
5027 "Label name %s is being renamed to an anonymous label due "
5028 "to disambiguation "
5029 "which is not supported right now. Please use unique names "
5030 "for explicit labels." % (self.name)
5031 )
5033 e._propagate_attrs = selectable._propagate_attrs
5034 e._proxies.append(self)
5035 if self.type is not None:
5036 e.type = self.type
5038 return self.key, e
5041class ColumnClause(
5042 roles.DDLReferredColumnRole,
5043 roles.LabeledColumnExprRole[_T],
5044 roles.StrAsPlainColumnRole,
5045 Immutable,
5046 NamedColumn[_T],
5047):
5048 """Represents a column expression from any textual string.
5050 The :class:`.ColumnClause`, a lightweight analogue to the
5051 :class:`_schema.Column` class, is typically invoked using the
5052 :func:`_expression.column` function, as in::
5054 from sqlalchemy import column
5056 id, name = column("id"), column("name")
5057 stmt = select(id, name).select_from("user")
5059 The above statement would produce SQL like:
5061 .. sourcecode:: sql
5063 SELECT id, name FROM user
5065 :class:`.ColumnClause` is the immediate superclass of the schema-specific
5066 :class:`_schema.Column` object. While the :class:`_schema.Column`
5067 class has all the
5068 same capabilities as :class:`.ColumnClause`, the :class:`.ColumnClause`
5069 class is usable by itself in those cases where behavioral requirements
5070 are limited to simple SQL expression generation. The object has none of
5071 the associations with schema-level metadata or with execution-time
5072 behavior that :class:`_schema.Column` does,
5073 so in that sense is a "lightweight"
5074 version of :class:`_schema.Column`.
5076 Full details on :class:`.ColumnClause` usage is at
5077 :func:`_expression.column`.
5079 .. seealso::
5081 :func:`_expression.column`
5083 :class:`_schema.Column`
5085 """
5087 table: Optional[FromClause]
5088 is_literal: bool
5090 __visit_name__ = "column"
5092 _traverse_internals: _TraverseInternalsType = [
5093 ("name", InternalTraversal.dp_anon_name),
5094 ("type", InternalTraversal.dp_type),
5095 ("table", InternalTraversal.dp_clauseelement),
5096 ("is_literal", InternalTraversal.dp_boolean),
5097 ]
5099 onupdate: Optional[DefaultGenerator] = None
5100 default: Optional[DefaultGenerator] = None
5101 server_default: Optional[FetchedValue] = None
5102 server_onupdate: Optional[FetchedValue] = None
5104 _is_multiparam_column = False
5106 @property
5107 def _is_star(self): # type: ignore[override]
5108 return self.is_literal and self.name == "*"
5110 def __init__(
5111 self,
5112 text: str,
5113 type_: Optional[_TypeEngineArgument[_T]] = None,
5114 is_literal: bool = False,
5115 _selectable: Optional[FromClause] = None,
5116 ):
5117 self.key = self.name = text
5118 self.table = _selectable
5120 # if type is None, we get NULLTYPE, which is our _T. But I don't
5121 # know how to get the overloads to express that correctly
5122 self.type = type_api.to_instance(type_) # type: ignore
5124 self.is_literal = is_literal
5126 def get_children(self, *, column_tables=False, **kw):
5127 # override base get_children() to not return the Table
5128 # or selectable that is parent to this column. Traversals
5129 # expect the columns of tables and subqueries to be leaf nodes.
5130 return []
5132 @property
5133 def entity_namespace(self):
5134 if self.table is not None:
5135 return self.table.entity_namespace
5136 else:
5137 return super().entity_namespace
5139 def _clone(self, detect_subquery_cols=False, **kw):
5140 if (
5141 detect_subquery_cols
5142 and self.table is not None
5143 and self.table._is_subquery
5144 ):
5145 clone = kw.pop("clone")
5146 table = clone(self.table, **kw)
5147 new = table.c.corresponding_column(self)
5148 return new
5150 return super()._clone(**kw)
5152 @HasMemoized_ro_memoized_attribute
5153 def _from_objects(self) -> List[FromClause]:
5154 t = self.table
5155 if t is not None:
5156 return [t]
5157 else:
5158 return []
5160 @HasMemoized.memoized_attribute
5161 def _render_label_in_columns_clause(self):
5162 return self.table is not None
5164 @property
5165 def _ddl_label(self):
5166 return self._gen_tq_label(self.name, dedupe_on_key=False)
5168 def _compare_name_for_result(self, other):
5169 if (
5170 self.is_literal
5171 or self.table is None
5172 or self.table._is_textual
5173 or not hasattr(other, "proxy_set")
5174 or (
5175 isinstance(other, ColumnClause)
5176 and (
5177 other.is_literal
5178 or other.table is None
5179 or other.table._is_textual
5180 )
5181 )
5182 ):
5183 return (hasattr(other, "name") and self.name == other.name) or (
5184 hasattr(other, "_tq_label")
5185 and self._tq_label == other._tq_label
5186 )
5187 else:
5188 return other.proxy_set.intersection(self.proxy_set)
5190 def _gen_tq_label(
5191 self, name: str, dedupe_on_key: bool = True
5192 ) -> Optional[str]:
5193 """generate table-qualified label
5195 for a table-bound column this is <tablename>_<columnname>.
5197 used primarily for LABEL_STYLE_TABLENAME_PLUS_COL
5198 as well as the .columns collection on a Join object.
5200 """
5201 label: str
5202 t = self.table
5203 if self.is_literal:
5204 return None
5205 elif t is not None and is_named_from_clause(t):
5206 if has_schema_attr(t) and t.schema:
5207 label = (
5208 t.schema.replace(".", "_") + "_" + t.name + ("_" + name)
5209 )
5210 else:
5211 assert not TYPE_CHECKING or isinstance(t, NamedFromClause)
5212 label = t.name + ("_" + name)
5214 # propagate name quoting rules for labels.
5215 if is_quoted_name(name) and name.quote is not None:
5216 if is_quoted_name(label):
5217 label.quote = name.quote
5218 else:
5219 label = quoted_name(label, name.quote)
5220 elif is_quoted_name(t.name) and t.name.quote is not None:
5221 # can't get this situation to occur, so let's
5222 # assert false on it for now
5223 assert not isinstance(label, quoted_name)
5224 label = quoted_name(label, t.name.quote)
5226 if dedupe_on_key:
5227 # ensure the label name doesn't conflict with that of an
5228 # existing column. note that this implies that any Column
5229 # must **not** set up its _label before its parent table has
5230 # all of its other Column objects set up. There are several
5231 # tables in the test suite which will fail otherwise; example:
5232 # table "owner" has columns "name" and "owner_name". Therefore
5233 # column owner.name cannot use the label "owner_name", it has
5234 # to be "owner_name_1".
5235 if label in t.c:
5236 _label = label
5237 counter = 1
5238 while _label in t.c:
5239 _label = label + f"_{counter}"
5240 counter += 1
5241 label = _label
5243 return coercions.expect(roles.TruncatedLabelRole, label)
5245 else:
5246 return name
5248 def _make_proxy(
5249 self,
5250 selectable: FromClause,
5251 *,
5252 primary_key: ColumnSet,
5253 foreign_keys: Set[KeyedColumnElement[Any]],
5254 name: Optional[str] = None,
5255 key: Optional[str] = None,
5256 name_is_truncatable: bool = False,
5257 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5258 disallow_is_literal: bool = False,
5259 **kw: Any,
5260 ) -> typing_Tuple[str, ColumnClause[_T]]:
5261 # the "is_literal" flag normally should never be propagated; a proxied
5262 # column is always a SQL identifier and never the actual expression
5263 # being evaluated. however, there is a case where the "is_literal" flag
5264 # might be used to allow the given identifier to have a fixed quoting
5265 # pattern already, so maintain the flag for the proxy unless a
5266 # :class:`.Label` object is creating the proxy. See [ticket:4730].
5267 is_literal = (
5268 not disallow_is_literal
5269 and self.is_literal
5270 and (
5271 # note this does not accommodate for quoted_name differences
5272 # right now
5273 name is None
5274 or name == self.name
5275 )
5276 )
5277 c = self._constructor(
5278 (
5279 coercions.expect(roles.TruncatedLabelRole, name or self.name)
5280 if name_is_truncatable
5281 else (name or self.name)
5282 ),
5283 type_=self.type,
5284 _selectable=selectable,
5285 is_literal=is_literal,
5286 )
5287 c._propagate_attrs = selectable._propagate_attrs
5288 if name is None:
5289 c.key = self.key
5290 if compound_select_cols:
5291 c._proxies = list(compound_select_cols)
5292 else:
5293 c._proxies = [self]
5295 if selectable._is_clone_of is not None:
5296 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
5297 return c.key, c
5300class TableValuedColumn(NamedColumn[_T]):
5301 __visit_name__ = "table_valued_column"
5303 _traverse_internals: _TraverseInternalsType = [
5304 ("name", InternalTraversal.dp_anon_name),
5305 ("type", InternalTraversal.dp_type),
5306 ("scalar_alias", InternalTraversal.dp_clauseelement),
5307 ]
5309 def __init__(self, scalar_alias: NamedFromClause, type_: TypeEngine[_T]):
5310 self.scalar_alias = scalar_alias
5311 self.key = self.name = scalar_alias.name
5312 self.type = type_
5314 def _copy_internals(
5315 self, clone: _CloneCallableType = _clone, **kw: Any
5316 ) -> None:
5317 self.scalar_alias = clone(self.scalar_alias, **kw)
5318 self.key = self.name = self.scalar_alias.name
5320 @util.ro_non_memoized_property
5321 def _from_objects(self) -> List[FromClause]:
5322 return [self.scalar_alias]
5325class CollationClause(ColumnElement[str]):
5326 __visit_name__ = "collation"
5328 _traverse_internals: _TraverseInternalsType = [
5329 ("collation", InternalTraversal.dp_string)
5330 ]
5332 @classmethod
5333 @util.preload_module("sqlalchemy.sql.sqltypes")
5334 def _create_collation_expression(
5335 cls, expression: _ColumnExpressionArgument[str], collation: str
5336 ) -> BinaryExpression[str]:
5338 sqltypes = util.preloaded.sql_sqltypes
5340 expr = coercions.expect(roles.ExpressionElementRole[str], expression)
5342 if expr.type._type_affinity is sqltypes.String:
5343 collate_type = expr.type._with_collation(collation)
5344 else:
5345 collate_type = expr.type
5347 return BinaryExpression(
5348 expr,
5349 CollationClause(collation),
5350 operators.collate,
5351 type_=collate_type,
5352 )
5354 def __init__(self, collation):
5355 self.collation = collation
5358class _IdentifiedClause(Executable, ClauseElement):
5359 __visit_name__ = "identified"
5361 def __init__(self, ident):
5362 self.ident = ident
5365class SavepointClause(_IdentifiedClause):
5366 __visit_name__ = "savepoint"
5367 inherit_cache = False
5370class RollbackToSavepointClause(_IdentifiedClause):
5371 __visit_name__ = "rollback_to_savepoint"
5372 inherit_cache = False
5375class ReleaseSavepointClause(_IdentifiedClause):
5376 __visit_name__ = "release_savepoint"
5377 inherit_cache = False
5380class quoted_name(util.MemoizedSlots, str):
5381 """Represent a SQL identifier combined with quoting preferences.
5383 :class:`.quoted_name` is a Python unicode/str subclass which
5384 represents a particular identifier name along with a
5385 ``quote`` flag. This ``quote`` flag, when set to
5386 ``True`` or ``False``, overrides automatic quoting behavior
5387 for this identifier in order to either unconditionally quote
5388 or to not quote the name. If left at its default of ``None``,
5389 quoting behavior is applied to the identifier on a per-backend basis
5390 based on an examination of the token itself.
5392 A :class:`.quoted_name` object with ``quote=True`` is also
5393 prevented from being modified in the case of a so-called
5394 "name normalize" option. Certain database backends, such as
5395 Oracle Database, Firebird, and DB2 "normalize" case-insensitive names
5396 as uppercase. The SQLAlchemy dialects for these backends
5397 convert from SQLAlchemy's lower-case-means-insensitive convention
5398 to the upper-case-means-insensitive conventions of those backends.
5399 The ``quote=True`` flag here will prevent this conversion from occurring
5400 to support an identifier that's quoted as all lower case against
5401 such a backend.
5403 The :class:`.quoted_name` object is normally created automatically
5404 when specifying the name for key schema constructs such as
5405 :class:`_schema.Table`, :class:`_schema.Column`, and others.
5406 The class can also be
5407 passed explicitly as the name to any function that receives a name which
5408 can be quoted. Such as to use the :meth:`_engine.Engine.has_table`
5409 method with
5410 an unconditionally quoted name::
5412 from sqlalchemy import create_engine
5413 from sqlalchemy import inspect
5414 from sqlalchemy.sql import quoted_name
5416 engine = create_engine("oracle+oracledb://some_dsn")
5417 print(inspect(engine).has_table(quoted_name("some_table", True)))
5419 The above logic will run the "has table" logic against the Oracle Database
5420 backend, passing the name exactly as ``"some_table"`` without converting to
5421 upper case.
5423 """
5425 __slots__ = "quote", "lower", "upper"
5427 quote: Optional[bool]
5429 @overload
5430 @classmethod
5431 def construct(cls, value: str, quote: Optional[bool]) -> quoted_name: ...
5433 @overload
5434 @classmethod
5435 def construct(cls, value: None, quote: Optional[bool]) -> None: ...
5437 @classmethod
5438 def construct(
5439 cls, value: Optional[str], quote: Optional[bool]
5440 ) -> Optional[quoted_name]:
5441 if value is None:
5442 return None
5443 else:
5444 return quoted_name(value, quote)
5446 def __new__(cls, value: str, quote: Optional[bool]) -> quoted_name:
5447 assert (
5448 value is not None
5449 ), "use quoted_name.construct() for None passthrough"
5450 if isinstance(value, cls) and (quote is None or value.quote == quote):
5451 return value
5452 self = super().__new__(cls, value)
5454 self.quote = quote
5455 return self
5457 def __reduce__(self):
5458 return quoted_name, (str(self), self.quote)
5460 def _memoized_method_lower(self):
5461 if self.quote:
5462 return self
5463 else:
5464 return str(self).lower()
5466 def _memoized_method_upper(self):
5467 if self.quote:
5468 return self
5469 else:
5470 return str(self).upper()
5473def _find_columns(clause: ClauseElement) -> Set[ColumnClause[Any]]:
5474 """locate Column objects within the given expression."""
5476 cols: Set[ColumnClause[Any]] = set()
5477 traverse(clause, {}, {"column": cols.add})
5478 return cols
5481def _type_from_args(args: Sequence[ColumnElement[_T]]) -> TypeEngine[_T]:
5482 for a in args:
5483 if not a.type._isnull:
5484 return a.type
5485 else:
5486 return type_api.NULLTYPE # type: ignore
5489def _corresponding_column_or_error(fromclause, column, require_embedded=False):
5490 c = fromclause.corresponding_column(
5491 column, require_embedded=require_embedded
5492 )
5493 if c is None:
5494 raise exc.InvalidRequestError(
5495 "Given column '%s', attached to table '%s', "
5496 "failed to locate a corresponding column from table '%s'"
5497 % (column, getattr(column, "table", None), fromclause.description)
5498 )
5499 return c
5502class _memoized_property_but_not_nulltype(
5503 util.memoized_property["TypeEngine[_T]"]
5504):
5505 """memoized property, but dont memoize NullType"""
5507 def __get__(self, obj, cls):
5508 if obj is None:
5509 return self
5510 result = self.fget(obj)
5511 if not result._isnull:
5512 obj.__dict__[self.__name__] = result
5513 return result
5516class AnnotatedColumnElement(Annotated):
5517 _Annotated__element: ColumnElement[Any]
5519 def __init__(self, element, values):
5520 Annotated.__init__(self, element, values)
5521 for attr in (
5522 "comparator",
5523 "_proxy_key",
5524 "_tq_key_label",
5525 "_tq_label",
5526 "_non_anon_label",
5527 "type",
5528 ):
5529 self.__dict__.pop(attr, None)
5530 for attr in ("name", "key", "table"):
5531 if self.__dict__.get(attr, False) is None:
5532 self.__dict__.pop(attr)
5534 def _with_annotations(self, values):
5535 clone = super()._with_annotations(values)
5536 for attr in (
5537 "comparator",
5538 "_proxy_key",
5539 "_tq_key_label",
5540 "_tq_label",
5541 "_non_anon_label",
5542 ):
5543 clone.__dict__.pop(attr, None)
5544 return clone
5546 @util.memoized_property
5547 def name(self):
5548 """pull 'name' from parent, if not present"""
5549 return self._Annotated__element.name
5551 @_memoized_property_but_not_nulltype
5552 def type(self):
5553 """pull 'type' from parent and don't cache if null.
5555 type is routinely changed on existing columns within the
5556 mapped_column() initialization process, and "type" is also consulted
5557 during the creation of SQL expressions. Therefore it can change after
5558 it was already retrieved. At the same time we don't want annotated
5559 objects having overhead when expressions are produced, so continue
5560 to memoize, but only when we have a non-null type.
5562 """
5563 return self._Annotated__element.type
5565 @util.memoized_property
5566 def table(self):
5567 """pull 'table' from parent, if not present"""
5568 return self._Annotated__element.table
5570 @util.memoized_property
5571 def key(self):
5572 """pull 'key' from parent, if not present"""
5573 return self._Annotated__element.key
5575 @util.memoized_property
5576 def info(self) -> _InfoType:
5577 if TYPE_CHECKING:
5578 assert isinstance(self._Annotated__element, Column)
5579 return self._Annotated__element.info
5581 @util.memoized_property
5582 def _anon_name_label(self) -> str:
5583 return self._Annotated__element._anon_name_label
5586class _truncated_label(quoted_name):
5587 """A unicode subclass used to identify symbolic "
5588 "names that may require truncation."""
5590 __slots__ = ()
5592 def __new__(cls, value: str, quote: Optional[bool] = None) -> Any:
5593 quote = getattr(value, "quote", quote)
5594 # return super(_truncated_label, cls).__new__(cls, value, quote, True)
5595 return super().__new__(cls, value, quote)
5597 def __reduce__(self) -> Any:
5598 return self.__class__, (str(self), self.quote)
5600 def apply_map(self, map_: Mapping[str, Any]) -> str:
5601 return self
5604class conv(_truncated_label):
5605 """Mark a string indicating that a name has already been converted
5606 by a naming convention.
5608 This is a string subclass that indicates a name that should not be
5609 subject to any further naming conventions.
5611 E.g. when we create a :class:`.Constraint` using a naming convention
5612 as follows::
5614 m = MetaData(
5615 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5616 )
5617 t = Table(
5618 "t", m, Column("x", Integer), CheckConstraint("x > 5", name="x5")
5619 )
5621 The name of the above constraint will be rendered as ``"ck_t_x5"``.
5622 That is, the existing name ``x5`` is used in the naming convention as the
5623 ``constraint_name`` token.
5625 In some situations, such as in migration scripts, we may be rendering
5626 the above :class:`.CheckConstraint` with a name that's already been
5627 converted. In order to make sure the name isn't double-modified, the
5628 new name is applied using the :func:`_schema.conv` marker. We can
5629 use this explicitly as follows::
5632 m = MetaData(
5633 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5634 )
5635 t = Table(
5636 "t",
5637 m,
5638 Column("x", Integer),
5639 CheckConstraint("x > 5", name=conv("ck_t_x5")),
5640 )
5642 Where above, the :func:`_schema.conv` marker indicates that the constraint
5643 name here is final, and the name will render as ``"ck_t_x5"`` and not
5644 ``"ck_t_ck_t_x5"``
5646 .. seealso::
5648 :ref:`constraint_naming_conventions`
5650 """
5652 __slots__ = ()
5655# for backwards compatibility in case
5656# someone is re-implementing the
5657# _truncated_identifier() sequence in a custom
5658# compiler
5659_generated_label = _truncated_label
5660_anonymous_label_escape = re.compile(r"[%\(\) \$]+")
5663class _anonymous_label(_truncated_label):
5664 """A unicode subclass used to identify anonymously
5665 generated names."""
5667 __slots__ = ()
5669 @classmethod
5670 def safe_construct_with_key(
5671 cls, seed: int, body: str, sanitize_key: bool = False
5672 ) -> typing_Tuple[_anonymous_label, str]:
5673 # need to escape chars that interfere with format
5674 # strings in any case, issue #8724
5675 body = _anonymous_label_escape.sub("_", body)
5677 if sanitize_key:
5678 # sanitize_key is then an extra step used by BindParameter
5679 body = body.strip("_")
5681 key = f"{seed} {body.replace('%', '%%')}"
5682 label = _anonymous_label(f"%({key})s")
5683 return label, key
5685 @classmethod
5686 def safe_construct(
5687 cls, seed: int, body: str, sanitize_key: bool = False
5688 ) -> _anonymous_label:
5689 # need to escape chars that interfere with format
5690 # strings in any case, issue #8724
5691 body = _anonymous_label_escape.sub("_", body)
5693 if sanitize_key:
5694 # sanitize_key is then an extra step used by BindParameter
5695 body = body.strip("_")
5697 return _anonymous_label(f"%({seed} {body.replace('%', '%%')})s")
5699 def __add__(self, other: str) -> _anonymous_label:
5700 if "%" in other and not isinstance(other, _anonymous_label):
5701 other = str(other).replace("%", "%%")
5702 else:
5703 other = str(other)
5705 return _anonymous_label(
5706 quoted_name(
5707 str.__add__(self, other),
5708 self.quote,
5709 )
5710 )
5712 def __radd__(self, other: str) -> _anonymous_label:
5713 if "%" in other and not isinstance(other, _anonymous_label):
5714 other = str(other).replace("%", "%%")
5715 else:
5716 other = str(other)
5718 return _anonymous_label(
5719 quoted_name(
5720 str.__add__(other, self),
5721 self.quote,
5722 )
5723 )
5725 def apply_map(self, map_: Mapping[str, Any]) -> str:
5726 if self.quote is not None:
5727 # preserve quoting only if necessary
5728 return quoted_name(self % map_, self.quote)
5729 else:
5730 # else skip the constructor call
5731 return self % map_