Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# sql/elements.py
2# Copyright (C) 2005-2025 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7# mypy: allow-untyped-defs, allow-untyped-calls
9"""Core SQL expression elements, including :class:`_expression.ClauseElement`,
10:class:`_expression.ColumnElement`, and derived classes.
12"""
14from __future__ import annotations
16from decimal import Decimal
17from enum import Enum
18import itertools
19import operator
20import re
21import typing
22from typing import AbstractSet
23from typing import Any
24from typing import Callable
25from typing import cast
26from typing import Dict
27from typing import FrozenSet
28from typing import Generic
29from typing import Iterable
30from typing import Iterator
31from typing import List
32from typing import Literal
33from typing import Mapping
34from typing import Optional
35from typing import overload
36from typing import ParamSpec
37from typing import Sequence
38from typing import Set
39from typing import Tuple as typing_Tuple
40from typing import Type
41from typing import TYPE_CHECKING
42from typing import TypeVar
43from typing import Union
45from . import coercions
46from . import operators
47from . import roles
48from . import traversals
49from . import type_api
50from ._typing import has_schema_attr
51from ._typing import is_named_from_clause
52from ._typing import is_quoted_name
53from ._typing import is_tuple_type
54from .annotation import Annotated
55from .annotation import SupportsWrappingAnnotations
56from .base import _clone
57from .base import _expand_cloned
58from .base import _generative
59from .base import _NoArg
60from .base import Executable
61from .base import Generative
62from .base import HasMemoized
63from .base import Immutable
64from .base import NO_ARG
65from .base import SingletonConstant
66from .cache_key import MemoizedHasCacheKey
67from .cache_key import NO_CACHE
68from .coercions import _document_text_coercion # noqa
69from .operators import ColumnOperators
70from .operators import OperatorClass
71from .traversals import HasCopyInternals
72from .visitors import cloned_traverse
73from .visitors import ExternallyTraversible
74from .visitors import InternalTraversal
75from .visitors import traverse
76from .visitors import Visitable
77from .. import exc
78from .. import inspection
79from .. import util
80from ..util import HasMemoized_ro_memoized_attribute
81from ..util import TypingOnly
82from ..util.typing import Self
83from ..util.typing import TupleAny
84from ..util.typing import Unpack
87if typing.TYPE_CHECKING:
88 from ._typing import _ByArgument
89 from ._typing import _ColumnExpressionArgument
90 from ._typing import _ColumnExpressionOrStrLabelArgument
91 from ._typing import _DMLOnlyColumnArgument
92 from ._typing import _HasDialect
93 from ._typing import _InfoType
94 from ._typing import _PropagateAttrsType
95 from ._typing import _TypeEngineArgument
96 from .base import _EntityNamespace
97 from .base import ColumnSet
98 from .cache_key import _CacheKeyTraversalType
99 from .cache_key import CacheKey
100 from .compiler import Compiled
101 from .compiler import SQLCompiler
102 from .functions import FunctionElement
103 from .operators import OperatorType
104 from .schema import Column
105 from .schema import DefaultGenerator
106 from .schema import FetchedValue
107 from .schema import ForeignKey
108 from .selectable import _SelectIterable
109 from .selectable import FromClause
110 from .selectable import NamedFromClause
111 from .selectable import TextualSelect
112 from .sqltypes import TupleType
113 from .type_api import TypeEngine
114 from .visitors import _CloneCallableType
115 from .visitors import _TraverseInternalsType
116 from .visitors import anon_map
117 from ..engine import Connection
118 from ..engine import Dialect
119 from ..engine.interfaces import _CoreMultiExecuteParams
120 from ..engine.interfaces import CacheStats
121 from ..engine.interfaces import CompiledCacheType
122 from ..engine.interfaces import CoreExecuteOptionsParameter
123 from ..engine.interfaces import SchemaTranslateMapType
124 from ..engine.result import Result
127_NUMERIC = Union[float, Decimal]
128_NUMBER = Union[float, int, Decimal]
130_T = TypeVar("_T", bound="Any")
131_T_co = TypeVar("_T_co", bound=Any, covariant=True)
132_OPT = TypeVar("_OPT", bound="Any")
133_NT = TypeVar("_NT", bound="_NUMERIC")
135_NMT = TypeVar("_NMT", bound="_NUMBER")
138@overload
139def literal(
140 value: Any,
141 type_: _TypeEngineArgument[_T],
142 literal_execute: bool = False,
143) -> BindParameter[_T]: ...
146@overload
147def literal(
148 value: _T,
149 type_: None = None,
150 literal_execute: bool = False,
151) -> BindParameter[_T]: ...
154@overload
155def literal(
156 value: Any,
157 type_: Optional[_TypeEngineArgument[Any]] = None,
158 literal_execute: bool = False,
159) -> BindParameter[Any]: ...
162def literal(
163 value: Any,
164 type_: Optional[_TypeEngineArgument[Any]] = None,
165 literal_execute: bool = False,
166) -> BindParameter[Any]:
167 r"""Return a literal clause, bound to a bind parameter.
169 Literal clauses are created automatically when non-
170 :class:`_expression.ClauseElement` objects (such as strings, ints, dates,
171 etc.) are
172 used in a comparison operation with a :class:`_expression.ColumnElement`
173 subclass,
174 such as a :class:`~sqlalchemy.schema.Column` object. Use this function
175 to force the generation of a literal clause, which will be created as a
176 :class:`BindParameter` with a bound value.
178 :param value: the value to be bound. Can be any Python object supported by
179 the underlying DB-API, or is translatable via the given type argument.
181 :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine` which will
182 provide bind-parameter translation for this literal.
184 :param literal_execute: optional bool, when True, the SQL engine will
185 attempt to render the bound value directly in the SQL statement at
186 execution time rather than providing as a parameter value.
188 .. versionadded:: 2.0
190 """
191 return coercions.expect(
192 roles.LiteralValueRole,
193 value,
194 type_=type_,
195 literal_execute=literal_execute,
196 )
199def literal_column(
200 text: str, type_: Optional[_TypeEngineArgument[_T]] = None
201) -> ColumnClause[_T]:
202 r"""Produce a :class:`.ColumnClause` object that has the
203 :paramref:`_expression.column.is_literal` flag set to True.
205 :func:`_expression.literal_column` is similar to
206 :func:`_expression.column`, except that
207 it is more often used as a "standalone" column expression that renders
208 exactly as stated; while :func:`_expression.column`
209 stores a string name that
210 will be assumed to be part of a table and may be quoted as such,
211 :func:`_expression.literal_column` can be that,
212 or any other arbitrary column-oriented
213 expression.
215 :param text: the text of the expression; can be any SQL expression.
216 Quoting rules will not be applied. To specify a column-name expression
217 which should be subject to quoting rules, use the :func:`column`
218 function.
220 :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine`
221 object which will
222 provide result-set translation and additional expression semantics for
223 this column. If left as ``None`` the type will be :class:`.NullType`.
225 .. seealso::
227 :func:`_expression.column`
229 :func:`_expression.text`
231 :ref:`tutorial_select_arbitrary_text`
233 """
234 return ColumnClause(text, type_=type_, is_literal=True)
237class CompilerElement(Visitable):
238 """base class for SQL elements that can be compiled to produce a
239 SQL string.
241 .. versionadded:: 2.0
243 """
245 __slots__ = ()
246 __visit_name__ = "compiler_element"
248 supports_execution = False
250 stringify_dialect = "default"
252 @util.preload_module("sqlalchemy.engine.default")
253 @util.preload_module("sqlalchemy.engine.url")
254 def compile(
255 self,
256 bind: Optional[_HasDialect] = None,
257 dialect: Optional[Dialect] = None,
258 **kw: Any,
259 ) -> Compiled:
260 """Compile this SQL expression.
262 The return value is a :class:`~.Compiled` object.
263 Calling ``str()`` or ``unicode()`` on the returned value will yield a
264 string representation of the result. The
265 :class:`~.Compiled` object also can return a
266 dictionary of bind parameter names and values
267 using the ``params`` accessor.
269 :param bind: An :class:`.Connection` or :class:`.Engine` which
270 can provide a :class:`.Dialect` in order to generate a
271 :class:`.Compiled` object. If the ``bind`` and
272 ``dialect`` parameters are both omitted, a default SQL compiler
273 is used.
275 :param column_keys: Used for INSERT and UPDATE statements, a list of
276 column names which should be present in the VALUES clause of the
277 compiled statement. If ``None``, all columns from the target table
278 object are rendered.
280 :param dialect: A :class:`.Dialect` instance which can generate
281 a :class:`.Compiled` object. This argument takes precedence over
282 the ``bind`` argument.
284 :param compile_kwargs: optional dictionary of additional parameters
285 that will be passed through to the compiler within all "visit"
286 methods. This allows any custom flag to be passed through to
287 a custom compilation construct, for example. It is also used
288 for the case of passing the ``literal_binds`` flag through::
290 from sqlalchemy.sql import table, column, select
292 t = table("t", column("x"))
294 s = select(t).where(t.c.x == 5)
296 print(s.compile(compile_kwargs={"literal_binds": True}))
298 .. seealso::
300 :ref:`faq_sql_expression_string`
302 """
304 if dialect is None:
305 if bind:
306 dialect = bind.dialect
307 elif self.stringify_dialect == "default":
308 dialect = self._default_dialect()
309 else:
310 url = util.preloaded.engine_url
311 dialect = url.URL.create(
312 self.stringify_dialect
313 ).get_dialect()()
315 return self._compiler(dialect, **kw)
317 def _default_dialect(self):
318 default = util.preloaded.engine_default
319 return default.StrCompileDialect()
321 def _compiler(self, dialect: Dialect, **kw: Any) -> Compiled:
322 """Return a compiler appropriate for this ClauseElement, given a
323 Dialect."""
325 if TYPE_CHECKING:
326 assert isinstance(self, ClauseElement)
327 return dialect.statement_compiler(dialect, self, **kw)
329 def __str__(self) -> str:
330 return str(self.compile())
333@inspection._self_inspects
334class ClauseElement(
335 SupportsWrappingAnnotations,
336 MemoizedHasCacheKey,
337 HasCopyInternals,
338 ExternallyTraversible,
339 CompilerElement,
340):
341 """Base class for elements of a programmatically constructed SQL
342 expression.
344 """
346 __visit_name__ = "clause"
348 if TYPE_CHECKING:
350 @util.memoized_property
351 def _propagate_attrs(self) -> _PropagateAttrsType:
352 """like annotations, however these propagate outwards liberally
353 as SQL constructs are built, and are set up at construction time.
355 """
356 ...
358 else:
359 _propagate_attrs = util.EMPTY_DICT
361 @util.ro_memoized_property
362 def description(self) -> Optional[str]:
363 return None
365 _is_clone_of: Optional[Self] = None
367 is_clause_element = True
368 is_selectable = False
369 is_dml = False
370 _is_column_element = False
371 _is_keyed_column_element = False
372 _is_table = False
373 _gen_static_annotations_cache_key = False
374 _is_textual = False
375 _is_from_clause = False
376 _is_returns_rows = False
377 _is_text_clause = False
378 _is_from_container = False
379 _is_select_container = False
380 _is_select_base = False
381 _is_select_statement = False
382 _is_bind_parameter = False
383 _is_clause_list = False
384 _is_lambda_element = False
385 _is_singleton_constant = False
386 _is_immutable = False
387 _is_star = False
389 @property
390 def _order_by_label_element(self) -> Optional[Label[Any]]:
391 return None
393 _cache_key_traversal: _CacheKeyTraversalType = None
395 negation_clause: ColumnElement[bool]
397 if typing.TYPE_CHECKING:
399 def get_children(
400 self, *, omit_attrs: typing_Tuple[str, ...] = ..., **kw: Any
401 ) -> Iterable[ClauseElement]: ...
403 @util.ro_non_memoized_property
404 def _from_objects(self) -> List[FromClause]:
405 return []
407 def _set_propagate_attrs(self, values: Mapping[str, Any]) -> Self:
408 # usually, self._propagate_attrs is empty here. one case where it's
409 # not is a subquery against ORM select, that is then pulled as a
410 # property of an aliased class. should all be good
412 # assert not self._propagate_attrs
414 self._propagate_attrs = util.immutabledict(values)
415 return self
417 def _default_compiler(self) -> SQLCompiler:
418 dialect = self._default_dialect()
419 return dialect.statement_compiler(dialect, self) # type: ignore
421 def _clone(self, **kw: Any) -> Self:
422 """Create a shallow copy of this ClauseElement.
424 This method may be used by a generative API. Its also used as
425 part of the "deep" copy afforded by a traversal that combines
426 the _copy_internals() method.
428 """
430 skip = self._memoized_keys
431 c = self.__class__.__new__(self.__class__)
433 if skip:
434 # ensure this iteration remains atomic
435 c.__dict__ = {
436 k: v for k, v in self.__dict__.copy().items() if k not in skip
437 }
438 else:
439 c.__dict__ = self.__dict__.copy()
441 # this is a marker that helps to "equate" clauses to each other
442 # when a Select returns its list of FROM clauses. the cloning
443 # process leaves around a lot of remnants of the previous clause
444 # typically in the form of column expressions still attached to the
445 # old table.
446 cc = self._is_clone_of
447 c._is_clone_of = cc if cc is not None else self
448 return c
450 def _negate_in_binary(self, negated_op, original_op):
451 """a hook to allow the right side of a binary expression to respond
452 to a negation of the binary expression.
454 Used for the special case of expanding bind parameter with IN.
456 """
457 return self
459 def _with_binary_element_type(self, type_):
460 """in the context of binary expression, convert the type of this
461 object to the one given.
463 applies only to :class:`_expression.ColumnElement` classes.
465 """
466 return self
468 @property
469 def _constructor(self): # type: ignore[override]
470 """return the 'constructor' for this ClauseElement.
472 This is for the purposes for creating a new object of
473 this type. Usually, its just the element's __class__.
474 However, the "Annotated" version of the object overrides
475 to return the class of its proxied element.
477 """
478 return self.__class__
480 @HasMemoized.memoized_attribute
481 def _cloned_set(self):
482 """Return the set consisting all cloned ancestors of this
483 ClauseElement.
485 Includes this ClauseElement. This accessor tends to be used for
486 FromClause objects to identify 'equivalent' FROM clauses, regardless
487 of transformative operations.
489 """
490 s = util.column_set()
491 f: Optional[ClauseElement] = self
493 # note this creates a cycle, asserted in test_memusage. however,
494 # turning this into a plain @property adds tends of thousands of method
495 # calls to Core / ORM performance tests, so the small overhead
496 # introduced by the relatively small amount of short term cycles
497 # produced here is preferable
498 while f is not None:
499 s.add(f)
500 f = f._is_clone_of
501 return s
503 def _de_clone(self):
504 while self._is_clone_of is not None:
505 self = self._is_clone_of
506 return self
508 @util.ro_non_memoized_property
509 def entity_namespace(self) -> _EntityNamespace:
510 raise AttributeError(
511 "This SQL expression has no entity namespace "
512 "with which to filter from."
513 )
515 def __getstate__(self):
516 d = self.__dict__.copy()
517 d.pop("_is_clone_of", None)
518 d.pop("_generate_cache_key", None)
519 return d
521 def _execute_on_connection(
522 self,
523 connection: Connection,
524 distilled_params: _CoreMultiExecuteParams,
525 execution_options: CoreExecuteOptionsParameter,
526 ) -> Result[Unpack[TupleAny]]:
527 if self.supports_execution:
528 if TYPE_CHECKING:
529 assert isinstance(self, Executable)
530 return connection._execute_clauseelement(
531 self, distilled_params, execution_options
532 )
533 else:
534 raise exc.ObjectNotExecutableError(self)
536 def _execute_on_scalar(
537 self,
538 connection: Connection,
539 distilled_params: _CoreMultiExecuteParams,
540 execution_options: CoreExecuteOptionsParameter,
541 ) -> Any:
542 """an additional hook for subclasses to provide a different
543 implementation for connection.scalar() vs. connection.execute().
545 .. versionadded:: 2.0
547 """
548 return self._execute_on_connection(
549 connection, distilled_params, execution_options
550 ).scalar()
552 def _get_embedded_bindparams(self) -> Sequence[BindParameter[Any]]:
553 """Return the list of :class:`.BindParameter` objects embedded in the
554 object.
556 This accomplishes the same purpose as ``visitors.traverse()`` or
557 similar would provide, however by making use of the cache key
558 it takes advantage of memoization of the key to result in fewer
559 net method calls, assuming the statement is also going to be
560 executed.
562 """
564 key = self._generate_cache_key()
565 if key is None:
566 bindparams: List[BindParameter[Any]] = []
568 traverse(self, {}, {"bindparam": bindparams.append})
569 return bindparams
571 else:
572 return key.bindparams
574 def unique_params(
575 self,
576 __optionaldict: Optional[Dict[str, Any]] = None,
577 /,
578 **kwargs: Any,
579 ) -> Self:
580 """Return a copy with :func:`_expression.bindparam` elements
581 replaced.
583 Same functionality as :meth:`_expression.ClauseElement.params`,
584 except adds `unique=True`
585 to affected bind parameters so that multiple statements can be
586 used.
588 """
589 return self._replace_params(True, __optionaldict, kwargs)
591 def params(
592 self,
593 __optionaldict: Optional[Mapping[str, Any]] = None,
594 /,
595 **kwargs: Any,
596 ) -> Self:
597 """Return a copy with :func:`_expression.bindparam` elements
598 replaced.
600 Returns a copy of this ClauseElement with
601 :func:`_expression.bindparam`
602 elements replaced with values taken from the given dictionary::
604 >>> clause = column("x") + bindparam("foo")
605 >>> print(clause.compile().params)
606 {'foo':None}
607 >>> print(clause.params({"foo": 7}).compile().params)
608 {'foo':7}
610 """
611 return self._replace_params(False, __optionaldict, kwargs)
613 def _replace_params(
614 self,
615 unique: bool,
616 optionaldict: Optional[Mapping[str, Any]],
617 kwargs: Dict[str, Any],
618 ) -> Self:
619 if optionaldict:
620 kwargs.update(optionaldict)
622 def visit_bindparam(bind: BindParameter[Any]) -> None:
623 if bind.key in kwargs:
624 bind.value = kwargs[bind.key]
625 bind.required = False
626 if unique:
627 bind._convert_to_unique()
629 return cloned_traverse(
630 self,
631 {"maintain_key": True, "detect_subquery_cols": True},
632 {"bindparam": visit_bindparam},
633 )
635 def compare(self, other: ClauseElement, **kw: Any) -> bool:
636 r"""Compare this :class:`_expression.ClauseElement` to
637 the given :class:`_expression.ClauseElement`.
639 Subclasses should override the default behavior, which is a
640 straight identity comparison.
642 \**kw are arguments consumed by subclass ``compare()`` methods and
643 may be used to modify the criteria for comparison
644 (see :class:`_expression.ColumnElement`).
646 """
647 return traversals.compare(self, other, **kw)
649 def self_group(
650 self, against: Optional[OperatorType] = None
651 ) -> ClauseElement:
652 """Apply a 'grouping' to this :class:`_expression.ClauseElement`.
654 This method is overridden by subclasses to return a "grouping"
655 construct, i.e. parenthesis. In particular it's used by "binary"
656 expressions to provide a grouping around themselves when placed into a
657 larger expression, as well as by :func:`_expression.select`
658 constructs when placed into the FROM clause of another
659 :func:`_expression.select`. (Note that subqueries should be
660 normally created using the :meth:`_expression.Select.alias` method,
661 as many
662 platforms require nested SELECT statements to be named).
664 As expressions are composed together, the application of
665 :meth:`self_group` is automatic - end-user code should never
666 need to use this method directly. Note that SQLAlchemy's
667 clause constructs take operator precedence into account -
668 so parenthesis might not be needed, for example, in
669 an expression like ``x OR (y AND z)`` - AND takes precedence
670 over OR.
672 The base :meth:`self_group` method of
673 :class:`_expression.ClauseElement`
674 just returns self.
675 """
676 return self
678 def _ungroup(self) -> ClauseElement:
679 """Return this :class:`_expression.ClauseElement`
680 without any groupings.
681 """
683 return self
685 def _compile_w_cache(
686 self,
687 dialect: Dialect,
688 *,
689 compiled_cache: Optional[CompiledCacheType],
690 column_keys: List[str],
691 for_executemany: bool = False,
692 schema_translate_map: Optional[SchemaTranslateMapType] = None,
693 **kw: Any,
694 ) -> typing_Tuple[
695 Compiled, Optional[Sequence[BindParameter[Any]]], CacheStats
696 ]:
697 elem_cache_key: Optional[CacheKey]
699 if compiled_cache is not None and dialect._supports_statement_cache:
700 elem_cache_key = self._generate_cache_key()
701 else:
702 elem_cache_key = None
704 extracted_params: Optional[Sequence[BindParameter[Any]]]
705 if elem_cache_key is not None:
706 if TYPE_CHECKING:
707 assert compiled_cache is not None
709 cache_key, extracted_params = elem_cache_key
710 key = (
711 dialect,
712 cache_key,
713 tuple(column_keys),
714 bool(schema_translate_map),
715 for_executemany,
716 )
717 compiled_sql = compiled_cache.get(key)
719 if compiled_sql is None:
720 cache_hit = dialect.CACHE_MISS
721 compiled_sql = self._compiler(
722 dialect,
723 cache_key=elem_cache_key,
724 column_keys=column_keys,
725 for_executemany=for_executemany,
726 schema_translate_map=schema_translate_map,
727 **kw,
728 )
729 compiled_cache[key] = compiled_sql
730 else:
731 cache_hit = dialect.CACHE_HIT
732 else:
733 extracted_params = None
734 compiled_sql = self._compiler(
735 dialect,
736 cache_key=elem_cache_key,
737 column_keys=column_keys,
738 for_executemany=for_executemany,
739 schema_translate_map=schema_translate_map,
740 **kw,
741 )
743 if not dialect._supports_statement_cache:
744 cache_hit = dialect.NO_DIALECT_SUPPORT
745 elif compiled_cache is None:
746 cache_hit = dialect.CACHING_DISABLED
747 else:
748 cache_hit = dialect.NO_CACHE_KEY
750 return compiled_sql, extracted_params, cache_hit
752 def __invert__(self):
753 # undocumented element currently used by the ORM for
754 # relationship.contains()
755 if hasattr(self, "negation_clause"):
756 return self.negation_clause
757 else:
758 return self._negate()
760 def _negate(self) -> ClauseElement:
761 # TODO: this code is uncovered and in all likelihood is not included
762 # in any codepath. So this should raise NotImplementedError in 2.1
763 grouped = self.self_group(against=operators.inv)
764 assert isinstance(grouped, ColumnElement)
765 return UnaryExpression(grouped, operator=operators.inv)
767 def __bool__(self):
768 raise TypeError("Boolean value of this clause is not defined")
770 def __repr__(self):
771 friendly = self.description
772 if friendly is None:
773 return object.__repr__(self)
774 else:
775 return "<%s.%s at 0x%x; %s>" % (
776 self.__module__,
777 self.__class__.__name__,
778 id(self),
779 friendly,
780 )
783class DQLDMLClauseElement(ClauseElement):
784 """represents a :class:`.ClauseElement` that compiles to a DQL or DML
785 expression, not DDL.
787 .. versionadded:: 2.0
789 """
791 if typing.TYPE_CHECKING:
793 def _compiler(self, dialect: Dialect, **kw: Any) -> SQLCompiler:
794 """Return a compiler appropriate for this ClauseElement, given a
795 Dialect."""
796 ...
798 def compile( # noqa: A001
799 self,
800 bind: Optional[_HasDialect] = None,
801 dialect: Optional[Dialect] = None,
802 **kw: Any,
803 ) -> SQLCompiler: ...
806class CompilerColumnElement(
807 roles.DMLColumnRole,
808 roles.DDLConstraintColumnRole,
809 roles.ColumnsClauseRole,
810 CompilerElement,
811):
812 """A compiler-only column element used for ad-hoc string compilations.
814 .. versionadded:: 2.0
816 """
818 __slots__ = ()
820 _propagate_attrs = util.EMPTY_DICT
821 _is_collection_aggregate = False
822 _is_implicitly_boolean = False
824 def _with_binary_element_type(self, type_):
825 raise NotImplementedError()
827 def _gen_cache_key(self, anon_map, bindparams):
828 raise NotImplementedError()
830 @property
831 def _from_objects(self) -> List[FromClause]:
832 raise NotImplementedError()
835# SQLCoreOperations should be suiting the ExpressionElementRole
836# and ColumnsClauseRole. however the MRO issues become too elaborate
837# at the moment.
838class SQLCoreOperations(Generic[_T_co], ColumnOperators, TypingOnly):
839 __slots__ = ()
841 # annotations for comparison methods
842 # these are from operators->Operators / ColumnOperators,
843 # redefined with the specific types returned by ColumnElement hierarchies
844 if typing.TYPE_CHECKING:
846 @util.non_memoized_property
847 def _propagate_attrs(self) -> _PropagateAttrsType: ...
849 def operate(
850 self, op: OperatorType, *other: Any, **kwargs: Any
851 ) -> ColumnElement[Any]: ...
853 def reverse_operate(
854 self, op: OperatorType, other: Any, **kwargs: Any
855 ) -> ColumnElement[Any]: ...
857 @overload
858 def op(
859 self,
860 opstring: str,
861 precedence: int = ...,
862 is_comparison: bool = ...,
863 *,
864 return_type: _TypeEngineArgument[_OPT],
865 python_impl: Optional[Callable[..., Any]] = None,
866 operator_class: OperatorClass = ...,
867 visit_name: Optional[str] = ...,
868 ) -> Callable[[Any], BinaryExpression[_OPT]]: ...
870 @overload
871 def op(
872 self,
873 opstring: str,
874 precedence: int = ...,
875 is_comparison: bool = ...,
876 return_type: Optional[_TypeEngineArgument[Any]] = ...,
877 python_impl: Optional[Callable[..., Any]] = ...,
878 operator_class: OperatorClass = ...,
879 visit_name: Optional[str] = ...,
880 ) -> Callable[[Any], BinaryExpression[Any]]: ...
882 def op(
883 self,
884 opstring: str,
885 precedence: int = 0,
886 is_comparison: bool = False,
887 return_type: Optional[_TypeEngineArgument[Any]] = None,
888 python_impl: Optional[Callable[..., Any]] = None,
889 operator_class: OperatorClass = OperatorClass.BASE,
890 visit_name: Optional[str] = None,
891 ) -> Callable[[Any], BinaryExpression[Any]]: ...
893 def bool_op(
894 self,
895 opstring: str,
896 precedence: int = 0,
897 python_impl: Optional[Callable[..., Any]] = None,
898 ) -> Callable[[Any], BinaryExpression[bool]]: ...
900 def __and__(self, other: Any) -> BooleanClauseList: ...
902 def __or__(self, other: Any) -> BooleanClauseList: ...
904 def __invert__(self) -> ColumnElement[_T_co]: ...
906 def __lt__(self, other: Any) -> ColumnElement[bool]: ...
908 def __le__(self, other: Any) -> ColumnElement[bool]: ...
910 # declare also that this class has an hash method otherwise
911 # it may be assumed to be None by type checkers since the
912 # object defines __eq__ and python sets it to None in that case:
913 # https://docs.python.org/3/reference/datamodel.html#object.__hash__
914 def __hash__(self) -> int: ...
916 def __eq__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
917 ...
919 def __ne__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
920 ...
922 def is_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
924 def is_not_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
926 def __gt__(self, other: Any) -> ColumnElement[bool]: ...
928 def __ge__(self, other: Any) -> ColumnElement[bool]: ...
930 def __neg__(self) -> UnaryExpression[_T_co]: ...
932 def __contains__(self, other: Any) -> ColumnElement[bool]: ...
934 def __getitem__(self, index: Any) -> ColumnElement[Any]: ...
936 @overload
937 def __lshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
939 @overload
940 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
942 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
944 @overload
945 def __rlshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
947 @overload
948 def __rlshift__(self, other: Any) -> ColumnElement[Any]: ...
950 def __rlshift__(self, other: Any) -> ColumnElement[Any]: ...
952 @overload
953 def __rshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
955 @overload
956 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
958 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
960 @overload
961 def __rrshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
963 @overload
964 def __rrshift__(self, other: Any) -> ColumnElement[Any]: ...
966 def __rrshift__(self, other: Any) -> ColumnElement[Any]: ...
968 def __matmul__(self, other: Any) -> ColumnElement[Any]: ...
970 def __rmatmul__(self, other: Any) -> ColumnElement[Any]: ...
972 @overload
973 def concat(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
975 @overload
976 def concat(self, other: Any) -> ColumnElement[Any]: ...
978 def concat(self, other: Any) -> ColumnElement[Any]: ...
980 def like(
981 self, other: Any, escape: Optional[str] = None
982 ) -> BinaryExpression[bool]: ...
984 def ilike(
985 self, other: Any, escape: Optional[str] = None
986 ) -> BinaryExpression[bool]: ...
988 def bitwise_xor(self, other: Any) -> BinaryExpression[Any]: ...
990 def bitwise_or(self, other: Any) -> BinaryExpression[Any]: ...
992 def bitwise_and(self, other: Any) -> BinaryExpression[Any]: ...
994 def bitwise_not(self) -> UnaryExpression[_T_co]: ...
996 def bitwise_lshift(self, other: Any) -> BinaryExpression[Any]: ...
998 def bitwise_rshift(self, other: Any) -> BinaryExpression[Any]: ...
1000 def in_(
1001 self,
1002 other: Union[
1003 Iterable[Any], BindParameter[Any], roles.InElementRole
1004 ],
1005 ) -> BinaryExpression[bool]: ...
1007 def not_in(
1008 self,
1009 other: Union[
1010 Iterable[Any], BindParameter[Any], roles.InElementRole
1011 ],
1012 ) -> BinaryExpression[bool]: ...
1014 def notin_(
1015 self,
1016 other: Union[
1017 Iterable[Any], BindParameter[Any], roles.InElementRole
1018 ],
1019 ) -> BinaryExpression[bool]: ...
1021 def not_like(
1022 self, other: Any, escape: Optional[str] = None
1023 ) -> BinaryExpression[bool]: ...
1025 def notlike(
1026 self, other: Any, escape: Optional[str] = None
1027 ) -> BinaryExpression[bool]: ...
1029 def not_ilike(
1030 self, other: Any, escape: Optional[str] = None
1031 ) -> BinaryExpression[bool]: ...
1033 def notilike(
1034 self, other: Any, escape: Optional[str] = None
1035 ) -> BinaryExpression[bool]: ...
1037 def is_(self, other: Any) -> BinaryExpression[bool]: ...
1039 def is_not(self, other: Any) -> BinaryExpression[bool]: ...
1041 def isnot(self, other: Any) -> BinaryExpression[bool]: ...
1043 def startswith(
1044 self,
1045 other: Any,
1046 escape: Optional[str] = None,
1047 autoescape: bool = False,
1048 ) -> ColumnElement[bool]: ...
1050 def istartswith(
1051 self,
1052 other: Any,
1053 escape: Optional[str] = None,
1054 autoescape: bool = False,
1055 ) -> ColumnElement[bool]: ...
1057 def endswith(
1058 self,
1059 other: Any,
1060 escape: Optional[str] = None,
1061 autoescape: bool = False,
1062 ) -> ColumnElement[bool]: ...
1064 def iendswith(
1065 self,
1066 other: Any,
1067 escape: Optional[str] = None,
1068 autoescape: bool = False,
1069 ) -> ColumnElement[bool]: ...
1071 def contains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1073 def icontains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1075 def match(self, other: Any, **kwargs: Any) -> ColumnElement[bool]: ...
1077 def regexp_match(
1078 self, pattern: Any, flags: Optional[str] = None
1079 ) -> ColumnElement[bool]: ...
1081 def regexp_replace(
1082 self, pattern: Any, replacement: Any, flags: Optional[str] = None
1083 ) -> ColumnElement[str]: ...
1085 def desc(self) -> UnaryExpression[_T_co]: ...
1087 def asc(self) -> UnaryExpression[_T_co]: ...
1089 def nulls_first(self) -> UnaryExpression[_T_co]: ...
1091 def nullsfirst(self) -> UnaryExpression[_T_co]: ...
1093 def nulls_last(self) -> UnaryExpression[_T_co]: ...
1095 def nullslast(self) -> UnaryExpression[_T_co]: ...
1097 def collate(self, collation: str) -> CollationClause: ...
1099 def between(
1100 self, cleft: Any, cright: Any, symmetric: bool = False
1101 ) -> BinaryExpression[bool]: ...
1103 def distinct(self: _SQO[_T_co]) -> UnaryExpression[_T_co]: ...
1105 def any_(self) -> CollectionAggregate[Any]: ...
1107 def all_(self) -> CollectionAggregate[Any]: ...
1109 # numeric overloads. These need more tweaking
1110 # in particular they all need to have a variant for Optiona[_T]
1111 # because Optional only applies to the data side, not the expression
1112 # side
1114 @overload
1115 def __add__(
1116 self: _SQO[_NMT],
1117 other: Any,
1118 ) -> ColumnElement[_NMT]: ...
1120 @overload
1121 def __add__(
1122 self: _SQO[str],
1123 other: Any,
1124 ) -> ColumnElement[str]: ...
1126 @overload
1127 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1129 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1131 @overload
1132 def __radd__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1134 @overload
1135 def __radd__(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
1137 def __radd__(self, other: Any) -> ColumnElement[Any]: ...
1139 @overload
1140 def __sub__(
1141 self: _SQO[_NMT],
1142 other: Any,
1143 ) -> ColumnElement[_NMT]: ...
1145 @overload
1146 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1148 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1150 @overload
1151 def __rsub__(
1152 self: _SQO[_NMT],
1153 other: Any,
1154 ) -> ColumnElement[_NMT]: ...
1156 @overload
1157 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1159 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1161 @overload
1162 def __mul__(
1163 self: _SQO[_NMT],
1164 other: Any,
1165 ) -> ColumnElement[_NMT]: ...
1167 @overload
1168 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1170 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1172 @overload
1173 def __rmul__(
1174 self: _SQO[_NMT],
1175 other: Any,
1176 ) -> ColumnElement[_NMT]: ...
1178 @overload
1179 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1181 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1183 @overload
1184 def __mod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1186 @overload
1187 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1189 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1191 @overload
1192 def __rmod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1194 @overload
1195 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1197 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1199 @overload
1200 def __truediv__(
1201 self: _SQO[int], other: Any
1202 ) -> ColumnElement[_NUMERIC]: ...
1204 @overload
1205 def __truediv__(self: _SQO[_NT], other: Any) -> ColumnElement[_NT]: ...
1207 @overload
1208 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1210 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1212 @overload
1213 def __rtruediv__(
1214 self: _SQO[_NMT], other: Any
1215 ) -> ColumnElement[_NUMERIC]: ...
1217 @overload
1218 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1220 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1222 @overload
1223 def __floordiv__(
1224 self: _SQO[_NMT], other: Any
1225 ) -> ColumnElement[_NMT]: ...
1227 @overload
1228 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1230 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1232 @overload
1233 def __rfloordiv__(
1234 self: _SQO[_NMT], other: Any
1235 ) -> ColumnElement[_NMT]: ...
1237 @overload
1238 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1240 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1243class SQLColumnExpression(
1244 SQLCoreOperations[_T_co], roles.ExpressionElementRole[_T_co], TypingOnly
1245):
1246 """A type that may be used to indicate any SQL column element or object
1247 that acts in place of one.
1249 :class:`.SQLColumnExpression` is a base of
1250 :class:`.ColumnElement`, as well as within the bases of ORM elements
1251 such as :class:`.InstrumentedAttribute`, and may be used in :pep:`484`
1252 typing to indicate arguments or return values that should behave
1253 as column expressions.
1255 .. versionadded:: 2.0.0b4
1258 """
1260 __slots__ = ()
1263_SQO = SQLCoreOperations
1266class ColumnElement(
1267 roles.ColumnArgumentOrKeyRole,
1268 roles.StatementOptionRole,
1269 roles.WhereHavingRole,
1270 roles.BinaryElementRole[_T],
1271 roles.OrderByRole,
1272 roles.ColumnsClauseRole,
1273 roles.LimitOffsetRole,
1274 roles.DMLColumnRole,
1275 roles.DDLConstraintColumnRole,
1276 roles.DDLExpressionRole,
1277 SQLColumnExpression[_T],
1278 DQLDMLClauseElement,
1279):
1280 """Represent a column-oriented SQL expression suitable for usage in the
1281 "columns" clause, WHERE clause etc. of a statement.
1283 While the most familiar kind of :class:`_expression.ColumnElement` is the
1284 :class:`_schema.Column` object, :class:`_expression.ColumnElement`
1285 serves as the basis
1286 for any unit that may be present in a SQL expression, including
1287 the expressions themselves, SQL functions, bound parameters,
1288 literal expressions, keywords such as ``NULL``, etc.
1289 :class:`_expression.ColumnElement`
1290 is the ultimate base class for all such elements.
1292 A wide variety of SQLAlchemy Core functions work at the SQL expression
1293 level, and are intended to accept instances of
1294 :class:`_expression.ColumnElement` as
1295 arguments. These functions will typically document that they accept a
1296 "SQL expression" as an argument. What this means in terms of SQLAlchemy
1297 usually refers to an input which is either already in the form of a
1298 :class:`_expression.ColumnElement` object,
1299 or a value which can be **coerced** into
1300 one. The coercion rules followed by most, but not all, SQLAlchemy Core
1301 functions with regards to SQL expressions are as follows:
1303 * a literal Python value, such as a string, integer or floating
1304 point value, boolean, datetime, ``Decimal`` object, or virtually
1305 any other Python object, will be coerced into a "literal bound
1306 value". This generally means that a :func:`.bindparam` will be
1307 produced featuring the given value embedded into the construct; the
1308 resulting :class:`.BindParameter` object is an instance of
1309 :class:`_expression.ColumnElement`.
1310 The Python value will ultimately be sent
1311 to the DBAPI at execution time as a parameterized argument to the
1312 ``execute()`` or ``executemany()`` methods, after SQLAlchemy
1313 type-specific converters (e.g. those provided by any associated
1314 :class:`.TypeEngine` objects) are applied to the value.
1316 * any special object value, typically ORM-level constructs, which
1317 feature an accessor called ``__clause_element__()``. The Core
1318 expression system looks for this method when an object of otherwise
1319 unknown type is passed to a function that is looking to coerce the
1320 argument into a :class:`_expression.ColumnElement` and sometimes a
1321 :class:`_expression.SelectBase` expression.
1322 It is used within the ORM to
1323 convert from ORM-specific objects like mapped classes and
1324 mapped attributes into Core expression objects.
1326 * The Python ``None`` value is typically interpreted as ``NULL``,
1327 which in SQLAlchemy Core produces an instance of :func:`.null`.
1329 A :class:`_expression.ColumnElement` provides the ability to generate new
1330 :class:`_expression.ColumnElement`
1331 objects using Python expressions. This means that Python operators
1332 such as ``==``, ``!=`` and ``<`` are overloaded to mimic SQL operations,
1333 and allow the instantiation of further :class:`_expression.ColumnElement`
1334 instances
1335 which are composed from other, more fundamental
1336 :class:`_expression.ColumnElement`
1337 objects. For example, two :class:`.ColumnClause` objects can be added
1338 together with the addition operator ``+`` to produce
1339 a :class:`.BinaryExpression`.
1340 Both :class:`.ColumnClause` and :class:`.BinaryExpression` are subclasses
1341 of :class:`_expression.ColumnElement`:
1343 .. sourcecode:: pycon+sql
1345 >>> from sqlalchemy.sql import column
1346 >>> column("a") + column("b")
1347 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
1348 >>> print(column("a") + column("b"))
1349 {printsql}a + b
1351 .. seealso::
1353 :class:`_schema.Column`
1355 :func:`_expression.column`
1357 """
1359 __visit_name__ = "column_element"
1361 primary_key: bool = False
1362 _is_clone_of: Optional[ColumnElement[_T]]
1363 _is_column_element = True
1364 _insert_sentinel: bool = False
1365 _omit_from_statements = False
1366 _is_collection_aggregate = False
1368 foreign_keys: AbstractSet[ForeignKey] = frozenset()
1370 @util.memoized_property
1371 def _proxies(self) -> List[ColumnElement[Any]]:
1372 return []
1374 @util.non_memoized_property
1375 def _tq_label(self) -> Optional[str]:
1376 """The named label that can be used to target
1377 this column in a result set in a "table qualified" context.
1379 This label is almost always the label used when
1380 rendering <expr> AS <label> in a SELECT statement when using
1381 the LABEL_STYLE_TABLENAME_PLUS_COL label style, which is what the
1382 legacy ORM ``Query`` object uses as well.
1384 For a regular Column bound to a Table, this is typically the label
1385 <tablename>_<columnname>. For other constructs, different rules
1386 may apply, such as anonymized labels and others.
1388 .. versionchanged:: 1.4.21 renamed from ``._label``
1390 """
1391 return None
1393 key: Optional[str] = None
1394 """The 'key' that in some circumstances refers to this object in a
1395 Python namespace.
1397 This typically refers to the "key" of the column as present in the
1398 ``.c`` collection of a selectable, e.g. ``sometable.c["somekey"]`` would
1399 return a :class:`_schema.Column` with a ``.key`` of "somekey".
1401 """
1403 @HasMemoized.memoized_attribute
1404 def _tq_key_label(self) -> Optional[str]:
1405 """A label-based version of 'key' that in some circumstances refers
1406 to this object in a Python namespace.
1409 _tq_key_label comes into play when a select() statement is constructed
1410 with apply_labels(); in this case, all Column objects in the ``.c``
1411 collection are rendered as <tablename>_<columnname> in SQL; this is
1412 essentially the value of ._label. But to locate those columns in the
1413 ``.c`` collection, the name is along the lines of <tablename>_<key>;
1414 that's the typical value of .key_label.
1416 .. versionchanged:: 1.4.21 renamed from ``._key_label``
1418 """
1419 return self._proxy_key
1421 @property
1422 def _key_label(self) -> Optional[str]:
1423 """legacy; renamed to _tq_key_label"""
1424 return self._tq_key_label
1426 @property
1427 def _label(self) -> Optional[str]:
1428 """legacy; renamed to _tq_label"""
1429 return self._tq_label
1431 @property
1432 def _non_anon_label(self) -> Optional[str]:
1433 """the 'name' that naturally applies this element when rendered in
1434 SQL.
1436 Concretely, this is the "name" of a column or a label in a
1437 SELECT statement; ``<columnname>`` and ``<labelname>`` below:
1439 .. sourcecode:: sql
1441 SELECT <columnmame> FROM table
1443 SELECT column AS <labelname> FROM table
1445 Above, the two names noted will be what's present in the DBAPI
1446 ``cursor.description`` as the names.
1448 If this attribute returns ``None``, it means that the SQL element as
1449 written does not have a 100% fully predictable "name" that would appear
1450 in the ``cursor.description``. Examples include SQL functions, CAST
1451 functions, etc. While such things do return names in
1452 ``cursor.description``, they are only predictable on a
1453 database-specific basis; e.g. an expression like ``MAX(table.col)`` may
1454 appear as the string ``max`` on one database (like PostgreSQL) or may
1455 appear as the whole expression ``max(table.col)`` on SQLite.
1457 The default implementation looks for a ``.name`` attribute on the
1458 object, as has been the precedent established in SQLAlchemy for many
1459 years. An exception is made on the ``FunctionElement`` subclass
1460 so that the return value is always ``None``.
1462 .. versionadded:: 1.4.21
1466 """
1467 return getattr(self, "name", None)
1469 _render_label_in_columns_clause = True
1470 """A flag used by select._columns_plus_names that helps to determine
1471 we are actually going to render in terms of "SELECT <col> AS <label>".
1472 This flag can be returned as False for some Column objects that want
1473 to be rendered as simple "SELECT <col>"; typically columns that don't have
1474 any parent table and are named the same as what the label would be
1475 in any case.
1477 """
1479 _allow_label_resolve = True
1480 """A flag that can be flipped to prevent a column from being resolvable
1481 by string label name.
1483 The joined eager loader strategy in the ORM uses this, for example.
1485 """
1487 _is_implicitly_boolean = False
1489 _alt_names: Sequence[str] = ()
1491 if TYPE_CHECKING:
1493 def _ungroup(self) -> ColumnElement[_T]: ...
1495 @overload
1496 def self_group(self, against: None = None) -> ColumnElement[_T]: ...
1498 @overload
1499 def self_group(
1500 self, against: Optional[OperatorType] = None
1501 ) -> ColumnElement[Any]: ...
1503 def self_group(
1504 self, against: Optional[OperatorType] = None
1505 ) -> ColumnElement[Any]:
1506 if (
1507 against in (operators.and_, operators.or_, operators._asbool)
1508 and self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity
1509 ):
1510 return AsBoolean(self, operators.is_true, operators.is_false)
1511 elif against in (operators.any_op, operators.all_op):
1512 return Grouping(self)
1513 else:
1514 return self
1516 @overload
1517 def _negate(self: ColumnElement[bool]) -> ColumnElement[bool]: ...
1519 @overload
1520 def _negate(self: ColumnElement[_T]) -> ColumnElement[_T]: ...
1522 def _negate(self) -> ColumnElement[Any]:
1523 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
1524 return AsBoolean(self, operators.is_false, operators.is_true)
1525 else:
1526 grouped = self.self_group(against=operators.inv)
1527 assert isinstance(grouped, ColumnElement)
1528 return UnaryExpression(
1529 grouped,
1530 operator=operators.inv,
1531 )
1533 type: TypeEngine[_T]
1535 if not TYPE_CHECKING:
1537 @util.memoized_property
1538 def type(self) -> TypeEngine[_T]: # noqa: A001
1539 # used for delayed setup of
1540 # type_api
1541 return type_api.NULLTYPE
1543 @HasMemoized.memoized_attribute
1544 def comparator(self) -> TypeEngine.Comparator[_T]:
1545 try:
1546 comparator_factory = self.type.comparator_factory
1547 except AttributeError as err:
1548 raise TypeError(
1549 "Object %r associated with '.type' attribute "
1550 "is not a TypeEngine class or object" % self.type
1551 ) from err
1552 else:
1553 return comparator_factory(self)
1555 def __setstate__(self, state):
1556 self.__dict__.update(state)
1558 def __getattr__(self, key: str) -> Any:
1559 try:
1560 return getattr(self.comparator, key)
1561 except AttributeError as err:
1562 raise AttributeError(
1563 "Neither %r object nor %r object has an attribute %r"
1564 % (
1565 type(self).__name__,
1566 type(self.comparator).__name__,
1567 key,
1568 )
1569 ) from err
1571 def operate(
1572 self,
1573 op: operators.OperatorType,
1574 *other: Any,
1575 **kwargs: Any,
1576 ) -> ColumnElement[Any]:
1577 return op(self.comparator, *other, **kwargs) # type: ignore[no-any-return] # noqa: E501
1579 def reverse_operate(
1580 self, op: operators.OperatorType, other: Any, **kwargs: Any
1581 ) -> ColumnElement[Any]:
1582 return op(other, self.comparator, **kwargs) # type: ignore[no-any-return] # noqa: E501
1584 def _bind_param(
1585 self,
1586 operator: operators.OperatorType,
1587 obj: Any,
1588 type_: Optional[TypeEngine[_T]] = None,
1589 expanding: bool = False,
1590 ) -> BindParameter[_T]:
1591 return BindParameter(
1592 None,
1593 obj,
1594 _compared_to_operator=operator,
1595 type_=type_,
1596 _compared_to_type=self.type,
1597 unique=True,
1598 expanding=expanding,
1599 )
1601 @property
1602 def expression(self) -> ColumnElement[Any]:
1603 """Return a column expression.
1605 Part of the inspection interface; returns self.
1607 """
1608 return self
1610 @property
1611 def _select_iterable(self) -> _SelectIterable:
1612 return (self,)
1614 @util.memoized_property
1615 def base_columns(self) -> FrozenSet[ColumnElement[Any]]:
1616 return frozenset(c for c in self.proxy_set if not c._proxies)
1618 @util.memoized_property
1619 def proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1620 """set of all columns we are proxying
1622 as of 2.0 this is explicitly deannotated columns. previously it was
1623 effectively deannotated columns but wasn't enforced. annotated
1624 columns should basically not go into sets if at all possible because
1625 their hashing behavior is very non-performant.
1627 """
1628 return frozenset([self._deannotate()]).union(
1629 itertools.chain(*[c.proxy_set for c in self._proxies])
1630 )
1632 @util.memoized_property
1633 def _expanded_proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1634 return frozenset(_expand_cloned(self.proxy_set))
1636 def _uncached_proxy_list(self) -> List[ColumnElement[Any]]:
1637 """An 'uncached' version of proxy set.
1639 This list includes annotated columns which perform very poorly in
1640 set operations.
1642 """
1644 return [self] + list(
1645 itertools.chain(*[c._uncached_proxy_list() for c in self._proxies])
1646 )
1648 def shares_lineage(self, othercolumn: ColumnElement[Any]) -> bool:
1649 """Return True if the given :class:`_expression.ColumnElement`
1650 has a common ancestor to this :class:`_expression.ColumnElement`."""
1652 return bool(self.proxy_set.intersection(othercolumn.proxy_set))
1654 def _compare_name_for_result(self, other: ColumnElement[Any]) -> bool:
1655 """Return True if the given column element compares to this one
1656 when targeting within a result row."""
1658 return (
1659 hasattr(other, "name")
1660 and hasattr(self, "name")
1661 and other.name == self.name
1662 )
1664 @HasMemoized.memoized_attribute
1665 def _proxy_key(self) -> Optional[str]:
1666 if self._annotations and "proxy_key" in self._annotations:
1667 return cast(str, self._annotations["proxy_key"])
1669 name = self.key
1670 if not name:
1671 # there's a bit of a seeming contradiction which is that the
1672 # "_non_anon_label" of a column can in fact be an
1673 # "_anonymous_label"; this is when it's on a column that is
1674 # proxying for an anonymous expression in a subquery.
1675 name = self._non_anon_label
1677 if isinstance(name, _anonymous_label):
1678 return None
1679 else:
1680 return name
1682 @HasMemoized.memoized_attribute
1683 def _expression_label(self) -> Optional[str]:
1684 """a suggested label to use in the case that the column has no name,
1685 which should be used if possible as the explicit 'AS <label>'
1686 where this expression would normally have an anon label.
1688 this is essentially mostly what _proxy_key does except it returns
1689 None if the column has a normal name that can be used.
1691 """
1693 if getattr(self, "name", None) is not None:
1694 return None
1695 elif self._annotations and "proxy_key" in self._annotations:
1696 return cast(str, self._annotations["proxy_key"])
1697 else:
1698 return None
1700 def _make_proxy(
1701 self,
1702 selectable: FromClause,
1703 *,
1704 primary_key: ColumnSet,
1705 foreign_keys: Set[KeyedColumnElement[Any]],
1706 name: Optional[str] = None,
1707 key: Optional[str] = None,
1708 name_is_truncatable: bool = False,
1709 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
1710 **kw: Any,
1711 ) -> typing_Tuple[str, ColumnClause[_T]]:
1712 """Create a new :class:`_expression.ColumnElement` representing this
1713 :class:`_expression.ColumnElement` as it appears in the select list of
1714 a descending selectable.
1716 """
1717 if name is None:
1718 name = self._anon_name_label
1719 if key is None:
1720 key = self._proxy_key
1721 else:
1722 key = name
1724 assert key is not None
1726 co: ColumnClause[_T] = ColumnClause(
1727 (
1728 coercions.expect(roles.TruncatedLabelRole, name)
1729 if name_is_truncatable
1730 else name
1731 ),
1732 type_=getattr(self, "type", None),
1733 _selectable=selectable,
1734 )
1736 co._propagate_attrs = selectable._propagate_attrs
1737 if compound_select_cols:
1738 co._proxies = list(compound_select_cols)
1739 else:
1740 co._proxies = [self]
1741 if selectable._is_clone_of is not None:
1742 co._is_clone_of = selectable._is_clone_of.columns.get(key)
1743 return key, co
1745 def cast(self, type_: _TypeEngineArgument[_OPT]) -> Cast[_OPT]:
1746 """Produce a type cast, i.e. ``CAST(<expression> AS <type>)``.
1748 This is a shortcut to the :func:`_expression.cast` function.
1750 .. seealso::
1752 :ref:`tutorial_casts`
1754 :func:`_expression.cast`
1756 :func:`_expression.type_coerce`
1758 """
1759 return Cast(self, type_)
1761 def label(self, name: Optional[str]) -> Label[_T]:
1762 """Produce a column label, i.e. ``<columnname> AS <name>``.
1764 This is a shortcut to the :func:`_expression.label` function.
1766 If 'name' is ``None``, an anonymous label name will be generated.
1768 """
1769 return Label(name, self, self.type)
1771 def _anon_label(
1772 self, seed: Optional[str], add_hash: Optional[int] = None
1773 ) -> _anonymous_label:
1774 while self._is_clone_of is not None:
1775 self = self._is_clone_of
1777 # as of 1.4 anonymous label for ColumnElement uses hash(), not id(),
1778 # as the identifier, because a column and its annotated version are
1779 # the same thing in a SQL statement
1780 hash_value = hash(self)
1782 if add_hash:
1783 # this path is used for disambiguating anon labels that would
1784 # otherwise be the same name for the same element repeated.
1785 # an additional numeric value is factored in for each label.
1787 # shift hash(self) (which is id(self), typically 8 byte integer)
1788 # 16 bits leftward. fill extra add_hash on right
1789 assert add_hash < (2 << 15)
1790 assert seed
1791 hash_value = (hash_value << 16) | add_hash
1793 # extra underscore is added for labels with extra hash
1794 # values, to isolate the "deduped anon" namespace from the
1795 # regular namespace. eliminates chance of these
1796 # manufactured hash values overlapping with regular ones for some
1797 # undefined python interpreter
1798 seed = seed + "_"
1800 if isinstance(seed, _anonymous_label):
1801 # NOTE: the space after the hash is required
1802 return _anonymous_label(f"{seed}%({hash_value} )s")
1804 return _anonymous_label.safe_construct(hash_value, seed or "anon")
1806 @util.memoized_property
1807 def _anon_name_label(self) -> str:
1808 """Provides a constant 'anonymous label' for this ColumnElement.
1810 This is a label() expression which will be named at compile time.
1811 The same label() is returned each time ``anon_label`` is called so
1812 that expressions can reference ``anon_label`` multiple times,
1813 producing the same label name at compile time.
1815 The compiler uses this function automatically at compile time
1816 for expressions that are known to be 'unnamed' like binary
1817 expressions and function calls.
1819 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1820 public and is renamed to _anon_name_label. anon_name exists
1821 for backwards compat
1823 """
1824 name = getattr(self, "name", None)
1825 return self._anon_label(name)
1827 @util.memoized_property
1828 def _anon_key_label(self) -> _anonymous_label:
1829 """Provides a constant 'anonymous key label' for this ColumnElement.
1831 Compare to ``anon_label``, except that the "key" of the column,
1832 if available, is used to generate the label.
1834 This is used when a deduplicating key is placed into the columns
1835 collection of a selectable.
1837 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1838 public and is renamed to _anon_key_label. anon_key_label exists
1839 for backwards compat
1841 """
1842 return self._anon_label(self._proxy_key)
1844 @property
1845 @util.deprecated(
1846 "1.4",
1847 "The :attr:`_expression.ColumnElement.anon_label` attribute is now "
1848 "private, and the public accessor is deprecated.",
1849 )
1850 def anon_label(self) -> str:
1851 return self._anon_name_label
1853 @property
1854 @util.deprecated(
1855 "1.4",
1856 "The :attr:`_expression.ColumnElement.anon_key_label` attribute is "
1857 "now private, and the public accessor is deprecated.",
1858 )
1859 def anon_key_label(self) -> str:
1860 return self._anon_key_label
1862 def _dedupe_anon_label_idx(self, idx: int) -> str:
1863 """label to apply to a column that is anon labeled, but repeated
1864 in the SELECT, so that we have to make an "extra anon" label that
1865 disambiguates it from the previous appearance.
1867 these labels come out like "foo_bar_id__1" and have double underscores
1868 in them.
1870 """
1871 label = getattr(self, "name", None)
1873 # current convention is that if the element doesn't have a
1874 # ".name" (usually because it is not NamedColumn), we try to
1875 # use a "table qualified" form for the "dedupe anon" label,
1876 # based on the notion that a label like
1877 # "CAST(casttest.v1 AS DECIMAL) AS casttest_v1__1" looks better than
1878 # "CAST(casttest.v1 AS DECIMAL) AS anon__1"
1880 if label is None:
1881 return self._dedupe_anon_tq_label_idx(idx)
1882 else:
1883 return self._anon_label(label, add_hash=idx)
1885 @util.memoized_property
1886 def _anon_tq_label(self) -> _anonymous_label:
1887 return self._anon_label(getattr(self, "_tq_label", None))
1889 @util.memoized_property
1890 def _anon_tq_key_label(self) -> _anonymous_label:
1891 return self._anon_label(getattr(self, "_tq_key_label", None))
1893 def _dedupe_anon_tq_label_idx(self, idx: int) -> _anonymous_label:
1894 label = getattr(self, "_tq_label", None) or "anon"
1896 return self._anon_label(label, add_hash=idx)
1899class KeyedColumnElement(ColumnElement[_T]):
1900 """ColumnElement where ``.key`` is non-None."""
1902 _is_keyed_column_element = True
1904 key: str
1907class WrapsColumnExpression(ColumnElement[_T]):
1908 """Mixin that defines a :class:`_expression.ColumnElement`
1909 as a wrapper with special
1910 labeling behavior for an expression that already has a name.
1912 .. versionadded:: 1.4
1914 .. seealso::
1916 :ref:`change_4449`
1919 """
1921 @property
1922 def wrapped_column_expression(self) -> ColumnElement[_T]:
1923 raise NotImplementedError()
1925 @util.non_memoized_property
1926 def _tq_label(self) -> Optional[str]:
1927 wce = self.wrapped_column_expression
1928 if hasattr(wce, "_tq_label"):
1929 return wce._tq_label
1930 else:
1931 return None
1933 @property
1934 def _label(self) -> Optional[str]:
1935 return self._tq_label
1937 @property
1938 def _non_anon_label(self) -> Optional[str]:
1939 return None
1941 @util.non_memoized_property
1942 def _anon_name_label(self) -> str:
1943 wce = self.wrapped_column_expression
1945 # this logic tries to get the WrappedColumnExpression to render
1946 # with "<expr> AS <name>", where "<name>" is the natural name
1947 # within the expression itself. e.g. "CAST(table.foo) AS foo".
1948 if not wce._is_text_clause:
1949 nal = wce._non_anon_label
1950 if nal:
1951 return nal
1952 elif hasattr(wce, "_anon_name_label"):
1953 return wce._anon_name_label
1954 return super()._anon_name_label
1956 def _dedupe_anon_label_idx(self, idx: int) -> str:
1957 wce = self.wrapped_column_expression
1958 nal = wce._non_anon_label
1959 if nal:
1960 return self._anon_label(nal + "_")
1961 else:
1962 return self._dedupe_anon_tq_label_idx(idx)
1964 @property
1965 def _proxy_key(self):
1966 wce = self.wrapped_column_expression
1968 if not wce._is_text_clause:
1969 return wce._proxy_key
1970 return super()._proxy_key
1973class DMLTargetCopy(roles.InElementRole, KeyedColumnElement[_T]):
1974 """Refer to another column's VALUES or SET expression in an INSERT or
1975 UPDATE statement.
1977 See the public-facing :func:`_sql.from_dml_column` constructor for
1978 background.
1980 .. versionadded:: 2.1
1983 """
1985 def __init__(self, column: _DMLOnlyColumnArgument[_T]):
1986 self.column = coercions.expect(roles.ColumnArgumentRole, column)
1987 self.type = self.column.type
1989 __visit_name__ = "dmltargetcopy"
1991 _traverse_internals: _TraverseInternalsType = [
1992 ("column", InternalTraversal.dp_clauseelement),
1993 ]
1996class BindParameter(roles.InElementRole, KeyedColumnElement[_T]):
1997 r"""Represent a "bound expression".
1999 :class:`.BindParameter` is invoked explicitly using the
2000 :func:`.bindparam` function, as in::
2002 from sqlalchemy import bindparam
2004 stmt = select(users_table).where(
2005 users_table.c.name == bindparam("username")
2006 )
2008 Detailed discussion of how :class:`.BindParameter` is used is
2009 at :func:`.bindparam`.
2011 .. seealso::
2013 :func:`.bindparam`
2015 """
2017 __visit_name__ = "bindparam"
2019 _traverse_internals: _TraverseInternalsType = [
2020 ("key", InternalTraversal.dp_anon_name),
2021 ("type", InternalTraversal.dp_type),
2022 ("callable", InternalTraversal.dp_plain_dict),
2023 ("value", InternalTraversal.dp_plain_obj),
2024 ("literal_execute", InternalTraversal.dp_boolean),
2025 ]
2027 key: str
2028 _anon_map_key: Optional[str] = None
2029 type: TypeEngine[_T]
2030 value: Optional[_T]
2032 _is_crud = False
2033 _is_bind_parameter = True
2035 # bindparam implements its own _gen_cache_key() method however
2036 # we check subclasses for this flag, else no cache key is generated
2037 inherit_cache = True
2039 def __init__(
2040 self,
2041 key: Optional[str],
2042 value: Any = _NoArg.NO_ARG,
2043 type_: Optional[_TypeEngineArgument[_T]] = None,
2044 unique: bool = False,
2045 required: Union[bool, Literal[_NoArg.NO_ARG]] = _NoArg.NO_ARG,
2046 quote: Optional[bool] = None,
2047 callable_: Optional[Callable[[], Any]] = None,
2048 expanding: bool = False,
2049 isoutparam: bool = False,
2050 literal_execute: bool = False,
2051 _compared_to_operator: Optional[OperatorType] = None,
2052 _compared_to_type: Optional[TypeEngine[Any]] = None,
2053 _is_crud: bool = False,
2054 ):
2055 if required is _NoArg.NO_ARG:
2056 required = value is _NoArg.NO_ARG and callable_ is None
2057 if value is _NoArg.NO_ARG:
2058 value = None
2060 if quote is not None:
2061 key = quoted_name.construct(key, quote)
2063 if unique:
2064 self.key, self._anon_map_key = (
2065 _anonymous_label.safe_construct_with_key(
2066 id(self),
2067 (
2068 key
2069 if key is not None
2070 and not isinstance(key, _anonymous_label)
2071 else "param"
2072 ),
2073 sanitize_key=True,
2074 )
2075 )
2076 elif key:
2077 self.key = key
2078 else:
2079 self.key, self._anon_map_key = (
2080 _anonymous_label.safe_construct_with_key(id(self), "param")
2081 )
2083 # identifying key that won't change across
2084 # clones, used to identify the bind's logical
2085 # identity
2086 self._identifying_key = self.key
2088 # key that was passed in the first place, used to
2089 # generate new keys
2090 self._orig_key = key or "param"
2092 self.unique = unique
2093 self.value = value
2094 self.callable = callable_
2095 self.isoutparam = isoutparam
2096 self.required = required
2098 # indicate an "expanding" parameter; the compiler sets this
2099 # automatically in the compiler _render_in_expr_w_bindparam method
2100 # for an IN expression
2101 self.expanding = expanding
2103 # this is another hint to help w/ expanding and is typically
2104 # set in the compiler _render_in_expr_w_bindparam method for an
2105 # IN expression
2106 self.expand_op = None
2108 self.literal_execute = literal_execute
2109 if _is_crud:
2110 self._is_crud = True
2112 if type_ is None:
2113 if expanding:
2114 if value:
2115 check_value = value[0]
2116 else:
2117 check_value = type_api._NO_VALUE_IN_LIST
2118 else:
2119 check_value = value
2120 if _compared_to_type is not None:
2121 self.type = _compared_to_type.coerce_compared_value(
2122 _compared_to_operator, check_value
2123 )
2124 else:
2125 self.type = type_api._resolve_value_to_type(check_value)
2126 elif isinstance(type_, type):
2127 self.type = type_()
2128 elif is_tuple_type(type_):
2129 if value:
2130 if expanding:
2131 check_value = value[0]
2132 else:
2133 check_value = value
2134 cast("BindParameter[TupleAny]", self).type = (
2135 type_._resolve_values_to_types(check_value)
2136 )
2137 else:
2138 cast("BindParameter[TupleAny]", self).type = type_
2139 else:
2140 self.type = type_
2142 def _with_value(self, value, maintain_key=False, required=NO_ARG):
2143 """Return a copy of this :class:`.BindParameter` with the given value
2144 set.
2145 """
2146 cloned = self._clone(maintain_key=maintain_key)
2147 cloned.value = value
2148 cloned.callable = None
2149 cloned.required = required if required is not NO_ARG else self.required
2150 if cloned.type is type_api.NULLTYPE:
2151 cloned.type = type_api._resolve_value_to_type(value)
2152 return cloned
2154 @property
2155 def effective_value(self) -> Optional[_T]:
2156 """Return the value of this bound parameter,
2157 taking into account if the ``callable`` parameter
2158 was set.
2160 The ``callable`` value will be evaluated
2161 and returned if present, else ``value``.
2163 """
2164 if self.callable:
2165 # TODO: set up protocol for bind parameter callable
2166 return self.callable() # type: ignore
2167 else:
2168 return self.value
2170 def render_literal_execute(self) -> Self:
2171 """Produce a copy of this bound parameter that will enable the
2172 :paramref:`_sql.BindParameter.literal_execute` flag.
2174 The :paramref:`_sql.BindParameter.literal_execute` flag will
2175 have the effect of the parameter rendered in the compiled SQL
2176 string using ``[POSTCOMPILE]`` form, which is a special form that
2177 is converted to be a rendering of the literal value of the parameter
2178 at SQL execution time. The rationale is to support caching
2179 of SQL statement strings that can embed per-statement literal values,
2180 such as LIMIT and OFFSET parameters, in the final SQL string that
2181 is passed to the DBAPI. Dialects in particular may want to use
2182 this method within custom compilation schemes.
2184 .. versionadded:: 1.4.5
2186 .. seealso::
2188 :ref:`engine_thirdparty_caching`
2190 """
2191 c: Self = ClauseElement._clone(self)
2192 c.literal_execute = True
2193 return c
2195 def _negate_in_binary(self, negated_op, original_op):
2196 if self.expand_op is original_op:
2197 bind = self._clone()
2198 bind.expand_op = negated_op
2199 return bind
2200 else:
2201 return self
2203 def _with_binary_element_type(self, type_: TypeEngine[Any]) -> Self:
2204 c: Self = ClauseElement._clone(self)
2205 c.type = type_
2206 return c
2208 def _clone(self, maintain_key: bool = False, **kw: Any) -> Self:
2209 c: Self = ClauseElement._clone(self, **kw)
2210 # ensure all the BindParameter objects stay in cloned set.
2211 # in #7823, we changed "clone" so that a clone only keeps a reference
2212 # to the "original" element, since for column correspondence, that's
2213 # all we need. However, for BindParam, _cloned_set is used by
2214 # the "cache key bind match" lookup, which means if any of those
2215 # interim BindParameter objects became part of a cache key in the
2216 # cache, we need it. So here, make sure all clones keep carrying
2217 # forward.
2218 c._cloned_set.update(self._cloned_set)
2219 if not maintain_key and self.unique:
2220 c.key, c._anon_map_key = _anonymous_label.safe_construct_with_key(
2221 id(c), c._orig_key or "param", sanitize_key=True
2222 )
2223 return c
2225 def _gen_cache_key(self, anon_map, bindparams):
2226 _gen_cache_ok = self.__class__.__dict__.get("inherit_cache", False)
2228 if not _gen_cache_ok:
2229 if anon_map is not None:
2230 anon_map[NO_CACHE] = True
2231 return None
2233 id_, found = anon_map.get_anon(self)
2234 if found:
2235 return (id_, self.__class__)
2237 if bindparams is not None:
2238 bindparams.append(self)
2240 return (
2241 id_,
2242 self.__class__,
2243 self.type._static_cache_key,
2244 (
2245 anon_map[self._anon_map_key]
2246 if self._anon_map_key is not None
2247 else self.key
2248 ),
2249 self.literal_execute,
2250 )
2252 def _convert_to_unique(self):
2253 if not self.unique:
2254 self.unique = True
2255 self.key, self._anon_map_key = (
2256 _anonymous_label.safe_construct_with_key(
2257 id(self), self._orig_key or "param", sanitize_key=True
2258 )
2259 )
2261 def __getstate__(self):
2262 """execute a deferred value for serialization purposes."""
2264 d = self.__dict__.copy()
2265 v = self.value
2266 if self.callable:
2267 v = self.callable()
2268 d["callable"] = None
2269 d["value"] = v
2270 return d
2272 def __setstate__(self, state):
2273 if state.get("unique", False):
2274 anon_and_key = _anonymous_label.safe_construct_with_key(
2275 id(self), state.get("_orig_key", "param"), sanitize_key=True
2276 )
2277 state["key"], state["_anon_map_key"] = anon_and_key
2278 self.__dict__.update(state)
2280 def __repr__(self):
2281 return "%s(%r, %r, type_=%r)" % (
2282 self.__class__.__name__,
2283 self.key,
2284 self.value,
2285 self.type,
2286 )
2289class TypeClause(DQLDMLClauseElement):
2290 """Handle a type keyword in a SQL statement.
2292 Used by the ``Case`` statement.
2294 """
2296 __visit_name__ = "typeclause"
2298 _traverse_internals: _TraverseInternalsType = [
2299 ("type", InternalTraversal.dp_type)
2300 ]
2301 type: TypeEngine[Any]
2303 def __init__(self, type_: TypeEngine[Any]):
2304 self.type = type_
2307class TextClause(
2308 roles.DDLConstraintColumnRole,
2309 roles.DDLExpressionRole,
2310 roles.StatementOptionRole,
2311 roles.WhereHavingRole,
2312 roles.OrderByRole,
2313 roles.FromClauseRole,
2314 roles.SelectStatementRole,
2315 roles.InElementRole,
2316 Generative,
2317 Executable,
2318 DQLDMLClauseElement,
2319 roles.BinaryElementRole[Any],
2320 inspection.Inspectable["TextClause"],
2321):
2322 """Represent a literal SQL text fragment.
2324 E.g.::
2326 from sqlalchemy import text
2328 t = text("SELECT * FROM users")
2329 result = connection.execute(t)
2331 The :class:`_expression.TextClause` construct is produced using the
2332 :func:`_expression.text`
2333 function; see that function for full documentation.
2335 .. seealso::
2337 :func:`_expression.text`
2339 """
2341 __visit_name__ = "textclause"
2343 _traverse_internals: _TraverseInternalsType = [
2344 ("_bindparams", InternalTraversal.dp_string_clauseelement_dict),
2345 ("text", InternalTraversal.dp_string),
2346 ] + Executable._executable_traverse_internals
2348 _is_text_clause = True
2350 _is_textual = True
2352 _bind_params_regex = re.compile(r"(?<![:\w\x5c]):(\w+)(?!:)", re.UNICODE)
2353 _is_implicitly_boolean = False
2355 _render_label_in_columns_clause = False
2357 _omit_from_statements = False
2359 _is_collection_aggregate = False
2361 @property
2362 def _hide_froms(self) -> Iterable[FromClause]:
2363 return ()
2365 def __and__(self, other):
2366 # support use in select.where(), query.filter()
2367 return and_(self, other)
2369 @property
2370 def _select_iterable(self) -> _SelectIterable:
2371 return (self,)
2373 # help in those cases where text() is
2374 # interpreted in a column expression situation
2375 key: Optional[str] = None
2376 _label: Optional[str] = None
2378 _allow_label_resolve = False
2380 @property
2381 def _is_star(self): # type: ignore[override]
2382 return self.text == "*"
2384 def __init__(self, text: str):
2385 self._bindparams: Dict[str, BindParameter[Any]] = {}
2387 def repl(m):
2388 self._bindparams[m.group(1)] = BindParameter(m.group(1))
2389 return ":%s" % m.group(1)
2391 # scan the string and search for bind parameter names, add them
2392 # to the list of bindparams
2393 self.text = self._bind_params_regex.sub(repl, text)
2395 @_generative
2396 def bindparams(
2397 self,
2398 *binds: BindParameter[Any],
2399 **names_to_values: Any,
2400 ) -> Self:
2401 """Establish the values and/or types of bound parameters within
2402 this :class:`_expression.TextClause` construct.
2404 Given a text construct such as::
2406 from sqlalchemy import text
2408 stmt = text(
2409 "SELECT id, name FROM user WHERE name=:name AND timestamp=:timestamp"
2410 )
2412 the :meth:`_expression.TextClause.bindparams`
2413 method can be used to establish
2414 the initial value of ``:name`` and ``:timestamp``,
2415 using simple keyword arguments::
2417 stmt = stmt.bindparams(
2418 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2419 )
2421 Where above, new :class:`.BindParameter` objects
2422 will be generated with the names ``name`` and ``timestamp``, and
2423 values of ``jack`` and ``datetime.datetime(2012, 10, 8, 15, 12, 5)``,
2424 respectively. The types will be
2425 inferred from the values given, in this case :class:`.String` and
2426 :class:`.DateTime`.
2428 When specific typing behavior is needed, the positional ``*binds``
2429 argument can be used in which to specify :func:`.bindparam` constructs
2430 directly. These constructs must include at least the ``key``
2431 argument, then an optional value and type::
2433 from sqlalchemy import bindparam
2435 stmt = stmt.bindparams(
2436 bindparam("name", value="jack", type_=String),
2437 bindparam("timestamp", type_=DateTime),
2438 )
2440 Above, we specified the type of :class:`.DateTime` for the
2441 ``timestamp`` bind, and the type of :class:`.String` for the ``name``
2442 bind. In the case of ``name`` we also set the default value of
2443 ``"jack"``.
2445 Additional bound parameters can be supplied at statement execution
2446 time, e.g.::
2448 result = connection.execute(
2449 stmt, timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2450 )
2452 The :meth:`_expression.TextClause.bindparams`
2453 method can be called repeatedly,
2454 where it will re-use existing :class:`.BindParameter` objects to add
2455 new information. For example, we can call
2456 :meth:`_expression.TextClause.bindparams`
2457 first with typing information, and a
2458 second time with value information, and it will be combined::
2460 stmt = text(
2461 "SELECT id, name FROM user WHERE name=:name "
2462 "AND timestamp=:timestamp"
2463 )
2464 stmt = stmt.bindparams(
2465 bindparam("name", type_=String), bindparam("timestamp", type_=DateTime)
2466 )
2467 stmt = stmt.bindparams(
2468 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2469 )
2471 The :meth:`_expression.TextClause.bindparams`
2472 method also supports the concept of
2473 **unique** bound parameters. These are parameters that are
2474 "uniquified" on name at statement compilation time, so that multiple
2475 :func:`_expression.text`
2476 constructs may be combined together without the names
2477 conflicting. To use this feature, specify the
2478 :paramref:`.BindParameter.unique` flag on each :func:`.bindparam`
2479 object::
2481 stmt1 = text("select id from table where name=:name").bindparams(
2482 bindparam("name", value="name1", unique=True)
2483 )
2484 stmt2 = text("select id from table where name=:name").bindparams(
2485 bindparam("name", value="name2", unique=True)
2486 )
2488 union = union_all(stmt1.columns(column("id")), stmt2.columns(column("id")))
2490 The above statement will render as:
2492 .. sourcecode:: sql
2494 select id from table where name=:name_1
2495 UNION ALL select id from table where name=:name_2
2497 """ # noqa: E501
2498 self._bindparams = new_params = self._bindparams.copy()
2500 for bind in binds:
2501 try:
2502 # the regex used for text() currently will not match
2503 # a unique/anonymous key in any case, so use the _orig_key
2504 # so that a text() construct can support unique parameters
2505 existing = new_params[bind._orig_key]
2506 except KeyError as err:
2507 raise exc.ArgumentError(
2508 "This text() construct doesn't define a "
2509 "bound parameter named %r" % bind._orig_key
2510 ) from err
2511 else:
2512 new_params[existing._orig_key] = bind
2514 for key, value in names_to_values.items():
2515 try:
2516 existing = new_params[key]
2517 except KeyError as err:
2518 raise exc.ArgumentError(
2519 "This text() construct doesn't define a "
2520 "bound parameter named %r" % key
2521 ) from err
2522 else:
2523 new_params[key] = existing._with_value(value, required=False)
2524 return self
2526 @util.preload_module("sqlalchemy.sql.selectable")
2527 def columns(
2528 self,
2529 *cols: _ColumnExpressionArgument[Any],
2530 **types: _TypeEngineArgument[Any],
2531 ) -> TextualSelect:
2532 r"""Turn this :class:`_expression.TextClause` object into a
2533 :class:`_expression.TextualSelect`
2534 object that serves the same role as a SELECT
2535 statement.
2537 The :class:`_expression.TextualSelect` is part of the
2538 :class:`_expression.SelectBase`
2539 hierarchy and can be embedded into another statement by using the
2540 :meth:`_expression.TextualSelect.subquery` method to produce a
2541 :class:`.Subquery`
2542 object, which can then be SELECTed from.
2544 This function essentially bridges the gap between an entirely
2545 textual SELECT statement and the SQL expression language concept
2546 of a "selectable"::
2548 from sqlalchemy.sql import column, text
2550 stmt = text("SELECT id, name FROM some_table")
2551 stmt = stmt.columns(column("id"), column("name")).subquery("st")
2553 stmt = (
2554 select(mytable)
2555 .select_from(mytable.join(stmt, mytable.c.name == stmt.c.name))
2556 .where(stmt.c.id > 5)
2557 )
2559 Above, we pass a series of :func:`_expression.column` elements to the
2560 :meth:`_expression.TextClause.columns` method positionally. These
2561 :func:`_expression.column`
2562 elements now become first class elements upon the
2563 :attr:`_expression.TextualSelect.selected_columns` column collection,
2564 which then
2565 become part of the :attr:`.Subquery.c` collection after
2566 :meth:`_expression.TextualSelect.subquery` is invoked.
2568 The column expressions we pass to
2569 :meth:`_expression.TextClause.columns` may
2570 also be typed; when we do so, these :class:`.TypeEngine` objects become
2571 the effective return type of the column, so that SQLAlchemy's
2572 result-set-processing systems may be used on the return values.
2573 This is often needed for types such as date or boolean types, as well
2574 as for unicode processing on some dialect configurations::
2576 stmt = text("SELECT id, name, timestamp FROM some_table")
2577 stmt = stmt.columns(
2578 column("id", Integer),
2579 column("name", Unicode),
2580 column("timestamp", DateTime),
2581 )
2583 for id, name, timestamp in connection.execute(stmt):
2584 print(id, name, timestamp)
2586 As a shortcut to the above syntax, keyword arguments referring to
2587 types alone may be used, if only type conversion is needed::
2589 stmt = text("SELECT id, name, timestamp FROM some_table")
2590 stmt = stmt.columns(id=Integer, name=Unicode, timestamp=DateTime)
2592 for id, name, timestamp in connection.execute(stmt):
2593 print(id, name, timestamp)
2595 The positional form of :meth:`_expression.TextClause.columns`
2596 also provides the
2597 unique feature of **positional column targeting**, which is
2598 particularly useful when using the ORM with complex textual queries. If
2599 we specify the columns from our model to
2600 :meth:`_expression.TextClause.columns`,
2601 the result set will match to those columns positionally, meaning the
2602 name or origin of the column in the textual SQL doesn't matter::
2604 stmt = text(
2605 "SELECT users.id, addresses.id, users.id, "
2606 "users.name, addresses.email_address AS email "
2607 "FROM users JOIN addresses ON users.id=addresses.user_id "
2608 "WHERE users.id = 1"
2609 ).columns(
2610 User.id,
2611 Address.id,
2612 Address.user_id,
2613 User.name,
2614 Address.email_address,
2615 )
2617 query = (
2618 session.query(User)
2619 .from_statement(stmt)
2620 .options(contains_eager(User.addresses))
2621 )
2623 The :meth:`_expression.TextClause.columns` method provides a direct
2624 route to calling :meth:`_expression.FromClause.subquery` as well as
2625 :meth:`_expression.SelectBase.cte`
2626 against a textual SELECT statement::
2628 stmt = stmt.columns(id=Integer, name=String).cte("st")
2630 stmt = select(sometable).where(sometable.c.id == stmt.c.id)
2632 :param \*cols: A series of :class:`_expression.ColumnElement` objects,
2633 typically
2634 :class:`_schema.Column` objects from a :class:`_schema.Table`
2635 or ORM level
2636 column-mapped attributes, representing a set of columns that this
2637 textual string will SELECT from.
2639 :param \**types: A mapping of string names to :class:`.TypeEngine`
2640 type objects indicating the datatypes to use for names that are
2641 SELECTed from the textual string. Prefer to use the ``*cols``
2642 argument as it also indicates positional ordering.
2644 """
2645 selectable = util.preloaded.sql_selectable
2647 input_cols: List[NamedColumn[Any]] = [
2648 coercions.expect(roles.LabeledColumnExprRole, col) for col in cols
2649 ]
2651 positional_input_cols = [
2652 (
2653 ColumnClause(col.key, types.pop(col.key))
2654 if col.key in types
2655 else col
2656 )
2657 for col in input_cols
2658 ]
2659 keyed_input_cols: List[NamedColumn[Any]] = [
2660 ColumnClause(key, type_) for key, type_ in types.items()
2661 ]
2663 elem = selectable.TextualSelect.__new__(selectable.TextualSelect)
2664 elem._init(
2665 self,
2666 positional_input_cols + keyed_input_cols,
2667 positional=bool(positional_input_cols) and not keyed_input_cols,
2668 )
2669 return elem
2671 @property
2672 def type(self) -> TypeEngine[Any]:
2673 return type_api.NULLTYPE
2675 @property
2676 def comparator(self):
2677 # TODO: this seems wrong, it seems like we might not
2678 # be using this method.
2679 return self.type.comparator_factory(self) # type: ignore
2681 def self_group(
2682 self, against: Optional[OperatorType] = None
2683 ) -> Union[Self, Grouping[Any]]:
2684 if against is operators.in_op:
2685 return Grouping(self)
2686 else:
2687 return self
2690class Null(SingletonConstant, roles.ConstExprRole[None], ColumnElement[None]):
2691 """Represent the NULL keyword in a SQL statement.
2693 :class:`.Null` is accessed as a constant via the
2694 :func:`.null` function.
2696 """
2698 __visit_name__ = "null"
2700 _traverse_internals: _TraverseInternalsType = []
2701 _singleton: Null
2703 if not TYPE_CHECKING:
2705 @util.memoized_property
2706 def type(self) -> TypeEngine[_T]: # noqa: A001
2707 return type_api.NULLTYPE
2709 @classmethod
2710 def _instance(cls) -> Null:
2711 """Return a constant :class:`.Null` construct."""
2713 return Null._singleton
2716Null._create_singleton()
2719class False_(
2720 SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]
2721):
2722 """Represent the ``false`` keyword, or equivalent, in a SQL statement.
2724 :class:`.False_` is accessed as a constant via the
2725 :func:`.false` function.
2727 """
2729 __visit_name__ = "false"
2730 _traverse_internals: _TraverseInternalsType = []
2731 _singleton: False_
2733 if not TYPE_CHECKING:
2735 @util.memoized_property
2736 def type(self) -> TypeEngine[_T]: # noqa: A001
2737 return type_api.BOOLEANTYPE
2739 def _negate(self) -> True_:
2740 return True_._singleton
2742 @classmethod
2743 def _instance(cls) -> False_:
2744 return False_._singleton
2747False_._create_singleton()
2750class True_(SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]):
2751 """Represent the ``true`` keyword, or equivalent, in a SQL statement.
2753 :class:`.True_` is accessed as a constant via the
2754 :func:`.true` function.
2756 """
2758 __visit_name__ = "true"
2760 _traverse_internals: _TraverseInternalsType = []
2761 _singleton: True_
2763 if not TYPE_CHECKING:
2765 @util.memoized_property
2766 def type(self) -> TypeEngine[_T]: # noqa: A001
2767 return type_api.BOOLEANTYPE
2769 def _negate(self) -> False_:
2770 return False_._singleton
2772 @classmethod
2773 def _ifnone(
2774 cls, other: Optional[ColumnElement[Any]]
2775 ) -> ColumnElement[Any]:
2776 if other is None:
2777 return cls._instance()
2778 else:
2779 return other
2781 @classmethod
2782 def _instance(cls) -> True_:
2783 return True_._singleton
2786True_._create_singleton()
2789class ElementList(DQLDMLClauseElement):
2790 """Describe a list of clauses that will be space separated.
2792 This is a minimal version of :class:`.ClauseList` which is used by
2793 the :class:`.HasSyntaxExtension` class. It does not do any coercions
2794 so should be used internally only.
2796 .. versionadded:: 2.1
2798 """
2800 __visit_name__ = "element_list"
2802 _traverse_internals: _TraverseInternalsType = [
2803 ("clauses", InternalTraversal.dp_clauseelement_tuple),
2804 ]
2806 clauses: typing_Tuple[ClauseElement, ...]
2808 def __init__(self, clauses: Sequence[ClauseElement]):
2809 self.clauses = tuple(clauses)
2812class OrderByList(
2813 roles.OrderByRole,
2814 operators.OrderingOperators,
2815 DQLDMLClauseElement,
2816):
2817 """Describe a list of clauses that will be comma separated to nest
2818 within an ORDER BY.
2820 .. versionadded:: 2.1
2822 """
2824 __visit_name__ = "order_by_list"
2826 _traverse_internals: _TraverseInternalsType = [
2827 ("clauses", InternalTraversal.dp_clauseelement_tuple),
2828 ]
2830 clauses: List[ColumnElement[Any]]
2832 def __init__(
2833 self,
2834 clauses: Iterable[Union[OrderByList, _ColumnExpressionArgument[Any]]],
2835 ):
2836 text_converter_role: Type[roles.SQLRole] = roles.ByOfRole
2837 self._text_converter_role = text_converter_role
2839 self.clauses = [
2840 coercions.expect(
2841 text_converter_role, clause, apply_propagate_attrs=self
2842 )
2843 for clause in clauses
2844 ]
2846 def __iter__(self) -> Iterator[ColumnElement[Any]]:
2847 return iter(self.clauses)
2849 def __len__(self) -> int:
2850 return len(self.clauses)
2852 @property
2853 def _select_iterable(self) -> _SelectIterable:
2854 return itertools.chain.from_iterable(
2855 [elem._select_iterable for elem in self.clauses]
2856 )
2858 @util.ro_non_memoized_property
2859 def _from_objects(self) -> List[FromClause]:
2860 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
2862 def self_group(
2863 self, against: Optional[OperatorType] = None
2864 ) -> Union[Self, Grouping[Any]]:
2865 return self
2867 def desc(self) -> OrderByList:
2868 return OrderByList([e.desc() for e in self.clauses])
2870 def asc(self) -> OrderByList:
2871 return OrderByList([e.asc() for e in self.clauses])
2873 def nulls_first(self) -> OrderByList:
2874 return OrderByList([e.nulls_first() for e in self.clauses])
2876 def nulls_last(self) -> OrderByList:
2877 return OrderByList([e.nulls_last() for e in self.clauses])
2880class ClauseList(
2881 roles.InElementRole,
2882 roles.OrderByRole,
2883 roles.ColumnsClauseRole,
2884 roles.DMLColumnRole,
2885 DQLDMLClauseElement,
2886):
2887 """Describe a list of clauses, separated by an operator.
2889 By default, is comma-separated, such as a column listing.
2891 """
2893 __visit_name__ = "clauselist"
2895 # Used by ORM context.py to identify ClauseList objects in legacy
2896 # composite attribute queries (see test_query_cols_legacy test)
2897 _is_clause_list = True
2899 _traverse_internals: _TraverseInternalsType = [
2900 ("clauses", InternalTraversal.dp_clauseelement_list),
2901 ("operator", InternalTraversal.dp_operator),
2902 ]
2904 clauses: List[ColumnElement[Any]]
2906 def __init__(
2907 self,
2908 *clauses: _ColumnExpressionArgument[Any],
2909 operator: OperatorType = operators.comma_op,
2910 group: bool = True,
2911 group_contents: bool = True,
2912 _literal_as_text_role: Type[roles.SQLRole] = roles.WhereHavingRole,
2913 ):
2914 self.operator = operator
2915 self.group = group
2916 self.group_contents = group_contents
2917 clauses_iterator: Iterable[_ColumnExpressionArgument[Any]] = clauses
2918 text_converter_role: Type[roles.SQLRole] = _literal_as_text_role
2919 self._text_converter_role = text_converter_role
2921 if self.group_contents:
2922 self.clauses = [
2923 coercions.expect(
2924 text_converter_role, clause, apply_propagate_attrs=self
2925 ).self_group(against=self.operator)
2926 for clause in clauses_iterator
2927 ]
2928 else:
2929 self.clauses = [
2930 coercions.expect(
2931 text_converter_role, clause, apply_propagate_attrs=self
2932 )
2933 for clause in clauses_iterator
2934 ]
2935 self._is_implicitly_boolean = operators.is_boolean(self.operator)
2937 @classmethod
2938 def _construct_raw(
2939 cls,
2940 operator: OperatorType,
2941 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
2942 ) -> ClauseList:
2943 self = cls.__new__(cls)
2944 self.clauses = list(clauses) if clauses else []
2945 self.group = True
2946 self.operator = operator
2947 self.group_contents = True
2948 self._is_implicitly_boolean = False
2949 return self
2951 def __iter__(self) -> Iterator[ColumnElement[Any]]:
2952 return iter(self.clauses)
2954 def __len__(self) -> int:
2955 return len(self.clauses)
2957 @property
2958 def _select_iterable(self) -> _SelectIterable:
2959 return itertools.chain.from_iterable(
2960 [elem._select_iterable for elem in self.clauses]
2961 )
2963 def append(self, clause):
2964 if self.group_contents:
2965 self.clauses.append(
2966 coercions.expect(self._text_converter_role, clause).self_group(
2967 against=self.operator
2968 )
2969 )
2970 else:
2971 self.clauses.append(
2972 coercions.expect(self._text_converter_role, clause)
2973 )
2975 @util.ro_non_memoized_property
2976 def _from_objects(self) -> List[FromClause]:
2977 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
2979 def self_group(
2980 self, against: Optional[OperatorType] = None
2981 ) -> Union[Self, Grouping[Any]]:
2982 if self.group and operators.is_precedent(self.operator, against):
2983 return Grouping(self)
2984 else:
2985 return self
2988class OperatorExpression(ColumnElement[_T]):
2989 """base for expressions that contain an operator and operands
2991 .. versionadded:: 2.0
2993 """
2995 operator: OperatorType
2996 type: TypeEngine[_T]
2998 group: bool = True
3000 @property
3001 def is_comparison(self):
3002 return operators.is_comparison(self.operator)
3004 def self_group(
3005 self, against: Optional[OperatorType] = None
3006 ) -> Union[Self, Grouping[_T]]:
3007 if (
3008 self.group
3009 and operators.is_precedent(self.operator, against)
3010 or (
3011 # a negate against a non-boolean operator
3012 # doesn't make too much sense but we should
3013 # group for that
3014 against is operators.inv
3015 and not operators.is_boolean(self.operator)
3016 )
3017 ):
3018 return Grouping(self)
3019 else:
3020 return self
3022 @property
3023 def _flattened_operator_clauses(
3024 self,
3025 ) -> typing_Tuple[ColumnElement[Any], ...]:
3026 raise NotImplementedError()
3028 @classmethod
3029 def _construct_for_op(
3030 cls,
3031 left: ColumnElement[Any],
3032 right: ColumnElement[Any],
3033 op: OperatorType,
3034 *,
3035 type_: TypeEngine[_T],
3036 negate: Optional[OperatorType] = None,
3037 modifiers: Optional[Mapping[str, Any]] = None,
3038 ) -> OperatorExpression[_T]:
3039 if operators.is_associative(op):
3040 assert (
3041 negate is None
3042 ), f"negate not supported for associative operator {op}"
3044 multi = False
3045 if getattr(
3046 left, "operator", None
3047 ) is op and type_._compare_type_affinity(left.type):
3048 multi = True
3049 left_flattened = left._flattened_operator_clauses
3050 else:
3051 left_flattened = (left,)
3053 if getattr(
3054 right, "operator", None
3055 ) is op and type_._compare_type_affinity(right.type):
3056 multi = True
3057 right_flattened = right._flattened_operator_clauses
3058 else:
3059 right_flattened = (right,)
3061 if multi:
3062 return ExpressionClauseList._construct_for_list(
3063 op,
3064 type_,
3065 *(left_flattened + right_flattened),
3066 )
3068 if right._is_collection_aggregate:
3069 negate = None
3071 return BinaryExpression(
3072 left, right, op, type_=type_, negate=negate, modifiers=modifiers
3073 )
3076class ExpressionClauseList(OperatorExpression[_T]):
3077 """Describe a list of clauses, separated by an operator,
3078 in a column expression context.
3080 :class:`.ExpressionClauseList` differs from :class:`.ClauseList` in that
3081 it represents a column-oriented DQL expression only, not an open ended
3082 list of anything comma separated.
3084 .. versionadded:: 2.0
3086 """
3088 __visit_name__ = "expression_clauselist"
3090 _traverse_internals: _TraverseInternalsType = [
3091 ("clauses", InternalTraversal.dp_clauseelement_tuple),
3092 ("operator", InternalTraversal.dp_operator),
3093 ]
3095 clauses: typing_Tuple[ColumnElement[Any], ...]
3097 group: bool
3099 def __init__(
3100 self,
3101 operator: OperatorType,
3102 *clauses: _ColumnExpressionArgument[Any],
3103 type_: Optional[_TypeEngineArgument[_T]] = None,
3104 ):
3105 self.operator = operator
3107 self.clauses = tuple(
3108 coercions.expect(
3109 roles.ExpressionElementRole, clause, apply_propagate_attrs=self
3110 )
3111 for clause in clauses
3112 )
3113 self._is_implicitly_boolean = operators.is_boolean(self.operator)
3114 self.type = type_api.to_instance(type_) # type: ignore
3116 @property
3117 def _flattened_operator_clauses(
3118 self,
3119 ) -> typing_Tuple[ColumnElement[Any], ...]:
3120 return self.clauses
3122 def __iter__(self) -> Iterator[ColumnElement[Any]]:
3123 return iter(self.clauses)
3125 def __len__(self) -> int:
3126 return len(self.clauses)
3128 @property
3129 def _select_iterable(self) -> _SelectIterable:
3130 return (self,)
3132 @util.ro_non_memoized_property
3133 def _from_objects(self) -> List[FromClause]:
3134 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
3136 def _append_inplace(self, clause: ColumnElement[Any]) -> None:
3137 self.clauses += (clause,)
3139 @classmethod
3140 def _construct_for_list(
3141 cls,
3142 operator: OperatorType,
3143 type_: TypeEngine[_T],
3144 *clauses: ColumnElement[Any],
3145 group: bool = True,
3146 ) -> ExpressionClauseList[_T]:
3147 self = cls.__new__(cls)
3148 self.group = group
3149 if group:
3150 self.clauses = tuple(
3151 c.self_group(against=operator) for c in clauses
3152 )
3153 else:
3154 self.clauses = clauses
3155 self.operator = operator
3156 self.type = type_
3157 for c in clauses:
3158 if c._propagate_attrs:
3159 self._propagate_attrs = c._propagate_attrs
3160 break
3161 return self
3163 def _negate(self) -> Any:
3164 grouped = self.self_group(against=operators.inv)
3165 assert isinstance(grouped, ColumnElement)
3166 return UnaryExpression(grouped, operator=operators.inv)
3169class BooleanClauseList(ExpressionClauseList[bool]):
3170 __visit_name__ = "expression_clauselist"
3171 inherit_cache = True
3173 def __init__(self, *arg, **kw):
3174 raise NotImplementedError(
3175 "BooleanClauseList has a private constructor"
3176 )
3178 @classmethod
3179 def _process_clauses_for_boolean(
3180 cls,
3181 operator: OperatorType,
3182 continue_on: Any,
3183 skip_on: Any,
3184 clauses: Iterable[ColumnElement[Any]],
3185 ) -> typing_Tuple[int, List[ColumnElement[Any]]]:
3186 has_continue_on = None
3188 convert_clauses = []
3190 against = operators._asbool
3191 lcc = 0
3193 for clause in clauses:
3194 if clause is continue_on:
3195 # instance of continue_on, like and_(x, y, True, z), store it
3196 # if we didn't find one already, we will use it if there
3197 # are no other expressions here.
3198 has_continue_on = clause
3199 elif clause is skip_on:
3200 # instance of skip_on, e.g. and_(x, y, False, z), cancels
3201 # the rest out
3202 convert_clauses = [clause]
3203 lcc = 1
3204 break
3205 else:
3206 if not lcc:
3207 lcc = 1
3208 else:
3209 against = operator
3210 # technically this would be len(convert_clauses) + 1
3211 # however this only needs to indicate "greater than one"
3212 lcc = 2
3213 convert_clauses.append(clause)
3215 if not convert_clauses and has_continue_on is not None:
3216 convert_clauses = [has_continue_on]
3217 lcc = 1
3219 return lcc, [c.self_group(against=against) for c in convert_clauses]
3221 @classmethod
3222 def _construct(
3223 cls,
3224 operator: OperatorType,
3225 continue_on: Any,
3226 skip_on: Any,
3227 initial_clause: Any = _NoArg.NO_ARG,
3228 *clauses: Any,
3229 **kw: Any,
3230 ) -> ColumnElement[Any]:
3231 if initial_clause is _NoArg.NO_ARG:
3232 # no elements period. deprecated use case. return an empty
3233 # ClauseList construct that generates nothing unless it has
3234 # elements added to it.
3235 name = operator.__name__
3237 util.warn_deprecated(
3238 f"Invoking {name}() without arguments is deprecated, and "
3239 f"will be disallowed in a future release. For an empty "
3240 f"""{name}() construct, use '{name}({
3241 'true()' if continue_on is True_._singleton else 'false()'
3242 }, *args)' """
3243 f"""or '{name}({
3244 'True' if continue_on is True_._singleton else 'False'
3245 }, *args)'.""",
3246 version="1.4",
3247 )
3248 return cls._construct_raw(operator)
3250 lcc, convert_clauses = cls._process_clauses_for_boolean(
3251 operator,
3252 continue_on,
3253 skip_on,
3254 [
3255 coercions.expect(roles.WhereHavingRole, clause)
3256 for clause in util.coerce_generator_arg(
3257 (initial_clause,) + clauses
3258 )
3259 ],
3260 )
3262 if lcc > 1:
3263 # multiple elements. Return regular BooleanClauseList
3264 # which will link elements against the operator.
3266 flattened_clauses = itertools.chain.from_iterable(
3267 (
3268 (c for c in to_flat._flattened_operator_clauses)
3269 if getattr(to_flat, "operator", None) is operator
3270 else (to_flat,)
3271 )
3272 for to_flat in convert_clauses
3273 )
3275 return cls._construct_raw(operator, flattened_clauses) # type: ignore # noqa: E501
3276 else:
3277 assert lcc
3278 # just one element. return it as a single boolean element,
3279 # not a list and discard the operator.
3280 return convert_clauses[0]
3282 @classmethod
3283 def _construct_for_whereclause(
3284 cls, clauses: Iterable[ColumnElement[Any]]
3285 ) -> Optional[ColumnElement[bool]]:
3286 operator, continue_on, skip_on = (
3287 operators.and_,
3288 True_._singleton,
3289 False_._singleton,
3290 )
3292 lcc, convert_clauses = cls._process_clauses_for_boolean(
3293 operator,
3294 continue_on,
3295 skip_on,
3296 clauses, # these are assumed to be coerced already
3297 )
3299 if lcc > 1:
3300 # multiple elements. Return regular BooleanClauseList
3301 # which will link elements against the operator.
3302 return cls._construct_raw(operator, convert_clauses)
3303 elif lcc == 1:
3304 # just one element. return it as a single boolean element,
3305 # not a list and discard the operator.
3306 return convert_clauses[0]
3307 else:
3308 return None
3310 @classmethod
3311 def _construct_raw(
3312 cls,
3313 operator: OperatorType,
3314 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
3315 ) -> BooleanClauseList:
3316 self = cls.__new__(cls)
3317 self.clauses = tuple(clauses) if clauses else ()
3318 self.group = True
3319 self.operator = operator
3320 self.type = type_api.BOOLEANTYPE
3321 self._is_implicitly_boolean = True
3322 return self
3324 @classmethod
3325 def and_(
3326 cls,
3327 initial_clause: Union[
3328 Literal[True], _ColumnExpressionArgument[bool], _NoArg
3329 ] = _NoArg.NO_ARG,
3330 *clauses: _ColumnExpressionArgument[bool],
3331 ) -> ColumnElement[bool]:
3332 r"""Produce a conjunction of expressions joined by ``AND``.
3334 See :func:`_sql.and_` for full documentation.
3335 """
3336 return cls._construct(
3337 operators.and_,
3338 True_._singleton,
3339 False_._singleton,
3340 initial_clause,
3341 *clauses,
3342 )
3344 @classmethod
3345 def or_(
3346 cls,
3347 initial_clause: Union[
3348 Literal[False], _ColumnExpressionArgument[bool], _NoArg
3349 ] = _NoArg.NO_ARG,
3350 *clauses: _ColumnExpressionArgument[bool],
3351 ) -> ColumnElement[bool]:
3352 """Produce a conjunction of expressions joined by ``OR``.
3354 See :func:`_sql.or_` for full documentation.
3355 """
3356 return cls._construct(
3357 operators.or_,
3358 False_._singleton,
3359 True_._singleton,
3360 initial_clause,
3361 *clauses,
3362 )
3364 @property
3365 def _select_iterable(self) -> _SelectIterable:
3366 return (self,)
3368 def self_group(
3369 self, against: Optional[OperatorType] = None
3370 ) -> Union[Self, Grouping[bool]]:
3371 if not self.clauses:
3372 return self
3373 else:
3374 return super().self_group(against=against)
3377and_ = BooleanClauseList.and_
3378or_ = BooleanClauseList.or_
3381class Tuple(ClauseList, ColumnElement[TupleAny]):
3382 """Represent a SQL tuple."""
3384 __visit_name__ = "tuple"
3386 _traverse_internals: _TraverseInternalsType = (
3387 ClauseList._traverse_internals + []
3388 )
3390 type: TupleType
3392 @util.preload_module("sqlalchemy.sql.sqltypes")
3393 def __init__(
3394 self,
3395 *clauses: _ColumnExpressionArgument[Any],
3396 types: Optional[Sequence[_TypeEngineArgument[Any]]] = None,
3397 ):
3398 sqltypes = util.preloaded.sql_sqltypes
3400 if types is None:
3401 init_clauses: List[ColumnElement[Any]] = [
3402 coercions.expect(roles.ExpressionElementRole, c)
3403 for c in clauses
3404 ]
3405 else:
3406 if len(types) != len(clauses):
3407 raise exc.ArgumentError(
3408 "Wrong number of elements for %d-tuple: %r "
3409 % (len(types), clauses)
3410 )
3411 init_clauses = [
3412 coercions.expect(
3413 roles.ExpressionElementRole,
3414 c,
3415 type_=typ if not typ._isnull else None,
3416 )
3417 for typ, c in zip(types, clauses)
3418 ]
3420 self.type = sqltypes.TupleType(*[arg.type for arg in init_clauses])
3421 super().__init__(*init_clauses)
3423 @property
3424 def _select_iterable(self) -> _SelectIterable:
3425 return (self,)
3427 def _bind_param(self, operator, obj, type_=None, expanding=False):
3428 if expanding:
3429 return BindParameter(
3430 None,
3431 value=obj,
3432 _compared_to_operator=operator,
3433 unique=True,
3434 expanding=True,
3435 type_=type_,
3436 _compared_to_type=self.type,
3437 )
3438 else:
3439 return Tuple(
3440 *[
3441 BindParameter(
3442 None,
3443 o,
3444 _compared_to_operator=operator,
3445 _compared_to_type=compared_to_type,
3446 unique=True,
3447 type_=type_,
3448 )
3449 for o, compared_to_type in zip(obj, self.type.types)
3450 ]
3451 )
3453 def self_group(self, against: Optional[OperatorType] = None) -> Self:
3454 # Tuple is parenthesized by definition.
3455 return self
3458class Case(ColumnElement[_T]):
3459 """Represent a ``CASE`` expression.
3461 :class:`.Case` is produced using the :func:`.case` factory function,
3462 as in::
3464 from sqlalchemy import case
3466 stmt = select(users_table).where(
3467 case(
3468 (users_table.c.name == "wendy", "W"),
3469 (users_table.c.name == "jack", "J"),
3470 else_="E",
3471 )
3472 )
3474 Details on :class:`.Case` usage is at :func:`.case`.
3476 .. seealso::
3478 :func:`.case`
3480 """
3482 __visit_name__ = "case"
3484 _traverse_internals: _TraverseInternalsType = [
3485 ("value", InternalTraversal.dp_clauseelement),
3486 ("whens", InternalTraversal.dp_clauseelement_tuples),
3487 ("else_", InternalTraversal.dp_clauseelement),
3488 ]
3490 # for case(), the type is derived from the whens. so for the moment
3491 # users would have to cast() the case to get a specific type
3493 whens: List[typing_Tuple[ColumnElement[bool], ColumnElement[_T]]]
3494 else_: Optional[ColumnElement[_T]]
3495 value: Optional[ColumnElement[Any]]
3497 def __init__(
3498 self,
3499 *whens: Union[
3500 typing_Tuple[_ColumnExpressionArgument[bool], Any],
3501 Mapping[Any, Any],
3502 ],
3503 value: Optional[Any] = None,
3504 else_: Optional[Any] = None,
3505 ):
3506 new_whens: Iterable[Any] = coercions._expression_collection_was_a_list(
3507 "whens", "case", whens
3508 )
3509 try:
3510 new_whens = util.dictlike_iteritems(new_whens)
3511 except TypeError:
3512 pass
3514 self.whens = [
3515 (
3516 coercions.expect(
3517 roles.ExpressionElementRole,
3518 c,
3519 apply_propagate_attrs=self,
3520 ).self_group(),
3521 coercions.expect(roles.ExpressionElementRole, r),
3522 )
3523 for (c, r) in new_whens
3524 ]
3526 if value is None:
3527 self.value = None
3528 else:
3529 self.value = coercions.expect(roles.ExpressionElementRole, value)
3531 if else_ is not None:
3532 self.else_ = coercions.expect(roles.ExpressionElementRole, else_)
3533 else:
3534 self.else_ = None
3536 type_ = next(
3537 (
3538 then.type
3539 # Iterate `whens` in reverse to match previous behaviour
3540 # where type of final element took priority
3541 for *_, then in reversed(self.whens)
3542 if not then.type._isnull
3543 ),
3544 self.else_.type if self.else_ is not None else type_api.NULLTYPE,
3545 )
3546 self.type = cast(_T, type_)
3548 @util.ro_non_memoized_property
3549 def _from_objects(self) -> List[FromClause]:
3550 return list(
3551 itertools.chain(*[x._from_objects for x in self.get_children()])
3552 )
3555class Cast(WrapsColumnExpression[_T]):
3556 """Represent a ``CAST`` expression.
3558 :class:`.Cast` is produced using the :func:`.cast` factory function,
3559 as in::
3561 from sqlalchemy import cast, Numeric
3563 stmt = select(cast(product_table.c.unit_price, Numeric(10, 4)))
3565 Details on :class:`.Cast` usage is at :func:`.cast`.
3567 .. seealso::
3569 :ref:`tutorial_casts`
3571 :func:`.cast`
3573 :func:`.try_cast`
3575 :func:`.type_coerce` - an alternative to CAST that coerces the type
3576 on the Python side only, which is often sufficient to generate the
3577 correct SQL and data coercion.
3579 """
3581 __visit_name__ = "cast"
3583 _traverse_internals: _TraverseInternalsType = [
3584 ("clause", InternalTraversal.dp_clauseelement),
3585 ("type", InternalTraversal.dp_type),
3586 ]
3588 clause: ColumnElement[Any]
3589 type: TypeEngine[_T]
3590 typeclause: TypeClause
3592 def __init__(
3593 self,
3594 expression: _ColumnExpressionArgument[Any],
3595 type_: _TypeEngineArgument[_T],
3596 ):
3597 self.type = type_api.to_instance(type_)
3598 self.clause = coercions.expect(
3599 roles.ExpressionElementRole,
3600 expression,
3601 type_=self.type,
3602 apply_propagate_attrs=self,
3603 )
3604 self.typeclause = TypeClause(self.type)
3606 @util.ro_non_memoized_property
3607 def _from_objects(self) -> List[FromClause]:
3608 return self.clause._from_objects
3610 @property
3611 def wrapped_column_expression(self):
3612 return self.clause
3615class TryCast(Cast[_T]):
3616 """Represent a TRY_CAST expression.
3618 Details on :class:`.TryCast` usage is at :func:`.try_cast`.
3620 .. seealso::
3622 :func:`.try_cast`
3624 :ref:`tutorial_casts`
3625 """
3627 __visit_name__ = "try_cast"
3628 inherit_cache = True
3631class TypeCoerce(WrapsColumnExpression[_T]):
3632 """Represent a Python-side type-coercion wrapper.
3634 :class:`.TypeCoerce` supplies the :func:`_expression.type_coerce`
3635 function; see that function for usage details.
3637 .. seealso::
3639 :func:`_expression.type_coerce`
3641 :func:`.cast`
3643 """
3645 __visit_name__ = "type_coerce"
3647 _traverse_internals: _TraverseInternalsType = [
3648 ("clause", InternalTraversal.dp_clauseelement),
3649 ("type", InternalTraversal.dp_type),
3650 ]
3652 clause: ColumnElement[Any]
3653 type: TypeEngine[_T]
3655 def __init__(
3656 self,
3657 expression: _ColumnExpressionArgument[Any],
3658 type_: _TypeEngineArgument[_T],
3659 ):
3660 self.type = type_api.to_instance(type_)
3661 self.clause = coercions.expect(
3662 roles.ExpressionElementRole,
3663 expression,
3664 type_=self.type,
3665 apply_propagate_attrs=self,
3666 )
3668 @util.ro_non_memoized_property
3669 def _from_objects(self) -> List[FromClause]:
3670 return self.clause._from_objects
3672 @HasMemoized.memoized_attribute
3673 def typed_expression(self):
3674 if isinstance(self.clause, BindParameter):
3675 bp = self.clause._clone()
3676 bp.type = self.type
3677 return bp
3678 else:
3679 return self.clause
3681 @property
3682 def wrapped_column_expression(self):
3683 return self.clause
3685 def self_group(
3686 self, against: Optional[OperatorType] = None
3687 ) -> TypeCoerce[_T]:
3688 grouped = self.clause.self_group(against=against)
3689 if grouped is not self.clause:
3690 return TypeCoerce(grouped, self.type)
3691 else:
3692 return self
3695class Extract(ColumnElement[int]):
3696 """Represent a SQL EXTRACT clause, ``extract(field FROM expr)``."""
3698 __visit_name__ = "extract"
3700 _traverse_internals: _TraverseInternalsType = [
3701 ("expr", InternalTraversal.dp_clauseelement),
3702 ("field", InternalTraversal.dp_string),
3703 ]
3705 expr: ColumnElement[Any]
3706 field: str
3708 def __init__(self, field: str, expr: _ColumnExpressionArgument[Any]):
3709 self.type = type_api.INTEGERTYPE
3710 self.field = field
3711 self.expr = coercions.expect(roles.ExpressionElementRole, expr)
3713 @util.ro_non_memoized_property
3714 def _from_objects(self) -> List[FromClause]:
3715 return self.expr._from_objects
3718class _label_reference(ColumnElement[_T]):
3719 """Wrap a column expression as it appears in a 'reference' context.
3721 This expression is any that includes an _order_by_label_element,
3722 which is a Label, or a DESC / ASC construct wrapping a Label.
3724 The production of _label_reference() should occur when an expression
3725 is added to this context; this includes the ORDER BY or GROUP BY of a
3726 SELECT statement, as well as a few other places, such as the ORDER BY
3727 within an OVER clause.
3729 """
3731 __visit_name__ = "label_reference"
3733 _traverse_internals: _TraverseInternalsType = [
3734 ("element", InternalTraversal.dp_clauseelement)
3735 ]
3737 element: ColumnElement[_T]
3739 def __init__(self, element: ColumnElement[_T]):
3740 self.element = element
3741 self._propagate_attrs = element._propagate_attrs
3743 @util.ro_non_memoized_property
3744 def _from_objects(self) -> List[FromClause]:
3745 return []
3748class _textual_label_reference(ColumnElement[Any]):
3749 __visit_name__ = "textual_label_reference"
3751 _traverse_internals: _TraverseInternalsType = [
3752 ("element", InternalTraversal.dp_string)
3753 ]
3755 def __init__(self, element: str):
3756 self.element = element
3758 @util.memoized_property
3759 def _text_clause(self) -> TextClause:
3760 return TextClause(self.element)
3763class UnaryExpression(ColumnElement[_T]):
3764 """Define a 'unary' expression.
3766 A unary expression has a single column expression
3767 and an operator. The operator can be placed on the left
3768 (where it is called the 'operator') or right (where it is called the
3769 'modifier') of the column expression.
3771 :class:`.UnaryExpression` is the basis for several unary operators
3772 including those used by :func:`.desc`, :func:`.asc`, :func:`.distinct`,
3773 :func:`.nulls_first` and :func:`.nulls_last`.
3775 """
3777 __visit_name__ = "unary"
3779 _traverse_internals: _TraverseInternalsType = [
3780 ("element", InternalTraversal.dp_clauseelement),
3781 ("operator", InternalTraversal.dp_operator),
3782 ("modifier", InternalTraversal.dp_operator),
3783 ]
3785 element: ColumnElement[Any]
3786 operator: Optional[OperatorType]
3787 modifier: Optional[OperatorType]
3789 def __init__(
3790 self,
3791 element: ColumnElement[Any],
3792 *,
3793 operator: Optional[OperatorType] = None,
3794 modifier: Optional[OperatorType] = None,
3795 type_: Optional[_TypeEngineArgument[_T]] = None,
3796 wraps_column_expression: bool = False, # legacy, not used as of 2.0.42
3797 ):
3798 self.operator = operator
3799 self.modifier = modifier
3800 self._propagate_attrs = element._propagate_attrs
3801 self.element = element.self_group(
3802 against=self.operator or self.modifier
3803 )
3805 # if type is None, we get NULLTYPE, which is our _T. But I don't
3806 # know how to get the overloads to express that correctly
3807 self.type = type_api.to_instance(type_) # type: ignore
3809 def _wraps_unnamed_column(self):
3810 ungrouped = self.element._ungroup()
3811 return (
3812 not isinstance(ungrouped, NamedColumn)
3813 or ungrouped._non_anon_label is None
3814 )
3816 @classmethod
3817 def _create_nulls_first(
3818 cls,
3819 column: _ColumnExpressionArgument[_T],
3820 ) -> UnaryExpression[_T]:
3821 return UnaryExpression(
3822 coercions.expect(roles.ByOfRole, column),
3823 modifier=operators.nulls_first_op,
3824 )
3826 @classmethod
3827 def _create_nulls_last(
3828 cls,
3829 column: _ColumnExpressionArgument[_T],
3830 ) -> UnaryExpression[_T]:
3831 return UnaryExpression(
3832 coercions.expect(roles.ByOfRole, column),
3833 modifier=operators.nulls_last_op,
3834 )
3836 @classmethod
3837 def _create_desc(
3838 cls, column: _ColumnExpressionOrStrLabelArgument[_T]
3839 ) -> UnaryExpression[_T]:
3841 return UnaryExpression(
3842 coercions.expect(roles.ByOfRole, column),
3843 modifier=operators.desc_op,
3844 )
3846 @classmethod
3847 def _create_asc(
3848 cls,
3849 column: _ColumnExpressionOrStrLabelArgument[_T],
3850 ) -> UnaryExpression[_T]:
3851 return UnaryExpression(
3852 coercions.expect(roles.ByOfRole, column),
3853 modifier=operators.asc_op,
3854 )
3856 @classmethod
3857 def _create_distinct(
3858 cls,
3859 expr: _ColumnExpressionArgument[_T],
3860 ) -> UnaryExpression[_T]:
3861 col_expr: ColumnElement[_T] = coercions.expect(
3862 roles.ExpressionElementRole, expr
3863 )
3864 return UnaryExpression(
3865 col_expr,
3866 operator=operators.distinct_op,
3867 type_=col_expr.type,
3868 )
3870 @classmethod
3871 def _create_bitwise_not(
3872 cls,
3873 expr: _ColumnExpressionArgument[_T],
3874 ) -> UnaryExpression[_T]:
3875 col_expr: ColumnElement[_T] = coercions.expect(
3876 roles.ExpressionElementRole, expr
3877 )
3878 return UnaryExpression(
3879 col_expr,
3880 operator=operators.bitwise_not_op,
3881 type_=col_expr.type,
3882 )
3884 @property
3885 def _order_by_label_element(self) -> Optional[Label[Any]]:
3886 if operators.is_order_by_modifier(self.modifier):
3887 return self.element._order_by_label_element
3888 else:
3889 return None
3891 @util.ro_non_memoized_property
3892 def _from_objects(self) -> List[FromClause]:
3893 return self.element._from_objects
3895 def _negate(self) -> ColumnElement[Any]:
3896 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
3897 return UnaryExpression(
3898 self.self_group(against=operators.inv),
3899 operator=operators.inv,
3900 type_=type_api.BOOLEANTYPE,
3901 )
3902 else:
3903 return ColumnElement._negate(self)
3905 def self_group(
3906 self, against: Optional[OperatorType] = None
3907 ) -> Union[Self, Grouping[_T]]:
3908 if self.operator and operators.is_precedent(self.operator, against):
3909 return Grouping(self)
3910 else:
3911 return self
3914class CollectionAggregate(UnaryExpression[_T]):
3915 """Forms the basis for right-hand collection operator modifiers
3916 ANY and ALL.
3918 The ANY and ALL keywords are available in different ways on different
3919 backends. On PostgreSQL, they only work for an ARRAY type. On
3920 MySQL, they only work for subqueries.
3922 """
3924 inherit_cache = True
3925 _is_collection_aggregate = True
3927 @classmethod
3928 def _create_any(
3929 cls, expr: _ColumnExpressionArgument[_T]
3930 ) -> CollectionAggregate[bool]:
3931 """create CollectionAggregate for the legacy
3932 ARRAY.Comparator.any() method"""
3933 col_expr: ColumnElement[_T] = coercions.expect(
3934 roles.ExpressionElementRole,
3935 expr,
3936 )
3937 col_expr = col_expr.self_group()
3938 return CollectionAggregate(
3939 col_expr,
3940 operator=operators.any_op,
3941 type_=type_api.BOOLEANTYPE,
3942 )
3944 @classmethod
3945 def _create_all(
3946 cls, expr: _ColumnExpressionArgument[_T]
3947 ) -> CollectionAggregate[bool]:
3948 """create CollectionAggregate for the legacy
3949 ARRAY.Comparator.all() method"""
3950 col_expr: ColumnElement[_T] = coercions.expect(
3951 roles.ExpressionElementRole,
3952 expr,
3953 )
3954 col_expr = col_expr.self_group()
3955 return CollectionAggregate(
3956 col_expr,
3957 operator=operators.all_op,
3958 type_=type_api.BOOLEANTYPE,
3959 )
3961 @util.preload_module("sqlalchemy.sql.sqltypes")
3962 def _bind_param(
3963 self,
3964 operator: operators.OperatorType,
3965 obj: Any,
3966 type_: Optional[TypeEngine[_T]] = None,
3967 expanding: bool = False,
3968 ) -> BindParameter[_T]:
3969 """For new style any_(), all_(), ensure compared literal value
3970 receives appropriate bound parameter type."""
3972 # a CollectionAggregate is specific to ARRAY or int
3973 # only. So for ARRAY case, make sure we use correct element type
3974 sqltypes = util.preloaded.sql_sqltypes
3975 if self.element.type._type_affinity is sqltypes.ARRAY:
3976 compared_to_type = cast(
3977 sqltypes.ARRAY[Any], self.element.type
3978 ).item_type
3979 else:
3980 compared_to_type = self.element.type
3982 return BindParameter(
3983 None,
3984 obj,
3985 _compared_to_operator=operator,
3986 type_=type_,
3987 _compared_to_type=compared_to_type,
3988 unique=True,
3989 expanding=expanding,
3990 )
3992 # operate and reverse_operate are hardwired to
3993 # dispatch onto the type comparator directly, so that we can
3994 # ensure "reversed" behavior.
3995 def operate(
3996 self, op: OperatorType, *other: Any, **kwargs: Any
3997 ) -> ColumnElement[_T]:
3998 if not operators.is_comparison(op):
3999 raise exc.ArgumentError(
4000 "Only comparison operators may be used with ANY/ALL"
4001 )
4002 kwargs["reverse"] = True
4003 return self.comparator.operate(operators.mirror(op), *other, **kwargs)
4005 def reverse_operate(
4006 self, op: OperatorType, other: Any, **kwargs: Any
4007 ) -> ColumnElement[_T]:
4008 # comparison operators should never call reverse_operate
4009 assert not operators.is_comparison(op)
4010 raise exc.ArgumentError(
4011 "Only comparison operators may be used with ANY/ALL"
4012 )
4015class AsBoolean(WrapsColumnExpression[bool], UnaryExpression[bool]):
4016 inherit_cache = True
4018 def __init__(self, element, operator, negate):
4019 self.element = element
4020 self.type = type_api.BOOLEANTYPE
4021 self.operator = operator
4022 self.negate = negate
4023 self.modifier = None
4024 self._is_implicitly_boolean = element._is_implicitly_boolean
4026 @property
4027 def wrapped_column_expression(self):
4028 return self.element
4030 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4031 return self
4033 def _negate(self):
4034 if isinstance(self.element, (True_, False_)):
4035 return self.element._negate()
4036 else:
4037 return AsBoolean(self.element, self.negate, self.operator)
4040class BinaryExpression(OperatorExpression[_T]):
4041 """Represent an expression that is ``LEFT <operator> RIGHT``.
4043 A :class:`.BinaryExpression` is generated automatically
4044 whenever two column expressions are used in a Python binary expression:
4046 .. sourcecode:: pycon+sql
4048 >>> from sqlalchemy.sql import column
4049 >>> column("a") + column("b")
4050 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
4051 >>> print(column("a") + column("b"))
4052 {printsql}a + b
4054 """
4056 __visit_name__ = "binary"
4058 _traverse_internals: _TraverseInternalsType = [
4059 ("left", InternalTraversal.dp_clauseelement),
4060 ("right", InternalTraversal.dp_clauseelement),
4061 ("operator", InternalTraversal.dp_operator),
4062 ("negate", InternalTraversal.dp_operator),
4063 ("modifiers", InternalTraversal.dp_plain_dict),
4064 (
4065 "type",
4066 InternalTraversal.dp_type,
4067 ),
4068 ]
4070 _cache_key_traversal = [
4071 ("left", InternalTraversal.dp_clauseelement),
4072 ("right", InternalTraversal.dp_clauseelement),
4073 ("operator", InternalTraversal.dp_operator),
4074 ("modifiers", InternalTraversal.dp_plain_dict),
4075 # "type" affects JSON CAST operators, so while redundant in most cases,
4076 # is needed for that one
4077 (
4078 "type",
4079 InternalTraversal.dp_type,
4080 ),
4081 ]
4083 _is_implicitly_boolean = True
4084 """Indicates that any database will know this is a boolean expression
4085 even if the database does not have an explicit boolean datatype.
4087 """
4089 left: ColumnElement[Any]
4090 right: ColumnElement[Any]
4091 modifiers: Mapping[str, Any]
4093 def __init__(
4094 self,
4095 left: ColumnElement[Any],
4096 right: ColumnElement[Any],
4097 operator: OperatorType,
4098 type_: Optional[_TypeEngineArgument[_T]] = None,
4099 negate: Optional[OperatorType] = None,
4100 modifiers: Optional[Mapping[str, Any]] = None,
4101 ):
4102 # allow compatibility with libraries that
4103 # refer to BinaryExpression directly and pass strings
4104 if isinstance(operator, str):
4105 operator = operators.custom_op(operator)
4106 self._orig = (left.__hash__(), right.__hash__())
4107 self._propagate_attrs = left._propagate_attrs or right._propagate_attrs
4108 self.left = left.self_group(against=operator)
4109 self.right = right.self_group(against=operator)
4110 self.operator = operator
4112 # if type is None, we get NULLTYPE, which is our _T. But I don't
4113 # know how to get the overloads to express that correctly
4114 self.type = type_api.to_instance(type_) # type: ignore
4116 self.negate = negate
4117 self._is_implicitly_boolean = operators.is_boolean(operator)
4119 if modifiers is None:
4120 self.modifiers = {}
4121 else:
4122 self.modifiers = modifiers
4124 @property
4125 def _flattened_operator_clauses(
4126 self,
4127 ) -> typing_Tuple[ColumnElement[Any], ...]:
4128 return (self.left, self.right)
4130 def __bool__(self):
4131 """Implement Python-side "bool" for BinaryExpression as a
4132 simple "identity" check for the left and right attributes,
4133 if the operator is "eq" or "ne". Otherwise the expression
4134 continues to not support "bool" like all other column expressions.
4136 The rationale here is so that ColumnElement objects can be hashable.
4137 What? Well, suppose you do this::
4139 c1, c2 = column("x"), column("y")
4140 s1 = set([c1, c2])
4142 We do that **a lot**, columns inside of sets is an extremely basic
4143 thing all over the ORM for example.
4145 So what happens if we do this? ::
4147 c1 in s1
4149 Hashing means it will normally use ``__hash__()`` of the object,
4150 but in case of hash collision, it's going to also do ``c1 == c1``
4151 and/or ``c1 == c2`` inside. Those operations need to return a
4152 True/False value. But because we override ``==`` and ``!=``, they're
4153 going to get a BinaryExpression. Hence we implement ``__bool__`` here
4154 so that these comparisons behave in this particular context mostly
4155 like regular object comparisons. Thankfully Python is OK with
4156 that! Otherwise we'd have to use special set classes for columns
4157 (which we used to do, decades ago).
4159 """
4160 if self.operator in (operators.eq, operators.ne):
4161 # this is using the eq/ne operator given int hash values,
4162 # rather than Operator, so that "bool" can be based on
4163 # identity
4164 return self.operator(*self._orig) # type: ignore
4165 else:
4166 raise TypeError("Boolean value of this clause is not defined")
4168 if typing.TYPE_CHECKING:
4170 def __invert__(
4171 self: BinaryExpression[_T],
4172 ) -> BinaryExpression[_T]: ...
4174 @util.ro_non_memoized_property
4175 def _from_objects(self) -> List[FromClause]:
4176 return self.left._from_objects + self.right._from_objects
4178 def _negate(self):
4179 if self.negate is not None:
4180 return BinaryExpression(
4181 self.left,
4182 self.right._negate_in_binary(self.negate, self.operator),
4183 self.negate,
4184 negate=self.operator,
4185 type_=self.type,
4186 modifiers=self.modifiers,
4187 )
4188 else:
4189 return self.self_group()._negate()
4192class Slice(ColumnElement[Any]):
4193 """Represent SQL for a Python array-slice object.
4195 This is not a specific SQL construct at this level, but
4196 may be interpreted by specific dialects, e.g. PostgreSQL.
4198 """
4200 __visit_name__ = "slice"
4202 _traverse_internals: _TraverseInternalsType = [
4203 ("start", InternalTraversal.dp_clauseelement),
4204 ("stop", InternalTraversal.dp_clauseelement),
4205 ("step", InternalTraversal.dp_clauseelement),
4206 ]
4208 def __init__(self, start, stop, step, _name=None):
4209 self.start = coercions.expect(
4210 roles.ExpressionElementRole,
4211 start,
4212 name=_name,
4213 type_=type_api.INTEGERTYPE,
4214 )
4215 self.stop = coercions.expect(
4216 roles.ExpressionElementRole,
4217 stop,
4218 name=_name,
4219 type_=type_api.INTEGERTYPE,
4220 )
4221 self.step = coercions.expect(
4222 roles.ExpressionElementRole,
4223 step,
4224 name=_name,
4225 type_=type_api.INTEGERTYPE,
4226 )
4227 self.type = type_api.NULLTYPE
4229 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4230 assert against is operator.getitem
4231 return self
4234class IndexExpression(BinaryExpression[Any]):
4235 """Represent the class of expressions that are like an "index"
4236 operation."""
4238 inherit_cache = True
4241class GroupedElement(DQLDMLClauseElement):
4242 """Represent any parenthesized expression"""
4244 __visit_name__ = "grouping"
4246 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4247 return self
4249 def _ungroup(self) -> ClauseElement:
4250 raise NotImplementedError()
4253class Grouping(GroupedElement, ColumnElement[_T]):
4254 """Represent a grouping within a column expression"""
4256 _traverse_internals: _TraverseInternalsType = [
4257 ("element", InternalTraversal.dp_clauseelement),
4258 ("type", InternalTraversal.dp_type),
4259 ]
4261 _cache_key_traversal = [
4262 ("element", InternalTraversal.dp_clauseelement),
4263 ]
4265 element: Union[
4266 TextClause, ClauseList, ColumnElement[_T], CompilerColumnElement
4267 ]
4269 def __init__(
4270 self,
4271 element: Union[
4272 TextClause, ClauseList, ColumnElement[_T], CompilerColumnElement
4273 ],
4274 ):
4275 self.element = element
4277 # nulltype assignment issue
4278 self.type = getattr(element, "type", type_api.NULLTYPE) # type: ignore
4279 self._propagate_attrs = element._propagate_attrs
4281 def _with_binary_element_type(self, type_):
4282 return self.__class__(self.element._with_binary_element_type(type_))
4284 def _ungroup(self) -> ColumnElement[_T]:
4285 assert isinstance(self.element, ColumnElement)
4286 return self.element._ungroup()
4288 @util.memoized_property
4289 def _is_implicitly_boolean(self):
4290 return self.element._is_implicitly_boolean
4292 @util.non_memoized_property
4293 def _tq_label(self) -> Optional[str]:
4294 return (
4295 getattr(self.element, "_tq_label", None) or self._anon_name_label
4296 )
4298 @util.non_memoized_property
4299 def _proxies(self) -> List[ColumnElement[Any]]:
4300 if isinstance(self.element, ColumnElement):
4301 return [self.element]
4302 else:
4303 return []
4305 @util.ro_non_memoized_property
4306 def _from_objects(self) -> List[FromClause]:
4307 return self.element._from_objects
4309 def __getattr__(self, attr):
4310 return getattr(self.element, attr)
4312 def __getstate__(self):
4313 return {"element": self.element, "type": self.type}
4315 def __setstate__(self, state):
4316 self.element = state["element"]
4317 self.type = state["type"]
4319 if TYPE_CHECKING:
4321 def self_group(
4322 self, against: Optional[OperatorType] = None
4323 ) -> Self: ...
4326class _OverrideBinds(Grouping[_T]):
4327 """used by cache_key->_apply_params_to_element to allow compilation /
4328 execution of a SQL element that's been cached, using an alternate set of
4329 bound parameter values.
4331 This is used by the ORM to swap new parameter values into expressions
4332 that are embedded into loader options like with_expression(),
4333 selectinload(). Previously, this task was accomplished using the
4334 .params() method which would perform a deep-copy instead. This deep
4335 copy proved to be too expensive for more complex expressions.
4337 See #11085
4339 """
4341 __visit_name__ = "override_binds"
4343 def __init__(
4344 self,
4345 element: ColumnElement[_T],
4346 bindparams: Sequence[BindParameter[Any]],
4347 replaces_params: Sequence[BindParameter[Any]],
4348 ):
4349 self.element = element
4350 self.translate = {
4351 k.key: v.value for k, v in zip(replaces_params, bindparams)
4352 }
4354 def _gen_cache_key(
4355 self, anon_map: anon_map, bindparams: List[BindParameter[Any]]
4356 ) -> Optional[typing_Tuple[Any, ...]]:
4357 """generate a cache key for the given element, substituting its bind
4358 values for the translation values present."""
4360 existing_bps: List[BindParameter[Any]] = []
4361 ck = self.element._gen_cache_key(anon_map, existing_bps)
4363 bindparams.extend(
4364 (
4365 bp._with_value(
4366 self.translate[bp.key], maintain_key=True, required=False
4367 )
4368 if bp.key in self.translate
4369 else bp
4370 )
4371 for bp in existing_bps
4372 )
4374 return ck
4377_FrameIntTuple = tuple[int | None, int | None]
4380class Over(ColumnElement[_T]):
4381 """Represent an OVER clause.
4383 This is a special operator against a so-called
4384 "window" function, as well as any aggregate function,
4385 which produces results relative to the result set
4386 itself. Most modern SQL backends now support window functions.
4388 """
4390 __visit_name__ = "over"
4392 _traverse_internals: _TraverseInternalsType = [
4393 ("element", InternalTraversal.dp_clauseelement),
4394 ("order_by", InternalTraversal.dp_clauseelement),
4395 ("partition_by", InternalTraversal.dp_clauseelement),
4396 ("range_", InternalTraversal.dp_clauseelement),
4397 ("rows", InternalTraversal.dp_clauseelement),
4398 ("groups", InternalTraversal.dp_clauseelement),
4399 ]
4401 order_by: Optional[ClauseList] = None
4402 partition_by: Optional[ClauseList] = None
4404 element: ColumnElement[_T]
4405 """The underlying expression object to which this :class:`.Over`
4406 object refers."""
4408 range_: FrameClause | None
4409 rows: FrameClause | None
4410 groups: FrameClause | None
4412 def __init__(
4413 self,
4414 element: ColumnElement[_T],
4415 partition_by: Optional[_ByArgument] = None,
4416 order_by: Optional[_ByArgument] = None,
4417 range_: _FrameIntTuple | FrameClause | None = None,
4418 rows: _FrameIntTuple | FrameClause | None = None,
4419 groups: _FrameIntTuple | FrameClause | None = None,
4420 ):
4421 self.element = element
4422 if order_by is not None:
4423 self.order_by = ClauseList(
4424 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4425 )
4426 if partition_by is not None:
4427 self.partition_by = ClauseList(
4428 *util.to_list(partition_by),
4429 _literal_as_text_role=roles.ByOfRole,
4430 )
4432 if sum(item is not None for item in (range_, rows, groups)) > 1:
4433 raise exc.ArgumentError(
4434 "only one of 'rows', 'range_', or 'groups' may be provided"
4435 )
4436 else:
4437 self.range_ = FrameClause._parse(range_, coerce_int=False)
4438 self.rows = FrameClause._parse(rows, coerce_int=True)
4439 self.groups = FrameClause._parse(groups, coerce_int=True)
4441 if not TYPE_CHECKING:
4443 @util.memoized_property
4444 def type(self) -> TypeEngine[_T]: # noqa: A001
4445 return self.element.type
4447 @util.ro_non_memoized_property
4448 def _from_objects(self) -> List[FromClause]:
4449 return list(
4450 itertools.chain(
4451 *[
4452 c._from_objects
4453 for c in (self.element, self.partition_by, self.order_by)
4454 if c is not None
4455 ]
4456 )
4457 )
4460class FrameClauseType(Enum):
4461 """Frame clause type enum for FrameClause lower_type and upper_type.
4463 .. versionadded:: 2.1
4465 """
4467 UNBOUNDED = 0
4468 """Produces an "UNBOUNDED PRECEDING" or "UNBOUNDED FOLLOWING" frame
4469 clause depending on the position.
4470 Requires a ``None`` value for the corresponding bound value.
4471 """
4472 CURRENT = 1
4473 """Produces a "CURRENT ROW" frame clause.
4474 Requires a ``None`` value for the corresponding bound value.
4475 """
4476 PRECEDING = 2
4477 """Produces a "PRECEDING" frame clause."""
4478 FOLLOWING = 3
4479 """Produces a "FOLLOWING" frame clause."""
4482_require_none = (
4483 FrameClauseType.CURRENT,
4484 FrameClauseType.UNBOUNDED,
4485)
4488class FrameClause(ClauseElement):
4489 """Indicate the 'rows' 'range' or 'group' field of a window function,
4490 e.g. using :class:`.Over`.
4492 .. versionadded:: 2.1
4494 """
4496 __visit_name__ = "frame_clause"
4498 _traverse_internals: _TraverseInternalsType = [
4499 ("lower_bind", InternalTraversal.dp_clauseelement),
4500 ("upper_bind", InternalTraversal.dp_clauseelement),
4501 ("lower_type", InternalTraversal.dp_plain_obj),
4502 ("upper_type", InternalTraversal.dp_plain_obj),
4503 ]
4505 def __init__(
4506 self,
4507 start: Any,
4508 end: Any,
4509 start_frame_type: FrameClauseType,
4510 end_frame_type: FrameClauseType,
4511 _validate: bool = True,
4512 ) -> None:
4513 """Creates a new FrameClause specifying the bounds of a window frame.
4515 :param start: The start value.
4516 :param end: The end value.
4517 :param start_frame_type: The :class:`FrameClauseType` for the
4518 start value.
4519 :param end_frame_type: The :class:`FrameClauseType` for the end value.
4520 """
4521 self.lower_bind = self._as_literal(start)
4522 self.upper_bind = self._as_literal(end)
4523 self.lower_type = FrameClauseType(start_frame_type)
4524 self.upper_type = FrameClauseType(end_frame_type)
4525 if _validate:
4526 if (
4527 self.lower_type in _require_none
4528 and self.lower_bind is not None
4529 ):
4530 raise exc.ArgumentError(
4531 "Cannot specify a value for start with frame type "
4532 f"{self.lower_type.name}"
4533 )
4534 if (
4535 self.upper_type in _require_none
4536 and self.upper_bind is not None
4537 ):
4538 raise exc.ArgumentError(
4539 "Cannot specify a value for end with frame type "
4540 f"{self.upper_type.name}"
4541 )
4543 @classmethod
4544 def _as_literal(cls, value: Any) -> BindParameter[Any] | None:
4545 if value is None:
4546 return None
4547 elif isinstance(value, int):
4548 return literal(value, type_api.INTEGERTYPE)
4549 elif isinstance(value, BindParameter):
4550 return value
4551 else:
4552 return literal(value) # let the default type resolution occur
4554 @classmethod
4555 def _handle_int(
4556 cls, value: Any | None, coerce_int: bool
4557 ) -> tuple[int | None, FrameClauseType]:
4558 if value is None:
4559 return None, FrameClauseType.UNBOUNDED
4561 if coerce_int:
4562 try:
4563 integer = int(value)
4564 except ValueError as err:
4565 raise exc.ArgumentError(
4566 "Integer or None expected for values in rows/groups frame"
4567 ) from err
4568 elif not isinstance(value, int):
4569 raise exc.ArgumentError(
4570 "When using a tuple to specify a range only integer or none "
4571 "values are allowed in the range frame. To specify a "
4572 "different type use the FrameClause directly."
4573 )
4574 else:
4575 integer = value
4576 if integer == 0:
4577 return None, FrameClauseType.CURRENT
4578 elif integer < 0:
4579 return abs(integer), FrameClauseType.PRECEDING
4580 else:
4581 return integer, FrameClauseType.FOLLOWING
4583 @classmethod
4584 def _parse(
4585 cls,
4586 range_: _FrameIntTuple | FrameClause | None,
4587 coerce_int: bool,
4588 ) -> FrameClause | None:
4589 if range_ is None or isinstance(range_, FrameClause):
4590 return range_
4592 try:
4593 r0, r1 = range_
4594 except (ValueError, TypeError) as ve:
4595 raise exc.ArgumentError(
4596 "2-tuple expected for range/rows/groups"
4597 ) from ve
4599 l_b, l_t = cls._handle_int(r0, coerce_int)
4600 u_b, u_t = cls._handle_int(r1, coerce_int)
4602 return FrameClause(
4603 start=l_b,
4604 end=u_b,
4605 start_frame_type=l_t,
4606 end_frame_type=u_t,
4607 _validate=False,
4608 )
4611class AggregateOrderBy(WrapsColumnExpression[_T]):
4612 """Represent an aggregate ORDER BY expression.
4614 This is a special operator against aggregate functions such as
4615 ``array_agg()``, ``json_arrayagg()`` ``string_agg()``, etc. that provides
4616 for an ORDER BY expression, using a syntax that's compatible with
4617 the backend.
4619 :class:`.AggregateOrderBy` is a generalized version of the
4620 :class:`.WithinGroup` construct, the latter of which always provides a
4621 "WITHIN GROUP (ORDER BY ...)" expression. :class:`.AggregateOrderBy` will
4622 also compile to "WITHIN GROUP (ORDER BY ...)" on backends such as Oracle
4623 and SQL Server that don't have another style of aggregate function
4624 ordering.
4626 .. versionadded:: 2.1
4629 """
4631 __visit_name__ = "aggregateorderby"
4633 _traverse_internals: _TraverseInternalsType = [
4634 ("element", InternalTraversal.dp_clauseelement),
4635 ("order_by", InternalTraversal.dp_clauseelement),
4636 ]
4638 order_by: ClauseList
4640 def __init__(
4641 self,
4642 element: Union[FunctionElement[_T], FunctionFilter[_T]],
4643 *order_by: _ColumnExpressionArgument[Any],
4644 ):
4645 self.element = element
4646 if not order_by:
4647 raise TypeError("at least one ORDER BY element is required")
4648 self.order_by = ClauseList(
4649 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4650 )
4652 if not TYPE_CHECKING:
4654 @util.memoized_property
4655 def type(self) -> TypeEngine[_T]: # noqa: A001
4656 return self.element.type
4658 @property
4659 def wrapped_column_expression(self) -> ColumnElement[_T]:
4660 return self.element
4662 def __reduce__(self):
4663 return self.__class__, (self.element,) + (
4664 tuple(self.order_by) if self.order_by is not None else ()
4665 )
4667 def over(
4668 self,
4669 *,
4670 partition_by: _ByArgument | None = None,
4671 order_by: _ByArgument | None = None,
4672 rows: _FrameIntTuple | FrameClause | None = None,
4673 range_: _FrameIntTuple | FrameClause | None = None,
4674 groups: _FrameIntTuple | FrameClause | None = None,
4675 ) -> Over[_T]:
4676 """Produce an OVER clause against this :class:`.WithinGroup`
4677 construct.
4679 This function has the same signature as that of
4680 :meth:`.FunctionElement.over`.
4682 """
4683 return Over(
4684 self,
4685 partition_by=partition_by,
4686 order_by=order_by,
4687 range_=range_,
4688 rows=rows,
4689 groups=groups,
4690 )
4692 @overload
4693 def filter(self) -> Self: ...
4695 @overload
4696 def filter(
4697 self,
4698 __criterion0: _ColumnExpressionArgument[bool],
4699 *criterion: _ColumnExpressionArgument[bool],
4700 ) -> FunctionFilter[_T]: ...
4702 def filter(
4703 self, *criterion: _ColumnExpressionArgument[bool]
4704 ) -> Union[Self, FunctionFilter[_T]]:
4705 """Produce a FILTER clause against this function."""
4706 if not criterion:
4707 return self
4708 return FunctionFilter(self, *criterion)
4710 @util.ro_non_memoized_property
4711 def _from_objects(self) -> List[FromClause]:
4712 return list(
4713 itertools.chain(
4714 *[
4715 c._from_objects
4716 for c in (self.element, self.order_by)
4717 if c is not None
4718 ]
4719 )
4720 )
4723class WithinGroup(AggregateOrderBy[_T]):
4724 """Represent a WITHIN GROUP (ORDER BY) clause.
4726 This is a special operator against so-called
4727 "ordered set aggregate" and "hypothetical
4728 set aggregate" functions, including ``percentile_cont()``,
4729 ``rank()``, ``dense_rank()``, etc.
4731 It's supported only by certain database backends, such as PostgreSQL,
4732 Oracle Database and MS SQL Server.
4734 The :class:`.WithinGroup` construct extracts its type from the
4735 method :meth:`.FunctionElement.within_group_type`. If this returns
4736 ``None``, the function's ``.type`` is used.
4738 """
4740 __visit_name__ = "withingroup"
4741 inherit_cache = True
4743 if not TYPE_CHECKING:
4745 @util.memoized_property
4746 def type(self) -> TypeEngine[_T]: # noqa: A001
4747 wgt = self.element.within_group_type(self)
4748 if wgt is not None:
4749 return wgt
4750 else:
4751 return self.element.type
4754class FunctionFilter(Generative, ColumnElement[_T]):
4755 """Represent a function FILTER clause.
4757 This is a special operator against aggregate and window functions,
4758 which controls which rows are passed to it.
4759 It's supported only by certain database backends.
4761 Invocation of :class:`.FunctionFilter` is via
4762 :meth:`.FunctionElement.filter`::
4764 func.count(1).filter(True)
4766 .. seealso::
4768 :meth:`.FunctionElement.filter`
4770 """
4772 __visit_name__ = "funcfilter"
4774 _traverse_internals: _TraverseInternalsType = [
4775 ("func", InternalTraversal.dp_clauseelement),
4776 ("criterion", InternalTraversal.dp_clauseelement),
4777 ]
4779 criterion: Optional[ColumnElement[bool]] = None
4781 def __init__(
4782 self,
4783 func: Union[FunctionElement[_T], AggregateOrderBy[_T]],
4784 *criterion: _ColumnExpressionArgument[bool],
4785 ):
4786 self.func = func
4787 self.filter.non_generative(self, *criterion) # type: ignore
4789 @_generative
4790 def filter(self, *criterion: _ColumnExpressionArgument[bool]) -> Self:
4791 """Produce an additional FILTER against the function.
4793 This method adds additional criteria to the initial criteria
4794 set up by :meth:`.FunctionElement.filter`.
4796 Multiple criteria are joined together at SQL render time
4797 via ``AND``.
4800 """
4802 for crit in list(criterion):
4803 crit = coercions.expect(roles.WhereHavingRole, crit)
4805 if self.criterion is not None:
4806 self.criterion = self.criterion & crit
4807 else:
4808 self.criterion = crit
4810 return self
4812 def over(
4813 self,
4814 partition_by: _ByArgument | None = None,
4815 order_by: _ByArgument | None = None,
4816 range_: _FrameIntTuple | FrameClause | None = None,
4817 rows: _FrameIntTuple | FrameClause | None = None,
4818 groups: _FrameIntTuple | FrameClause | None = None,
4819 ) -> Over[_T]:
4820 """Produce an OVER clause against this filtered function.
4822 Used against aggregate or so-called "window" functions,
4823 for database backends that support window functions.
4825 The expression::
4827 func.rank().filter(MyClass.y > 5).over(order_by="x")
4829 is shorthand for::
4831 from sqlalchemy import over, funcfilter
4833 over(funcfilter(func.rank(), MyClass.y > 5), order_by="x")
4835 See :func:`_expression.over` for a full description.
4837 """
4838 return Over(
4839 self,
4840 partition_by=partition_by,
4841 order_by=order_by,
4842 range_=range_,
4843 rows=rows,
4844 groups=groups,
4845 )
4847 def within_group(
4848 self, *order_by: _ColumnExpressionArgument[Any]
4849 ) -> WithinGroup[_T]:
4850 """Produce a WITHIN GROUP (ORDER BY expr) clause against
4851 this function.
4852 """
4853 return WithinGroup(self, *order_by)
4855 def within_group_type(
4856 self, within_group: WithinGroup[_T]
4857 ) -> Optional[TypeEngine[_T]]:
4858 return None
4860 def self_group(
4861 self, against: Optional[OperatorType] = None
4862 ) -> Union[Self, Grouping[_T]]:
4863 if operators.is_precedent(operators.filter_op, against):
4864 return Grouping(self)
4865 else:
4866 return self
4868 if not TYPE_CHECKING:
4870 @util.memoized_property
4871 def type(self) -> TypeEngine[_T]: # noqa: A001
4872 return self.func.type
4874 @util.ro_non_memoized_property
4875 def _from_objects(self) -> List[FromClause]:
4876 return list(
4877 itertools.chain(
4878 *[
4879 c._from_objects
4880 for c in (self.func, self.criterion)
4881 if c is not None
4882 ]
4883 )
4884 )
4887class NamedColumn(KeyedColumnElement[_T]):
4888 is_literal = False
4889 table: Optional[FromClause] = None
4890 name: str
4891 key: str
4893 def _compare_name_for_result(self, other):
4894 return (hasattr(other, "name") and self.name == other.name) or (
4895 hasattr(other, "_label") and self._label == other._label
4896 )
4898 @util.ro_memoized_property
4899 def description(self) -> str:
4900 return self.name
4902 @HasMemoized.memoized_attribute
4903 def _tq_key_label(self) -> Optional[str]:
4904 """table qualified label based on column key.
4906 for table-bound columns this is <tablename>_<column key/proxy key>;
4908 all other expressions it resolves to key/proxy key.
4910 """
4911 proxy_key = self._proxy_key
4912 if proxy_key and proxy_key != self.name:
4913 return self._gen_tq_label(proxy_key)
4914 else:
4915 return self._tq_label
4917 @HasMemoized.memoized_attribute
4918 def _tq_label(self) -> Optional[str]:
4919 """table qualified label based on column name.
4921 for table-bound columns this is <tablename>_<columnname>; all other
4922 expressions it resolves to .name.
4924 """
4925 return self._gen_tq_label(self.name)
4927 @HasMemoized.memoized_attribute
4928 def _render_label_in_columns_clause(self):
4929 return True
4931 @HasMemoized.memoized_attribute
4932 def _non_anon_label(self):
4933 return self.name
4935 def _gen_tq_label(
4936 self, name: str, dedupe_on_key: bool = True
4937 ) -> Optional[str]:
4938 return name
4940 def _bind_param(
4941 self,
4942 operator: OperatorType,
4943 obj: Any,
4944 type_: Optional[TypeEngine[_T]] = None,
4945 expanding: bool = False,
4946 ) -> BindParameter[_T]:
4947 return BindParameter(
4948 self.key,
4949 obj,
4950 _compared_to_operator=operator,
4951 _compared_to_type=self.type,
4952 type_=type_,
4953 unique=True,
4954 expanding=expanding,
4955 )
4957 def _make_proxy(
4958 self,
4959 selectable: FromClause,
4960 *,
4961 primary_key: ColumnSet,
4962 foreign_keys: Set[KeyedColumnElement[Any]],
4963 name: Optional[str] = None,
4964 key: Optional[str] = None,
4965 name_is_truncatable: bool = False,
4966 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
4967 disallow_is_literal: bool = False,
4968 **kw: Any,
4969 ) -> typing_Tuple[str, ColumnClause[_T]]:
4970 c = ColumnClause(
4971 (
4972 coercions.expect(roles.TruncatedLabelRole, name or self.name)
4973 if name_is_truncatable
4974 else (name or self.name)
4975 ),
4976 type_=self.type,
4977 _selectable=selectable,
4978 is_literal=False,
4979 )
4981 c._propagate_attrs = selectable._propagate_attrs
4982 if name is None:
4983 c.key = self.key
4984 if compound_select_cols:
4985 c._proxies = list(compound_select_cols)
4986 else:
4987 c._proxies = [self]
4989 if selectable._is_clone_of is not None:
4990 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
4991 return c.key, c
4994_PS = ParamSpec("_PS")
4997class Label(roles.LabeledColumnExprRole[_T], NamedColumn[_T]):
4998 """Represents a column label (AS).
5000 Represent a label, as typically applied to any column-level
5001 element using the ``AS`` sql keyword.
5003 """
5005 __visit_name__ = "label"
5007 _traverse_internals: _TraverseInternalsType = [
5008 ("name", InternalTraversal.dp_anon_name),
5009 ("type", InternalTraversal.dp_type),
5010 ("_element", InternalTraversal.dp_clauseelement),
5011 ]
5013 _cache_key_traversal = [
5014 ("name", InternalTraversal.dp_anon_name),
5015 ("_element", InternalTraversal.dp_clauseelement),
5016 ]
5018 _element: ColumnElement[_T]
5019 name: str
5021 def __init__(
5022 self,
5023 name: Optional[str],
5024 element: _ColumnExpressionArgument[_T],
5025 type_: Optional[_TypeEngineArgument[_T]] = None,
5026 ):
5027 orig_element = element
5028 element = coercions.expect(
5029 roles.ExpressionElementRole,
5030 element,
5031 apply_propagate_attrs=self,
5032 )
5033 while isinstance(element, Label):
5034 # TODO: this is only covered in test_text.py, but nothing
5035 # fails if it's removed. determine rationale
5036 element = element.element
5038 if name:
5039 self.name = name
5040 else:
5041 self.name = _anonymous_label.safe_construct(
5042 id(self), getattr(element, "name", "anon")
5043 )
5044 if isinstance(orig_element, Label):
5045 # TODO: no coverage for this block, again would be in
5046 # test_text.py where the resolve_label concept is important
5047 self._resolve_label = orig_element._label
5049 self.key = self._tq_label = self._tq_key_label = self.name
5050 self._element = element
5052 self.type = (
5053 type_api.to_instance(type_)
5054 if type_ is not None
5055 else self._element.type
5056 )
5058 self._proxies = [element]
5060 def __reduce__(self):
5061 return self.__class__, (self.name, self._element, self.type)
5063 @HasMemoized.memoized_attribute
5064 def _render_label_in_columns_clause(self):
5065 return True
5067 def _bind_param(self, operator, obj, type_=None, expanding=False):
5068 return BindParameter(
5069 None,
5070 obj,
5071 _compared_to_operator=operator,
5072 type_=type_,
5073 _compared_to_type=self.type,
5074 unique=True,
5075 expanding=expanding,
5076 )
5078 @util.memoized_property
5079 def _is_implicitly_boolean(self):
5080 return self.element._is_implicitly_boolean
5082 @HasMemoized.memoized_attribute
5083 def _allow_label_resolve(self):
5084 return self.element._allow_label_resolve
5086 @property
5087 def _order_by_label_element(self):
5088 return self
5090 def as_reference(self) -> _label_reference[_T]:
5091 """refer to this labeled expression in a clause such as GROUP BY,
5092 ORDER BY etc. as the label name itself, without expanding
5093 into the full expression.
5095 .. versionadded:: 2.1
5097 """
5098 return _label_reference(self)
5100 @HasMemoized.memoized_attribute
5101 def element(self) -> ColumnElement[_T]:
5102 return self._element.self_group(against=operators.as_)
5104 def self_group(self, against: Optional[OperatorType] = None) -> Label[_T]:
5105 return self._apply_to_inner(self._element.self_group, against=against)
5107 def _negate(self):
5108 return self._apply_to_inner(self._element._negate)
5110 def _apply_to_inner(
5111 self,
5112 fn: Callable[_PS, ColumnElement[_T]],
5113 *arg: _PS.args,
5114 **kw: _PS.kwargs,
5115 ) -> Label[_T]:
5116 sub_element = fn(*arg, **kw)
5117 if sub_element is not self._element:
5118 return Label(self.name, sub_element, type_=self.type)
5119 else:
5120 return self
5122 @property
5123 def primary_key(self): # type: ignore[override]
5124 return self.element.primary_key
5126 @property
5127 def foreign_keys(self): # type: ignore[override]
5128 return self.element.foreign_keys
5130 def _copy_internals(
5131 self,
5132 *,
5133 clone: _CloneCallableType = _clone,
5134 anonymize_labels: bool = False,
5135 **kw: Any,
5136 ) -> None:
5137 self._reset_memoizations()
5138 self._element = clone(self._element, **kw)
5139 if anonymize_labels:
5140 self.name = _anonymous_label.safe_construct(
5141 id(self), getattr(self.element, "name", "anon")
5142 )
5143 self.key = self._tq_label = self._tq_key_label = self.name
5145 @util.ro_non_memoized_property
5146 def _from_objects(self) -> List[FromClause]:
5147 return self.element._from_objects
5149 def _make_proxy(
5150 self,
5151 selectable: FromClause,
5152 *,
5153 primary_key: ColumnSet,
5154 foreign_keys: Set[KeyedColumnElement[Any]],
5155 name: Optional[str] = None,
5156 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5157 **kw: Any,
5158 ) -> typing_Tuple[str, ColumnClause[_T]]:
5159 name = self.name if not name else name
5161 key, e = self.element._make_proxy(
5162 selectable,
5163 name=name,
5164 disallow_is_literal=True,
5165 name_is_truncatable=isinstance(name, _truncated_label),
5166 compound_select_cols=compound_select_cols,
5167 primary_key=primary_key,
5168 foreign_keys=foreign_keys,
5169 )
5171 # there was a note here to remove this assertion, which was here
5172 # to determine if we later could support a use case where
5173 # the key and name of a label are separate. But I don't know what
5174 # that case was. For now, this is an unexpected case that occurs
5175 # when a label name conflicts with other columns and select()
5176 # is attempting to disambiguate an explicit label, which is not what
5177 # the user would want. See issue #6090.
5178 if key != self.name and not isinstance(self.name, _anonymous_label):
5179 raise exc.InvalidRequestError(
5180 "Label name %s is being renamed to an anonymous label due "
5181 "to disambiguation "
5182 "which is not supported right now. Please use unique names "
5183 "for explicit labels." % (self.name)
5184 )
5186 e._propagate_attrs = selectable._propagate_attrs
5187 e._proxies.append(self)
5188 if self.type is not None:
5189 e.type = self.type
5191 return self.key, e
5194class ColumnClause(
5195 roles.DDLReferredColumnRole,
5196 roles.LabeledColumnExprRole[_T],
5197 roles.StrAsPlainColumnRole,
5198 Immutable,
5199 NamedColumn[_T],
5200):
5201 """Represents a column expression from any textual string.
5203 The :class:`.ColumnClause`, a lightweight analogue to the
5204 :class:`_schema.Column` class, is typically invoked using the
5205 :func:`_expression.column` function, as in::
5207 from sqlalchemy import column
5209 id, name = column("id"), column("name")
5210 stmt = select(id, name).select_from("user")
5212 The above statement would produce SQL like:
5214 .. sourcecode:: sql
5216 SELECT id, name FROM user
5218 :class:`.ColumnClause` is the immediate superclass of the schema-specific
5219 :class:`_schema.Column` object. While the :class:`_schema.Column`
5220 class has all the
5221 same capabilities as :class:`.ColumnClause`, the :class:`.ColumnClause`
5222 class is usable by itself in those cases where behavioral requirements
5223 are limited to simple SQL expression generation. The object has none of
5224 the associations with schema-level metadata or with execution-time
5225 behavior that :class:`_schema.Column` does,
5226 so in that sense is a "lightweight"
5227 version of :class:`_schema.Column`.
5229 Full details on :class:`.ColumnClause` usage is at
5230 :func:`_expression.column`.
5232 .. seealso::
5234 :func:`_expression.column`
5236 :class:`_schema.Column`
5238 """
5240 table: Optional[FromClause]
5241 is_literal: bool
5243 __visit_name__ = "column"
5245 _traverse_internals: _TraverseInternalsType = [
5246 ("name", InternalTraversal.dp_anon_name),
5247 ("type", InternalTraversal.dp_type),
5248 ("table", InternalTraversal.dp_clauseelement),
5249 ("is_literal", InternalTraversal.dp_boolean),
5250 ]
5252 onupdate: Optional[DefaultGenerator] = None
5253 default: Optional[DefaultGenerator] = None
5254 server_default: Optional[FetchedValue] = None
5255 server_onupdate: Optional[FetchedValue] = None
5257 _is_multiparam_column = False
5259 @property
5260 def _is_star(self): # type: ignore[override]
5261 return self.is_literal and self.name == "*"
5263 def __init__(
5264 self,
5265 text: str,
5266 type_: Optional[_TypeEngineArgument[_T]] = None,
5267 is_literal: bool = False,
5268 _selectable: Optional[FromClause] = None,
5269 ):
5270 self.key = self.name = text
5271 self.table = _selectable
5273 # if type is None, we get NULLTYPE, which is our _T. But I don't
5274 # know how to get the overloads to express that correctly
5275 self.type = type_api.to_instance(type_) # type: ignore
5277 self.is_literal = is_literal
5279 def get_children(self, *, column_tables=False, **kw):
5280 # override base get_children() to not return the Table
5281 # or selectable that is parent to this column. Traversals
5282 # expect the columns of tables and subqueries to be leaf nodes.
5283 return []
5285 @property
5286 def entity_namespace(self):
5287 if self.table is not None:
5288 return self.table.entity_namespace
5289 else:
5290 return super().entity_namespace
5292 def _clone(self, detect_subquery_cols=False, **kw):
5293 if (
5294 detect_subquery_cols
5295 and self.table is not None
5296 and self.table._is_subquery
5297 ):
5298 clone = kw.pop("clone")
5299 table = clone(self.table, **kw)
5300 new = table.c.corresponding_column(self)
5301 return new
5303 return super()._clone(**kw)
5305 @HasMemoized_ro_memoized_attribute
5306 def _from_objects(self) -> List[FromClause]:
5307 t = self.table
5308 if t is not None:
5309 return [t]
5310 else:
5311 return []
5313 @HasMemoized.memoized_attribute
5314 def _render_label_in_columns_clause(self):
5315 return self.table is not None
5317 @property
5318 def _ddl_label(self):
5319 return self._gen_tq_label(self.name, dedupe_on_key=False)
5321 def _compare_name_for_result(self, other):
5322 if (
5323 self.is_literal
5324 or self.table is None
5325 or self.table._is_textual
5326 or not hasattr(other, "proxy_set")
5327 or (
5328 isinstance(other, ColumnClause)
5329 and (
5330 other.is_literal
5331 or other.table is None
5332 or other.table._is_textual
5333 )
5334 )
5335 ):
5336 return (hasattr(other, "name") and self.name == other.name) or (
5337 hasattr(other, "_tq_label")
5338 and self._tq_label == other._tq_label
5339 )
5340 else:
5341 return other.proxy_set.intersection(self.proxy_set)
5343 def _gen_tq_label(
5344 self, name: str, dedupe_on_key: bool = True
5345 ) -> Optional[str]:
5346 """generate table-qualified label
5348 for a table-bound column this is <tablename>_<columnname>.
5350 used primarily for LABEL_STYLE_TABLENAME_PLUS_COL
5351 as well as the .columns collection on a Join object.
5353 """
5354 label: str
5355 t = self.table
5356 if self.is_literal:
5357 return None
5358 elif t is not None and is_named_from_clause(t):
5359 if has_schema_attr(t) and t.schema:
5360 label = (
5361 t.schema.replace(".", "_") + "_" + t.name + ("_" + name)
5362 )
5363 else:
5364 assert not TYPE_CHECKING or isinstance(t, NamedFromClause)
5365 label = t.name + ("_" + name)
5367 # propagate name quoting rules for labels.
5368 if is_quoted_name(name) and name.quote is not None:
5369 if is_quoted_name(label):
5370 label.quote = name.quote
5371 else:
5372 label = quoted_name(label, name.quote)
5373 elif is_quoted_name(t.name) and t.name.quote is not None:
5374 # can't get this situation to occur, so let's
5375 # assert false on it for now
5376 assert not isinstance(label, quoted_name)
5377 label = quoted_name(label, t.name.quote)
5379 if dedupe_on_key:
5380 # ensure the label name doesn't conflict with that of an
5381 # existing column. note that this implies that any Column
5382 # must **not** set up its _label before its parent table has
5383 # all of its other Column objects set up. There are several
5384 # tables in the test suite which will fail otherwise; example:
5385 # table "owner" has columns "name" and "owner_name". Therefore
5386 # column owner.name cannot use the label "owner_name", it has
5387 # to be "owner_name_1".
5388 if label in t.c:
5389 _label = label
5390 counter = 1
5391 while _label in t.c:
5392 _label = label + f"_{counter}"
5393 counter += 1
5394 label = _label
5396 return coercions.expect(roles.TruncatedLabelRole, label)
5398 else:
5399 return name
5401 def _make_proxy(
5402 self,
5403 selectable: FromClause,
5404 *,
5405 primary_key: ColumnSet,
5406 foreign_keys: Set[KeyedColumnElement[Any]],
5407 name: Optional[str] = None,
5408 key: Optional[str] = None,
5409 name_is_truncatable: bool = False,
5410 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5411 disallow_is_literal: bool = False,
5412 **kw: Any,
5413 ) -> typing_Tuple[str, ColumnClause[_T]]:
5414 # the "is_literal" flag normally should never be propagated; a proxied
5415 # column is always a SQL identifier and never the actual expression
5416 # being evaluated. however, there is a case where the "is_literal" flag
5417 # might be used to allow the given identifier to have a fixed quoting
5418 # pattern already, so maintain the flag for the proxy unless a
5419 # :class:`.Label` object is creating the proxy. See [ticket:4730].
5420 is_literal = (
5421 not disallow_is_literal
5422 and self.is_literal
5423 and (
5424 # note this does not accommodate for quoted_name differences
5425 # right now
5426 name is None
5427 or name == self.name
5428 )
5429 )
5430 c = self._constructor(
5431 (
5432 coercions.expect(roles.TruncatedLabelRole, name or self.name)
5433 if name_is_truncatable
5434 else (name or self.name)
5435 ),
5436 type_=self.type,
5437 _selectable=selectable,
5438 is_literal=is_literal,
5439 )
5440 c._propagate_attrs = selectable._propagate_attrs
5441 if name is None:
5442 c.key = self.key
5443 if compound_select_cols:
5444 c._proxies = list(compound_select_cols)
5445 else:
5446 c._proxies = [self]
5448 if selectable._is_clone_of is not None:
5449 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
5450 return c.key, c
5453class TableValuedColumn(NamedColumn[_T]):
5454 __visit_name__ = "table_valued_column"
5456 _traverse_internals: _TraverseInternalsType = [
5457 ("name", InternalTraversal.dp_anon_name),
5458 ("type", InternalTraversal.dp_type),
5459 ("scalar_alias", InternalTraversal.dp_clauseelement),
5460 ]
5462 def __init__(self, scalar_alias: NamedFromClause, type_: TypeEngine[_T]):
5463 self.scalar_alias = scalar_alias
5464 self.key = self.name = scalar_alias.name
5465 self.type = type_
5467 def _copy_internals(
5468 self, clone: _CloneCallableType = _clone, **kw: Any
5469 ) -> None:
5470 self.scalar_alias = clone(self.scalar_alias, **kw)
5471 self.key = self.name = self.scalar_alias.name
5473 @util.ro_non_memoized_property
5474 def _from_objects(self) -> List[FromClause]:
5475 return [self.scalar_alias]
5478class CollationClause(ColumnElement[str]):
5479 __visit_name__ = "collation"
5481 _traverse_internals: _TraverseInternalsType = [
5482 ("collation", InternalTraversal.dp_string)
5483 ]
5485 @classmethod
5486 @util.preload_module("sqlalchemy.sql.sqltypes")
5487 def _create_collation_expression(
5488 cls, expression: _ColumnExpressionArgument[str], collation: str
5489 ) -> BinaryExpression[str]:
5491 sqltypes = util.preloaded.sql_sqltypes
5493 expr = coercions.expect(roles.ExpressionElementRole[str], expression)
5495 if expr.type._type_affinity is sqltypes.String:
5496 collate_type = expr.type._with_collation(collation)
5497 else:
5498 collate_type = expr.type
5500 return BinaryExpression(
5501 expr,
5502 CollationClause(collation),
5503 operators.collate,
5504 type_=collate_type,
5505 )
5507 def __init__(self, collation):
5508 self.collation = collation
5511class _IdentifiedClause(Executable, ClauseElement):
5512 __visit_name__ = "identified"
5514 def __init__(self, ident):
5515 self.ident = ident
5518class SavepointClause(_IdentifiedClause):
5519 __visit_name__ = "savepoint"
5520 inherit_cache = False
5523class RollbackToSavepointClause(_IdentifiedClause):
5524 __visit_name__ = "rollback_to_savepoint"
5525 inherit_cache = False
5528class ReleaseSavepointClause(_IdentifiedClause):
5529 __visit_name__ = "release_savepoint"
5530 inherit_cache = False
5533class quoted_name(util.MemoizedSlots, str):
5534 """Represent a SQL identifier combined with quoting preferences.
5536 :class:`.quoted_name` is a Python unicode/str subclass which
5537 represents a particular identifier name along with a
5538 ``quote`` flag. This ``quote`` flag, when set to
5539 ``True`` or ``False``, overrides automatic quoting behavior
5540 for this identifier in order to either unconditionally quote
5541 or to not quote the name. If left at its default of ``None``,
5542 quoting behavior is applied to the identifier on a per-backend basis
5543 based on an examination of the token itself.
5545 A :class:`.quoted_name` object with ``quote=True`` is also
5546 prevented from being modified in the case of a so-called
5547 "name normalize" option. Certain database backends, such as
5548 Oracle Database, Firebird, and DB2 "normalize" case-insensitive names
5549 as uppercase. The SQLAlchemy dialects for these backends
5550 convert from SQLAlchemy's lower-case-means-insensitive convention
5551 to the upper-case-means-insensitive conventions of those backends.
5552 The ``quote=True`` flag here will prevent this conversion from occurring
5553 to support an identifier that's quoted as all lower case against
5554 such a backend.
5556 The :class:`.quoted_name` object is normally created automatically
5557 when specifying the name for key schema constructs such as
5558 :class:`_schema.Table`, :class:`_schema.Column`, and others.
5559 The class can also be
5560 passed explicitly as the name to any function that receives a name which
5561 can be quoted. Such as to use the :meth:`_engine.Engine.has_table`
5562 method with
5563 an unconditionally quoted name::
5565 from sqlalchemy import create_engine
5566 from sqlalchemy import inspect
5567 from sqlalchemy.sql import quoted_name
5569 engine = create_engine("oracle+oracledb://some_dsn")
5570 print(inspect(engine).has_table(quoted_name("some_table", True)))
5572 The above logic will run the "has table" logic against the Oracle Database
5573 backend, passing the name exactly as ``"some_table"`` without converting to
5574 upper case.
5576 """
5578 __slots__ = "quote", "lower", "upper"
5580 quote: Optional[bool]
5582 @overload
5583 @classmethod
5584 def construct(cls, value: str, quote: Optional[bool]) -> quoted_name: ...
5586 @overload
5587 @classmethod
5588 def construct(cls, value: None, quote: Optional[bool]) -> None: ...
5590 @classmethod
5591 def construct(
5592 cls, value: Optional[str], quote: Optional[bool]
5593 ) -> Optional[quoted_name]:
5594 if value is None:
5595 return None
5596 else:
5597 return quoted_name(value, quote)
5599 def __new__(cls, value: str, quote: Optional[bool]) -> quoted_name:
5600 assert (
5601 value is not None
5602 ), "use quoted_name.construct() for None passthrough"
5603 if isinstance(value, cls) and (quote is None or value.quote == quote):
5604 return value
5605 self = super().__new__(cls, value)
5607 self.quote = quote
5608 return self
5610 def __reduce__(self):
5611 return quoted_name, (str(self), self.quote)
5613 def _memoized_method_lower(self):
5614 if self.quote:
5615 return self
5616 else:
5617 return str(self).lower()
5619 def _memoized_method_upper(self):
5620 if self.quote:
5621 return self
5622 else:
5623 return str(self).upper()
5626def _find_columns(clause: ClauseElement) -> Set[ColumnClause[Any]]:
5627 """locate Column objects within the given expression."""
5629 cols: Set[ColumnClause[Any]] = set()
5630 traverse(clause, {}, {"column": cols.add})
5631 return cols
5634def _type_from_args(args: Sequence[ColumnElement[_T]]) -> TypeEngine[_T]:
5635 for a in args:
5636 if not a.type._isnull:
5637 return a.type
5638 else:
5639 return type_api.NULLTYPE # type: ignore
5642def _corresponding_column_or_error(fromclause, column, require_embedded=False):
5643 c = fromclause.corresponding_column(
5644 column, require_embedded=require_embedded
5645 )
5646 if c is None:
5647 raise exc.InvalidRequestError(
5648 "Given column '%s', attached to table '%s', "
5649 "failed to locate a corresponding column from table '%s'"
5650 % (column, getattr(column, "table", None), fromclause.description)
5651 )
5652 return c
5655class _memoized_property_but_not_nulltype(
5656 util.memoized_property["TypeEngine[_T]"]
5657):
5658 """memoized property, but dont memoize NullType"""
5660 def __get__(self, obj, cls):
5661 if obj is None:
5662 return self
5663 result = self.fget(obj)
5664 if not result._isnull:
5665 obj.__dict__[self.__name__] = result
5666 return result
5669class AnnotatedColumnElement(Annotated):
5670 _Annotated__element: ColumnElement[Any]
5672 def __init__(self, element, values):
5673 Annotated.__init__(self, element, values)
5674 for attr in (
5675 "comparator",
5676 "_proxy_key",
5677 "_tq_key_label",
5678 "_tq_label",
5679 "_non_anon_label",
5680 "type",
5681 ):
5682 self.__dict__.pop(attr, None)
5683 for attr in ("name", "key", "table"):
5684 if self.__dict__.get(attr, False) is None:
5685 self.__dict__.pop(attr)
5687 def _with_annotations(self, values):
5688 clone = super()._with_annotations(values)
5689 for attr in (
5690 "comparator",
5691 "_proxy_key",
5692 "_tq_key_label",
5693 "_tq_label",
5694 "_non_anon_label",
5695 ):
5696 clone.__dict__.pop(attr, None)
5697 return clone
5699 @util.memoized_property
5700 def name(self):
5701 """pull 'name' from parent, if not present"""
5702 return self._Annotated__element.name
5704 @_memoized_property_but_not_nulltype
5705 def type(self):
5706 """pull 'type' from parent and don't cache if null.
5708 type is routinely changed on existing columns within the
5709 mapped_column() initialization process, and "type" is also consulted
5710 during the creation of SQL expressions. Therefore it can change after
5711 it was already retrieved. At the same time we don't want annotated
5712 objects having overhead when expressions are produced, so continue
5713 to memoize, but only when we have a non-null type.
5715 """
5716 return self._Annotated__element.type
5718 @util.memoized_property
5719 def table(self):
5720 """pull 'table' from parent, if not present"""
5721 return self._Annotated__element.table
5723 @util.memoized_property
5724 def key(self):
5725 """pull 'key' from parent, if not present"""
5726 return self._Annotated__element.key
5728 @util.memoized_property
5729 def info(self) -> _InfoType:
5730 if TYPE_CHECKING:
5731 assert isinstance(self._Annotated__element, Column)
5732 return self._Annotated__element.info
5734 @util.memoized_property
5735 def _anon_name_label(self) -> str:
5736 return self._Annotated__element._anon_name_label
5739class _truncated_label(quoted_name):
5740 """A unicode subclass used to identify symbolic "
5741 "names that may require truncation."""
5743 __slots__ = ()
5745 def __new__(cls, value: str, quote: Optional[bool] = None) -> Any:
5746 quote = getattr(value, "quote", quote)
5747 # return super(_truncated_label, cls).__new__(cls, value, quote, True)
5748 return super().__new__(cls, value, quote)
5750 def __reduce__(self) -> Any:
5751 return self.__class__, (str(self), self.quote)
5753 def apply_map(self, map_: Mapping[str, Any]) -> str:
5754 return self
5757class conv(_truncated_label):
5758 """Mark a string indicating that a name has already been converted
5759 by a naming convention.
5761 This is a string subclass that indicates a name that should not be
5762 subject to any further naming conventions.
5764 E.g. when we create a :class:`.Constraint` using a naming convention
5765 as follows::
5767 m = MetaData(
5768 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5769 )
5770 t = Table(
5771 "t", m, Column("x", Integer), CheckConstraint("x > 5", name="x5")
5772 )
5774 The name of the above constraint will be rendered as ``"ck_t_x5"``.
5775 That is, the existing name ``x5`` is used in the naming convention as the
5776 ``constraint_name`` token.
5778 In some situations, such as in migration scripts, we may be rendering
5779 the above :class:`.CheckConstraint` with a name that's already been
5780 converted. In order to make sure the name isn't double-modified, the
5781 new name is applied using the :func:`_schema.conv` marker. We can
5782 use this explicitly as follows::
5785 m = MetaData(
5786 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5787 )
5788 t = Table(
5789 "t",
5790 m,
5791 Column("x", Integer),
5792 CheckConstraint("x > 5", name=conv("ck_t_x5")),
5793 )
5795 Where above, the :func:`_schema.conv` marker indicates that the constraint
5796 name here is final, and the name will render as ``"ck_t_x5"`` and not
5797 ``"ck_t_ck_t_x5"``
5799 .. seealso::
5801 :ref:`constraint_naming_conventions`
5803 """
5805 __slots__ = ()
5808# for backwards compatibility in case
5809# someone is re-implementing the
5810# _truncated_identifier() sequence in a custom
5811# compiler
5812_generated_label = _truncated_label
5813_anonymous_label_escape = re.compile(r"[%\(\) \$]+")
5816class _anonymous_label(_truncated_label):
5817 """A unicode subclass used to identify anonymously
5818 generated names."""
5820 __slots__ = ()
5822 @classmethod
5823 def safe_construct_with_key(
5824 cls, seed: int, body: str, sanitize_key: bool = False
5825 ) -> typing_Tuple[_anonymous_label, str]:
5826 # need to escape chars that interfere with format
5827 # strings in any case, issue #8724
5828 body = _anonymous_label_escape.sub("_", body)
5830 if sanitize_key:
5831 # sanitize_key is then an extra step used by BindParameter
5832 body = body.strip("_")
5834 key = f"{seed} {body.replace('%', '%%')}"
5835 label = _anonymous_label(f"%({key})s")
5836 return label, key
5838 @classmethod
5839 def safe_construct(
5840 cls, seed: int, body: str, sanitize_key: bool = False
5841 ) -> _anonymous_label:
5842 # need to escape chars that interfere with format
5843 # strings in any case, issue #8724
5844 body = _anonymous_label_escape.sub("_", body)
5846 if sanitize_key:
5847 # sanitize_key is then an extra step used by BindParameter
5848 body = body.strip("_")
5850 return _anonymous_label(f"%({seed} {body.replace('%', '%%')})s")
5852 def __add__(self, other: str) -> _anonymous_label:
5853 if "%" in other and not isinstance(other, _anonymous_label):
5854 other = str(other).replace("%", "%%")
5855 else:
5856 other = str(other)
5858 return _anonymous_label(
5859 quoted_name(
5860 str.__add__(self, other),
5861 self.quote,
5862 )
5863 )
5865 def __radd__(self, other: str) -> _anonymous_label:
5866 if "%" in other and not isinstance(other, _anonymous_label):
5867 other = str(other).replace("%", "%%")
5868 else:
5869 other = str(other)
5871 return _anonymous_label(
5872 quoted_name(
5873 str.__add__(other, self),
5874 self.quote,
5875 )
5876 )
5878 def apply_map(self, map_: Mapping[str, Any]) -> str:
5879 if self.quote is not None:
5880 # preserve quoting only if necessary
5881 return quoted_name(self % map_, self.quote)
5882 else:
5883 # else skip the constructor call
5884 return self % map_