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 ExecutableStatement
62from .base import Generative
63from .base import HasMemoized
64from .base import Immutable
65from .base import NO_ARG
66from .base import SingletonConstant
67from .cache_key import MemoizedHasCacheKey
68from .cache_key import NO_CACHE
69from .coercions import _document_text_coercion # noqa
70from .operators import ColumnOperators
71from .operators import OperatorClass
72from .traversals import HasCopyInternals
73from .visitors import cloned_traverse
74from .visitors import ExternallyTraversible
75from .visitors import InternalTraversal
76from .visitors import traverse
77from .visitors import Visitable
78from .. import exc
79from .. import inspection
80from .. import util
81from ..util import deprecated
82from ..util import HasMemoized_ro_memoized_attribute
83from ..util import TypingOnly
84from ..util.typing import Self
85from ..util.typing import TupleAny
86from ..util.typing import Unpack
89if typing.TYPE_CHECKING:
90 from ._typing import _ByArgument
91 from ._typing import _ColumnExpressionArgument
92 from ._typing import _ColumnExpressionOrStrLabelArgument
93 from ._typing import _DMLOnlyColumnArgument
94 from ._typing import _HasDialect
95 from ._typing import _InfoType
96 from ._typing import _PropagateAttrsType
97 from ._typing import _TypeEngineArgument
98 from .base import _EntityNamespace
99 from .base import ColumnSet
100 from .cache_key import _CacheKeyTraversalType
101 from .cache_key import CacheKey
102 from .compiler import Compiled
103 from .compiler import SQLCompiler
104 from .functions import FunctionElement
105 from .operators import OperatorType
106 from .schema import Column
107 from .schema import DefaultGenerator
108 from .schema import FetchedValue
109 from .schema import ForeignKey
110 from .selectable import _SelectIterable
111 from .selectable import FromClause
112 from .selectable import NamedFromClause
113 from .selectable import TextualSelect
114 from .sqltypes import TupleType
115 from .type_api import TypeEngine
116 from .visitors import _CloneCallableType
117 from .visitors import _TraverseInternalsType
118 from .visitors import anon_map
119 from ..engine import Connection
120 from ..engine import Dialect
121 from ..engine.interfaces import _CoreMultiExecuteParams
122 from ..engine.interfaces import _CoreSingleExecuteParams
123 from ..engine.interfaces import CacheStats
124 from ..engine.interfaces import CompiledCacheType
125 from ..engine.interfaces import CoreExecuteOptionsParameter
126 from ..engine.interfaces import SchemaTranslateMapType
127 from ..engine.result import Result
130_NUMERIC = Union[float, Decimal]
131_NUMBER = Union[float, int, Decimal]
133_T = TypeVar("_T", bound="Any")
134_T_co = TypeVar("_T_co", bound=Any, covariant=True)
135_OPT = TypeVar("_OPT", bound="Any")
136_NT = TypeVar("_NT", bound="_NUMERIC")
138_NMT = TypeVar("_NMT", bound="_NUMBER")
141@overload
142def literal(
143 value: Any,
144 type_: _TypeEngineArgument[_T],
145 literal_execute: bool = False,
146) -> BindParameter[_T]: ...
149@overload
150def literal(
151 value: _T,
152 type_: None = None,
153 literal_execute: bool = False,
154) -> BindParameter[_T]: ...
157@overload
158def literal(
159 value: Any,
160 type_: Optional[_TypeEngineArgument[Any]] = None,
161 literal_execute: bool = False,
162) -> BindParameter[Any]: ...
165def literal(
166 value: Any,
167 type_: Optional[_TypeEngineArgument[Any]] = None,
168 literal_execute: bool = False,
169) -> BindParameter[Any]:
170 r"""Return a literal clause, bound to a bind parameter.
172 Literal clauses are created automatically when non-
173 :class:`_expression.ClauseElement` objects (such as strings, ints, dates,
174 etc.) are
175 used in a comparison operation with a :class:`_expression.ColumnElement`
176 subclass,
177 such as a :class:`~sqlalchemy.schema.Column` object. Use this function
178 to force the generation of a literal clause, which will be created as a
179 :class:`BindParameter` with a bound value.
181 :param value: the value to be bound. Can be any Python object supported by
182 the underlying DB-API, or is translatable via the given type argument.
184 :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine` which will
185 provide bind-parameter translation for this literal.
187 :param literal_execute: optional bool, when True, the SQL engine will
188 attempt to render the bound value directly in the SQL statement at
189 execution time rather than providing as a parameter value.
191 .. versionadded:: 2.0
193 """
194 return coercions.expect(
195 roles.LiteralValueRole,
196 value,
197 type_=type_,
198 literal_execute=literal_execute,
199 )
202def literal_column(
203 text: str, type_: Optional[_TypeEngineArgument[_T]] = None
204) -> ColumnClause[_T]:
205 r"""Produce a :class:`.ColumnClause` object that has the
206 :paramref:`_expression.column.is_literal` flag set to True.
208 :func:`_expression.literal_column` is similar to
209 :func:`_expression.column`, except that
210 it is more often used as a "standalone" column expression that renders
211 exactly as stated; while :func:`_expression.column`
212 stores a string name that
213 will be assumed to be part of a table and may be quoted as such,
214 :func:`_expression.literal_column` can be that,
215 or any other arbitrary column-oriented
216 expression.
218 :param text: the text of the expression; can be any SQL expression.
219 Quoting rules will not be applied. To specify a column-name expression
220 which should be subject to quoting rules, use the :func:`column`
221 function.
223 :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine`
224 object which will
225 provide result-set translation and additional expression semantics for
226 this column. If left as ``None`` the type will be :class:`.NullType`.
228 .. seealso::
230 :func:`_expression.column`
232 :func:`_expression.text`
234 :ref:`tutorial_select_arbitrary_text`
236 """
237 return ColumnClause(text, type_=type_, is_literal=True)
240class CompilerElement(Visitable):
241 """base class for SQL elements that can be compiled to produce a
242 SQL string.
244 .. versionadded:: 2.0
246 """
248 __slots__ = ()
249 __visit_name__ = "compiler_element"
251 supports_execution = False
253 stringify_dialect = "default"
255 @util.preload_module("sqlalchemy.engine.default")
256 @util.preload_module("sqlalchemy.engine.url")
257 def compile(
258 self,
259 bind: Optional[_HasDialect] = None,
260 dialect: Optional[Dialect] = None,
261 **kw: Any,
262 ) -> Compiled:
263 """Compile this SQL expression.
265 The return value is a :class:`~.Compiled` object.
266 Calling ``str()`` or ``unicode()`` on the returned value will yield a
267 string representation of the result. The
268 :class:`~.Compiled` object also can return a
269 dictionary of bind parameter names and values
270 using the ``params`` accessor.
272 :param bind: An :class:`.Connection` or :class:`.Engine` which
273 can provide a :class:`.Dialect` in order to generate a
274 :class:`.Compiled` object. If the ``bind`` and
275 ``dialect`` parameters are both omitted, a default SQL compiler
276 is used.
278 :param column_keys: Used for INSERT and UPDATE statements, a list of
279 column names which should be present in the VALUES clause of the
280 compiled statement. If ``None``, all columns from the target table
281 object are rendered.
283 :param dialect: A :class:`.Dialect` instance which can generate
284 a :class:`.Compiled` object. This argument takes precedence over
285 the ``bind`` argument.
287 :param compile_kwargs: optional dictionary of additional parameters
288 that will be passed through to the compiler within all "visit"
289 methods. This allows any custom flag to be passed through to
290 a custom compilation construct, for example. It is also used
291 for the case of passing the ``literal_binds`` flag through::
293 from sqlalchemy.sql import table, column, select
295 t = table("t", column("x"))
297 s = select(t).where(t.c.x == 5)
299 print(s.compile(compile_kwargs={"literal_binds": True}))
301 .. seealso::
303 :ref:`faq_sql_expression_string`
305 """
307 if dialect is None:
308 if bind:
309 dialect = bind.dialect
310 elif self.stringify_dialect == "default":
311 dialect = self._default_dialect()
312 else:
313 url = util.preloaded.engine_url
314 dialect = url.URL.create(
315 self.stringify_dialect
316 ).get_dialect()()
318 return self._compiler(dialect, **kw)
320 def _default_dialect(self):
321 default = util.preloaded.engine_default
322 return default.StrCompileDialect()
324 def _compiler(self, dialect: Dialect, **kw: Any) -> Compiled:
325 """Return a compiler appropriate for this ClauseElement, given a
326 Dialect."""
328 if TYPE_CHECKING:
329 assert isinstance(self, ClauseElement)
330 return dialect.statement_compiler(dialect, self, **kw)
332 def __str__(self) -> str:
333 return str(self.compile())
336@inspection._self_inspects
337class ClauseElement(
338 SupportsWrappingAnnotations,
339 MemoizedHasCacheKey,
340 HasCopyInternals,
341 ExternallyTraversible,
342 CompilerElement,
343):
344 """Base class for elements of a programmatically constructed SQL
345 expression.
347 """
349 __visit_name__ = "clause"
351 if TYPE_CHECKING:
353 @util.memoized_property
354 def _propagate_attrs(self) -> _PropagateAttrsType:
355 """like annotations, however these propagate outwards liberally
356 as SQL constructs are built, and are set up at construction time.
358 """
359 ...
361 else:
362 _propagate_attrs = util.EMPTY_DICT
364 @util.ro_memoized_property
365 def description(self) -> Optional[str]:
366 return None
368 _is_clone_of: Optional[Self] = None
370 is_clause_element = True
371 is_selectable = False
372 is_dml = False
373 _is_column_element = False
374 _is_keyed_column_element = False
375 _is_table = False
376 _gen_static_annotations_cache_key = False
377 _is_textual = False
378 _is_from_clause = False
379 _is_returns_rows = False
380 _is_text_clause = False
381 _is_from_container = False
382 _is_select_container = False
383 _is_select_base = False
384 _is_select_statement = False
385 _is_bind_parameter = False
386 _is_clause_list = False
387 _is_lambda_element = False
388 _is_singleton_constant = False
389 _is_immutable = False
390 _is_star = False
392 @property
393 def _order_by_label_element(self) -> Optional[Label[Any]]:
394 return None
396 _cache_key_traversal: _CacheKeyTraversalType = None
398 negation_clause: ColumnElement[bool]
400 if typing.TYPE_CHECKING:
402 def get_children(
403 self, *, omit_attrs: typing_Tuple[str, ...] = ..., **kw: Any
404 ) -> Iterable[ClauseElement]: ...
406 @util.ro_non_memoized_property
407 def _from_objects(self) -> List[FromClause]:
408 return []
410 def _set_propagate_attrs(self, values: Mapping[str, Any]) -> Self:
411 # usually, self._propagate_attrs is empty here. one case where it's
412 # not is a subquery against ORM select, that is then pulled as a
413 # property of an aliased class. should all be good
415 # assert not self._propagate_attrs
417 self._propagate_attrs = util.immutabledict(values)
418 return self
420 def _default_compiler(self) -> SQLCompiler:
421 dialect = self._default_dialect()
422 return dialect.statement_compiler(dialect, self) # type: ignore
424 def _clone(self, **kw: Any) -> Self:
425 """Create a shallow copy of this ClauseElement.
427 This method may be used by a generative API. Its also used as
428 part of the "deep" copy afforded by a traversal that combines
429 the _copy_internals() method.
431 """
433 skip = self._memoized_keys
434 c = self.__class__.__new__(self.__class__)
436 if skip:
437 # ensure this iteration remains atomic
438 c.__dict__ = {
439 k: v for k, v in self.__dict__.copy().items() if k not in skip
440 }
441 else:
442 c.__dict__ = self.__dict__.copy()
444 # this is a marker that helps to "equate" clauses to each other
445 # when a Select returns its list of FROM clauses. the cloning
446 # process leaves around a lot of remnants of the previous clause
447 # typically in the form of column expressions still attached to the
448 # old table.
449 cc = self._is_clone_of
450 c._is_clone_of = cc if cc is not None else self
451 return c
453 def _negate_in_binary(self, negated_op, original_op):
454 """a hook to allow the right side of a binary expression to respond
455 to a negation of the binary expression.
457 Used for the special case of expanding bind parameter with IN.
459 """
460 return self
462 def _with_binary_element_type(self, type_):
463 """in the context of binary expression, convert the type of this
464 object to the one given.
466 applies only to :class:`_expression.ColumnElement` classes.
468 """
469 return self
471 @property
472 def _constructor(self): # type: ignore[override]
473 """return the 'constructor' for this ClauseElement.
475 This is for the purposes for creating a new object of
476 this type. Usually, its just the element's __class__.
477 However, the "Annotated" version of the object overrides
478 to return the class of its proxied element.
480 """
481 return self.__class__
483 @HasMemoized.memoized_attribute
484 def _cloned_set(self):
485 """Return the set consisting all cloned ancestors of this
486 ClauseElement.
488 Includes this ClauseElement. This accessor tends to be used for
489 FromClause objects to identify 'equivalent' FROM clauses, regardless
490 of transformative operations.
492 """
493 s = util.column_set()
494 f: Optional[ClauseElement] = self
496 # note this creates a cycle, asserted in test_memusage. however,
497 # turning this into a plain @property adds tends of thousands of method
498 # calls to Core / ORM performance tests, so the small overhead
499 # introduced by the relatively small amount of short term cycles
500 # produced here is preferable
501 while f is not None:
502 s.add(f)
503 f = f._is_clone_of
504 return s
506 def _de_clone(self):
507 while self._is_clone_of is not None:
508 self = self._is_clone_of
509 return self
511 @util.ro_non_memoized_property
512 def entity_namespace(self) -> _EntityNamespace:
513 raise AttributeError(
514 "This SQL expression has no entity namespace "
515 "with which to filter from."
516 )
518 def __getstate__(self):
519 d = self.__dict__.copy()
520 d.pop("_is_clone_of", None)
521 d.pop("_generate_cache_key", None)
522 return d
524 def _execute_on_connection(
525 self,
526 connection: Connection,
527 distilled_params: _CoreMultiExecuteParams,
528 execution_options: CoreExecuteOptionsParameter,
529 ) -> Result[Unpack[TupleAny]]:
530 if self.supports_execution:
531 if TYPE_CHECKING:
532 assert isinstance(self, Executable)
533 return connection._execute_clauseelement(
534 self, distilled_params, execution_options
535 )
536 else:
537 raise exc.ObjectNotExecutableError(self)
539 def _execute_on_scalar(
540 self,
541 connection: Connection,
542 distilled_params: _CoreMultiExecuteParams,
543 execution_options: CoreExecuteOptionsParameter,
544 ) -> Any:
545 """an additional hook for subclasses to provide a different
546 implementation for connection.scalar() vs. connection.execute().
548 .. versionadded:: 2.0
550 """
551 return self._execute_on_connection(
552 connection, distilled_params, execution_options
553 ).scalar()
555 def _get_embedded_bindparams(self) -> Sequence[BindParameter[Any]]:
556 """Return the list of :class:`.BindParameter` objects embedded in the
557 object.
559 This accomplishes the same purpose as ``visitors.traverse()`` or
560 similar would provide, however by making use of the cache key
561 it takes advantage of memoization of the key to result in fewer
562 net method calls, assuming the statement is also going to be
563 executed.
565 """
567 key = self._generate_cache_key()
568 if key is None:
569 bindparams: List[BindParameter[Any]] = []
571 traverse(self, {}, {"bindparam": bindparams.append})
572 return bindparams
574 else:
575 return key.bindparams
577 def unique_params(
578 self,
579 __optionaldict: Optional[Dict[str, Any]] = None,
580 /,
581 **kwargs: Any,
582 ) -> Self:
583 """Return a copy with :func:`_expression.bindparam` elements
584 replaced.
586 Same functionality as :meth:`_expression.ClauseElement.params`,
587 except adds `unique=True`
588 to affected bind parameters so that multiple statements can be
589 used.
591 """
592 return self._replace_params(True, __optionaldict, kwargs)
594 def params(
595 self,
596 __optionaldict: Optional[Mapping[str, Any]] = None,
597 /,
598 **kwargs: Any,
599 ) -> Self:
600 """Return a copy with :func:`_expression.bindparam` elements
601 replaced.
603 Returns a copy of this ClauseElement with
604 :func:`_expression.bindparam`
605 elements replaced with values taken from the given dictionary::
607 >>> clause = column("x") + bindparam("foo")
608 >>> print(clause.compile().params)
609 {'foo':None}
610 >>> print(clause.params({"foo": 7}).compile().params)
611 {'foo':7}
613 """
614 return self._replace_params(False, __optionaldict, kwargs)
616 @deprecated(
617 "2.1",
618 "The params() and unique_params() methods on non-statement "
619 "ClauseElement objects are deprecated; params() is now limited to "
620 "statement level objects such as select(), insert(), union(), etc. ",
621 )
622 def _replace_params(
623 self,
624 unique: bool,
625 optionaldict: Optional[Mapping[str, Any]],
626 kwargs: Dict[str, Any],
627 ) -> Self:
628 if optionaldict:
629 kwargs.update(optionaldict)
631 def visit_bindparam(bind: BindParameter[Any]) -> None:
632 if bind.key in kwargs:
633 bind.value = kwargs[bind.key]
634 bind.required = False
635 if unique:
636 bind._convert_to_unique()
638 return cloned_traverse(
639 self,
640 {"maintain_key": True, "detect_subquery_cols": True},
641 {"bindparam": visit_bindparam},
642 )
644 def compare(self, other: ClauseElement, **kw: Any) -> bool:
645 r"""Compare this :class:`_expression.ClauseElement` to
646 the given :class:`_expression.ClauseElement`.
648 Subclasses should override the default behavior, which is a
649 straight identity comparison.
651 \**kw are arguments consumed by subclass ``compare()`` methods and
652 may be used to modify the criteria for comparison
653 (see :class:`_expression.ColumnElement`).
655 """
656 return traversals.compare(self, other, **kw)
658 def self_group(
659 self, against: Optional[OperatorType] = None
660 ) -> ClauseElement:
661 """Apply a 'grouping' to this :class:`_expression.ClauseElement`.
663 This method is overridden by subclasses to return a "grouping"
664 construct, i.e. parenthesis. In particular it's used by "binary"
665 expressions to provide a grouping around themselves when placed into a
666 larger expression, as well as by :func:`_expression.select`
667 constructs when placed into the FROM clause of another
668 :func:`_expression.select`. (Note that subqueries should be
669 normally created using the :meth:`_expression.Select.alias` method,
670 as many
671 platforms require nested SELECT statements to be named).
673 As expressions are composed together, the application of
674 :meth:`self_group` is automatic - end-user code should never
675 need to use this method directly. Note that SQLAlchemy's
676 clause constructs take operator precedence into account -
677 so parenthesis might not be needed, for example, in
678 an expression like ``x OR (y AND z)`` - AND takes precedence
679 over OR.
681 The base :meth:`self_group` method of
682 :class:`_expression.ClauseElement`
683 just returns self.
684 """
685 return self
687 def _ungroup(self) -> ClauseElement:
688 """Return this :class:`_expression.ClauseElement`
689 without any groupings.
690 """
692 return self
694 def _compile_w_cache(
695 self,
696 dialect: Dialect,
697 *,
698 compiled_cache: Optional[CompiledCacheType],
699 column_keys: List[str],
700 for_executemany: bool = False,
701 schema_translate_map: Optional[SchemaTranslateMapType] = None,
702 **kw: Any,
703 ) -> tuple[
704 Compiled,
705 Sequence[BindParameter[Any]] | None,
706 _CoreSingleExecuteParams | None,
707 CacheStats,
708 ]:
709 elem_cache_key: Optional[CacheKey]
711 if compiled_cache is not None and dialect._supports_statement_cache:
712 elem_cache_key = self._generate_cache_key()
713 else:
714 elem_cache_key = None
716 extracted_params: Optional[Sequence[BindParameter[Any]]]
717 if elem_cache_key is not None:
718 if TYPE_CHECKING:
719 assert compiled_cache is not None
721 cache_key, extracted_params, param_dict = elem_cache_key
722 key = (
723 dialect,
724 cache_key,
725 tuple(column_keys),
726 bool(schema_translate_map),
727 for_executemany,
728 )
729 compiled_sql = compiled_cache.get(key)
731 if compiled_sql is None:
732 cache_hit = dialect.CACHE_MISS
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 )
741 # ensure that params of the current statement are not
742 # left in the cache
743 assert not compiled_sql._collect_params # type: ignore[attr-defined] # noqa: E501
744 compiled_cache[key] = compiled_sql
745 else:
746 cache_hit = dialect.CACHE_HIT
747 else:
748 param_dict = None
749 extracted_params = None
750 compiled_sql = self._compiler(
751 dialect,
752 cache_key=None,
753 column_keys=column_keys,
754 for_executemany=for_executemany,
755 schema_translate_map=schema_translate_map,
756 **kw,
757 )
758 # here instead the params need to be extracted, since we don't
759 # have them otherwise
760 assert compiled_sql._collect_params # type: ignore[attr-defined] # noqa: E501
762 if not dialect._supports_statement_cache:
763 cache_hit = dialect.NO_DIALECT_SUPPORT
764 elif compiled_cache is None:
765 cache_hit = dialect.CACHING_DISABLED
766 else:
767 cache_hit = dialect.NO_CACHE_KEY
769 return compiled_sql, extracted_params, param_dict, cache_hit
771 def __invert__(self):
772 # undocumented element currently used by the ORM for
773 # relationship.contains()
774 if hasattr(self, "negation_clause"):
775 return self.negation_clause
776 else:
777 return self._negate()
779 def _negate(self) -> ClauseElement:
780 # TODO: this code is uncovered and in all likelihood is not included
781 # in any codepath. So this should raise NotImplementedError in 2.1
782 grouped = self.self_group(against=operators.inv)
783 assert isinstance(grouped, ColumnElement)
784 return UnaryExpression(grouped, operator=operators.inv)
786 def __bool__(self):
787 raise TypeError("Boolean value of this clause is not defined")
789 def __repr__(self):
790 friendly = self.description
791 if friendly is None:
792 return object.__repr__(self)
793 else:
794 return "<%s.%s at 0x%x; %s>" % (
795 self.__module__,
796 self.__class__.__name__,
797 id(self),
798 friendly,
799 )
802class DQLDMLClauseElement(ClauseElement):
803 """represents a :class:`.ClauseElement` that compiles to a DQL or DML
804 expression, not DDL.
806 .. versionadded:: 2.0
808 """
810 if typing.TYPE_CHECKING:
812 def _compiler(self, dialect: Dialect, **kw: Any) -> SQLCompiler:
813 """Return a compiler appropriate for this ClauseElement, given a
814 Dialect."""
815 ...
817 def compile( # noqa: A001
818 self,
819 bind: Optional[_HasDialect] = None,
820 dialect: Optional[Dialect] = None,
821 **kw: Any,
822 ) -> SQLCompiler: ...
825class CompilerColumnElement(
826 roles.DMLColumnRole,
827 roles.DDLConstraintColumnRole,
828 roles.ColumnsClauseRole,
829 CompilerElement,
830):
831 """A compiler-only column element used for ad-hoc string compilations.
833 .. versionadded:: 2.0
835 """
837 __slots__ = ()
839 _propagate_attrs = util.EMPTY_DICT
840 _is_collection_aggregate = False
841 _is_implicitly_boolean = False
843 def _with_binary_element_type(self, type_):
844 raise NotImplementedError()
846 def _gen_cache_key(self, anon_map, bindparams):
847 raise NotImplementedError()
849 @property
850 def _from_objects(self) -> List[FromClause]:
851 raise NotImplementedError()
854# SQLCoreOperations should be suiting the ExpressionElementRole
855# and ColumnsClauseRole. however the MRO issues become too elaborate
856# at the moment.
857class SQLCoreOperations(Generic[_T_co], ColumnOperators, TypingOnly):
858 __slots__ = ()
860 # annotations for comparison methods
861 # these are from operators->Operators / ColumnOperators,
862 # redefined with the specific types returned by ColumnElement hierarchies
863 if typing.TYPE_CHECKING:
865 @util.non_memoized_property
866 def _propagate_attrs(self) -> _PropagateAttrsType: ...
868 def operate(
869 self, op: OperatorType, *other: Any, **kwargs: Any
870 ) -> ColumnElement[Any]: ...
872 def reverse_operate(
873 self, op: OperatorType, other: Any, **kwargs: Any
874 ) -> ColumnElement[Any]: ...
876 @overload
877 def op(
878 self,
879 opstring: str,
880 precedence: int = ...,
881 is_comparison: bool = ...,
882 *,
883 return_type: _TypeEngineArgument[_OPT],
884 python_impl: Optional[Callable[..., Any]] = None,
885 operator_class: OperatorClass = ...,
886 visit_name: Optional[str] = ...,
887 ) -> Callable[[Any], BinaryExpression[_OPT]]: ...
889 @overload
890 def op(
891 self,
892 opstring: str,
893 precedence: int = ...,
894 is_comparison: bool = ...,
895 return_type: Optional[_TypeEngineArgument[Any]] = ...,
896 python_impl: Optional[Callable[..., Any]] = ...,
897 operator_class: OperatorClass = ...,
898 visit_name: Optional[str] = ...,
899 ) -> Callable[[Any], BinaryExpression[Any]]: ...
901 def op(
902 self,
903 opstring: str,
904 precedence: int = 0,
905 is_comparison: bool = False,
906 return_type: Optional[_TypeEngineArgument[Any]] = None,
907 python_impl: Optional[Callable[..., Any]] = None,
908 operator_class: OperatorClass = OperatorClass.BASE,
909 visit_name: Optional[str] = None,
910 ) -> Callable[[Any], BinaryExpression[Any]]: ...
912 def bool_op(
913 self,
914 opstring: str,
915 precedence: int = 0,
916 python_impl: Optional[Callable[..., Any]] = None,
917 ) -> Callable[[Any], BinaryExpression[bool]]: ...
919 def __and__(self, other: Any) -> BooleanClauseList: ...
921 def __or__(self, other: Any) -> BooleanClauseList: ...
923 def __invert__(self) -> ColumnElement[_T_co]: ...
925 def __lt__(self, other: Any) -> ColumnElement[bool]: ...
927 def __le__(self, other: Any) -> ColumnElement[bool]: ...
929 # declare also that this class has an hash method otherwise
930 # it may be assumed to be None by type checkers since the
931 # object defines __eq__ and python sets it to None in that case:
932 # https://docs.python.org/3/reference/datamodel.html#object.__hash__
933 def __hash__(self) -> int: ...
935 def __eq__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
936 ...
938 def __ne__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
939 ...
941 def is_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
943 def is_not_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
945 def __gt__(self, other: Any) -> ColumnElement[bool]: ...
947 def __ge__(self, other: Any) -> ColumnElement[bool]: ...
949 def __neg__(self) -> UnaryExpression[_T_co]: ...
951 def __contains__(self, other: Any) -> ColumnElement[bool]: ...
953 def __getitem__(self, index: Any) -> ColumnElement[Any]: ...
955 @overload
956 def __lshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
958 @overload
959 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
961 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
963 @overload
964 def __rlshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
966 @overload
967 def __rlshift__(self, other: Any) -> ColumnElement[Any]: ...
969 def __rlshift__(self, other: Any) -> ColumnElement[Any]: ...
971 @overload
972 def __rshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
974 @overload
975 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
977 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
979 @overload
980 def __rrshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
982 @overload
983 def __rrshift__(self, other: Any) -> ColumnElement[Any]: ...
985 def __rrshift__(self, other: Any) -> ColumnElement[Any]: ...
987 def __matmul__(self, other: Any) -> ColumnElement[Any]: ...
989 def __rmatmul__(self, other: Any) -> ColumnElement[Any]: ...
991 @overload
992 def concat(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
994 @overload
995 def concat(self, other: Any) -> ColumnElement[Any]: ...
997 def concat(self, other: Any) -> ColumnElement[Any]: ...
999 def like(
1000 self, other: Any, escape: Optional[str] = None
1001 ) -> BinaryExpression[bool]: ...
1003 def ilike(
1004 self, other: Any, escape: Optional[str] = None
1005 ) -> BinaryExpression[bool]: ...
1007 def bitwise_xor(self, other: Any) -> BinaryExpression[Any]: ...
1009 def bitwise_or(self, other: Any) -> BinaryExpression[Any]: ...
1011 def bitwise_and(self, other: Any) -> BinaryExpression[Any]: ...
1013 def bitwise_not(self) -> UnaryExpression[_T_co]: ...
1015 def bitwise_lshift(self, other: Any) -> BinaryExpression[Any]: ...
1017 def bitwise_rshift(self, other: Any) -> BinaryExpression[Any]: ...
1019 def in_(
1020 self,
1021 other: Union[
1022 Iterable[Any], BindParameter[Any], roles.InElementRole
1023 ],
1024 ) -> BinaryExpression[bool]: ...
1026 def not_in(
1027 self,
1028 other: Union[
1029 Iterable[Any], BindParameter[Any], roles.InElementRole
1030 ],
1031 ) -> BinaryExpression[bool]: ...
1033 def notin_(
1034 self,
1035 other: Union[
1036 Iterable[Any], BindParameter[Any], roles.InElementRole
1037 ],
1038 ) -> BinaryExpression[bool]: ...
1040 def not_like(
1041 self, other: Any, escape: Optional[str] = None
1042 ) -> BinaryExpression[bool]: ...
1044 def notlike(
1045 self, other: Any, escape: Optional[str] = None
1046 ) -> BinaryExpression[bool]: ...
1048 def not_ilike(
1049 self, other: Any, escape: Optional[str] = None
1050 ) -> BinaryExpression[bool]: ...
1052 def notilike(
1053 self, other: Any, escape: Optional[str] = None
1054 ) -> BinaryExpression[bool]: ...
1056 def is_(self, other: Any) -> BinaryExpression[bool]: ...
1058 def is_not(self, other: Any) -> BinaryExpression[bool]: ...
1060 def isnot(self, other: Any) -> BinaryExpression[bool]: ...
1062 def startswith(
1063 self,
1064 other: Any,
1065 escape: Optional[str] = None,
1066 autoescape: bool = False,
1067 ) -> ColumnElement[bool]: ...
1069 def istartswith(
1070 self,
1071 other: Any,
1072 escape: Optional[str] = None,
1073 autoescape: bool = False,
1074 ) -> ColumnElement[bool]: ...
1076 def endswith(
1077 self,
1078 other: Any,
1079 escape: Optional[str] = None,
1080 autoescape: bool = False,
1081 ) -> ColumnElement[bool]: ...
1083 def iendswith(
1084 self,
1085 other: Any,
1086 escape: Optional[str] = None,
1087 autoescape: bool = False,
1088 ) -> ColumnElement[bool]: ...
1090 def contains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1092 def icontains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1094 def match(self, other: Any, **kwargs: Any) -> ColumnElement[bool]: ...
1096 def regexp_match(
1097 self, pattern: Any, flags: Optional[str] = None
1098 ) -> ColumnElement[bool]: ...
1100 def regexp_replace(
1101 self, pattern: Any, replacement: Any, flags: Optional[str] = None
1102 ) -> ColumnElement[str]: ...
1104 def desc(self) -> UnaryExpression[_T_co]: ...
1106 def asc(self) -> UnaryExpression[_T_co]: ...
1108 def nulls_first(self) -> UnaryExpression[_T_co]: ...
1110 def nullsfirst(self) -> UnaryExpression[_T_co]: ...
1112 def nulls_last(self) -> UnaryExpression[_T_co]: ...
1114 def nullslast(self) -> UnaryExpression[_T_co]: ...
1116 def collate(self, collation: str) -> CollationClause: ...
1118 def between(
1119 self, cleft: Any, cright: Any, symmetric: bool = False
1120 ) -> BinaryExpression[bool]: ...
1122 def distinct(self: _SQO[_T_co]) -> UnaryExpression[_T_co]: ...
1124 def any_(self) -> CollectionAggregate[Any]: ...
1126 def all_(self) -> CollectionAggregate[Any]: ...
1128 # numeric overloads. These need more tweaking
1129 # in particular they all need to have a variant for Optiona[_T]
1130 # because Optional only applies to the data side, not the expression
1131 # side
1133 @overload
1134 def __add__(
1135 self: _SQO[_NMT],
1136 other: Any,
1137 ) -> ColumnElement[_NMT]: ...
1139 @overload
1140 def __add__(
1141 self: _SQO[str],
1142 other: Any,
1143 ) -> ColumnElement[str]: ...
1145 @overload
1146 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1148 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1150 @overload
1151 def __radd__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1153 @overload
1154 def __radd__(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
1156 def __radd__(self, other: Any) -> ColumnElement[Any]: ...
1158 @overload
1159 def __sub__(
1160 self: _SQO[_NMT],
1161 other: Any,
1162 ) -> ColumnElement[_NMT]: ...
1164 @overload
1165 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1167 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1169 @overload
1170 def __rsub__(
1171 self: _SQO[_NMT],
1172 other: Any,
1173 ) -> ColumnElement[_NMT]: ...
1175 @overload
1176 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1178 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1180 @overload
1181 def __mul__(
1182 self: _SQO[_NMT],
1183 other: Any,
1184 ) -> ColumnElement[_NMT]: ...
1186 @overload
1187 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1189 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1191 @overload
1192 def __rmul__(
1193 self: _SQO[_NMT],
1194 other: Any,
1195 ) -> ColumnElement[_NMT]: ...
1197 @overload
1198 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1200 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1202 @overload
1203 def __mod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1205 @overload
1206 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1208 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1210 @overload
1211 def __rmod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1213 @overload
1214 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1216 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1218 @overload
1219 def __truediv__(
1220 self: _SQO[int], other: Any
1221 ) -> ColumnElement[_NUMERIC]: ...
1223 @overload
1224 def __truediv__(self: _SQO[_NT], other: Any) -> ColumnElement[_NT]: ...
1226 @overload
1227 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1229 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1231 @overload
1232 def __rtruediv__(
1233 self: _SQO[_NMT], other: Any
1234 ) -> ColumnElement[_NUMERIC]: ...
1236 @overload
1237 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1239 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1241 @overload
1242 def __floordiv__(
1243 self: _SQO[_NMT], other: Any
1244 ) -> ColumnElement[_NMT]: ...
1246 @overload
1247 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1249 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1251 @overload
1252 def __rfloordiv__(
1253 self: _SQO[_NMT], other: Any
1254 ) -> ColumnElement[_NMT]: ...
1256 @overload
1257 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1259 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1262class SQLColumnExpression(
1263 SQLCoreOperations[_T_co], roles.ExpressionElementRole[_T_co], TypingOnly
1264):
1265 """A type that may be used to indicate any SQL column element or object
1266 that acts in place of one.
1268 :class:`.SQLColumnExpression` is a base of
1269 :class:`.ColumnElement`, as well as within the bases of ORM elements
1270 such as :class:`.InstrumentedAttribute`, and may be used in :pep:`484`
1271 typing to indicate arguments or return values that should behave
1272 as column expressions.
1274 .. versionadded:: 2.0.0b4
1277 """
1279 __slots__ = ()
1282_SQO = SQLCoreOperations
1285class ColumnElement(
1286 roles.ColumnArgumentOrKeyRole,
1287 roles.StatementOptionRole,
1288 roles.WhereHavingRole,
1289 roles.BinaryElementRole[_T],
1290 roles.OrderByRole,
1291 roles.ColumnsClauseRole,
1292 roles.LimitOffsetRole,
1293 roles.DMLColumnRole,
1294 roles.DDLConstraintColumnRole,
1295 roles.DDLExpressionRole,
1296 SQLColumnExpression[_T],
1297 DQLDMLClauseElement,
1298):
1299 """Represent a column-oriented SQL expression suitable for usage in the
1300 "columns" clause, WHERE clause etc. of a statement.
1302 While the most familiar kind of :class:`_expression.ColumnElement` is the
1303 :class:`_schema.Column` object, :class:`_expression.ColumnElement`
1304 serves as the basis
1305 for any unit that may be present in a SQL expression, including
1306 the expressions themselves, SQL functions, bound parameters,
1307 literal expressions, keywords such as ``NULL``, etc.
1308 :class:`_expression.ColumnElement`
1309 is the ultimate base class for all such elements.
1311 A wide variety of SQLAlchemy Core functions work at the SQL expression
1312 level, and are intended to accept instances of
1313 :class:`_expression.ColumnElement` as
1314 arguments. These functions will typically document that they accept a
1315 "SQL expression" as an argument. What this means in terms of SQLAlchemy
1316 usually refers to an input which is either already in the form of a
1317 :class:`_expression.ColumnElement` object,
1318 or a value which can be **coerced** into
1319 one. The coercion rules followed by most, but not all, SQLAlchemy Core
1320 functions with regards to SQL expressions are as follows:
1322 * a literal Python value, such as a string, integer or floating
1323 point value, boolean, datetime, ``Decimal`` object, or virtually
1324 any other Python object, will be coerced into a "literal bound
1325 value". This generally means that a :func:`.bindparam` will be
1326 produced featuring the given value embedded into the construct; the
1327 resulting :class:`.BindParameter` object is an instance of
1328 :class:`_expression.ColumnElement`.
1329 The Python value will ultimately be sent
1330 to the DBAPI at execution time as a parameterized argument to the
1331 ``execute()`` or ``executemany()`` methods, after SQLAlchemy
1332 type-specific converters (e.g. those provided by any associated
1333 :class:`.TypeEngine` objects) are applied to the value.
1335 * any special object value, typically ORM-level constructs, which
1336 feature an accessor called ``__clause_element__()``. The Core
1337 expression system looks for this method when an object of otherwise
1338 unknown type is passed to a function that is looking to coerce the
1339 argument into a :class:`_expression.ColumnElement` and sometimes a
1340 :class:`_expression.SelectBase` expression.
1341 It is used within the ORM to
1342 convert from ORM-specific objects like mapped classes and
1343 mapped attributes into Core expression objects.
1345 * The Python ``None`` value is typically interpreted as ``NULL``,
1346 which in SQLAlchemy Core produces an instance of :func:`.null`.
1348 A :class:`_expression.ColumnElement` provides the ability to generate new
1349 :class:`_expression.ColumnElement`
1350 objects using Python expressions. This means that Python operators
1351 such as ``==``, ``!=`` and ``<`` are overloaded to mimic SQL operations,
1352 and allow the instantiation of further :class:`_expression.ColumnElement`
1353 instances
1354 which are composed from other, more fundamental
1355 :class:`_expression.ColumnElement`
1356 objects. For example, two :class:`.ColumnClause` objects can be added
1357 together with the addition operator ``+`` to produce
1358 a :class:`.BinaryExpression`.
1359 Both :class:`.ColumnClause` and :class:`.BinaryExpression` are subclasses
1360 of :class:`_expression.ColumnElement`:
1362 .. sourcecode:: pycon+sql
1364 >>> from sqlalchemy.sql import column
1365 >>> column("a") + column("b")
1366 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
1367 >>> print(column("a") + column("b"))
1368 {printsql}a + b
1370 .. seealso::
1372 :class:`_schema.Column`
1374 :func:`_expression.column`
1376 """
1378 __visit_name__ = "column_element"
1380 primary_key: bool = False
1381 _is_clone_of: Optional[ColumnElement[_T]]
1382 _is_column_element = True
1383 _insert_sentinel: bool = False
1384 _omit_from_statements = False
1385 _is_collection_aggregate = False
1387 foreign_keys: AbstractSet[ForeignKey] = frozenset()
1389 @util.memoized_property
1390 def _proxies(self) -> List[ColumnElement[Any]]:
1391 return []
1393 @util.non_memoized_property
1394 def _tq_label(self) -> Optional[str]:
1395 """The named label that can be used to target
1396 this column in a result set in a "table qualified" context.
1398 This label is almost always the label used when
1399 rendering <expr> AS <label> in a SELECT statement when using
1400 the LABEL_STYLE_TABLENAME_PLUS_COL label style, which is what the
1401 legacy ORM ``Query`` object uses as well.
1403 For a regular Column bound to a Table, this is typically the label
1404 <tablename>_<columnname>. For other constructs, different rules
1405 may apply, such as anonymized labels and others.
1407 .. versionchanged:: 1.4.21 renamed from ``._label``
1409 """
1410 return None
1412 key: Optional[str] = None
1413 """The 'key' that in some circumstances refers to this object in a
1414 Python namespace.
1416 This typically refers to the "key" of the column as present in the
1417 ``.c`` collection of a selectable, e.g. ``sometable.c["somekey"]`` would
1418 return a :class:`_schema.Column` with a ``.key`` of "somekey".
1420 """
1422 @HasMemoized.memoized_attribute
1423 def _tq_key_label(self) -> Optional[str]:
1424 """A label-based version of 'key' that in some circumstances refers
1425 to this object in a Python namespace.
1428 _tq_key_label comes into play when a select() statement is constructed
1429 with apply_labels(); in this case, all Column objects in the ``.c``
1430 collection are rendered as <tablename>_<columnname> in SQL; this is
1431 essentially the value of ._label. But to locate those columns in the
1432 ``.c`` collection, the name is along the lines of <tablename>_<key>;
1433 that's the typical value of .key_label.
1435 .. versionchanged:: 1.4.21 renamed from ``._key_label``
1437 """
1438 return self._proxy_key
1440 @property
1441 def _key_label(self) -> Optional[str]:
1442 """legacy; renamed to _tq_key_label"""
1443 return self._tq_key_label
1445 @property
1446 def _label(self) -> Optional[str]:
1447 """legacy; renamed to _tq_label"""
1448 return self._tq_label
1450 @property
1451 def _non_anon_label(self) -> Optional[str]:
1452 """the 'name' that naturally applies this element when rendered in
1453 SQL.
1455 Concretely, this is the "name" of a column or a label in a
1456 SELECT statement; ``<columnname>`` and ``<labelname>`` below:
1458 .. sourcecode:: sql
1460 SELECT <columnmame> FROM table
1462 SELECT column AS <labelname> FROM table
1464 Above, the two names noted will be what's present in the DBAPI
1465 ``cursor.description`` as the names.
1467 If this attribute returns ``None``, it means that the SQL element as
1468 written does not have a 100% fully predictable "name" that would appear
1469 in the ``cursor.description``. Examples include SQL functions, CAST
1470 functions, etc. While such things do return names in
1471 ``cursor.description``, they are only predictable on a
1472 database-specific basis; e.g. an expression like ``MAX(table.col)`` may
1473 appear as the string ``max`` on one database (like PostgreSQL) or may
1474 appear as the whole expression ``max(table.col)`` on SQLite.
1476 The default implementation looks for a ``.name`` attribute on the
1477 object, as has been the precedent established in SQLAlchemy for many
1478 years. An exception is made on the ``FunctionElement`` subclass
1479 so that the return value is always ``None``.
1481 .. versionadded:: 1.4.21
1485 """
1486 return getattr(self, "name", None)
1488 _render_label_in_columns_clause = True
1489 """A flag used by select._columns_plus_names that helps to determine
1490 we are actually going to render in terms of "SELECT <col> AS <label>".
1491 This flag can be returned as False for some Column objects that want
1492 to be rendered as simple "SELECT <col>"; typically columns that don't have
1493 any parent table and are named the same as what the label would be
1494 in any case.
1496 """
1498 _allow_label_resolve = True
1499 """A flag that can be flipped to prevent a column from being resolvable
1500 by string label name.
1502 The joined eager loader strategy in the ORM uses this, for example.
1504 """
1506 _is_implicitly_boolean = False
1508 _alt_names: Sequence[str] = ()
1510 if TYPE_CHECKING:
1512 def _ungroup(self) -> ColumnElement[_T]: ...
1514 @overload
1515 def self_group(self, against: None = None) -> ColumnElement[_T]: ...
1517 @overload
1518 def self_group(
1519 self, against: Optional[OperatorType] = None
1520 ) -> ColumnElement[Any]: ...
1522 def self_group(
1523 self, against: Optional[OperatorType] = None
1524 ) -> ColumnElement[Any]:
1525 if (
1526 against in (operators.and_, operators.or_, operators._asbool)
1527 and self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity
1528 ):
1529 return AsBoolean(self, operators.is_true, operators.is_false)
1530 elif against in (operators.any_op, operators.all_op):
1531 return Grouping(self)
1532 else:
1533 return self
1535 @overload
1536 def _negate(self: ColumnElement[bool]) -> ColumnElement[bool]: ...
1538 @overload
1539 def _negate(self: ColumnElement[_T]) -> ColumnElement[_T]: ...
1541 def _negate(self) -> ColumnElement[Any]:
1542 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
1543 return AsBoolean(self, operators.is_false, operators.is_true)
1544 else:
1545 grouped = self.self_group(against=operators.inv)
1546 assert isinstance(grouped, ColumnElement)
1547 return UnaryExpression(
1548 grouped,
1549 operator=operators.inv,
1550 )
1552 type: TypeEngine[_T]
1554 if not TYPE_CHECKING:
1556 @util.memoized_property
1557 def type(self) -> TypeEngine[_T]: # noqa: A001
1558 # used for delayed setup of
1559 # type_api
1560 return type_api.NULLTYPE
1562 @HasMemoized.memoized_attribute
1563 def comparator(self) -> TypeEngine.Comparator[_T]:
1564 try:
1565 comparator_factory = self.type.comparator_factory
1566 except AttributeError as err:
1567 raise TypeError(
1568 "Object %r associated with '.type' attribute "
1569 "is not a TypeEngine class or object" % self.type
1570 ) from err
1571 else:
1572 return comparator_factory(self)
1574 def __setstate__(self, state):
1575 self.__dict__.update(state)
1577 def __getattr__(self, key: str) -> Any:
1578 try:
1579 return getattr(self.comparator, key)
1580 except AttributeError as err:
1581 raise AttributeError(
1582 "Neither %r object nor %r object has an attribute %r"
1583 % (
1584 type(self).__name__,
1585 type(self.comparator).__name__,
1586 key,
1587 )
1588 ) from err
1590 def operate(
1591 self,
1592 op: operators.OperatorType,
1593 *other: Any,
1594 **kwargs: Any,
1595 ) -> ColumnElement[Any]:
1596 return op(self.comparator, *other, **kwargs) # type: ignore[no-any-return] # noqa: E501
1598 def reverse_operate(
1599 self, op: operators.OperatorType, other: Any, **kwargs: Any
1600 ) -> ColumnElement[Any]:
1601 return op(other, self.comparator, **kwargs) # type: ignore[no-any-return] # noqa: E501
1603 def _bind_param(
1604 self,
1605 operator: operators.OperatorType,
1606 obj: Any,
1607 type_: Optional[TypeEngine[_T]] = None,
1608 expanding: bool = False,
1609 ) -> BindParameter[_T]:
1610 return BindParameter(
1611 None,
1612 obj,
1613 _compared_to_operator=operator,
1614 type_=type_,
1615 _compared_to_type=self.type,
1616 unique=True,
1617 expanding=expanding,
1618 )
1620 @property
1621 def expression(self) -> ColumnElement[Any]:
1622 """Return a column expression.
1624 Part of the inspection interface; returns self.
1626 """
1627 return self
1629 @property
1630 def _select_iterable(self) -> _SelectIterable:
1631 return (self,)
1633 @util.memoized_property
1634 def base_columns(self) -> FrozenSet[ColumnElement[Any]]:
1635 return frozenset(c for c in self.proxy_set if not c._proxies)
1637 @util.memoized_property
1638 def proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1639 """set of all columns we are proxying
1641 as of 2.0 this is explicitly deannotated columns. previously it was
1642 effectively deannotated columns but wasn't enforced. annotated
1643 columns should basically not go into sets if at all possible because
1644 their hashing behavior is very non-performant.
1646 """
1647 return frozenset([self._deannotate()]).union(
1648 itertools.chain(*[c.proxy_set for c in self._proxies])
1649 )
1651 @util.memoized_property
1652 def _expanded_proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1653 return frozenset(_expand_cloned(self.proxy_set))
1655 def _uncached_proxy_list(self) -> List[ColumnElement[Any]]:
1656 """An 'uncached' version of proxy set.
1658 This list includes annotated columns which perform very poorly in
1659 set operations.
1661 """
1663 return [self] + list(
1664 itertools.chain(*[c._uncached_proxy_list() for c in self._proxies])
1665 )
1667 def shares_lineage(self, othercolumn: ColumnElement[Any]) -> bool:
1668 """Return True if the given :class:`_expression.ColumnElement`
1669 has a common ancestor to this :class:`_expression.ColumnElement`."""
1671 return bool(self.proxy_set.intersection(othercolumn.proxy_set))
1673 def _compare_name_for_result(self, other: ColumnElement[Any]) -> bool:
1674 """Return True if the given column element compares to this one
1675 when targeting within a result row."""
1677 return (
1678 hasattr(other, "name")
1679 and hasattr(self, "name")
1680 and other.name == self.name
1681 )
1683 @HasMemoized.memoized_attribute
1684 def _proxy_key(self) -> Optional[str]:
1685 if self._annotations and "proxy_key" in self._annotations:
1686 return cast(str, self._annotations["proxy_key"])
1688 name = self.key
1689 if not name:
1690 # there's a bit of a seeming contradiction which is that the
1691 # "_non_anon_label" of a column can in fact be an
1692 # "_anonymous_label"; this is when it's on a column that is
1693 # proxying for an anonymous expression in a subquery.
1694 name = self._non_anon_label
1696 if isinstance(name, _anonymous_label):
1697 return None
1698 else:
1699 return name
1701 @HasMemoized.memoized_attribute
1702 def _expression_label(self) -> Optional[str]:
1703 """a suggested label to use in the case that the column has no name,
1704 which should be used if possible as the explicit 'AS <label>'
1705 where this expression would normally have an anon label.
1707 this is essentially mostly what _proxy_key does except it returns
1708 None if the column has a normal name that can be used.
1710 """
1712 if getattr(self, "name", None) is not None:
1713 return None
1714 elif self._annotations and "proxy_key" in self._annotations:
1715 return cast(str, self._annotations["proxy_key"])
1716 else:
1717 return None
1719 def _make_proxy(
1720 self,
1721 selectable: FromClause,
1722 *,
1723 primary_key: ColumnSet,
1724 foreign_keys: Set[KeyedColumnElement[Any]],
1725 name: Optional[str] = None,
1726 key: Optional[str] = None,
1727 name_is_truncatable: bool = False,
1728 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
1729 **kw: Any,
1730 ) -> typing_Tuple[str, ColumnClause[_T]]:
1731 """Create a new :class:`_expression.ColumnElement` representing this
1732 :class:`_expression.ColumnElement` as it appears in the select list of
1733 a descending selectable.
1735 """
1736 if name is None:
1737 name = self._anon_name_label
1738 if key is None:
1739 key = self._proxy_key
1740 else:
1741 key = name
1743 assert key is not None
1745 co: ColumnClause[_T] = ColumnClause(
1746 (
1747 coercions.expect(roles.TruncatedLabelRole, name)
1748 if name_is_truncatable
1749 else name
1750 ),
1751 type_=getattr(self, "type", None),
1752 _selectable=selectable,
1753 )
1755 co._propagate_attrs = selectable._propagate_attrs
1756 if compound_select_cols:
1757 co._proxies = list(compound_select_cols)
1758 else:
1759 co._proxies = [self]
1760 if selectable._is_clone_of is not None:
1761 co._is_clone_of = selectable._is_clone_of.columns.get(key)
1762 return key, co
1764 def cast(self, type_: _TypeEngineArgument[_OPT]) -> Cast[_OPT]:
1765 """Produce a type cast, i.e. ``CAST(<expression> AS <type>)``.
1767 This is a shortcut to the :func:`_expression.cast` function.
1769 .. seealso::
1771 :ref:`tutorial_casts`
1773 :func:`_expression.cast`
1775 :func:`_expression.type_coerce`
1777 """
1778 return Cast(self, type_)
1780 def label(self, name: Optional[str]) -> Label[_T]:
1781 """Produce a column label, i.e. ``<columnname> AS <name>``.
1783 This is a shortcut to the :func:`_expression.label` function.
1785 If 'name' is ``None``, an anonymous label name will be generated.
1787 """
1788 return Label(name, self, self.type)
1790 def _anon_label(
1791 self, seed: Optional[str], add_hash: Optional[int] = None
1792 ) -> _anonymous_label:
1793 while self._is_clone_of is not None:
1794 self = self._is_clone_of
1796 # as of 1.4 anonymous label for ColumnElement uses hash(), not id(),
1797 # as the identifier, because a column and its annotated version are
1798 # the same thing in a SQL statement
1799 hash_value = hash(self)
1801 if add_hash:
1802 # this path is used for disambiguating anon labels that would
1803 # otherwise be the same name for the same element repeated.
1804 # an additional numeric value is factored in for each label.
1806 # shift hash(self) (which is id(self), typically 8 byte integer)
1807 # 16 bits leftward. fill extra add_hash on right
1808 assert add_hash < (2 << 15)
1809 assert seed
1810 hash_value = (hash_value << 16) | add_hash
1812 # extra underscore is added for labels with extra hash
1813 # values, to isolate the "deduped anon" namespace from the
1814 # regular namespace. eliminates chance of these
1815 # manufactured hash values overlapping with regular ones for some
1816 # undefined python interpreter
1817 seed = seed + "_"
1819 if isinstance(seed, _anonymous_label):
1820 # NOTE: the space after the hash is required
1821 return _anonymous_label(f"{seed}%({hash_value} )s")
1823 return _anonymous_label.safe_construct(hash_value, seed or "anon")
1825 @util.memoized_property
1826 def _anon_name_label(self) -> str:
1827 """Provides a constant 'anonymous label' for this ColumnElement.
1829 This is a label() expression which will be named at compile time.
1830 The same label() is returned each time ``anon_label`` is called so
1831 that expressions can reference ``anon_label`` multiple times,
1832 producing the same label name at compile time.
1834 The compiler uses this function automatically at compile time
1835 for expressions that are known to be 'unnamed' like binary
1836 expressions and function calls.
1838 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1839 public and is renamed to _anon_name_label. anon_name exists
1840 for backwards compat
1842 """
1843 name = getattr(self, "name", None)
1844 return self._anon_label(name)
1846 @util.memoized_property
1847 def _anon_key_label(self) -> _anonymous_label:
1848 """Provides a constant 'anonymous key label' for this ColumnElement.
1850 Compare to ``anon_label``, except that the "key" of the column,
1851 if available, is used to generate the label.
1853 This is used when a deduplicating key is placed into the columns
1854 collection of a selectable.
1856 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1857 public and is renamed to _anon_key_label. anon_key_label exists
1858 for backwards compat
1860 """
1861 return self._anon_label(self._proxy_key)
1863 @property
1864 @util.deprecated(
1865 "1.4",
1866 "The :attr:`_expression.ColumnElement.anon_label` attribute is now "
1867 "private, and the public accessor is deprecated.",
1868 )
1869 def anon_label(self) -> str:
1870 return self._anon_name_label
1872 @property
1873 @util.deprecated(
1874 "1.4",
1875 "The :attr:`_expression.ColumnElement.anon_key_label` attribute is "
1876 "now private, and the public accessor is deprecated.",
1877 )
1878 def anon_key_label(self) -> str:
1879 return self._anon_key_label
1881 def _dedupe_anon_label_idx(self, idx: int) -> str:
1882 """label to apply to a column that is anon labeled, but repeated
1883 in the SELECT, so that we have to make an "extra anon" label that
1884 disambiguates it from the previous appearance.
1886 these labels come out like "foo_bar_id__1" and have double underscores
1887 in them.
1889 """
1890 label = getattr(self, "name", None)
1892 # current convention is that if the element doesn't have a
1893 # ".name" (usually because it is not NamedColumn), we try to
1894 # use a "table qualified" form for the "dedupe anon" label,
1895 # based on the notion that a label like
1896 # "CAST(casttest.v1 AS DECIMAL) AS casttest_v1__1" looks better than
1897 # "CAST(casttest.v1 AS DECIMAL) AS anon__1"
1899 if label is None:
1900 return self._dedupe_anon_tq_label_idx(idx)
1901 else:
1902 return self._anon_label(label, add_hash=idx)
1904 @util.memoized_property
1905 def _anon_tq_label(self) -> _anonymous_label:
1906 return self._anon_label(getattr(self, "_tq_label", None))
1908 @util.memoized_property
1909 def _anon_tq_key_label(self) -> _anonymous_label:
1910 return self._anon_label(getattr(self, "_tq_key_label", None))
1912 def _dedupe_anon_tq_label_idx(self, idx: int) -> _anonymous_label:
1913 label = getattr(self, "_tq_label", None) or "anon"
1915 return self._anon_label(label, add_hash=idx)
1918class KeyedColumnElement(ColumnElement[_T]):
1919 """ColumnElement where ``.key`` is non-None."""
1921 _is_keyed_column_element = True
1923 key: str
1926class WrapsColumnExpression(ColumnElement[_T]):
1927 """Mixin that defines a :class:`_expression.ColumnElement`
1928 as a wrapper with special
1929 labeling behavior for an expression that already has a name.
1931 .. versionadded:: 1.4
1933 .. seealso::
1935 :ref:`change_4449`
1938 """
1940 @property
1941 def wrapped_column_expression(self) -> ColumnElement[_T]:
1942 raise NotImplementedError()
1944 @util.non_memoized_property
1945 def _tq_label(self) -> Optional[str]:
1946 wce = self.wrapped_column_expression
1947 if hasattr(wce, "_tq_label"):
1948 return wce._tq_label
1949 else:
1950 return None
1952 @property
1953 def _label(self) -> Optional[str]:
1954 return self._tq_label
1956 @property
1957 def _non_anon_label(self) -> Optional[str]:
1958 return None
1960 @util.non_memoized_property
1961 def _anon_name_label(self) -> str:
1962 wce = self.wrapped_column_expression
1964 # this logic tries to get the WrappedColumnExpression to render
1965 # with "<expr> AS <name>", where "<name>" is the natural name
1966 # within the expression itself. e.g. "CAST(table.foo) AS foo".
1967 if not wce._is_text_clause:
1968 nal = wce._non_anon_label
1969 if nal:
1970 return nal
1971 elif hasattr(wce, "_anon_name_label"):
1972 return wce._anon_name_label
1973 return super()._anon_name_label
1975 def _dedupe_anon_label_idx(self, idx: int) -> str:
1976 wce = self.wrapped_column_expression
1977 nal = wce._non_anon_label
1978 if nal:
1979 return self._anon_label(nal + "_")
1980 else:
1981 return self._dedupe_anon_tq_label_idx(idx)
1983 @property
1984 def _proxy_key(self):
1985 wce = self.wrapped_column_expression
1987 if not wce._is_text_clause:
1988 return wce._proxy_key
1989 return super()._proxy_key
1992class DMLTargetCopy(roles.InElementRole, KeyedColumnElement[_T]):
1993 """Refer to another column's VALUES or SET expression in an INSERT or
1994 UPDATE statement.
1996 See the public-facing :func:`_sql.from_dml_column` constructor for
1997 background.
1999 .. versionadded:: 2.1
2002 """
2004 def __init__(self, column: _DMLOnlyColumnArgument[_T]):
2005 self.column = coercions.expect(roles.ColumnArgumentRole, column)
2006 self.type = self.column.type
2008 __visit_name__ = "dmltargetcopy"
2010 _traverse_internals: _TraverseInternalsType = [
2011 ("column", InternalTraversal.dp_clauseelement),
2012 ]
2015class BindParameter(roles.InElementRole, KeyedColumnElement[_T]):
2016 r"""Represent a "bound expression".
2018 :class:`.BindParameter` is invoked explicitly using the
2019 :func:`.bindparam` function, as in::
2021 from sqlalchemy import bindparam
2023 stmt = select(users_table).where(
2024 users_table.c.name == bindparam("username")
2025 )
2027 Detailed discussion of how :class:`.BindParameter` is used is
2028 at :func:`.bindparam`.
2030 .. seealso::
2032 :func:`.bindparam`
2034 """
2036 __visit_name__ = "bindparam"
2038 _traverse_internals: _TraverseInternalsType = [
2039 ("key", InternalTraversal.dp_anon_name),
2040 ("type", InternalTraversal.dp_type),
2041 ("callable", InternalTraversal.dp_plain_dict),
2042 ("value", InternalTraversal.dp_plain_obj),
2043 ("literal_execute", InternalTraversal.dp_boolean),
2044 ]
2046 key: str
2047 _anon_map_key: Optional[str] = None
2048 type: TypeEngine[_T]
2049 value: Optional[_T]
2051 _is_crud = False
2052 _is_bind_parameter = True
2054 # bindparam implements its own _gen_cache_key() method however
2055 # we check subclasses for this flag, else no cache key is generated
2056 inherit_cache = True
2058 def __init__(
2059 self,
2060 key: Optional[str],
2061 value: Any = _NoArg.NO_ARG,
2062 type_: Optional[_TypeEngineArgument[_T]] = None,
2063 unique: bool = False,
2064 required: Union[bool, Literal[_NoArg.NO_ARG]] = _NoArg.NO_ARG,
2065 quote: Optional[bool] = None,
2066 callable_: Optional[Callable[[], Any]] = None,
2067 expanding: bool = False,
2068 isoutparam: bool = False,
2069 literal_execute: bool = False,
2070 _compared_to_operator: Optional[OperatorType] = None,
2071 _compared_to_type: Optional[TypeEngine[Any]] = None,
2072 _is_crud: bool = False,
2073 ):
2074 if required is _NoArg.NO_ARG:
2075 required = value is _NoArg.NO_ARG and callable_ is None
2076 if value is _NoArg.NO_ARG:
2077 value = None
2079 if quote is not None:
2080 key = quoted_name.construct(key, quote)
2082 if unique:
2083 self.key, self._anon_map_key = (
2084 _anonymous_label.safe_construct_with_key(
2085 id(self),
2086 (
2087 key
2088 if key is not None
2089 and not isinstance(key, _anonymous_label)
2090 else "param"
2091 ),
2092 sanitize_key=True,
2093 )
2094 )
2095 elif key:
2096 self.key = key
2097 else:
2098 self.key, self._anon_map_key = (
2099 _anonymous_label.safe_construct_with_key(id(self), "param")
2100 )
2102 # identifying key that won't change across
2103 # clones, used to identify the bind's logical
2104 # identity
2105 self._identifying_key = self.key
2107 # key that was passed in the first place, used to
2108 # generate new keys
2109 self._orig_key = key or "param"
2111 self.unique = unique
2112 self.value = value
2113 self.callable = callable_
2114 self.isoutparam = isoutparam
2115 self.required = required
2117 # indicate an "expanding" parameter; the compiler sets this
2118 # automatically in the compiler _render_in_expr_w_bindparam method
2119 # for an IN expression
2120 self.expanding = expanding
2122 # this is another hint to help w/ expanding and is typically
2123 # set in the compiler _render_in_expr_w_bindparam method for an
2124 # IN expression
2125 self.expand_op = None
2127 self.literal_execute = literal_execute
2128 if _is_crud:
2129 self._is_crud = True
2131 if type_ is None:
2132 if expanding:
2133 if value:
2134 check_value = value[0]
2135 else:
2136 check_value = type_api._NO_VALUE_IN_LIST
2137 else:
2138 check_value = value
2139 if _compared_to_type is not None:
2140 self.type = _compared_to_type.coerce_compared_value(
2141 _compared_to_operator, check_value
2142 )
2143 else:
2144 self.type = type_api._resolve_value_to_type(check_value)
2145 elif isinstance(type_, type):
2146 self.type = type_()
2147 elif is_tuple_type(type_):
2148 if value:
2149 if expanding:
2150 check_value = value[0]
2151 else:
2152 check_value = value
2153 cast("BindParameter[TupleAny]", self).type = (
2154 type_._resolve_values_to_types(check_value)
2155 )
2156 else:
2157 cast("BindParameter[TupleAny]", self).type = type_
2158 else:
2159 self.type = type_
2161 def _with_value(self, value, maintain_key=False, required=NO_ARG):
2162 """Return a copy of this :class:`.BindParameter` with the given value
2163 set.
2164 """
2165 cloned = self._clone(maintain_key=maintain_key)
2166 cloned.value = value
2167 cloned.callable = None
2168 cloned.required = required if required is not NO_ARG else self.required
2169 if cloned.type is type_api.NULLTYPE:
2170 cloned.type = type_api._resolve_value_to_type(value)
2171 return cloned
2173 @property
2174 def effective_value(self) -> Optional[_T]:
2175 """Return the value of this bound parameter,
2176 taking into account if the ``callable`` parameter
2177 was set.
2179 The ``callable`` value will be evaluated
2180 and returned if present, else ``value``.
2182 """
2183 if self.callable:
2184 # TODO: set up protocol for bind parameter callable
2185 return self.callable() # type: ignore
2186 else:
2187 return self.value
2189 def render_literal_execute(self) -> Self:
2190 """Produce a copy of this bound parameter that will enable the
2191 :paramref:`_sql.BindParameter.literal_execute` flag.
2193 The :paramref:`_sql.BindParameter.literal_execute` flag will
2194 have the effect of the parameter rendered in the compiled SQL
2195 string using ``[POSTCOMPILE]`` form, which is a special form that
2196 is converted to be a rendering of the literal value of the parameter
2197 at SQL execution time. The rationale is to support caching
2198 of SQL statement strings that can embed per-statement literal values,
2199 such as LIMIT and OFFSET parameters, in the final SQL string that
2200 is passed to the DBAPI. Dialects in particular may want to use
2201 this method within custom compilation schemes.
2203 .. versionadded:: 1.4.5
2205 .. seealso::
2207 :ref:`engine_thirdparty_caching`
2209 """
2210 c: Self = ClauseElement._clone(self)
2211 c.literal_execute = True
2212 return c
2214 def _negate_in_binary(self, negated_op, original_op):
2215 if self.expand_op is original_op:
2216 bind = self._clone()
2217 bind.expand_op = negated_op
2218 return bind
2219 else:
2220 return self
2222 def _with_binary_element_type(self, type_: TypeEngine[Any]) -> Self:
2223 c: Self = ClauseElement._clone(self)
2224 c.type = type_
2225 return c
2227 def _clone(self, maintain_key: bool = False, **kw: Any) -> Self:
2228 c: Self = ClauseElement._clone(self, **kw)
2229 # ensure all the BindParameter objects stay in cloned set.
2230 # in #7823, we changed "clone" so that a clone only keeps a reference
2231 # to the "original" element, since for column correspondence, that's
2232 # all we need. However, for BindParam, _cloned_set is used by
2233 # the "cache key bind match" lookup, which means if any of those
2234 # interim BindParameter objects became part of a cache key in the
2235 # cache, we need it. So here, make sure all clones keep carrying
2236 # forward.
2237 c._cloned_set.update(self._cloned_set)
2238 if not maintain_key and self.unique:
2239 c.key, c._anon_map_key = _anonymous_label.safe_construct_with_key(
2240 id(c), c._orig_key or "param", sanitize_key=True
2241 )
2242 return c
2244 def _gen_cache_key(self, anon_map, bindparams):
2245 _gen_cache_ok = self.__class__.__dict__.get("inherit_cache", False)
2247 if not _gen_cache_ok:
2248 if anon_map is not None:
2249 anon_map[NO_CACHE] = True
2250 return None
2252 id_, found = anon_map.get_anon(self)
2253 if found:
2254 return (id_, self.__class__)
2256 if bindparams is not None:
2257 bindparams.append(self)
2259 return (
2260 id_,
2261 self.__class__,
2262 self.type._static_cache_key,
2263 (
2264 anon_map[self._anon_map_key]
2265 if self._anon_map_key is not None
2266 else self.key
2267 ),
2268 self.literal_execute,
2269 )
2271 def _convert_to_unique(self):
2272 if not self.unique:
2273 self.unique = True
2274 self.key, self._anon_map_key = (
2275 _anonymous_label.safe_construct_with_key(
2276 id(self), self._orig_key or "param", sanitize_key=True
2277 )
2278 )
2280 def __getstate__(self):
2281 """execute a deferred value for serialization purposes."""
2283 d = self.__dict__.copy()
2284 v = self.value
2285 if self.callable:
2286 v = self.callable()
2287 d["callable"] = None
2288 d["value"] = v
2289 return d
2291 def __setstate__(self, state):
2292 if state.get("unique", False):
2293 anon_and_key = _anonymous_label.safe_construct_with_key(
2294 id(self), state.get("_orig_key", "param"), sanitize_key=True
2295 )
2296 state["key"], state["_anon_map_key"] = anon_and_key
2297 self.__dict__.update(state)
2299 def __repr__(self):
2300 return "%s(%r, %r, type_=%r)" % (
2301 self.__class__.__name__,
2302 self.key,
2303 self.value,
2304 self.type,
2305 )
2308class TypeClause(DQLDMLClauseElement):
2309 """Handle a type keyword in a SQL statement.
2311 Used by the ``Case`` statement.
2313 """
2315 __visit_name__ = "typeclause"
2317 _traverse_internals: _TraverseInternalsType = [
2318 ("type", InternalTraversal.dp_type)
2319 ]
2320 type: TypeEngine[Any]
2322 def __init__(self, type_: TypeEngine[Any]):
2323 self.type = type_
2326class TextClause(
2327 roles.DDLConstraintColumnRole,
2328 roles.DDLExpressionRole,
2329 roles.StatementOptionRole,
2330 roles.WhereHavingRole,
2331 roles.OrderByRole,
2332 roles.FromClauseRole,
2333 roles.SelectStatementRole,
2334 roles.InElementRole,
2335 Generative,
2336 ExecutableStatement,
2337 DQLDMLClauseElement,
2338 roles.BinaryElementRole[Any],
2339 inspection.Inspectable["TextClause"],
2340):
2341 """Represent a literal SQL text fragment.
2343 E.g.::
2345 from sqlalchemy import text
2347 t = text("SELECT * FROM users")
2348 result = connection.execute(t)
2350 The :class:`_expression.TextClause` construct is produced using the
2351 :func:`_expression.text`
2352 function; see that function for full documentation.
2354 .. seealso::
2356 :func:`_expression.text`
2358 """
2360 __visit_name__ = "textclause"
2362 _traverse_internals: _TraverseInternalsType = [
2363 ("_bindparams", InternalTraversal.dp_string_clauseelement_dict),
2364 ("text", InternalTraversal.dp_string),
2365 ] + ExecutableStatement._executable_traverse_internals
2367 _is_text_clause = True
2369 _is_textual = True
2371 _bind_params_regex = re.compile(r"(?<![:\w\x5c]):(\w+)(?!:)", re.UNICODE)
2372 _is_implicitly_boolean = False
2374 _render_label_in_columns_clause = False
2376 _omit_from_statements = False
2378 _is_collection_aggregate = False
2380 @property
2381 def _hide_froms(self) -> Iterable[FromClause]:
2382 return ()
2384 def __and__(self, other):
2385 # support use in select.where(), query.filter()
2386 return and_(self, other)
2388 @property
2389 def _select_iterable(self) -> _SelectIterable:
2390 return (self,)
2392 # help in those cases where text() is
2393 # interpreted in a column expression situation
2394 key: Optional[str] = None
2395 _label: Optional[str] = None
2397 _allow_label_resolve = False
2399 @property
2400 def _is_star(self): # type: ignore[override]
2401 return self.text == "*"
2403 def __init__(self, text: str):
2404 self._bindparams: Dict[str, BindParameter[Any]] = {}
2406 def repl(m):
2407 self._bindparams[m.group(1)] = BindParameter(m.group(1))
2408 return ":%s" % m.group(1)
2410 # scan the string and search for bind parameter names, add them
2411 # to the list of bindparams
2412 self.text = self._bind_params_regex.sub(repl, text)
2414 @_generative
2415 def bindparams(
2416 self,
2417 *binds: BindParameter[Any],
2418 **names_to_values: Any,
2419 ) -> Self:
2420 """Establish the values and/or types of bound parameters within
2421 this :class:`_expression.TextClause` construct.
2423 Given a text construct such as::
2425 from sqlalchemy import text
2427 stmt = text(
2428 "SELECT id, name FROM user WHERE name=:name AND timestamp=:timestamp"
2429 )
2431 the :meth:`_expression.TextClause.bindparams`
2432 method can be used to establish
2433 the initial value of ``:name`` and ``:timestamp``,
2434 using simple keyword arguments::
2436 stmt = stmt.bindparams(
2437 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2438 )
2440 Where above, new :class:`.BindParameter` objects
2441 will be generated with the names ``name`` and ``timestamp``, and
2442 values of ``jack`` and ``datetime.datetime(2012, 10, 8, 15, 12, 5)``,
2443 respectively. The types will be
2444 inferred from the values given, in this case :class:`.String` and
2445 :class:`.DateTime`.
2447 When specific typing behavior is needed, the positional ``*binds``
2448 argument can be used in which to specify :func:`.bindparam` constructs
2449 directly. These constructs must include at least the ``key``
2450 argument, then an optional value and type::
2452 from sqlalchemy import bindparam
2454 stmt = stmt.bindparams(
2455 bindparam("name", value="jack", type_=String),
2456 bindparam("timestamp", type_=DateTime),
2457 )
2459 Above, we specified the type of :class:`.DateTime` for the
2460 ``timestamp`` bind, and the type of :class:`.String` for the ``name``
2461 bind. In the case of ``name`` we also set the default value of
2462 ``"jack"``.
2464 Additional bound parameters can be supplied at statement execution
2465 time, e.g.::
2467 result = connection.execute(
2468 stmt, timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2469 )
2471 The :meth:`_expression.TextClause.bindparams`
2472 method can be called repeatedly,
2473 where it will re-use existing :class:`.BindParameter` objects to add
2474 new information. For example, we can call
2475 :meth:`_expression.TextClause.bindparams`
2476 first with typing information, and a
2477 second time with value information, and it will be combined::
2479 stmt = text(
2480 "SELECT id, name FROM user WHERE name=:name "
2481 "AND timestamp=:timestamp"
2482 )
2483 stmt = stmt.bindparams(
2484 bindparam("name", type_=String), bindparam("timestamp", type_=DateTime)
2485 )
2486 stmt = stmt.bindparams(
2487 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2488 )
2490 The :meth:`_expression.TextClause.bindparams`
2491 method also supports the concept of
2492 **unique** bound parameters. These are parameters that are
2493 "uniquified" on name at statement compilation time, so that multiple
2494 :func:`_expression.text`
2495 constructs may be combined together without the names
2496 conflicting. To use this feature, specify the
2497 :paramref:`.BindParameter.unique` flag on each :func:`.bindparam`
2498 object::
2500 stmt1 = text("select id from table where name=:name").bindparams(
2501 bindparam("name", value="name1", unique=True)
2502 )
2503 stmt2 = text("select id from table where name=:name").bindparams(
2504 bindparam("name", value="name2", unique=True)
2505 )
2507 union = union_all(stmt1.columns(column("id")), stmt2.columns(column("id")))
2509 The above statement will render as:
2511 .. sourcecode:: sql
2513 select id from table where name=:name_1
2514 UNION ALL select id from table where name=:name_2
2516 """ # noqa: E501
2517 self._bindparams = new_params = self._bindparams.copy()
2519 for bind in binds:
2520 try:
2521 # the regex used for text() currently will not match
2522 # a unique/anonymous key in any case, so use the _orig_key
2523 # so that a text() construct can support unique parameters
2524 existing = new_params[bind._orig_key]
2525 except KeyError as err:
2526 raise exc.ArgumentError(
2527 "This text() construct doesn't define a "
2528 "bound parameter named %r" % bind._orig_key
2529 ) from err
2530 else:
2531 new_params[existing._orig_key] = bind
2533 for key, value in names_to_values.items():
2534 try:
2535 existing = new_params[key]
2536 except KeyError as err:
2537 raise exc.ArgumentError(
2538 "This text() construct doesn't define a "
2539 "bound parameter named %r" % key
2540 ) from err
2541 else:
2542 new_params[key] = existing._with_value(value, required=False)
2543 return self
2545 @util.preload_module("sqlalchemy.sql.selectable")
2546 def columns(
2547 self,
2548 *cols: _ColumnExpressionArgument[Any],
2549 **types: _TypeEngineArgument[Any],
2550 ) -> TextualSelect:
2551 r"""Turn this :class:`_expression.TextClause` object into a
2552 :class:`_expression.TextualSelect`
2553 object that serves the same role as a SELECT
2554 statement.
2556 The :class:`_expression.TextualSelect` is part of the
2557 :class:`_expression.SelectBase`
2558 hierarchy and can be embedded into another statement by using the
2559 :meth:`_expression.TextualSelect.subquery` method to produce a
2560 :class:`.Subquery`
2561 object, which can then be SELECTed from.
2563 This function essentially bridges the gap between an entirely
2564 textual SELECT statement and the SQL expression language concept
2565 of a "selectable"::
2567 from sqlalchemy.sql import column, text
2569 stmt = text("SELECT id, name FROM some_table")
2570 stmt = stmt.columns(column("id"), column("name")).subquery("st")
2572 stmt = (
2573 select(mytable)
2574 .select_from(mytable.join(stmt, mytable.c.name == stmt.c.name))
2575 .where(stmt.c.id > 5)
2576 )
2578 Above, we pass a series of :func:`_expression.column` elements to the
2579 :meth:`_expression.TextClause.columns` method positionally. These
2580 :func:`_expression.column`
2581 elements now become first class elements upon the
2582 :attr:`_expression.TextualSelect.selected_columns` column collection,
2583 which then
2584 become part of the :attr:`.Subquery.c` collection after
2585 :meth:`_expression.TextualSelect.subquery` is invoked.
2587 The column expressions we pass to
2588 :meth:`_expression.TextClause.columns` may
2589 also be typed; when we do so, these :class:`.TypeEngine` objects become
2590 the effective return type of the column, so that SQLAlchemy's
2591 result-set-processing systems may be used on the return values.
2592 This is often needed for types such as date or boolean types, as well
2593 as for unicode processing on some dialect configurations::
2595 stmt = text("SELECT id, name, timestamp FROM some_table")
2596 stmt = stmt.columns(
2597 column("id", Integer),
2598 column("name", Unicode),
2599 column("timestamp", DateTime),
2600 )
2602 for id, name, timestamp in connection.execute(stmt):
2603 print(id, name, timestamp)
2605 As a shortcut to the above syntax, keyword arguments referring to
2606 types alone may be used, if only type conversion is needed::
2608 stmt = text("SELECT id, name, timestamp FROM some_table")
2609 stmt = stmt.columns(id=Integer, name=Unicode, timestamp=DateTime)
2611 for id, name, timestamp in connection.execute(stmt):
2612 print(id, name, timestamp)
2614 The positional form of :meth:`_expression.TextClause.columns`
2615 also provides the
2616 unique feature of **positional column targeting**, which is
2617 particularly useful when using the ORM with complex textual queries. If
2618 we specify the columns from our model to
2619 :meth:`_expression.TextClause.columns`,
2620 the result set will match to those columns positionally, meaning the
2621 name or origin of the column in the textual SQL doesn't matter::
2623 stmt = text(
2624 "SELECT users.id, addresses.id, users.id, "
2625 "users.name, addresses.email_address AS email "
2626 "FROM users JOIN addresses ON users.id=addresses.user_id "
2627 "WHERE users.id = 1"
2628 ).columns(
2629 User.id,
2630 Address.id,
2631 Address.user_id,
2632 User.name,
2633 Address.email_address,
2634 )
2636 query = (
2637 session.query(User)
2638 .from_statement(stmt)
2639 .options(contains_eager(User.addresses))
2640 )
2642 The :meth:`_expression.TextClause.columns` method provides a direct
2643 route to calling :meth:`_expression.FromClause.subquery` as well as
2644 :meth:`_expression.SelectBase.cte`
2645 against a textual SELECT statement::
2647 stmt = stmt.columns(id=Integer, name=String).cte("st")
2649 stmt = select(sometable).where(sometable.c.id == stmt.c.id)
2651 :param \*cols: A series of :class:`_expression.ColumnElement` objects,
2652 typically
2653 :class:`_schema.Column` objects from a :class:`_schema.Table`
2654 or ORM level
2655 column-mapped attributes, representing a set of columns that this
2656 textual string will SELECT from.
2658 :param \**types: A mapping of string names to :class:`.TypeEngine`
2659 type objects indicating the datatypes to use for names that are
2660 SELECTed from the textual string. Prefer to use the ``*cols``
2661 argument as it also indicates positional ordering.
2663 """
2664 selectable = util.preloaded.sql_selectable
2666 input_cols: List[NamedColumn[Any]] = [
2667 coercions.expect(roles.LabeledColumnExprRole, col) for col in cols
2668 ]
2670 positional_input_cols = [
2671 (
2672 ColumnClause(col.key, types.pop(col.key))
2673 if col.key in types
2674 else col
2675 )
2676 for col in input_cols
2677 ]
2678 keyed_input_cols: List[NamedColumn[Any]] = [
2679 ColumnClause(key, type_) for key, type_ in types.items()
2680 ]
2682 elem = selectable.TextualSelect.__new__(selectable.TextualSelect)
2683 elem._init(
2684 self,
2685 positional_input_cols + keyed_input_cols,
2686 positional=bool(positional_input_cols) and not keyed_input_cols,
2687 )
2688 return elem
2690 @property
2691 def type(self) -> TypeEngine[Any]:
2692 return type_api.NULLTYPE
2694 @property
2695 def comparator(self):
2696 # TODO: this seems wrong, it seems like we might not
2697 # be using this method.
2698 return self.type.comparator_factory(self) # type: ignore
2700 def self_group(
2701 self, against: Optional[OperatorType] = None
2702 ) -> Union[Self, Grouping[Any]]:
2703 if against is operators.in_op:
2704 return Grouping(self)
2705 else:
2706 return self
2709class Null(SingletonConstant, roles.ConstExprRole[None], ColumnElement[None]):
2710 """Represent the NULL keyword in a SQL statement.
2712 :class:`.Null` is accessed as a constant via the
2713 :func:`.null` function.
2715 """
2717 __visit_name__ = "null"
2719 _traverse_internals: _TraverseInternalsType = []
2720 _singleton: Null
2722 if not TYPE_CHECKING:
2724 @util.memoized_property
2725 def type(self) -> TypeEngine[_T]: # noqa: A001
2726 return type_api.NULLTYPE
2728 @classmethod
2729 def _instance(cls) -> Null:
2730 """Return a constant :class:`.Null` construct."""
2732 return Null._singleton
2735Null._create_singleton()
2738class False_(
2739 SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]
2740):
2741 """Represent the ``false`` keyword, or equivalent, in a SQL statement.
2743 :class:`.False_` is accessed as a constant via the
2744 :func:`.false` function.
2746 """
2748 __visit_name__ = "false"
2749 _traverse_internals: _TraverseInternalsType = []
2750 _singleton: False_
2752 if not TYPE_CHECKING:
2754 @util.memoized_property
2755 def type(self) -> TypeEngine[_T]: # noqa: A001
2756 return type_api.BOOLEANTYPE
2758 def _negate(self) -> True_:
2759 return True_._singleton
2761 @classmethod
2762 def _instance(cls) -> False_:
2763 return False_._singleton
2766False_._create_singleton()
2769class True_(SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]):
2770 """Represent the ``true`` keyword, or equivalent, in a SQL statement.
2772 :class:`.True_` is accessed as a constant via the
2773 :func:`.true` function.
2775 """
2777 __visit_name__ = "true"
2779 _traverse_internals: _TraverseInternalsType = []
2780 _singleton: True_
2782 if not TYPE_CHECKING:
2784 @util.memoized_property
2785 def type(self) -> TypeEngine[_T]: # noqa: A001
2786 return type_api.BOOLEANTYPE
2788 def _negate(self) -> False_:
2789 return False_._singleton
2791 @classmethod
2792 def _ifnone(
2793 cls, other: Optional[ColumnElement[Any]]
2794 ) -> ColumnElement[Any]:
2795 if other is None:
2796 return cls._instance()
2797 else:
2798 return other
2800 @classmethod
2801 def _instance(cls) -> True_:
2802 return True_._singleton
2805True_._create_singleton()
2808class ElementList(DQLDMLClauseElement):
2809 """Describe a list of clauses that will be space separated.
2811 This is a minimal version of :class:`.ClauseList` which is used by
2812 the :class:`.HasSyntaxExtension` class. It does not do any coercions
2813 so should be used internally only.
2815 .. versionadded:: 2.1
2817 """
2819 __visit_name__ = "element_list"
2821 _traverse_internals: _TraverseInternalsType = [
2822 ("clauses", InternalTraversal.dp_clauseelement_tuple),
2823 ]
2825 clauses: typing_Tuple[ClauseElement, ...]
2827 def __init__(self, clauses: Sequence[ClauseElement]):
2828 self.clauses = tuple(clauses)
2831class OrderByList(
2832 roles.OrderByRole,
2833 operators.OrderingOperators,
2834 DQLDMLClauseElement,
2835):
2836 """Describe a list of clauses that will be comma separated to nest
2837 within an ORDER BY.
2839 .. versionadded:: 2.1
2841 """
2843 __visit_name__ = "order_by_list"
2845 _traverse_internals: _TraverseInternalsType = [
2846 ("clauses", InternalTraversal.dp_clauseelement_tuple),
2847 ]
2849 clauses: List[ColumnElement[Any]]
2851 def __init__(
2852 self,
2853 clauses: Iterable[Union[OrderByList, _ColumnExpressionArgument[Any]]],
2854 ):
2855 text_converter_role: Type[roles.SQLRole] = roles.ByOfRole
2856 self._text_converter_role = text_converter_role
2858 self.clauses = [
2859 coercions.expect(
2860 text_converter_role, clause, apply_propagate_attrs=self
2861 )
2862 for clause in clauses
2863 ]
2865 def __iter__(self) -> Iterator[ColumnElement[Any]]:
2866 return iter(self.clauses)
2868 def __len__(self) -> int:
2869 return len(self.clauses)
2871 @property
2872 def _select_iterable(self) -> _SelectIterable:
2873 return itertools.chain.from_iterable(
2874 [elem._select_iterable for elem in self.clauses]
2875 )
2877 @util.ro_non_memoized_property
2878 def _from_objects(self) -> List[FromClause]:
2879 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
2881 def self_group(
2882 self, against: Optional[OperatorType] = None
2883 ) -> Union[Self, Grouping[Any]]:
2884 return self
2886 def desc(self) -> OrderByList:
2887 return OrderByList([e.desc() for e in self.clauses])
2889 def asc(self) -> OrderByList:
2890 return OrderByList([e.asc() for e in self.clauses])
2892 def nulls_first(self) -> OrderByList:
2893 return OrderByList([e.nulls_first() for e in self.clauses])
2895 def nulls_last(self) -> OrderByList:
2896 return OrderByList([e.nulls_last() for e in self.clauses])
2899class ClauseList(
2900 roles.InElementRole,
2901 roles.OrderByRole,
2902 roles.ColumnsClauseRole,
2903 roles.DMLColumnRole,
2904 DQLDMLClauseElement,
2905):
2906 """Describe a list of clauses, separated by an operator.
2908 By default, is comma-separated, such as a column listing.
2910 """
2912 __visit_name__ = "clauselist"
2914 # Used by ORM context.py to identify ClauseList objects in legacy
2915 # composite attribute queries (see test_query_cols_legacy test)
2916 _is_clause_list = True
2918 _traverse_internals: _TraverseInternalsType = [
2919 ("clauses", InternalTraversal.dp_clauseelement_list),
2920 ("operator", InternalTraversal.dp_operator),
2921 ]
2923 clauses: List[ColumnElement[Any]]
2925 def __init__(
2926 self,
2927 *clauses: _ColumnExpressionArgument[Any],
2928 operator: OperatorType = operators.comma_op,
2929 group: bool = True,
2930 group_contents: bool = True,
2931 _literal_as_text_role: Type[roles.SQLRole] = roles.WhereHavingRole,
2932 ):
2933 self.operator = operator
2934 self.group = group
2935 self.group_contents = group_contents
2936 clauses_iterator: Iterable[_ColumnExpressionArgument[Any]] = clauses
2937 text_converter_role: Type[roles.SQLRole] = _literal_as_text_role
2938 self._text_converter_role = text_converter_role
2940 if self.group_contents:
2941 self.clauses = [
2942 coercions.expect(
2943 text_converter_role, clause, apply_propagate_attrs=self
2944 ).self_group(against=self.operator)
2945 for clause in clauses_iterator
2946 ]
2947 else:
2948 self.clauses = [
2949 coercions.expect(
2950 text_converter_role, clause, apply_propagate_attrs=self
2951 )
2952 for clause in clauses_iterator
2953 ]
2954 self._is_implicitly_boolean = operators.is_boolean(self.operator)
2956 @classmethod
2957 def _construct_raw(
2958 cls,
2959 operator: OperatorType,
2960 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
2961 ) -> ClauseList:
2962 self = cls.__new__(cls)
2963 self.clauses = list(clauses) if clauses else []
2964 self.group = True
2965 self.operator = operator
2966 self.group_contents = True
2967 self._is_implicitly_boolean = False
2968 return self
2970 def __iter__(self) -> Iterator[ColumnElement[Any]]:
2971 return iter(self.clauses)
2973 def __len__(self) -> int:
2974 return len(self.clauses)
2976 @property
2977 def _select_iterable(self) -> _SelectIterable:
2978 return itertools.chain.from_iterable(
2979 [elem._select_iterable for elem in self.clauses]
2980 )
2982 def append(self, clause):
2983 if self.group_contents:
2984 self.clauses.append(
2985 coercions.expect(self._text_converter_role, clause).self_group(
2986 against=self.operator
2987 )
2988 )
2989 else:
2990 self.clauses.append(
2991 coercions.expect(self._text_converter_role, clause)
2992 )
2994 @util.ro_non_memoized_property
2995 def _from_objects(self) -> List[FromClause]:
2996 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
2998 def self_group(
2999 self, against: Optional[OperatorType] = None
3000 ) -> Union[Self, Grouping[Any]]:
3001 if self.group and operators.is_precedent(self.operator, against):
3002 return Grouping(self)
3003 else:
3004 return self
3007class OperatorExpression(ColumnElement[_T]):
3008 """base for expressions that contain an operator and operands
3010 .. versionadded:: 2.0
3012 """
3014 operator: OperatorType
3015 type: TypeEngine[_T]
3017 group: bool = True
3019 @property
3020 def is_comparison(self):
3021 return operators.is_comparison(self.operator)
3023 def self_group(
3024 self, against: Optional[OperatorType] = None
3025 ) -> Union[Self, Grouping[_T]]:
3026 if (
3027 self.group
3028 and operators.is_precedent(self.operator, against)
3029 or (
3030 # a negate against a non-boolean operator
3031 # doesn't make too much sense but we should
3032 # group for that
3033 against is operators.inv
3034 and not operators.is_boolean(self.operator)
3035 )
3036 ):
3037 return Grouping(self)
3038 else:
3039 return self
3041 @property
3042 def _flattened_operator_clauses(
3043 self,
3044 ) -> typing_Tuple[ColumnElement[Any], ...]:
3045 raise NotImplementedError()
3047 @classmethod
3048 def _construct_for_op(
3049 cls,
3050 left: ColumnElement[Any],
3051 right: ColumnElement[Any],
3052 op: OperatorType,
3053 *,
3054 type_: TypeEngine[_T],
3055 negate: Optional[OperatorType] = None,
3056 modifiers: Optional[Mapping[str, Any]] = None,
3057 ) -> OperatorExpression[_T]:
3058 if operators.is_associative(op):
3059 assert (
3060 negate is None
3061 ), f"negate not supported for associative operator {op}"
3063 multi = False
3064 if getattr(
3065 left, "operator", None
3066 ) is op and type_._compare_type_affinity(left.type):
3067 multi = True
3068 left_flattened = left._flattened_operator_clauses
3069 else:
3070 left_flattened = (left,)
3072 if getattr(
3073 right, "operator", None
3074 ) is op and type_._compare_type_affinity(right.type):
3075 multi = True
3076 right_flattened = right._flattened_operator_clauses
3077 else:
3078 right_flattened = (right,)
3080 if multi:
3081 return ExpressionClauseList._construct_for_list(
3082 op,
3083 type_,
3084 *(left_flattened + right_flattened),
3085 )
3087 if right._is_collection_aggregate:
3088 negate = None
3090 return BinaryExpression(
3091 left, right, op, type_=type_, negate=negate, modifiers=modifiers
3092 )
3095class ExpressionClauseList(OperatorExpression[_T]):
3096 """Describe a list of clauses, separated by an operator,
3097 in a column expression context.
3099 :class:`.ExpressionClauseList` differs from :class:`.ClauseList` in that
3100 it represents a column-oriented DQL expression only, not an open ended
3101 list of anything comma separated.
3103 .. versionadded:: 2.0
3105 """
3107 __visit_name__ = "expression_clauselist"
3109 _traverse_internals: _TraverseInternalsType = [
3110 ("clauses", InternalTraversal.dp_clauseelement_tuple),
3111 ("operator", InternalTraversal.dp_operator),
3112 ]
3114 clauses: typing_Tuple[ColumnElement[Any], ...]
3116 group: bool
3118 def __init__(
3119 self,
3120 operator: OperatorType,
3121 *clauses: _ColumnExpressionArgument[Any],
3122 type_: Optional[_TypeEngineArgument[_T]] = None,
3123 ):
3124 self.operator = operator
3126 self.clauses = tuple(
3127 coercions.expect(
3128 roles.ExpressionElementRole, clause, apply_propagate_attrs=self
3129 )
3130 for clause in clauses
3131 )
3132 self._is_implicitly_boolean = operators.is_boolean(self.operator)
3133 self.type = type_api.to_instance(type_) # type: ignore
3135 @property
3136 def _flattened_operator_clauses(
3137 self,
3138 ) -> typing_Tuple[ColumnElement[Any], ...]:
3139 return self.clauses
3141 def __iter__(self) -> Iterator[ColumnElement[Any]]:
3142 return iter(self.clauses)
3144 def __len__(self) -> int:
3145 return len(self.clauses)
3147 @property
3148 def _select_iterable(self) -> _SelectIterable:
3149 return (self,)
3151 @util.ro_non_memoized_property
3152 def _from_objects(self) -> List[FromClause]:
3153 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
3155 def _append_inplace(self, clause: ColumnElement[Any]) -> None:
3156 self.clauses += (clause,)
3158 @classmethod
3159 def _construct_for_list(
3160 cls,
3161 operator: OperatorType,
3162 type_: TypeEngine[_T],
3163 *clauses: ColumnElement[Any],
3164 group: bool = True,
3165 ) -> ExpressionClauseList[_T]:
3166 self = cls.__new__(cls)
3167 self.group = group
3168 if group:
3169 self.clauses = tuple(
3170 c.self_group(against=operator) for c in clauses
3171 )
3172 else:
3173 self.clauses = clauses
3174 self.operator = operator
3175 self.type = type_
3176 for c in clauses:
3177 if c._propagate_attrs:
3178 self._propagate_attrs = c._propagate_attrs
3179 break
3180 return self
3182 def _negate(self) -> Any:
3183 grouped = self.self_group(against=operators.inv)
3184 assert isinstance(grouped, ColumnElement)
3185 return UnaryExpression(grouped, operator=operators.inv)
3188class BooleanClauseList(ExpressionClauseList[bool]):
3189 __visit_name__ = "expression_clauselist"
3190 inherit_cache = True
3192 def __init__(self, *arg, **kw):
3193 raise NotImplementedError(
3194 "BooleanClauseList has a private constructor"
3195 )
3197 @classmethod
3198 def _process_clauses_for_boolean(
3199 cls,
3200 operator: OperatorType,
3201 continue_on: Any,
3202 skip_on: Any,
3203 clauses: Iterable[ColumnElement[Any]],
3204 ) -> typing_Tuple[int, List[ColumnElement[Any]]]:
3205 has_continue_on = None
3207 convert_clauses = []
3209 against = operators._asbool
3210 lcc = 0
3212 for clause in clauses:
3213 if clause is continue_on:
3214 # instance of continue_on, like and_(x, y, True, z), store it
3215 # if we didn't find one already, we will use it if there
3216 # are no other expressions here.
3217 has_continue_on = clause
3218 elif clause is skip_on:
3219 # instance of skip_on, e.g. and_(x, y, False, z), cancels
3220 # the rest out
3221 convert_clauses = [clause]
3222 lcc = 1
3223 break
3224 else:
3225 if not lcc:
3226 lcc = 1
3227 else:
3228 against = operator
3229 # technically this would be len(convert_clauses) + 1
3230 # however this only needs to indicate "greater than one"
3231 lcc = 2
3232 convert_clauses.append(clause)
3234 if not convert_clauses and has_continue_on is not None:
3235 convert_clauses = [has_continue_on]
3236 lcc = 1
3238 return lcc, [c.self_group(against=against) for c in convert_clauses]
3240 @classmethod
3241 def _construct(
3242 cls,
3243 operator: OperatorType,
3244 continue_on: Any,
3245 skip_on: Any,
3246 initial_clause: Any = _NoArg.NO_ARG,
3247 *clauses: Any,
3248 **kw: Any,
3249 ) -> ColumnElement[Any]:
3250 if initial_clause is _NoArg.NO_ARG:
3251 # no elements period. deprecated use case. return an empty
3252 # ClauseList construct that generates nothing unless it has
3253 # elements added to it.
3254 name = operator.__name__
3256 util.warn_deprecated(
3257 f"Invoking {name}() without arguments is deprecated, and "
3258 f"will be disallowed in a future release. For an empty "
3259 f"""{name}() construct, use '{name}({
3260 'true()' if continue_on is True_._singleton else 'false()'
3261 }, *args)' """
3262 f"""or '{name}({
3263 'True' if continue_on is True_._singleton else 'False'
3264 }, *args)'.""",
3265 version="1.4",
3266 )
3267 return cls._construct_raw(operator)
3269 lcc, convert_clauses = cls._process_clauses_for_boolean(
3270 operator,
3271 continue_on,
3272 skip_on,
3273 [
3274 coercions.expect(roles.WhereHavingRole, clause)
3275 for clause in util.coerce_generator_arg(
3276 (initial_clause,) + clauses
3277 )
3278 ],
3279 )
3281 if lcc > 1:
3282 # multiple elements. Return regular BooleanClauseList
3283 # which will link elements against the operator.
3285 flattened_clauses = itertools.chain.from_iterable(
3286 (
3287 (c for c in to_flat._flattened_operator_clauses)
3288 if getattr(to_flat, "operator", None) is operator
3289 else (to_flat,)
3290 )
3291 for to_flat in convert_clauses
3292 )
3294 return cls._construct_raw(operator, flattened_clauses) # type: ignore # noqa: E501
3295 else:
3296 assert lcc
3297 # just one element. return it as a single boolean element,
3298 # not a list and discard the operator.
3299 return convert_clauses[0]
3301 @classmethod
3302 def _construct_for_whereclause(
3303 cls, clauses: Iterable[ColumnElement[Any]]
3304 ) -> Optional[ColumnElement[bool]]:
3305 operator, continue_on, skip_on = (
3306 operators.and_,
3307 True_._singleton,
3308 False_._singleton,
3309 )
3311 lcc, convert_clauses = cls._process_clauses_for_boolean(
3312 operator,
3313 continue_on,
3314 skip_on,
3315 clauses, # these are assumed to be coerced already
3316 )
3318 if lcc > 1:
3319 # multiple elements. Return regular BooleanClauseList
3320 # which will link elements against the operator.
3321 return cls._construct_raw(operator, convert_clauses)
3322 elif lcc == 1:
3323 # just one element. return it as a single boolean element,
3324 # not a list and discard the operator.
3325 return convert_clauses[0]
3326 else:
3327 return None
3329 @classmethod
3330 def _construct_raw(
3331 cls,
3332 operator: OperatorType,
3333 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
3334 ) -> BooleanClauseList:
3335 self = cls.__new__(cls)
3336 self.clauses = tuple(clauses) if clauses else ()
3337 self.group = True
3338 self.operator = operator
3339 self.type = type_api.BOOLEANTYPE
3340 self._is_implicitly_boolean = True
3341 return self
3343 @classmethod
3344 def and_(
3345 cls,
3346 initial_clause: Union[
3347 Literal[True], _ColumnExpressionArgument[bool], _NoArg
3348 ] = _NoArg.NO_ARG,
3349 *clauses: _ColumnExpressionArgument[bool],
3350 ) -> ColumnElement[bool]:
3351 r"""Produce a conjunction of expressions joined by ``AND``.
3353 See :func:`_sql.and_` for full documentation.
3354 """
3355 return cls._construct(
3356 operators.and_,
3357 True_._singleton,
3358 False_._singleton,
3359 initial_clause,
3360 *clauses,
3361 )
3363 @classmethod
3364 def or_(
3365 cls,
3366 initial_clause: Union[
3367 Literal[False], _ColumnExpressionArgument[bool], _NoArg
3368 ] = _NoArg.NO_ARG,
3369 *clauses: _ColumnExpressionArgument[bool],
3370 ) -> ColumnElement[bool]:
3371 """Produce a conjunction of expressions joined by ``OR``.
3373 See :func:`_sql.or_` for full documentation.
3374 """
3375 return cls._construct(
3376 operators.or_,
3377 False_._singleton,
3378 True_._singleton,
3379 initial_clause,
3380 *clauses,
3381 )
3383 @property
3384 def _select_iterable(self) -> _SelectIterable:
3385 return (self,)
3387 def self_group(
3388 self, against: Optional[OperatorType] = None
3389 ) -> Union[Self, Grouping[bool]]:
3390 if not self.clauses:
3391 return self
3392 else:
3393 return super().self_group(against=against)
3396and_ = BooleanClauseList.and_
3397or_ = BooleanClauseList.or_
3400class Tuple(ClauseList, ColumnElement[TupleAny]):
3401 """Represent a SQL tuple."""
3403 __visit_name__ = "tuple"
3405 _traverse_internals: _TraverseInternalsType = (
3406 ClauseList._traverse_internals + []
3407 )
3409 type: TupleType
3411 @util.preload_module("sqlalchemy.sql.sqltypes")
3412 def __init__(
3413 self,
3414 *clauses: _ColumnExpressionArgument[Any],
3415 types: Optional[Sequence[_TypeEngineArgument[Any]]] = None,
3416 ):
3417 sqltypes = util.preloaded.sql_sqltypes
3419 if types is None:
3420 init_clauses: List[ColumnElement[Any]] = [
3421 coercions.expect(roles.ExpressionElementRole, c)
3422 for c in clauses
3423 ]
3424 else:
3425 if len(types) != len(clauses):
3426 raise exc.ArgumentError(
3427 "Wrong number of elements for %d-tuple: %r "
3428 % (len(types), clauses)
3429 )
3430 init_clauses = [
3431 coercions.expect(
3432 roles.ExpressionElementRole,
3433 c,
3434 type_=typ if not typ._isnull else None,
3435 )
3436 for typ, c in zip(types, clauses)
3437 ]
3439 self.type = sqltypes.TupleType(*[arg.type for arg in init_clauses])
3440 super().__init__(*init_clauses)
3442 @property
3443 def _select_iterable(self) -> _SelectIterable:
3444 return (self,)
3446 def _bind_param(self, operator, obj, type_=None, expanding=False):
3447 if expanding:
3448 return BindParameter(
3449 None,
3450 value=obj,
3451 _compared_to_operator=operator,
3452 unique=True,
3453 expanding=True,
3454 type_=type_,
3455 _compared_to_type=self.type,
3456 )
3457 else:
3458 return Tuple(
3459 *[
3460 BindParameter(
3461 None,
3462 o,
3463 _compared_to_operator=operator,
3464 _compared_to_type=compared_to_type,
3465 unique=True,
3466 type_=type_,
3467 )
3468 for o, compared_to_type in zip(obj, self.type.types)
3469 ]
3470 )
3472 def self_group(self, against: Optional[OperatorType] = None) -> Self:
3473 # Tuple is parenthesized by definition.
3474 return self
3477class Case(ColumnElement[_T]):
3478 """Represent a ``CASE`` expression.
3480 :class:`.Case` is produced using the :func:`.case` factory function,
3481 as in::
3483 from sqlalchemy import case
3485 stmt = select(users_table).where(
3486 case(
3487 (users_table.c.name == "wendy", "W"),
3488 (users_table.c.name == "jack", "J"),
3489 else_="E",
3490 )
3491 )
3493 Details on :class:`.Case` usage is at :func:`.case`.
3495 .. seealso::
3497 :func:`.case`
3499 """
3501 __visit_name__ = "case"
3503 _traverse_internals: _TraverseInternalsType = [
3504 ("value", InternalTraversal.dp_clauseelement),
3505 ("whens", InternalTraversal.dp_clauseelement_tuples),
3506 ("else_", InternalTraversal.dp_clauseelement),
3507 ]
3509 # for case(), the type is derived from the whens. so for the moment
3510 # users would have to cast() the case to get a specific type
3512 whens: List[typing_Tuple[ColumnElement[bool], ColumnElement[_T]]]
3513 else_: Optional[ColumnElement[_T]]
3514 value: Optional[ColumnElement[Any]]
3516 def __init__(
3517 self,
3518 *whens: Union[
3519 typing_Tuple[_ColumnExpressionArgument[bool], Any],
3520 Mapping[Any, Any],
3521 ],
3522 value: Optional[Any] = None,
3523 else_: Optional[Any] = None,
3524 ):
3525 new_whens: Iterable[Any] = coercions._expression_collection_was_a_list(
3526 "whens", "case", whens
3527 )
3528 try:
3529 new_whens = util.dictlike_iteritems(new_whens)
3530 except TypeError:
3531 pass
3533 self.whens = [
3534 (
3535 coercions.expect(
3536 roles.ExpressionElementRole,
3537 c,
3538 apply_propagate_attrs=self,
3539 ).self_group(),
3540 coercions.expect(roles.ExpressionElementRole, r),
3541 )
3542 for (c, r) in new_whens
3543 ]
3545 if value is None:
3546 self.value = None
3547 else:
3548 self.value = coercions.expect(roles.ExpressionElementRole, value)
3550 if else_ is not None:
3551 self.else_ = coercions.expect(roles.ExpressionElementRole, else_)
3552 else:
3553 self.else_ = None
3555 type_ = next(
3556 (
3557 then.type
3558 # Iterate `whens` in reverse to match previous behaviour
3559 # where type of final element took priority
3560 for *_, then in reversed(self.whens)
3561 if not then.type._isnull
3562 ),
3563 self.else_.type if self.else_ is not None else type_api.NULLTYPE,
3564 )
3565 self.type = cast(_T, type_)
3567 @util.ro_non_memoized_property
3568 def _from_objects(self) -> List[FromClause]:
3569 return list(
3570 itertools.chain(*[x._from_objects for x in self.get_children()])
3571 )
3574class Cast(WrapsColumnExpression[_T]):
3575 """Represent a ``CAST`` expression.
3577 :class:`.Cast` is produced using the :func:`.cast` factory function,
3578 as in::
3580 from sqlalchemy import cast, Numeric
3582 stmt = select(cast(product_table.c.unit_price, Numeric(10, 4)))
3584 Details on :class:`.Cast` usage is at :func:`.cast`.
3586 .. seealso::
3588 :ref:`tutorial_casts`
3590 :func:`.cast`
3592 :func:`.try_cast`
3594 :func:`.type_coerce` - an alternative to CAST that coerces the type
3595 on the Python side only, which is often sufficient to generate the
3596 correct SQL and data coercion.
3598 """
3600 __visit_name__ = "cast"
3602 _traverse_internals: _TraverseInternalsType = [
3603 ("clause", InternalTraversal.dp_clauseelement),
3604 ("type", InternalTraversal.dp_type),
3605 ]
3607 clause: ColumnElement[Any]
3608 type: TypeEngine[_T]
3609 typeclause: TypeClause
3611 def __init__(
3612 self,
3613 expression: _ColumnExpressionArgument[Any],
3614 type_: _TypeEngineArgument[_T],
3615 ):
3616 self.type = type_api.to_instance(type_)
3617 self.clause = coercions.expect(
3618 roles.ExpressionElementRole,
3619 expression,
3620 type_=self.type,
3621 apply_propagate_attrs=self,
3622 )
3623 self.typeclause = TypeClause(self.type)
3625 @util.ro_non_memoized_property
3626 def _from_objects(self) -> List[FromClause]:
3627 return self.clause._from_objects
3629 @property
3630 def wrapped_column_expression(self):
3631 return self.clause
3634class TryCast(Cast[_T]):
3635 """Represent a TRY_CAST expression.
3637 Details on :class:`.TryCast` usage is at :func:`.try_cast`.
3639 .. seealso::
3641 :func:`.try_cast`
3643 :ref:`tutorial_casts`
3644 """
3646 __visit_name__ = "try_cast"
3647 inherit_cache = True
3650class TypeCoerce(WrapsColumnExpression[_T]):
3651 """Represent a Python-side type-coercion wrapper.
3653 :class:`.TypeCoerce` supplies the :func:`_expression.type_coerce`
3654 function; see that function for usage details.
3656 .. seealso::
3658 :func:`_expression.type_coerce`
3660 :func:`.cast`
3662 """
3664 __visit_name__ = "type_coerce"
3666 _traverse_internals: _TraverseInternalsType = [
3667 ("clause", InternalTraversal.dp_clauseelement),
3668 ("type", InternalTraversal.dp_type),
3669 ]
3671 clause: ColumnElement[Any]
3672 type: TypeEngine[_T]
3674 def __init__(
3675 self,
3676 expression: _ColumnExpressionArgument[Any],
3677 type_: _TypeEngineArgument[_T],
3678 ):
3679 self.type = type_api.to_instance(type_)
3680 self.clause = coercions.expect(
3681 roles.ExpressionElementRole,
3682 expression,
3683 type_=self.type,
3684 apply_propagate_attrs=self,
3685 )
3687 @util.ro_non_memoized_property
3688 def _from_objects(self) -> List[FromClause]:
3689 return self.clause._from_objects
3691 @HasMemoized.memoized_attribute
3692 def typed_expression(self):
3693 if isinstance(self.clause, BindParameter):
3694 bp = self.clause._clone()
3695 bp.type = self.type
3696 return bp
3697 else:
3698 return self.clause
3700 @property
3701 def wrapped_column_expression(self):
3702 return self.clause
3704 def self_group(
3705 self, against: Optional[OperatorType] = None
3706 ) -> TypeCoerce[_T]:
3707 grouped = self.clause.self_group(against=against)
3708 if grouped is not self.clause:
3709 return TypeCoerce(grouped, self.type)
3710 else:
3711 return self
3714class Extract(ColumnElement[int]):
3715 """Represent a SQL EXTRACT clause, ``extract(field FROM expr)``."""
3717 __visit_name__ = "extract"
3719 _traverse_internals: _TraverseInternalsType = [
3720 ("expr", InternalTraversal.dp_clauseelement),
3721 ("field", InternalTraversal.dp_string),
3722 ]
3724 expr: ColumnElement[Any]
3725 field: str
3727 def __init__(self, field: str, expr: _ColumnExpressionArgument[Any]):
3728 self.type = type_api.INTEGERTYPE
3729 self.field = field
3730 self.expr = coercions.expect(roles.ExpressionElementRole, expr)
3732 @util.ro_non_memoized_property
3733 def _from_objects(self) -> List[FromClause]:
3734 return self.expr._from_objects
3737class _label_reference(ColumnElement[_T]):
3738 """Wrap a column expression as it appears in a 'reference' context.
3740 This expression is any that includes an _order_by_label_element,
3741 which is a Label, or a DESC / ASC construct wrapping a Label.
3743 The production of _label_reference() should occur when an expression
3744 is added to this context; this includes the ORDER BY or GROUP BY of a
3745 SELECT statement, as well as a few other places, such as the ORDER BY
3746 within an OVER clause.
3748 """
3750 __visit_name__ = "label_reference"
3752 _traverse_internals: _TraverseInternalsType = [
3753 ("element", InternalTraversal.dp_clauseelement)
3754 ]
3756 element: ColumnElement[_T]
3758 def __init__(self, element: ColumnElement[_T]):
3759 self.element = element
3760 self._propagate_attrs = element._propagate_attrs
3762 @util.ro_non_memoized_property
3763 def _from_objects(self) -> List[FromClause]:
3764 return []
3767class _textual_label_reference(ColumnElement[Any]):
3768 __visit_name__ = "textual_label_reference"
3770 _traverse_internals: _TraverseInternalsType = [
3771 ("element", InternalTraversal.dp_string)
3772 ]
3774 def __init__(self, element: str):
3775 self.element = element
3777 @util.memoized_property
3778 def _text_clause(self) -> TextClause:
3779 return TextClause(self.element)
3782class UnaryExpression(ColumnElement[_T]):
3783 """Define a 'unary' expression.
3785 A unary expression has a single column expression
3786 and an operator. The operator can be placed on the left
3787 (where it is called the 'operator') or right (where it is called the
3788 'modifier') of the column expression.
3790 :class:`.UnaryExpression` is the basis for several unary operators
3791 including those used by :func:`.desc`, :func:`.asc`, :func:`.distinct`,
3792 :func:`.nulls_first` and :func:`.nulls_last`.
3794 """
3796 __visit_name__ = "unary"
3798 _traverse_internals: _TraverseInternalsType = [
3799 ("element", InternalTraversal.dp_clauseelement),
3800 ("operator", InternalTraversal.dp_operator),
3801 ("modifier", InternalTraversal.dp_operator),
3802 ]
3804 element: ColumnElement[Any]
3805 operator: Optional[OperatorType]
3806 modifier: Optional[OperatorType]
3808 def __init__(
3809 self,
3810 element: ColumnElement[Any],
3811 *,
3812 operator: Optional[OperatorType] = None,
3813 modifier: Optional[OperatorType] = None,
3814 type_: Optional[_TypeEngineArgument[_T]] = None,
3815 wraps_column_expression: bool = False, # legacy, not used as of 2.0.42
3816 ):
3817 self.operator = operator
3818 self.modifier = modifier
3819 self._propagate_attrs = element._propagate_attrs
3820 self.element = element.self_group(
3821 against=self.operator or self.modifier
3822 )
3824 # if type is None, we get NULLTYPE, which is our _T. But I don't
3825 # know how to get the overloads to express that correctly
3826 self.type = type_api.to_instance(type_) # type: ignore
3828 def _wraps_unnamed_column(self):
3829 ungrouped = self.element._ungroup()
3830 return (
3831 not isinstance(ungrouped, NamedColumn)
3832 or ungrouped._non_anon_label is None
3833 )
3835 @classmethod
3836 def _create_nulls_first(
3837 cls,
3838 column: _ColumnExpressionArgument[_T],
3839 ) -> UnaryExpression[_T]:
3840 return UnaryExpression(
3841 coercions.expect(roles.ByOfRole, column),
3842 modifier=operators.nulls_first_op,
3843 )
3845 @classmethod
3846 def _create_nulls_last(
3847 cls,
3848 column: _ColumnExpressionArgument[_T],
3849 ) -> UnaryExpression[_T]:
3850 return UnaryExpression(
3851 coercions.expect(roles.ByOfRole, column),
3852 modifier=operators.nulls_last_op,
3853 )
3855 @classmethod
3856 def _create_desc(
3857 cls, column: _ColumnExpressionOrStrLabelArgument[_T]
3858 ) -> UnaryExpression[_T]:
3860 return UnaryExpression(
3861 coercions.expect(roles.ByOfRole, column),
3862 modifier=operators.desc_op,
3863 )
3865 @classmethod
3866 def _create_asc(
3867 cls,
3868 column: _ColumnExpressionOrStrLabelArgument[_T],
3869 ) -> UnaryExpression[_T]:
3870 return UnaryExpression(
3871 coercions.expect(roles.ByOfRole, column),
3872 modifier=operators.asc_op,
3873 )
3875 @classmethod
3876 def _create_distinct(
3877 cls,
3878 expr: _ColumnExpressionArgument[_T],
3879 ) -> UnaryExpression[_T]:
3880 col_expr: ColumnElement[_T] = coercions.expect(
3881 roles.ExpressionElementRole, expr
3882 )
3883 return UnaryExpression(
3884 col_expr,
3885 operator=operators.distinct_op,
3886 type_=col_expr.type,
3887 )
3889 @classmethod
3890 def _create_bitwise_not(
3891 cls,
3892 expr: _ColumnExpressionArgument[_T],
3893 ) -> UnaryExpression[_T]:
3894 col_expr: ColumnElement[_T] = coercions.expect(
3895 roles.ExpressionElementRole, expr
3896 )
3897 return UnaryExpression(
3898 col_expr,
3899 operator=operators.bitwise_not_op,
3900 type_=col_expr.type,
3901 )
3903 @property
3904 def _order_by_label_element(self) -> Optional[Label[Any]]:
3905 if operators.is_order_by_modifier(self.modifier):
3906 return self.element._order_by_label_element
3907 else:
3908 return None
3910 @util.ro_non_memoized_property
3911 def _from_objects(self) -> List[FromClause]:
3912 return self.element._from_objects
3914 def _negate(self) -> ColumnElement[Any]:
3915 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
3916 return UnaryExpression(
3917 self.self_group(against=operators.inv),
3918 operator=operators.inv,
3919 type_=type_api.BOOLEANTYPE,
3920 )
3921 else:
3922 return ColumnElement._negate(self)
3924 def self_group(
3925 self, against: Optional[OperatorType] = None
3926 ) -> Union[Self, Grouping[_T]]:
3927 if self.operator and operators.is_precedent(self.operator, against):
3928 return Grouping(self)
3929 else:
3930 return self
3933class CollectionAggregate(UnaryExpression[_T]):
3934 """Forms the basis for right-hand collection operator modifiers
3935 ANY and ALL.
3937 The ANY and ALL keywords are available in different ways on different
3938 backends. On PostgreSQL, they only work for an ARRAY type. On
3939 MySQL, they only work for subqueries.
3941 """
3943 inherit_cache = True
3944 _is_collection_aggregate = True
3946 @classmethod
3947 def _create_any(
3948 cls, expr: _ColumnExpressionArgument[_T]
3949 ) -> CollectionAggregate[bool]:
3950 """create CollectionAggregate for the legacy
3951 ARRAY.Comparator.any() method"""
3952 col_expr: ColumnElement[_T] = coercions.expect(
3953 roles.ExpressionElementRole,
3954 expr,
3955 )
3956 col_expr = col_expr.self_group()
3957 return CollectionAggregate(
3958 col_expr,
3959 operator=operators.any_op,
3960 type_=type_api.BOOLEANTYPE,
3961 )
3963 @classmethod
3964 def _create_all(
3965 cls, expr: _ColumnExpressionArgument[_T]
3966 ) -> CollectionAggregate[bool]:
3967 """create CollectionAggregate for the legacy
3968 ARRAY.Comparator.all() method"""
3969 col_expr: ColumnElement[_T] = coercions.expect(
3970 roles.ExpressionElementRole,
3971 expr,
3972 )
3973 col_expr = col_expr.self_group()
3974 return CollectionAggregate(
3975 col_expr,
3976 operator=operators.all_op,
3977 type_=type_api.BOOLEANTYPE,
3978 )
3980 @util.preload_module("sqlalchemy.sql.sqltypes")
3981 def _bind_param(
3982 self,
3983 operator: operators.OperatorType,
3984 obj: Any,
3985 type_: Optional[TypeEngine[_T]] = None,
3986 expanding: bool = False,
3987 ) -> BindParameter[_T]:
3988 """For new style any_(), all_(), ensure compared literal value
3989 receives appropriate bound parameter type."""
3991 # a CollectionAggregate is specific to ARRAY or int
3992 # only. So for ARRAY case, make sure we use correct element type
3993 sqltypes = util.preloaded.sql_sqltypes
3994 if self.element.type._type_affinity is sqltypes.ARRAY:
3995 compared_to_type = cast(
3996 sqltypes.ARRAY[Any], self.element.type
3997 ).item_type
3998 else:
3999 compared_to_type = self.element.type
4001 return BindParameter(
4002 None,
4003 obj,
4004 _compared_to_operator=operator,
4005 type_=type_,
4006 _compared_to_type=compared_to_type,
4007 unique=True,
4008 expanding=expanding,
4009 )
4011 # operate and reverse_operate are hardwired to
4012 # dispatch onto the type comparator directly, so that we can
4013 # ensure "reversed" behavior.
4014 def operate(
4015 self, op: OperatorType, *other: Any, **kwargs: Any
4016 ) -> ColumnElement[_T]:
4017 if not operators.is_comparison(op):
4018 raise exc.ArgumentError(
4019 "Only comparison operators may be used with ANY/ALL"
4020 )
4021 kwargs["reverse"] = True
4022 return self.comparator.operate(operators.mirror(op), *other, **kwargs)
4024 def reverse_operate(
4025 self, op: OperatorType, other: Any, **kwargs: Any
4026 ) -> ColumnElement[_T]:
4027 # comparison operators should never call reverse_operate
4028 assert not operators.is_comparison(op)
4029 raise exc.ArgumentError(
4030 "Only comparison operators may be used with ANY/ALL"
4031 )
4034class AsBoolean(WrapsColumnExpression[bool], UnaryExpression[bool]):
4035 inherit_cache = True
4037 def __init__(self, element, operator, negate):
4038 self.element = element
4039 self.type = type_api.BOOLEANTYPE
4040 self.operator = operator
4041 self.negate = negate
4042 self.modifier = None
4043 self._is_implicitly_boolean = element._is_implicitly_boolean
4045 @property
4046 def wrapped_column_expression(self):
4047 return self.element
4049 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4050 return self
4052 def _negate(self):
4053 if isinstance(self.element, (True_, False_)):
4054 return self.element._negate()
4055 else:
4056 return AsBoolean(self.element, self.negate, self.operator)
4059class BinaryExpression(OperatorExpression[_T]):
4060 """Represent an expression that is ``LEFT <operator> RIGHT``.
4062 A :class:`.BinaryExpression` is generated automatically
4063 whenever two column expressions are used in a Python binary expression:
4065 .. sourcecode:: pycon+sql
4067 >>> from sqlalchemy.sql import column
4068 >>> column("a") + column("b")
4069 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
4070 >>> print(column("a") + column("b"))
4071 {printsql}a + b
4073 """
4075 __visit_name__ = "binary"
4077 _traverse_internals: _TraverseInternalsType = [
4078 ("left", InternalTraversal.dp_clauseelement),
4079 ("right", InternalTraversal.dp_clauseelement),
4080 ("operator", InternalTraversal.dp_operator),
4081 ("negate", InternalTraversal.dp_operator),
4082 ("modifiers", InternalTraversal.dp_plain_dict),
4083 (
4084 "type",
4085 InternalTraversal.dp_type,
4086 ),
4087 ]
4089 _cache_key_traversal = [
4090 ("left", InternalTraversal.dp_clauseelement),
4091 ("right", InternalTraversal.dp_clauseelement),
4092 ("operator", InternalTraversal.dp_operator),
4093 ("modifiers", InternalTraversal.dp_plain_dict),
4094 # "type" affects JSON CAST operators, so while redundant in most cases,
4095 # is needed for that one
4096 (
4097 "type",
4098 InternalTraversal.dp_type,
4099 ),
4100 ]
4102 _is_implicitly_boolean = True
4103 """Indicates that any database will know this is a boolean expression
4104 even if the database does not have an explicit boolean datatype.
4106 """
4108 left: ColumnElement[Any]
4109 right: ColumnElement[Any]
4110 modifiers: Mapping[str, Any]
4112 def __init__(
4113 self,
4114 left: ColumnElement[Any],
4115 right: ColumnElement[Any],
4116 operator: OperatorType,
4117 type_: Optional[_TypeEngineArgument[_T]] = None,
4118 negate: Optional[OperatorType] = None,
4119 modifiers: Optional[Mapping[str, Any]] = None,
4120 ):
4121 # allow compatibility with libraries that
4122 # refer to BinaryExpression directly and pass strings
4123 if isinstance(operator, str):
4124 operator = operators.custom_op(operator)
4125 self._orig = (left.__hash__(), right.__hash__())
4126 self._propagate_attrs = left._propagate_attrs or right._propagate_attrs
4127 self.left = left.self_group(against=operator)
4128 self.right = right.self_group(against=operator)
4129 self.operator = operator
4131 # if type is None, we get NULLTYPE, which is our _T. But I don't
4132 # know how to get the overloads to express that correctly
4133 self.type = type_api.to_instance(type_) # type: ignore
4135 self.negate = negate
4136 self._is_implicitly_boolean = operators.is_boolean(operator)
4138 if modifiers is None:
4139 self.modifiers = {}
4140 else:
4141 self.modifiers = modifiers
4143 @property
4144 def _flattened_operator_clauses(
4145 self,
4146 ) -> typing_Tuple[ColumnElement[Any], ...]:
4147 return (self.left, self.right)
4149 def __bool__(self):
4150 """Implement Python-side "bool" for BinaryExpression as a
4151 simple "identity" check for the left and right attributes,
4152 if the operator is "eq" or "ne". Otherwise the expression
4153 continues to not support "bool" like all other column expressions.
4155 The rationale here is so that ColumnElement objects can be hashable.
4156 What? Well, suppose you do this::
4158 c1, c2 = column("x"), column("y")
4159 s1 = set([c1, c2])
4161 We do that **a lot**, columns inside of sets is an extremely basic
4162 thing all over the ORM for example.
4164 So what happens if we do this? ::
4166 c1 in s1
4168 Hashing means it will normally use ``__hash__()`` of the object,
4169 but in case of hash collision, it's going to also do ``c1 == c1``
4170 and/or ``c1 == c2`` inside. Those operations need to return a
4171 True/False value. But because we override ``==`` and ``!=``, they're
4172 going to get a BinaryExpression. Hence we implement ``__bool__`` here
4173 so that these comparisons behave in this particular context mostly
4174 like regular object comparisons. Thankfully Python is OK with
4175 that! Otherwise we'd have to use special set classes for columns
4176 (which we used to do, decades ago).
4178 """
4179 if self.operator in (operators.eq, operators.ne):
4180 # this is using the eq/ne operator given int hash values,
4181 # rather than Operator, so that "bool" can be based on
4182 # identity
4183 return self.operator(*self._orig) # type: ignore
4184 else:
4185 raise TypeError("Boolean value of this clause is not defined")
4187 if typing.TYPE_CHECKING:
4189 def __invert__(
4190 self: BinaryExpression[_T],
4191 ) -> BinaryExpression[_T]: ...
4193 @util.ro_non_memoized_property
4194 def _from_objects(self) -> List[FromClause]:
4195 return self.left._from_objects + self.right._from_objects
4197 def _negate(self):
4198 if self.negate is not None:
4199 return BinaryExpression(
4200 self.left,
4201 self.right._negate_in_binary(self.negate, self.operator),
4202 self.negate,
4203 negate=self.operator,
4204 type_=self.type,
4205 modifiers=self.modifiers,
4206 )
4207 else:
4208 return self.self_group()._negate()
4211class Slice(ColumnElement[Any]):
4212 """Represent SQL for a Python array-slice object.
4214 This is not a specific SQL construct at this level, but
4215 may be interpreted by specific dialects, e.g. PostgreSQL.
4217 """
4219 __visit_name__ = "slice"
4221 _traverse_internals: _TraverseInternalsType = [
4222 ("start", InternalTraversal.dp_clauseelement),
4223 ("stop", InternalTraversal.dp_clauseelement),
4224 ("step", InternalTraversal.dp_clauseelement),
4225 ]
4227 def __init__(self, start, stop, step, _name=None):
4228 self.start = coercions.expect(
4229 roles.ExpressionElementRole,
4230 start,
4231 name=_name,
4232 type_=type_api.INTEGERTYPE,
4233 )
4234 self.stop = coercions.expect(
4235 roles.ExpressionElementRole,
4236 stop,
4237 name=_name,
4238 type_=type_api.INTEGERTYPE,
4239 )
4240 self.step = coercions.expect(
4241 roles.ExpressionElementRole,
4242 step,
4243 name=_name,
4244 type_=type_api.INTEGERTYPE,
4245 )
4246 self.type = type_api.NULLTYPE
4248 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4249 assert against is operator.getitem
4250 return self
4253class IndexExpression(BinaryExpression[Any]):
4254 """Represent the class of expressions that are like an "index"
4255 operation."""
4257 inherit_cache = True
4260class GroupedElement(DQLDMLClauseElement):
4261 """Represent any parenthesized expression"""
4263 __visit_name__ = "grouping"
4265 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4266 return self
4268 def _ungroup(self) -> ClauseElement:
4269 raise NotImplementedError()
4272class Grouping(GroupedElement, ColumnElement[_T]):
4273 """Represent a grouping within a column expression"""
4275 _traverse_internals: _TraverseInternalsType = [
4276 ("element", InternalTraversal.dp_clauseelement),
4277 ("type", InternalTraversal.dp_type),
4278 ]
4280 _cache_key_traversal = [
4281 ("element", InternalTraversal.dp_clauseelement),
4282 ]
4284 element: Union[
4285 TextClause, ClauseList, ColumnElement[_T], CompilerColumnElement
4286 ]
4288 def __init__(
4289 self,
4290 element: Union[
4291 TextClause, ClauseList, ColumnElement[_T], CompilerColumnElement
4292 ],
4293 ):
4294 self.element = element
4296 # nulltype assignment issue
4297 self.type = getattr(element, "type", type_api.NULLTYPE) # type: ignore
4298 self._propagate_attrs = element._propagate_attrs
4300 def _with_binary_element_type(self, type_):
4301 return self.__class__(self.element._with_binary_element_type(type_))
4303 def _ungroup(self) -> ColumnElement[_T]:
4304 assert isinstance(self.element, ColumnElement)
4305 return self.element._ungroup()
4307 @util.memoized_property
4308 def _is_implicitly_boolean(self):
4309 return self.element._is_implicitly_boolean
4311 @util.non_memoized_property
4312 def _tq_label(self) -> Optional[str]:
4313 return (
4314 getattr(self.element, "_tq_label", None) or self._anon_name_label
4315 )
4317 @util.non_memoized_property
4318 def _proxies(self) -> List[ColumnElement[Any]]:
4319 if isinstance(self.element, ColumnElement):
4320 return [self.element]
4321 else:
4322 return []
4324 @util.ro_non_memoized_property
4325 def _from_objects(self) -> List[FromClause]:
4326 return self.element._from_objects
4328 def __getattr__(self, attr):
4329 return getattr(self.element, attr)
4331 def __getstate__(self):
4332 return {"element": self.element, "type": self.type}
4334 def __setstate__(self, state):
4335 self.element = state["element"]
4336 self.type = state["type"]
4338 if TYPE_CHECKING:
4340 def self_group(
4341 self, against: Optional[OperatorType] = None
4342 ) -> Self: ...
4345class _OverrideBinds(Grouping[_T]):
4346 """used by cache_key->_apply_params_to_element to allow compilation /
4347 execution of a SQL element that's been cached, using an alternate set of
4348 bound parameter values.
4350 This is used by the ORM to swap new parameter values into expressions
4351 that are embedded into loader options like with_expression(),
4352 selectinload(). Previously, this task was accomplished using the
4353 .params() method which would perform a deep-copy instead. This deep
4354 copy proved to be too expensive for more complex expressions.
4356 See #11085
4358 """
4360 __visit_name__ = "override_binds"
4362 def __init__(
4363 self,
4364 element: ColumnElement[_T],
4365 bindparams: Sequence[BindParameter[Any]],
4366 replaces_params: Sequence[BindParameter[Any]],
4367 ):
4368 self.element = element
4369 self.translate = {
4370 k.key: v.value for k, v in zip(replaces_params, bindparams)
4371 }
4373 def _gen_cache_key(
4374 self, anon_map: anon_map, bindparams: List[BindParameter[Any]]
4375 ) -> Optional[typing_Tuple[Any, ...]]:
4376 """generate a cache key for the given element, substituting its bind
4377 values for the translation values present."""
4379 existing_bps: List[BindParameter[Any]] = []
4380 ck = self.element._gen_cache_key(anon_map, existing_bps)
4382 bindparams.extend(
4383 (
4384 bp._with_value(
4385 self.translate[bp.key], maintain_key=True, required=False
4386 )
4387 if bp.key in self.translate
4388 else bp
4389 )
4390 for bp in existing_bps
4391 )
4393 return ck
4396_FrameIntTuple = tuple[int | None, int | None]
4399class Over(ColumnElement[_T]):
4400 """Represent an OVER clause.
4402 This is a special operator against a so-called
4403 "window" function, as well as any aggregate function,
4404 which produces results relative to the result set
4405 itself. Most modern SQL backends now support window functions.
4407 """
4409 __visit_name__ = "over"
4411 _traverse_internals: _TraverseInternalsType = [
4412 ("element", InternalTraversal.dp_clauseelement),
4413 ("order_by", InternalTraversal.dp_clauseelement),
4414 ("partition_by", InternalTraversal.dp_clauseelement),
4415 ("range_", InternalTraversal.dp_clauseelement),
4416 ("rows", InternalTraversal.dp_clauseelement),
4417 ("groups", InternalTraversal.dp_clauseelement),
4418 ]
4420 order_by: Optional[ClauseList] = None
4421 partition_by: Optional[ClauseList] = None
4423 element: ColumnElement[_T]
4424 """The underlying expression object to which this :class:`.Over`
4425 object refers."""
4427 range_: FrameClause | None
4428 rows: FrameClause | None
4429 groups: FrameClause | None
4431 def __init__(
4432 self,
4433 element: ColumnElement[_T],
4434 partition_by: Optional[_ByArgument] = None,
4435 order_by: Optional[_ByArgument] = None,
4436 range_: _FrameIntTuple | FrameClause | None = None,
4437 rows: _FrameIntTuple | FrameClause | None = None,
4438 groups: _FrameIntTuple | FrameClause | None = None,
4439 ):
4440 self.element = element
4441 if order_by is not None:
4442 self.order_by = ClauseList(
4443 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4444 )
4445 if partition_by is not None:
4446 self.partition_by = ClauseList(
4447 *util.to_list(partition_by),
4448 _literal_as_text_role=roles.ByOfRole,
4449 )
4451 if sum(item is not None for item in (range_, rows, groups)) > 1:
4452 raise exc.ArgumentError(
4453 "only one of 'rows', 'range_', or 'groups' may be provided"
4454 )
4455 else:
4456 self.range_ = FrameClause._parse(range_, coerce_int=False)
4457 self.rows = FrameClause._parse(rows, coerce_int=True)
4458 self.groups = FrameClause._parse(groups, coerce_int=True)
4460 if not TYPE_CHECKING:
4462 @util.memoized_property
4463 def type(self) -> TypeEngine[_T]: # noqa: A001
4464 return self.element.type
4466 @util.ro_non_memoized_property
4467 def _from_objects(self) -> List[FromClause]:
4468 return list(
4469 itertools.chain(
4470 *[
4471 c._from_objects
4472 for c in (self.element, self.partition_by, self.order_by)
4473 if c is not None
4474 ]
4475 )
4476 )
4479class FrameClauseType(Enum):
4480 """Frame clause type enum for FrameClause lower_type and upper_type.
4482 .. versionadded:: 2.1
4484 """
4486 UNBOUNDED = 0
4487 """Produces an "UNBOUNDED PRECEDING" or "UNBOUNDED FOLLOWING" frame
4488 clause depending on the position.
4489 Requires a ``None`` value for the corresponding bound value.
4490 """
4491 CURRENT = 1
4492 """Produces a "CURRENT ROW" frame clause.
4493 Requires a ``None`` value for the corresponding bound value.
4494 """
4495 PRECEDING = 2
4496 """Produces a "PRECEDING" frame clause."""
4497 FOLLOWING = 3
4498 """Produces a "FOLLOWING" frame clause."""
4501_require_none = (
4502 FrameClauseType.CURRENT,
4503 FrameClauseType.UNBOUNDED,
4504)
4507class FrameClause(ClauseElement):
4508 """Indicate the 'rows' 'range' or 'group' field of a window function,
4509 e.g. using :class:`.Over`.
4511 .. versionadded:: 2.1
4513 """
4515 __visit_name__ = "frame_clause"
4517 _traverse_internals: _TraverseInternalsType = [
4518 ("lower_bind", InternalTraversal.dp_clauseelement),
4519 ("upper_bind", InternalTraversal.dp_clauseelement),
4520 ("lower_type", InternalTraversal.dp_plain_obj),
4521 ("upper_type", InternalTraversal.dp_plain_obj),
4522 ]
4524 def __init__(
4525 self,
4526 start: Any,
4527 end: Any,
4528 start_frame_type: FrameClauseType,
4529 end_frame_type: FrameClauseType,
4530 _validate: bool = True,
4531 ) -> None:
4532 """Creates a new FrameClause specifying the bounds of a window frame.
4534 :param start: The start value.
4535 :param end: The end value.
4536 :param start_frame_type: The :class:`FrameClauseType` for the
4537 start value.
4538 :param end_frame_type: The :class:`FrameClauseType` for the end value.
4539 """
4540 self.lower_bind = self._as_literal(start)
4541 self.upper_bind = self._as_literal(end)
4542 self.lower_type = FrameClauseType(start_frame_type)
4543 self.upper_type = FrameClauseType(end_frame_type)
4544 if _validate:
4545 if (
4546 self.lower_type in _require_none
4547 and self.lower_bind is not None
4548 ):
4549 raise exc.ArgumentError(
4550 "Cannot specify a value for start with frame type "
4551 f"{self.lower_type.name}"
4552 )
4553 if (
4554 self.upper_type in _require_none
4555 and self.upper_bind is not None
4556 ):
4557 raise exc.ArgumentError(
4558 "Cannot specify a value for end with frame type "
4559 f"{self.upper_type.name}"
4560 )
4562 @classmethod
4563 def _as_literal(cls, value: Any) -> BindParameter[Any] | None:
4564 if value is None:
4565 return None
4566 elif isinstance(value, int):
4567 return literal(value, type_api.INTEGERTYPE)
4568 elif isinstance(value, BindParameter):
4569 return value
4570 else:
4571 return literal(value) # let the default type resolution occur
4573 @classmethod
4574 def _handle_int(
4575 cls, value: Any | None, coerce_int: bool
4576 ) -> tuple[int | None, FrameClauseType]:
4577 if value is None:
4578 return None, FrameClauseType.UNBOUNDED
4580 if coerce_int:
4581 try:
4582 integer = int(value)
4583 except ValueError as err:
4584 raise exc.ArgumentError(
4585 "Integer or None expected for values in rows/groups frame"
4586 ) from err
4587 elif not isinstance(value, int):
4588 raise exc.ArgumentError(
4589 "When using a tuple to specify a range only integer or none "
4590 "values are allowed in the range frame. To specify a "
4591 "different type use the FrameClause directly."
4592 )
4593 else:
4594 integer = value
4595 if integer == 0:
4596 return None, FrameClauseType.CURRENT
4597 elif integer < 0:
4598 return abs(integer), FrameClauseType.PRECEDING
4599 else:
4600 return integer, FrameClauseType.FOLLOWING
4602 @classmethod
4603 def _parse(
4604 cls,
4605 range_: _FrameIntTuple | FrameClause | None,
4606 coerce_int: bool,
4607 ) -> FrameClause | None:
4608 if range_ is None or isinstance(range_, FrameClause):
4609 return range_
4611 try:
4612 r0, r1 = range_
4613 except (ValueError, TypeError) as ve:
4614 raise exc.ArgumentError(
4615 "2-tuple expected for range/rows/groups"
4616 ) from ve
4618 l_b, l_t = cls._handle_int(r0, coerce_int)
4619 u_b, u_t = cls._handle_int(r1, coerce_int)
4621 return FrameClause(
4622 start=l_b,
4623 end=u_b,
4624 start_frame_type=l_t,
4625 end_frame_type=u_t,
4626 _validate=False,
4627 )
4630class AggregateOrderBy(WrapsColumnExpression[_T]):
4631 """Represent an aggregate ORDER BY expression.
4633 This is a special operator against aggregate functions such as
4634 ``array_agg()``, ``json_arrayagg()`` ``string_agg()``, etc. that provides
4635 for an ORDER BY expression, using a syntax that's compatible with
4636 the backend.
4638 :class:`.AggregateOrderBy` is a generalized version of the
4639 :class:`.WithinGroup` construct, the latter of which always provides a
4640 "WITHIN GROUP (ORDER BY ...)" expression. :class:`.AggregateOrderBy` will
4641 also compile to "WITHIN GROUP (ORDER BY ...)" on backends such as Oracle
4642 and SQL Server that don't have another style of aggregate function
4643 ordering.
4645 .. versionadded:: 2.1
4648 """
4650 __visit_name__ = "aggregateorderby"
4652 _traverse_internals: _TraverseInternalsType = [
4653 ("element", InternalTraversal.dp_clauseelement),
4654 ("order_by", InternalTraversal.dp_clauseelement),
4655 ]
4657 order_by: ClauseList
4659 def __init__(
4660 self,
4661 element: Union[FunctionElement[_T], FunctionFilter[_T]],
4662 *order_by: _ColumnExpressionArgument[Any],
4663 ):
4664 self.element = element
4665 if not order_by:
4666 raise TypeError("at least one ORDER BY element is required")
4667 self.order_by = ClauseList(
4668 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4669 )
4671 if not TYPE_CHECKING:
4673 @util.memoized_property
4674 def type(self) -> TypeEngine[_T]: # noqa: A001
4675 return self.element.type
4677 @property
4678 def wrapped_column_expression(self) -> ColumnElement[_T]:
4679 return self.element
4681 def __reduce__(self):
4682 return self.__class__, (self.element,) + (
4683 tuple(self.order_by) if self.order_by is not None else ()
4684 )
4686 def over(
4687 self,
4688 *,
4689 partition_by: _ByArgument | None = None,
4690 order_by: _ByArgument | None = None,
4691 rows: _FrameIntTuple | FrameClause | None = None,
4692 range_: _FrameIntTuple | FrameClause | None = None,
4693 groups: _FrameIntTuple | FrameClause | None = None,
4694 ) -> Over[_T]:
4695 """Produce an OVER clause against this :class:`.WithinGroup`
4696 construct.
4698 This function has the same signature as that of
4699 :meth:`.FunctionElement.over`.
4701 """
4702 return Over(
4703 self,
4704 partition_by=partition_by,
4705 order_by=order_by,
4706 range_=range_,
4707 rows=rows,
4708 groups=groups,
4709 )
4711 @overload
4712 def filter(self) -> Self: ...
4714 @overload
4715 def filter(
4716 self,
4717 __criterion0: _ColumnExpressionArgument[bool],
4718 *criterion: _ColumnExpressionArgument[bool],
4719 ) -> FunctionFilter[_T]: ...
4721 def filter(
4722 self, *criterion: _ColumnExpressionArgument[bool]
4723 ) -> Union[Self, FunctionFilter[_T]]:
4724 """Produce a FILTER clause against this function."""
4725 if not criterion:
4726 return self
4727 return FunctionFilter(self, *criterion)
4729 @util.ro_non_memoized_property
4730 def _from_objects(self) -> List[FromClause]:
4731 return list(
4732 itertools.chain(
4733 *[
4734 c._from_objects
4735 for c in (self.element, self.order_by)
4736 if c is not None
4737 ]
4738 )
4739 )
4742class WithinGroup(AggregateOrderBy[_T]):
4743 """Represent a WITHIN GROUP (ORDER BY) clause.
4745 This is a special operator against so-called
4746 "ordered set aggregate" and "hypothetical
4747 set aggregate" functions, including ``percentile_cont()``,
4748 ``rank()``, ``dense_rank()``, etc.
4750 It's supported only by certain database backends, such as PostgreSQL,
4751 Oracle Database and MS SQL Server.
4753 The :class:`.WithinGroup` construct extracts its type from the
4754 method :meth:`.FunctionElement.within_group_type`. If this returns
4755 ``None``, the function's ``.type`` is used.
4757 """
4759 __visit_name__ = "withingroup"
4760 inherit_cache = True
4762 if not TYPE_CHECKING:
4764 @util.memoized_property
4765 def type(self) -> TypeEngine[_T]: # noqa: A001
4766 wgt = self.element.within_group_type(self)
4767 if wgt is not None:
4768 return wgt
4769 else:
4770 return self.element.type
4773class FunctionFilter(Generative, ColumnElement[_T]):
4774 """Represent a function FILTER clause.
4776 This is a special operator against aggregate and window functions,
4777 which controls which rows are passed to it.
4778 It's supported only by certain database backends.
4780 Invocation of :class:`.FunctionFilter` is via
4781 :meth:`.FunctionElement.filter`::
4783 func.count(1).filter(True)
4785 .. seealso::
4787 :meth:`.FunctionElement.filter`
4789 """
4791 __visit_name__ = "funcfilter"
4793 _traverse_internals: _TraverseInternalsType = [
4794 ("func", InternalTraversal.dp_clauseelement),
4795 ("criterion", InternalTraversal.dp_clauseelement),
4796 ]
4798 criterion: Optional[ColumnElement[bool]] = None
4800 def __init__(
4801 self,
4802 func: Union[FunctionElement[_T], AggregateOrderBy[_T]],
4803 *criterion: _ColumnExpressionArgument[bool],
4804 ):
4805 self.func = func
4806 self.filter.non_generative(self, *criterion) # type: ignore
4808 @_generative
4809 def filter(self, *criterion: _ColumnExpressionArgument[bool]) -> Self:
4810 """Produce an additional FILTER against the function.
4812 This method adds additional criteria to the initial criteria
4813 set up by :meth:`.FunctionElement.filter`.
4815 Multiple criteria are joined together at SQL render time
4816 via ``AND``.
4819 """
4821 for crit in list(criterion):
4822 crit = coercions.expect(roles.WhereHavingRole, crit)
4824 if self.criterion is not None:
4825 self.criterion = self.criterion & crit
4826 else:
4827 self.criterion = crit
4829 return self
4831 def over(
4832 self,
4833 partition_by: _ByArgument | None = None,
4834 order_by: _ByArgument | None = None,
4835 range_: _FrameIntTuple | FrameClause | None = None,
4836 rows: _FrameIntTuple | FrameClause | None = None,
4837 groups: _FrameIntTuple | FrameClause | None = None,
4838 ) -> Over[_T]:
4839 """Produce an OVER clause against this filtered function.
4841 Used against aggregate or so-called "window" functions,
4842 for database backends that support window functions.
4844 The expression::
4846 func.rank().filter(MyClass.y > 5).over(order_by="x")
4848 is shorthand for::
4850 from sqlalchemy import over, funcfilter
4852 over(funcfilter(func.rank(), MyClass.y > 5), order_by="x")
4854 See :func:`_expression.over` for a full description.
4856 """
4857 return Over(
4858 self,
4859 partition_by=partition_by,
4860 order_by=order_by,
4861 range_=range_,
4862 rows=rows,
4863 groups=groups,
4864 )
4866 def within_group(
4867 self, *order_by: _ColumnExpressionArgument[Any]
4868 ) -> WithinGroup[_T]:
4869 """Produce a WITHIN GROUP (ORDER BY expr) clause against
4870 this function.
4871 """
4872 return WithinGroup(self, *order_by)
4874 def within_group_type(
4875 self, within_group: WithinGroup[_T]
4876 ) -> Optional[TypeEngine[_T]]:
4877 return None
4879 def self_group(
4880 self, against: Optional[OperatorType] = None
4881 ) -> Union[Self, Grouping[_T]]:
4882 if operators.is_precedent(operators.filter_op, against):
4883 return Grouping(self)
4884 else:
4885 return self
4887 if not TYPE_CHECKING:
4889 @util.memoized_property
4890 def type(self) -> TypeEngine[_T]: # noqa: A001
4891 return self.func.type
4893 @util.ro_non_memoized_property
4894 def _from_objects(self) -> List[FromClause]:
4895 return list(
4896 itertools.chain(
4897 *[
4898 c._from_objects
4899 for c in (self.func, self.criterion)
4900 if c is not None
4901 ]
4902 )
4903 )
4906class NamedColumn(KeyedColumnElement[_T]):
4907 is_literal = False
4908 table: Optional[FromClause] = None
4909 name: str
4910 key: str
4912 def _compare_name_for_result(self, other):
4913 return (hasattr(other, "name") and self.name == other.name) or (
4914 hasattr(other, "_label") and self._label == other._label
4915 )
4917 @util.ro_memoized_property
4918 def description(self) -> str:
4919 return self.name
4921 @HasMemoized.memoized_attribute
4922 def _tq_key_label(self) -> Optional[str]:
4923 """table qualified label based on column key.
4925 for table-bound columns this is <tablename>_<column key/proxy key>;
4927 all other expressions it resolves to key/proxy key.
4929 """
4930 proxy_key = self._proxy_key
4931 if proxy_key and proxy_key != self.name:
4932 return self._gen_tq_label(proxy_key)
4933 else:
4934 return self._tq_label
4936 @HasMemoized.memoized_attribute
4937 def _tq_label(self) -> Optional[str]:
4938 """table qualified label based on column name.
4940 for table-bound columns this is <tablename>_<columnname>; all other
4941 expressions it resolves to .name.
4943 """
4944 return self._gen_tq_label(self.name)
4946 @HasMemoized.memoized_attribute
4947 def _render_label_in_columns_clause(self):
4948 return True
4950 @HasMemoized.memoized_attribute
4951 def _non_anon_label(self):
4952 return self.name
4954 def _gen_tq_label(
4955 self, name: str, dedupe_on_key: bool = True
4956 ) -> Optional[str]:
4957 return name
4959 def _bind_param(
4960 self,
4961 operator: OperatorType,
4962 obj: Any,
4963 type_: Optional[TypeEngine[_T]] = None,
4964 expanding: bool = False,
4965 ) -> BindParameter[_T]:
4966 return BindParameter(
4967 self.key,
4968 obj,
4969 _compared_to_operator=operator,
4970 _compared_to_type=self.type,
4971 type_=type_,
4972 unique=True,
4973 expanding=expanding,
4974 )
4976 def _make_proxy(
4977 self,
4978 selectable: FromClause,
4979 *,
4980 primary_key: ColumnSet,
4981 foreign_keys: Set[KeyedColumnElement[Any]],
4982 name: Optional[str] = None,
4983 key: Optional[str] = None,
4984 name_is_truncatable: bool = False,
4985 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
4986 disallow_is_literal: bool = False,
4987 **kw: Any,
4988 ) -> typing_Tuple[str, ColumnClause[_T]]:
4989 c = ColumnClause(
4990 (
4991 coercions.expect(roles.TruncatedLabelRole, name or self.name)
4992 if name_is_truncatable
4993 else (name or self.name)
4994 ),
4995 type_=self.type,
4996 _selectable=selectable,
4997 is_literal=False,
4998 )
5000 c._propagate_attrs = selectable._propagate_attrs
5001 if name is None:
5002 c.key = self.key
5003 if compound_select_cols:
5004 c._proxies = list(compound_select_cols)
5005 else:
5006 c._proxies = [self]
5008 if selectable._is_clone_of is not None:
5009 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
5010 return c.key, c
5013_PS = ParamSpec("_PS")
5016class Label(roles.LabeledColumnExprRole[_T], NamedColumn[_T]):
5017 """Represents a column label (AS).
5019 Represent a label, as typically applied to any column-level
5020 element using the ``AS`` sql keyword.
5022 """
5024 __visit_name__ = "label"
5026 _traverse_internals: _TraverseInternalsType = [
5027 ("name", InternalTraversal.dp_anon_name),
5028 ("type", InternalTraversal.dp_type),
5029 ("_element", InternalTraversal.dp_clauseelement),
5030 ]
5032 _cache_key_traversal = [
5033 ("name", InternalTraversal.dp_anon_name),
5034 ("_element", InternalTraversal.dp_clauseelement),
5035 ]
5037 _element: ColumnElement[_T]
5038 name: str
5040 def __init__(
5041 self,
5042 name: Optional[str],
5043 element: _ColumnExpressionArgument[_T],
5044 type_: Optional[_TypeEngineArgument[_T]] = None,
5045 ):
5046 orig_element = element
5047 element = coercions.expect(
5048 roles.ExpressionElementRole,
5049 element,
5050 apply_propagate_attrs=self,
5051 )
5052 while isinstance(element, Label):
5053 # TODO: this is only covered in test_text.py, but nothing
5054 # fails if it's removed. determine rationale
5055 element = element.element
5057 if name:
5058 self.name = name
5059 else:
5060 self.name = _anonymous_label.safe_construct(
5061 id(self), getattr(element, "name", "anon")
5062 )
5063 if isinstance(orig_element, Label):
5064 # TODO: no coverage for this block, again would be in
5065 # test_text.py where the resolve_label concept is important
5066 self._resolve_label = orig_element._label
5068 self.key = self._tq_label = self._tq_key_label = self.name
5069 self._element = element
5071 self.type = (
5072 type_api.to_instance(type_)
5073 if type_ is not None
5074 else self._element.type
5075 )
5077 self._proxies = [element]
5079 def __reduce__(self):
5080 return self.__class__, (self.name, self._element, self.type)
5082 @HasMemoized.memoized_attribute
5083 def _render_label_in_columns_clause(self):
5084 return True
5086 def _bind_param(self, operator, obj, type_=None, expanding=False):
5087 return BindParameter(
5088 None,
5089 obj,
5090 _compared_to_operator=operator,
5091 type_=type_,
5092 _compared_to_type=self.type,
5093 unique=True,
5094 expanding=expanding,
5095 )
5097 @util.memoized_property
5098 def _is_implicitly_boolean(self):
5099 return self.element._is_implicitly_boolean
5101 @HasMemoized.memoized_attribute
5102 def _allow_label_resolve(self):
5103 return self.element._allow_label_resolve
5105 @property
5106 def _order_by_label_element(self):
5107 return self
5109 def as_reference(self) -> _label_reference[_T]:
5110 """refer to this labeled expression in a clause such as GROUP BY,
5111 ORDER BY etc. as the label name itself, without expanding
5112 into the full expression.
5114 .. versionadded:: 2.1
5116 """
5117 return _label_reference(self)
5119 @HasMemoized.memoized_attribute
5120 def element(self) -> ColumnElement[_T]:
5121 return self._element.self_group(against=operators.as_)
5123 def self_group(self, against: Optional[OperatorType] = None) -> Label[_T]:
5124 return self._apply_to_inner(self._element.self_group, against=against)
5126 def _negate(self):
5127 return self._apply_to_inner(self._element._negate)
5129 def _apply_to_inner(
5130 self,
5131 fn: Callable[_PS, ColumnElement[_T]],
5132 *arg: _PS.args,
5133 **kw: _PS.kwargs,
5134 ) -> Label[_T]:
5135 sub_element = fn(*arg, **kw)
5136 if sub_element is not self._element:
5137 return Label(self.name, sub_element, type_=self.type)
5138 else:
5139 return self
5141 @property
5142 def primary_key(self): # type: ignore[override]
5143 return self.element.primary_key
5145 @property
5146 def foreign_keys(self): # type: ignore[override]
5147 return self.element.foreign_keys
5149 def _copy_internals(
5150 self,
5151 *,
5152 clone: _CloneCallableType = _clone,
5153 anonymize_labels: bool = False,
5154 **kw: Any,
5155 ) -> None:
5156 self._reset_memoizations()
5157 self._element = clone(self._element, **kw)
5158 if anonymize_labels:
5159 self.name = _anonymous_label.safe_construct(
5160 id(self), getattr(self.element, "name", "anon")
5161 )
5162 self.key = self._tq_label = self._tq_key_label = self.name
5164 @util.ro_non_memoized_property
5165 def _from_objects(self) -> List[FromClause]:
5166 return self.element._from_objects
5168 def _make_proxy(
5169 self,
5170 selectable: FromClause,
5171 *,
5172 primary_key: ColumnSet,
5173 foreign_keys: Set[KeyedColumnElement[Any]],
5174 name: Optional[str] = None,
5175 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5176 **kw: Any,
5177 ) -> typing_Tuple[str, ColumnClause[_T]]:
5178 name = self.name if not name else name
5180 key, e = self.element._make_proxy(
5181 selectable,
5182 name=name,
5183 disallow_is_literal=True,
5184 name_is_truncatable=isinstance(name, _truncated_label),
5185 compound_select_cols=compound_select_cols,
5186 primary_key=primary_key,
5187 foreign_keys=foreign_keys,
5188 )
5190 # there was a note here to remove this assertion, which was here
5191 # to determine if we later could support a use case where
5192 # the key and name of a label are separate. But I don't know what
5193 # that case was. For now, this is an unexpected case that occurs
5194 # when a label name conflicts with other columns and select()
5195 # is attempting to disambiguate an explicit label, which is not what
5196 # the user would want. See issue #6090.
5197 if key != self.name and not isinstance(self.name, _anonymous_label):
5198 raise exc.InvalidRequestError(
5199 "Label name %s is being renamed to an anonymous label due "
5200 "to disambiguation "
5201 "which is not supported right now. Please use unique names "
5202 "for explicit labels." % (self.name)
5203 )
5205 e._propagate_attrs = selectable._propagate_attrs
5206 e._proxies.append(self)
5207 if self.type is not None:
5208 e.type = self.type
5210 return self.key, e
5213class ColumnClause(
5214 roles.DDLReferredColumnRole,
5215 roles.LabeledColumnExprRole[_T],
5216 roles.StrAsPlainColumnRole,
5217 Immutable,
5218 NamedColumn[_T],
5219):
5220 """Represents a column expression from any textual string.
5222 The :class:`.ColumnClause`, a lightweight analogue to the
5223 :class:`_schema.Column` class, is typically invoked using the
5224 :func:`_expression.column` function, as in::
5226 from sqlalchemy import column
5228 id, name = column("id"), column("name")
5229 stmt = select(id, name).select_from("user")
5231 The above statement would produce SQL like:
5233 .. sourcecode:: sql
5235 SELECT id, name FROM user
5237 :class:`.ColumnClause` is the immediate superclass of the schema-specific
5238 :class:`_schema.Column` object. While the :class:`_schema.Column`
5239 class has all the
5240 same capabilities as :class:`.ColumnClause`, the :class:`.ColumnClause`
5241 class is usable by itself in those cases where behavioral requirements
5242 are limited to simple SQL expression generation. The object has none of
5243 the associations with schema-level metadata or with execution-time
5244 behavior that :class:`_schema.Column` does,
5245 so in that sense is a "lightweight"
5246 version of :class:`_schema.Column`.
5248 Full details on :class:`.ColumnClause` usage is at
5249 :func:`_expression.column`.
5251 .. seealso::
5253 :func:`_expression.column`
5255 :class:`_schema.Column`
5257 """
5259 table: Optional[FromClause]
5260 is_literal: bool
5262 __visit_name__ = "column"
5264 _traverse_internals: _TraverseInternalsType = [
5265 ("name", InternalTraversal.dp_anon_name),
5266 ("type", InternalTraversal.dp_type),
5267 ("table", InternalTraversal.dp_clauseelement),
5268 ("is_literal", InternalTraversal.dp_boolean),
5269 ]
5271 onupdate: Optional[DefaultGenerator] = None
5272 default: Optional[DefaultGenerator] = None
5273 server_default: Optional[FetchedValue] = None
5274 server_onupdate: Optional[FetchedValue] = None
5276 _is_multiparam_column = False
5278 @property
5279 def _is_star(self): # type: ignore[override]
5280 return self.is_literal and self.name == "*"
5282 def __init__(
5283 self,
5284 text: str,
5285 type_: Optional[_TypeEngineArgument[_T]] = None,
5286 is_literal: bool = False,
5287 _selectable: Optional[FromClause] = None,
5288 ):
5289 self.key = self.name = text
5290 self.table = _selectable
5292 # if type is None, we get NULLTYPE, which is our _T. But I don't
5293 # know how to get the overloads to express that correctly
5294 self.type = type_api.to_instance(type_) # type: ignore
5296 self.is_literal = is_literal
5298 def get_children(self, *, column_tables=False, **kw):
5299 # override base get_children() to not return the Table
5300 # or selectable that is parent to this column. Traversals
5301 # expect the columns of tables and subqueries to be leaf nodes.
5302 return []
5304 @property
5305 def entity_namespace(self):
5306 if self.table is not None:
5307 return self.table.entity_namespace
5308 else:
5309 return super().entity_namespace
5311 def _clone(self, detect_subquery_cols=False, **kw):
5312 if (
5313 detect_subquery_cols
5314 and self.table is not None
5315 and self.table._is_subquery
5316 ):
5317 clone = kw.pop("clone")
5318 table = clone(self.table, **kw)
5319 new = table.c.corresponding_column(self)
5320 return new
5322 return super()._clone(**kw)
5324 @HasMemoized_ro_memoized_attribute
5325 def _from_objects(self) -> List[FromClause]:
5326 t = self.table
5327 if t is not None:
5328 return [t]
5329 else:
5330 return []
5332 @HasMemoized.memoized_attribute
5333 def _render_label_in_columns_clause(self):
5334 return self.table is not None
5336 @property
5337 def _ddl_label(self):
5338 return self._gen_tq_label(self.name, dedupe_on_key=False)
5340 def _compare_name_for_result(self, other):
5341 if (
5342 self.is_literal
5343 or self.table is None
5344 or self.table._is_textual
5345 or not hasattr(other, "proxy_set")
5346 or (
5347 isinstance(other, ColumnClause)
5348 and (
5349 other.is_literal
5350 or other.table is None
5351 or other.table._is_textual
5352 )
5353 )
5354 ):
5355 return (hasattr(other, "name") and self.name == other.name) or (
5356 hasattr(other, "_tq_label")
5357 and self._tq_label == other._tq_label
5358 )
5359 else:
5360 return other.proxy_set.intersection(self.proxy_set)
5362 def _gen_tq_label(
5363 self, name: str, dedupe_on_key: bool = True
5364 ) -> Optional[str]:
5365 """generate table-qualified label
5367 for a table-bound column this is <tablename>_<columnname>.
5369 used primarily for LABEL_STYLE_TABLENAME_PLUS_COL
5370 as well as the .columns collection on a Join object.
5372 """
5373 label: str
5374 t = self.table
5375 if self.is_literal:
5376 return None
5377 elif t is not None and is_named_from_clause(t):
5378 if has_schema_attr(t) and t.schema:
5379 label = (
5380 t.schema.replace(".", "_") + "_" + t.name + ("_" + name)
5381 )
5382 else:
5383 assert not TYPE_CHECKING or isinstance(t, NamedFromClause)
5384 label = t.name + ("_" + name)
5386 # propagate name quoting rules for labels.
5387 if is_quoted_name(name) and name.quote is not None:
5388 if is_quoted_name(label):
5389 label.quote = name.quote
5390 else:
5391 label = quoted_name(label, name.quote)
5392 elif is_quoted_name(t.name) and t.name.quote is not None:
5393 # can't get this situation to occur, so let's
5394 # assert false on it for now
5395 assert not isinstance(label, quoted_name)
5396 label = quoted_name(label, t.name.quote)
5398 if dedupe_on_key:
5399 # ensure the label name doesn't conflict with that of an
5400 # existing column. note that this implies that any Column
5401 # must **not** set up its _label before its parent table has
5402 # all of its other Column objects set up. There are several
5403 # tables in the test suite which will fail otherwise; example:
5404 # table "owner" has columns "name" and "owner_name". Therefore
5405 # column owner.name cannot use the label "owner_name", it has
5406 # to be "owner_name_1".
5407 if label in t.c:
5408 _label = label
5409 counter = 1
5410 while _label in t.c:
5411 _label = label + f"_{counter}"
5412 counter += 1
5413 label = _label
5415 return coercions.expect(roles.TruncatedLabelRole, label)
5417 else:
5418 return name
5420 def _make_proxy(
5421 self,
5422 selectable: FromClause,
5423 *,
5424 primary_key: ColumnSet,
5425 foreign_keys: Set[KeyedColumnElement[Any]],
5426 name: Optional[str] = None,
5427 key: Optional[str] = None,
5428 name_is_truncatable: bool = False,
5429 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5430 disallow_is_literal: bool = False,
5431 **kw: Any,
5432 ) -> typing_Tuple[str, ColumnClause[_T]]:
5433 # the "is_literal" flag normally should never be propagated; a proxied
5434 # column is always a SQL identifier and never the actual expression
5435 # being evaluated. however, there is a case where the "is_literal" flag
5436 # might be used to allow the given identifier to have a fixed quoting
5437 # pattern already, so maintain the flag for the proxy unless a
5438 # :class:`.Label` object is creating the proxy. See [ticket:4730].
5439 is_literal = (
5440 not disallow_is_literal
5441 and self.is_literal
5442 and (
5443 # note this does not accommodate for quoted_name differences
5444 # right now
5445 name is None
5446 or name == self.name
5447 )
5448 )
5449 c = self._constructor(
5450 (
5451 coercions.expect(roles.TruncatedLabelRole, name or self.name)
5452 if name_is_truncatable
5453 else (name or self.name)
5454 ),
5455 type_=self.type,
5456 _selectable=selectable,
5457 is_literal=is_literal,
5458 )
5459 c._propagate_attrs = selectable._propagate_attrs
5460 if name is None:
5461 c.key = self.key
5462 if compound_select_cols:
5463 c._proxies = list(compound_select_cols)
5464 else:
5465 c._proxies = [self]
5467 if selectable._is_clone_of is not None:
5468 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
5469 return c.key, c
5472class TableValuedColumn(NamedColumn[_T]):
5473 __visit_name__ = "table_valued_column"
5475 _traverse_internals: _TraverseInternalsType = [
5476 ("name", InternalTraversal.dp_anon_name),
5477 ("type", InternalTraversal.dp_type),
5478 ("scalar_alias", InternalTraversal.dp_clauseelement),
5479 ]
5481 def __init__(self, scalar_alias: NamedFromClause, type_: TypeEngine[_T]):
5482 self.scalar_alias = scalar_alias
5483 self.key = self.name = scalar_alias.name
5484 self.type = type_
5486 def _copy_internals(
5487 self, clone: _CloneCallableType = _clone, **kw: Any
5488 ) -> None:
5489 self.scalar_alias = clone(self.scalar_alias, **kw)
5490 self.key = self.name = self.scalar_alias.name
5492 @util.ro_non_memoized_property
5493 def _from_objects(self) -> List[FromClause]:
5494 return [self.scalar_alias]
5497class CollationClause(ColumnElement[str]):
5498 __visit_name__ = "collation"
5500 _traverse_internals: _TraverseInternalsType = [
5501 ("collation", InternalTraversal.dp_string)
5502 ]
5504 @classmethod
5505 @util.preload_module("sqlalchemy.sql.sqltypes")
5506 def _create_collation_expression(
5507 cls, expression: _ColumnExpressionArgument[str], collation: str
5508 ) -> BinaryExpression[str]:
5510 sqltypes = util.preloaded.sql_sqltypes
5512 expr = coercions.expect(roles.ExpressionElementRole[str], expression)
5514 if expr.type._type_affinity is sqltypes.String:
5515 collate_type = expr.type._with_collation(collation)
5516 else:
5517 collate_type = expr.type
5519 return BinaryExpression(
5520 expr,
5521 CollationClause(collation),
5522 operators.collate,
5523 type_=collate_type,
5524 )
5526 def __init__(self, collation):
5527 self.collation = collation
5530class _IdentifiedClause(Executable, ClauseElement):
5531 __visit_name__ = "identified"
5533 def __init__(self, ident):
5534 self.ident = ident
5537class SavepointClause(_IdentifiedClause):
5538 __visit_name__ = "savepoint"
5539 inherit_cache = False
5542class RollbackToSavepointClause(_IdentifiedClause):
5543 __visit_name__ = "rollback_to_savepoint"
5544 inherit_cache = False
5547class ReleaseSavepointClause(_IdentifiedClause):
5548 __visit_name__ = "release_savepoint"
5549 inherit_cache = False
5552class quoted_name(util.MemoizedSlots, str):
5553 """Represent a SQL identifier combined with quoting preferences.
5555 :class:`.quoted_name` is a Python unicode/str subclass which
5556 represents a particular identifier name along with a
5557 ``quote`` flag. This ``quote`` flag, when set to
5558 ``True`` or ``False``, overrides automatic quoting behavior
5559 for this identifier in order to either unconditionally quote
5560 or to not quote the name. If left at its default of ``None``,
5561 quoting behavior is applied to the identifier on a per-backend basis
5562 based on an examination of the token itself.
5564 A :class:`.quoted_name` object with ``quote=True`` is also
5565 prevented from being modified in the case of a so-called
5566 "name normalize" option. Certain database backends, such as
5567 Oracle Database, Firebird, and DB2 "normalize" case-insensitive names
5568 as uppercase. The SQLAlchemy dialects for these backends
5569 convert from SQLAlchemy's lower-case-means-insensitive convention
5570 to the upper-case-means-insensitive conventions of those backends.
5571 The ``quote=True`` flag here will prevent this conversion from occurring
5572 to support an identifier that's quoted as all lower case against
5573 such a backend.
5575 The :class:`.quoted_name` object is normally created automatically
5576 when specifying the name for key schema constructs such as
5577 :class:`_schema.Table`, :class:`_schema.Column`, and others.
5578 The class can also be
5579 passed explicitly as the name to any function that receives a name which
5580 can be quoted. Such as to use the :meth:`_engine.Engine.has_table`
5581 method with
5582 an unconditionally quoted name::
5584 from sqlalchemy import create_engine
5585 from sqlalchemy import inspect
5586 from sqlalchemy.sql import quoted_name
5588 engine = create_engine("oracle+oracledb://some_dsn")
5589 print(inspect(engine).has_table(quoted_name("some_table", True)))
5591 The above logic will run the "has table" logic against the Oracle Database
5592 backend, passing the name exactly as ``"some_table"`` without converting to
5593 upper case.
5595 """
5597 __slots__ = "quote", "lower", "upper"
5599 quote: Optional[bool]
5601 @overload
5602 @classmethod
5603 def construct(cls, value: str, quote: Optional[bool]) -> quoted_name: ...
5605 @overload
5606 @classmethod
5607 def construct(cls, value: None, quote: Optional[bool]) -> None: ...
5609 @classmethod
5610 def construct(
5611 cls, value: Optional[str], quote: Optional[bool]
5612 ) -> Optional[quoted_name]:
5613 if value is None:
5614 return None
5615 else:
5616 return quoted_name(value, quote)
5618 def __new__(cls, value: str, quote: Optional[bool]) -> quoted_name:
5619 assert (
5620 value is not None
5621 ), "use quoted_name.construct() for None passthrough"
5622 if isinstance(value, cls) and (quote is None or value.quote == quote):
5623 return value
5624 self = super().__new__(cls, value)
5626 self.quote = quote
5627 return self
5629 def __reduce__(self):
5630 return quoted_name, (str(self), self.quote)
5632 def _memoized_method_lower(self):
5633 if self.quote:
5634 return self
5635 else:
5636 return str(self).lower()
5638 def _memoized_method_upper(self):
5639 if self.quote:
5640 return self
5641 else:
5642 return str(self).upper()
5645def _find_columns(clause: ClauseElement) -> Set[ColumnClause[Any]]:
5646 """locate Column objects within the given expression."""
5648 cols: Set[ColumnClause[Any]] = set()
5649 traverse(clause, {}, {"column": cols.add})
5650 return cols
5653def _type_from_args(args: Sequence[ColumnElement[_T]]) -> TypeEngine[_T]:
5654 for a in args:
5655 if not a.type._isnull:
5656 return a.type
5657 else:
5658 return type_api.NULLTYPE # type: ignore
5661def _corresponding_column_or_error(fromclause, column, require_embedded=False):
5662 c = fromclause.corresponding_column(
5663 column, require_embedded=require_embedded
5664 )
5665 if c is None:
5666 raise exc.InvalidRequestError(
5667 "Given column '%s', attached to table '%s', "
5668 "failed to locate a corresponding column from table '%s'"
5669 % (column, getattr(column, "table", None), fromclause.description)
5670 )
5671 return c
5674class _memoized_property_but_not_nulltype(
5675 util.memoized_property["TypeEngine[_T]"]
5676):
5677 """memoized property, but dont memoize NullType"""
5679 def __get__(self, obj, cls):
5680 if obj is None:
5681 return self
5682 result = self.fget(obj)
5683 if not result._isnull:
5684 obj.__dict__[self.__name__] = result
5685 return result
5688class AnnotatedColumnElement(Annotated):
5689 _Annotated__element: ColumnElement[Any]
5691 def __init__(self, element, values):
5692 Annotated.__init__(self, element, values)
5693 for attr in (
5694 "comparator",
5695 "_proxy_key",
5696 "_tq_key_label",
5697 "_tq_label",
5698 "_non_anon_label",
5699 "type",
5700 ):
5701 self.__dict__.pop(attr, None)
5702 for attr in ("name", "key", "table"):
5703 if self.__dict__.get(attr, False) is None:
5704 self.__dict__.pop(attr)
5706 def _with_annotations(self, values):
5707 clone = super()._with_annotations(values)
5708 for attr in (
5709 "comparator",
5710 "_proxy_key",
5711 "_tq_key_label",
5712 "_tq_label",
5713 "_non_anon_label",
5714 ):
5715 clone.__dict__.pop(attr, None)
5716 return clone
5718 @util.memoized_property
5719 def name(self):
5720 """pull 'name' from parent, if not present"""
5721 return self._Annotated__element.name
5723 @_memoized_property_but_not_nulltype
5724 def type(self):
5725 """pull 'type' from parent and don't cache if null.
5727 type is routinely changed on existing columns within the
5728 mapped_column() initialization process, and "type" is also consulted
5729 during the creation of SQL expressions. Therefore it can change after
5730 it was already retrieved. At the same time we don't want annotated
5731 objects having overhead when expressions are produced, so continue
5732 to memoize, but only when we have a non-null type.
5734 """
5735 return self._Annotated__element.type
5737 @util.memoized_property
5738 def table(self):
5739 """pull 'table' from parent, if not present"""
5740 return self._Annotated__element.table
5742 @util.memoized_property
5743 def key(self):
5744 """pull 'key' from parent, if not present"""
5745 return self._Annotated__element.key
5747 @util.memoized_property
5748 def info(self) -> _InfoType:
5749 if TYPE_CHECKING:
5750 assert isinstance(self._Annotated__element, Column)
5751 return self._Annotated__element.info
5753 @util.memoized_property
5754 def _anon_name_label(self) -> str:
5755 return self._Annotated__element._anon_name_label
5758class _truncated_label(quoted_name):
5759 """A unicode subclass used to identify symbolic "
5760 "names that may require truncation."""
5762 __slots__ = ()
5764 def __new__(cls, value: str, quote: Optional[bool] = None) -> Any:
5765 quote = getattr(value, "quote", quote)
5766 # return super(_truncated_label, cls).__new__(cls, value, quote, True)
5767 return super().__new__(cls, value, quote)
5769 def __reduce__(self) -> Any:
5770 return self.__class__, (str(self), self.quote)
5772 def apply_map(self, map_: Mapping[str, Any]) -> str:
5773 return self
5776class conv(_truncated_label):
5777 """Mark a string indicating that a name has already been converted
5778 by a naming convention.
5780 This is a string subclass that indicates a name that should not be
5781 subject to any further naming conventions.
5783 E.g. when we create a :class:`.Constraint` using a naming convention
5784 as follows::
5786 m = MetaData(
5787 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5788 )
5789 t = Table(
5790 "t", m, Column("x", Integer), CheckConstraint("x > 5", name="x5")
5791 )
5793 The name of the above constraint will be rendered as ``"ck_t_x5"``.
5794 That is, the existing name ``x5`` is used in the naming convention as the
5795 ``constraint_name`` token.
5797 In some situations, such as in migration scripts, we may be rendering
5798 the above :class:`.CheckConstraint` with a name that's already been
5799 converted. In order to make sure the name isn't double-modified, the
5800 new name is applied using the :func:`_schema.conv` marker. We can
5801 use this explicitly as follows::
5804 m = MetaData(
5805 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5806 )
5807 t = Table(
5808 "t",
5809 m,
5810 Column("x", Integer),
5811 CheckConstraint("x > 5", name=conv("ck_t_x5")),
5812 )
5814 Where above, the :func:`_schema.conv` marker indicates that the constraint
5815 name here is final, and the name will render as ``"ck_t_x5"`` and not
5816 ``"ck_t_ck_t_x5"``
5818 .. seealso::
5820 :ref:`constraint_naming_conventions`
5822 """
5824 __slots__ = ()
5827# for backwards compatibility in case
5828# someone is re-implementing the
5829# _truncated_identifier() sequence in a custom
5830# compiler
5831_generated_label = _truncated_label
5832_anonymous_label_escape = re.compile(r"[%\(\) \$]+")
5835class _anonymous_label(_truncated_label):
5836 """A unicode subclass used to identify anonymously
5837 generated names."""
5839 __slots__ = ()
5841 @classmethod
5842 def safe_construct_with_key(
5843 cls, seed: int, body: str, sanitize_key: bool = False
5844 ) -> typing_Tuple[_anonymous_label, str]:
5845 # need to escape chars that interfere with format
5846 # strings in any case, issue #8724
5847 body = _anonymous_label_escape.sub("_", body)
5849 if sanitize_key:
5850 # sanitize_key is then an extra step used by BindParameter
5851 body = body.strip("_")
5853 key = f"{seed} {body.replace('%', '%%')}"
5854 label = _anonymous_label(f"%({key})s")
5855 return label, key
5857 @classmethod
5858 def safe_construct(
5859 cls, seed: int, body: str, sanitize_key: bool = False
5860 ) -> _anonymous_label:
5861 # need to escape chars that interfere with format
5862 # strings in any case, issue #8724
5863 body = _anonymous_label_escape.sub("_", body)
5865 if sanitize_key:
5866 # sanitize_key is then an extra step used by BindParameter
5867 body = body.strip("_")
5869 return _anonymous_label(f"%({seed} {body.replace('%', '%%')})s")
5871 def __add__(self, other: str) -> _anonymous_label:
5872 if "%" in other and not isinstance(other, _anonymous_label):
5873 other = str(other).replace("%", "%%")
5874 else:
5875 other = str(other)
5877 return _anonymous_label(
5878 quoted_name(
5879 str.__add__(self, other),
5880 self.quote,
5881 )
5882 )
5884 def __radd__(self, other: str) -> _anonymous_label:
5885 if "%" in other and not isinstance(other, _anonymous_label):
5886 other = str(other).replace("%", "%%")
5887 else:
5888 other = str(other)
5890 return _anonymous_label(
5891 quoted_name(
5892 str.__add__(other, self),
5893 self.quote,
5894 )
5895 )
5897 def apply_map(self, map_: Mapping[str, Any]) -> str:
5898 if self.quote is not None:
5899 # preserve quoting only if necessary
5900 return quoted_name(self % map_, self.quote)
5901 else:
5902 # else skip the constructor call
5903 return self % map_