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 Mapping
33from typing import Optional
34from typing import overload
35from typing import Sequence
36from typing import Set
37from typing import Tuple as typing_Tuple
38from typing import Type
39from typing import TYPE_CHECKING
40from typing import TypeVar
41from typing import Union
43from . import coercions
44from . import operators
45from . import roles
46from . import traversals
47from . import type_api
48from ._typing import has_schema_attr
49from ._typing import is_named_from_clause
50from ._typing import is_quoted_name
51from ._typing import is_tuple_type
52from .annotation import Annotated
53from .annotation import SupportsWrappingAnnotations
54from .base import _clone
55from .base import _expand_cloned
56from .base import _generative
57from .base import _NoArg
58from .base import Executable
59from .base import Generative
60from .base import HasMemoized
61from .base import Immutable
62from .base import NO_ARG
63from .base import SingletonConstant
64from .cache_key import MemoizedHasCacheKey
65from .cache_key import NO_CACHE
66from .coercions import _document_text_coercion # noqa
67from .operators import ColumnOperators
68from .traversals import HasCopyInternals
69from .visitors import cloned_traverse
70from .visitors import ExternallyTraversible
71from .visitors import InternalTraversal
72from .visitors import traverse
73from .visitors import Visitable
74from .. import exc
75from .. import inspection
76from .. import util
77from ..util import HasMemoized_ro_memoized_attribute
78from ..util import TypingOnly
79from ..util.typing import Literal
80from ..util.typing import ParamSpec
81from ..util.typing import Self
84if typing.TYPE_CHECKING:
85 from ._typing import _ByArgument
86 from ._typing import _ColumnExpressionArgument
87 from ._typing import _ColumnExpressionOrStrLabelArgument
88 from ._typing import _HasDialect
89 from ._typing import _InfoType
90 from ._typing import _PropagateAttrsType
91 from ._typing import _TypeEngineArgument
92 from .base import _EntityNamespace
93 from .base import ColumnSet
94 from .cache_key import _CacheKeyTraversalType
95 from .cache_key import CacheKey
96 from .compiler import Compiled
97 from .compiler import SQLCompiler
98 from .functions import FunctionElement
99 from .operators import OperatorType
100 from .schema import Column
101 from .schema import DefaultGenerator
102 from .schema import FetchedValue
103 from .schema import ForeignKey
104 from .selectable import _SelectIterable
105 from .selectable import FromClause
106 from .selectable import NamedFromClause
107 from .selectable import TextualSelect
108 from .sqltypes import TupleType
109 from .type_api import TypeEngine
110 from .visitors import _CloneCallableType
111 from .visitors import _TraverseInternalsType
112 from .visitors import anon_map
113 from ..engine import Connection
114 from ..engine import Dialect
115 from ..engine.interfaces import _CoreMultiExecuteParams
116 from ..engine.interfaces import CacheStats
117 from ..engine.interfaces import CompiledCacheType
118 from ..engine.interfaces import CoreExecuteOptionsParameter
119 from ..engine.interfaces import SchemaTranslateMapType
120 from ..engine.result import Result
123_NUMERIC = Union[float, Decimal]
124_NUMBER = Union[float, int, Decimal]
126_T = TypeVar("_T", bound="Any")
127_T_co = TypeVar("_T_co", bound=Any, covariant=True)
128_OPT = TypeVar("_OPT", bound="Any")
129_NT = TypeVar("_NT", bound="_NUMERIC")
131_NMT = TypeVar("_NMT", bound="_NUMBER")
134@overload
135def literal(
136 value: Any,
137 type_: _TypeEngineArgument[_T],
138 literal_execute: bool = False,
139) -> BindParameter[_T]: ...
142@overload
143def literal(
144 value: _T,
145 type_: None = None,
146 literal_execute: bool = False,
147) -> BindParameter[_T]: ...
150@overload
151def literal(
152 value: Any,
153 type_: Optional[_TypeEngineArgument[Any]] = None,
154 literal_execute: bool = False,
155) -> BindParameter[Any]: ...
158def literal(
159 value: Any,
160 type_: Optional[_TypeEngineArgument[Any]] = None,
161 literal_execute: bool = False,
162) -> BindParameter[Any]:
163 r"""Return a literal clause, bound to a bind parameter.
165 Literal clauses are created automatically when non-
166 :class:`_expression.ClauseElement` objects (such as strings, ints, dates,
167 etc.) are
168 used in a comparison operation with a :class:`_expression.ColumnElement`
169 subclass,
170 such as a :class:`~sqlalchemy.schema.Column` object. Use this function
171 to force the generation of a literal clause, which will be created as a
172 :class:`BindParameter` with a bound value.
174 :param value: the value to be bound. Can be any Python object supported by
175 the underlying DB-API, or is translatable via the given type argument.
177 :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine` which will
178 provide bind-parameter translation for this literal.
180 :param literal_execute: optional bool, when True, the SQL engine will
181 attempt to render the bound value directly in the SQL statement at
182 execution time rather than providing as a parameter value.
184 .. versionadded:: 2.0
186 """
187 return coercions.expect(
188 roles.LiteralValueRole,
189 value,
190 type_=type_,
191 literal_execute=literal_execute,
192 )
195def literal_column(
196 text: str, type_: Optional[_TypeEngineArgument[_T]] = None
197) -> ColumnClause[_T]:
198 r"""Produce a :class:`.ColumnClause` object that has the
199 :paramref:`_expression.column.is_literal` flag set to True.
201 :func:`_expression.literal_column` is similar to
202 :func:`_expression.column`, except that
203 it is more often used as a "standalone" column expression that renders
204 exactly as stated; while :func:`_expression.column`
205 stores a string name that
206 will be assumed to be part of a table and may be quoted as such,
207 :func:`_expression.literal_column` can be that,
208 or any other arbitrary column-oriented
209 expression.
211 :param text: the text of the expression; can be any SQL expression.
212 Quoting rules will not be applied. To specify a column-name expression
213 which should be subject to quoting rules, use the :func:`column`
214 function.
216 :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine`
217 object which will
218 provide result-set translation and additional expression semantics for
219 this column. If left as ``None`` the type will be :class:`.NullType`.
221 .. seealso::
223 :func:`_expression.column`
225 :func:`_expression.text`
227 :ref:`tutorial_select_arbitrary_text`
229 """
230 return ColumnClause(text, type_=type_, is_literal=True)
233class CompilerElement(Visitable):
234 """base class for SQL elements that can be compiled to produce a
235 SQL string.
237 .. versionadded:: 2.0
239 """
241 __slots__ = ()
242 __visit_name__ = "compiler_element"
244 supports_execution = False
246 stringify_dialect = "default"
248 @util.preload_module("sqlalchemy.engine.default")
249 @util.preload_module("sqlalchemy.engine.url")
250 def compile(
251 self,
252 bind: Optional[_HasDialect] = None,
253 dialect: Optional[Dialect] = None,
254 **kw: Any,
255 ) -> Compiled:
256 """Compile this SQL expression.
258 The return value is a :class:`~.Compiled` object.
259 Calling ``str()`` or ``unicode()`` on the returned value will yield a
260 string representation of the result. The
261 :class:`~.Compiled` object also can return a
262 dictionary of bind parameter names and values
263 using the ``params`` accessor.
265 :param bind: An :class:`.Connection` or :class:`.Engine` which
266 can provide a :class:`.Dialect` in order to generate a
267 :class:`.Compiled` object. If the ``bind`` and
268 ``dialect`` parameters are both omitted, a default SQL compiler
269 is used.
271 :param column_keys: Used for INSERT and UPDATE statements, a list of
272 column names which should be present in the VALUES clause of the
273 compiled statement. If ``None``, all columns from the target table
274 object are rendered.
276 :param dialect: A :class:`.Dialect` instance which can generate
277 a :class:`.Compiled` object. This argument takes precedence over
278 the ``bind`` argument.
280 :param compile_kwargs: optional dictionary of additional parameters
281 that will be passed through to the compiler within all "visit"
282 methods. This allows any custom flag to be passed through to
283 a custom compilation construct, for example. It is also used
284 for the case of passing the ``literal_binds`` flag through::
286 from sqlalchemy.sql import table, column, select
288 t = table("t", column("x"))
290 s = select(t).where(t.c.x == 5)
292 print(s.compile(compile_kwargs={"literal_binds": True}))
294 .. seealso::
296 :ref:`faq_sql_expression_string`
298 """
300 if dialect is None:
301 if bind:
302 dialect = bind.dialect
303 elif self.stringify_dialect == "default":
304 dialect = self._default_dialect()
305 else:
306 url = util.preloaded.engine_url
307 dialect = url.URL.create(
308 self.stringify_dialect
309 ).get_dialect()()
311 return self._compiler(dialect, **kw)
313 def _default_dialect(self):
314 default = util.preloaded.engine_default
315 return default.StrCompileDialect()
317 def _compiler(self, dialect: Dialect, **kw: Any) -> Compiled:
318 """Return a compiler appropriate for this ClauseElement, given a
319 Dialect."""
321 if TYPE_CHECKING:
322 assert isinstance(self, ClauseElement)
323 return dialect.statement_compiler(dialect, self, **kw)
325 def __str__(self) -> str:
326 return str(self.compile())
329@inspection._self_inspects
330class ClauseElement(
331 SupportsWrappingAnnotations,
332 MemoizedHasCacheKey,
333 HasCopyInternals,
334 ExternallyTraversible,
335 CompilerElement,
336):
337 """Base class for elements of a programmatically constructed SQL
338 expression.
340 """
342 __visit_name__ = "clause"
344 if TYPE_CHECKING:
346 @util.memoized_property
347 def _propagate_attrs(self) -> _PropagateAttrsType:
348 """like annotations, however these propagate outwards liberally
349 as SQL constructs are built, and are set up at construction time.
351 """
352 ...
354 else:
355 _propagate_attrs = util.EMPTY_DICT
357 @util.ro_memoized_property
358 def description(self) -> Optional[str]:
359 return None
361 _is_clone_of: Optional[Self] = None
363 is_clause_element = True
364 is_selectable = False
365 is_dml = False
366 _is_column_element = False
367 _is_keyed_column_element = False
368 _is_table = False
369 _gen_static_annotations_cache_key = False
370 _is_textual = False
371 _is_from_clause = False
372 _is_returns_rows = False
373 _is_text_clause = False
374 _is_from_container = False
375 _is_select_container = False
376 _is_select_base = False
377 _is_select_statement = False
378 _is_bind_parameter = False
379 _is_clause_list = False
380 _is_lambda_element = False
381 _is_singleton_constant = False
382 _is_immutable = False
383 _is_star = False
385 @property
386 def _order_by_label_element(self) -> Optional[Label[Any]]:
387 return None
389 _cache_key_traversal: _CacheKeyTraversalType = None
391 negation_clause: ColumnElement[bool]
393 if typing.TYPE_CHECKING:
395 def get_children(
396 self, *, omit_attrs: typing_Tuple[str, ...] = ..., **kw: Any
397 ) -> Iterable[ClauseElement]: ...
399 @util.ro_non_memoized_property
400 def _from_objects(self) -> List[FromClause]:
401 return []
403 def _set_propagate_attrs(self, values: Mapping[str, Any]) -> Self:
404 # usually, self._propagate_attrs is empty here. one case where it's
405 # not is a subquery against ORM select, that is then pulled as a
406 # property of an aliased class. should all be good
408 # assert not self._propagate_attrs
410 self._propagate_attrs = util.immutabledict(values)
411 return self
413 def _default_compiler(self) -> SQLCompiler:
414 dialect = self._default_dialect()
415 return dialect.statement_compiler(dialect, self) # type: ignore
417 def _clone(self, **kw: Any) -> Self:
418 """Create a shallow copy of this ClauseElement.
420 This method may be used by a generative API. Its also used as
421 part of the "deep" copy afforded by a traversal that combines
422 the _copy_internals() method.
424 """
426 skip = self._memoized_keys
427 c = self.__class__.__new__(self.__class__)
429 if skip:
430 # ensure this iteration remains atomic
431 c.__dict__ = {
432 k: v for k, v in self.__dict__.copy().items() if k not in skip
433 }
434 else:
435 c.__dict__ = self.__dict__.copy()
437 # this is a marker that helps to "equate" clauses to each other
438 # when a Select returns its list of FROM clauses. the cloning
439 # process leaves around a lot of remnants of the previous clause
440 # typically in the form of column expressions still attached to the
441 # old table.
442 cc = self._is_clone_of
443 c._is_clone_of = cc if cc is not None else self
444 return c
446 def _negate_in_binary(self, negated_op, original_op):
447 """a hook to allow the right side of a binary expression to respond
448 to a negation of the binary expression.
450 Used for the special case of expanding bind parameter with IN.
452 """
453 return self
455 def _with_binary_element_type(self, type_):
456 """in the context of binary expression, convert the type of this
457 object to the one given.
459 applies only to :class:`_expression.ColumnElement` classes.
461 """
462 return self
464 @property
465 def _constructor(self): # type: ignore[override]
466 """return the 'constructor' for this ClauseElement.
468 This is for the purposes for creating a new object of
469 this type. Usually, its just the element's __class__.
470 However, the "Annotated" version of the object overrides
471 to return the class of its proxied element.
473 """
474 return self.__class__
476 @HasMemoized.memoized_attribute
477 def _cloned_set(self):
478 """Return the set consisting all cloned ancestors of this
479 ClauseElement.
481 Includes this ClauseElement. This accessor tends to be used for
482 FromClause objects to identify 'equivalent' FROM clauses, regardless
483 of transformative operations.
485 """
486 s = util.column_set()
487 f: Optional[ClauseElement] = self
489 # note this creates a cycle, asserted in test_memusage. however,
490 # turning this into a plain @property adds tends of thousands of method
491 # calls to Core / ORM performance tests, so the small overhead
492 # introduced by the relatively small amount of short term cycles
493 # produced here is preferable
494 while f is not None:
495 s.add(f)
496 f = f._is_clone_of
497 return s
499 def _de_clone(self):
500 while self._is_clone_of is not None:
501 self = self._is_clone_of
502 return self
504 @util.ro_non_memoized_property
505 def entity_namespace(self) -> _EntityNamespace:
506 raise AttributeError(
507 "This SQL expression has no entity namespace "
508 "with which to filter from."
509 )
511 def __getstate__(self):
512 d = self.__dict__.copy()
513 d.pop("_is_clone_of", None)
514 d.pop("_generate_cache_key", None)
515 return d
517 def _execute_on_connection(
518 self,
519 connection: Connection,
520 distilled_params: _CoreMultiExecuteParams,
521 execution_options: CoreExecuteOptionsParameter,
522 ) -> Result[Any]:
523 if self.supports_execution:
524 if TYPE_CHECKING:
525 assert isinstance(self, Executable)
526 return connection._execute_clauseelement(
527 self, distilled_params, execution_options
528 )
529 else:
530 raise exc.ObjectNotExecutableError(self)
532 def _execute_on_scalar(
533 self,
534 connection: Connection,
535 distilled_params: _CoreMultiExecuteParams,
536 execution_options: CoreExecuteOptionsParameter,
537 ) -> Any:
538 """an additional hook for subclasses to provide a different
539 implementation for connection.scalar() vs. connection.execute().
541 .. versionadded:: 2.0
543 """
544 return self._execute_on_connection(
545 connection, distilled_params, execution_options
546 ).scalar()
548 def _get_embedded_bindparams(self) -> Sequence[BindParameter[Any]]:
549 """Return the list of :class:`.BindParameter` objects embedded in the
550 object.
552 This accomplishes the same purpose as ``visitors.traverse()`` or
553 similar would provide, however by making use of the cache key
554 it takes advantage of memoization of the key to result in fewer
555 net method calls, assuming the statement is also going to be
556 executed.
558 """
560 key = self._generate_cache_key()
561 if key is None:
562 bindparams: List[BindParameter[Any]] = []
564 traverse(self, {}, {"bindparam": bindparams.append})
565 return bindparams
567 else:
568 return key.bindparams
570 def unique_params(
571 self,
572 __optionaldict: Optional[Dict[str, Any]] = None,
573 **kwargs: Any,
574 ) -> Self:
575 """Return a copy with :func:`_expression.bindparam` elements
576 replaced.
578 Same functionality as :meth:`_expression.ClauseElement.params`,
579 except adds `unique=True`
580 to affected bind parameters so that multiple statements can be
581 used.
583 """
584 return self._replace_params(True, __optionaldict, kwargs)
586 def params(
587 self,
588 __optionaldict: Optional[Mapping[str, Any]] = None,
589 **kwargs: Any,
590 ) -> Self:
591 """Return a copy with :func:`_expression.bindparam` elements
592 replaced.
594 Returns a copy of this ClauseElement with
595 :func:`_expression.bindparam`
596 elements replaced with values taken from the given dictionary::
598 >>> clause = column("x") + bindparam("foo")
599 >>> print(clause.compile().params)
600 {'foo':None}
601 >>> print(clause.params({"foo": 7}).compile().params)
602 {'foo':7}
604 """
605 return self._replace_params(False, __optionaldict, kwargs)
607 def _replace_params(
608 self,
609 unique: bool,
610 optionaldict: Optional[Mapping[str, Any]],
611 kwargs: Dict[str, Any],
612 ) -> Self:
613 if optionaldict:
614 kwargs.update(optionaldict)
616 def visit_bindparam(bind: BindParameter[Any]) -> None:
617 if bind.key in kwargs:
618 bind.value = kwargs[bind.key]
619 bind.required = False
620 if unique:
621 bind._convert_to_unique()
623 return cloned_traverse(
624 self,
625 {"maintain_key": True, "detect_subquery_cols": True},
626 {"bindparam": visit_bindparam},
627 )
629 def compare(self, other: ClauseElement, **kw: Any) -> bool:
630 r"""Compare this :class:`_expression.ClauseElement` to
631 the given :class:`_expression.ClauseElement`.
633 Subclasses should override the default behavior, which is a
634 straight identity comparison.
636 \**kw are arguments consumed by subclass ``compare()`` methods and
637 may be used to modify the criteria for comparison
638 (see :class:`_expression.ColumnElement`).
640 """
641 return traversals.compare(self, other, **kw)
643 def self_group(
644 self, against: Optional[OperatorType] = None
645 ) -> ClauseElement:
646 """Apply a 'grouping' to this :class:`_expression.ClauseElement`.
648 This method is overridden by subclasses to return a "grouping"
649 construct, i.e. parenthesis. In particular it's used by "binary"
650 expressions to provide a grouping around themselves when placed into a
651 larger expression, as well as by :func:`_expression.select`
652 constructs when placed into the FROM clause of another
653 :func:`_expression.select`. (Note that subqueries should be
654 normally created using the :meth:`_expression.Select.alias` method,
655 as many
656 platforms require nested SELECT statements to be named).
658 As expressions are composed together, the application of
659 :meth:`self_group` is automatic - end-user code should never
660 need to use this method directly. Note that SQLAlchemy's
661 clause constructs take operator precedence into account -
662 so parenthesis might not be needed, for example, in
663 an expression like ``x OR (y AND z)`` - AND takes precedence
664 over OR.
666 The base :meth:`self_group` method of
667 :class:`_expression.ClauseElement`
668 just returns self.
669 """
670 return self
672 def _ungroup(self) -> ClauseElement:
673 """Return this :class:`_expression.ClauseElement`
674 without any groupings.
675 """
677 return self
679 def _compile_w_cache(
680 self,
681 dialect: Dialect,
682 *,
683 compiled_cache: Optional[CompiledCacheType],
684 column_keys: List[str],
685 for_executemany: bool = False,
686 schema_translate_map: Optional[SchemaTranslateMapType] = None,
687 **kw: Any,
688 ) -> typing_Tuple[
689 Compiled, Optional[Sequence[BindParameter[Any]]], CacheStats
690 ]:
691 elem_cache_key: Optional[CacheKey]
693 if compiled_cache is not None and dialect._supports_statement_cache:
694 elem_cache_key = self._generate_cache_key()
695 else:
696 elem_cache_key = None
698 extracted_params: Optional[Sequence[BindParameter[Any]]]
699 if elem_cache_key is not None:
700 if TYPE_CHECKING:
701 assert compiled_cache is not None
703 cache_key, extracted_params = elem_cache_key
704 key = (
705 dialect,
706 cache_key,
707 tuple(column_keys),
708 bool(schema_translate_map),
709 for_executemany,
710 )
711 compiled_sql = compiled_cache.get(key)
713 if compiled_sql is None:
714 cache_hit = dialect.CACHE_MISS
715 compiled_sql = self._compiler(
716 dialect,
717 cache_key=elem_cache_key,
718 column_keys=column_keys,
719 for_executemany=for_executemany,
720 schema_translate_map=schema_translate_map,
721 **kw,
722 )
723 compiled_cache[key] = compiled_sql
724 else:
725 cache_hit = dialect.CACHE_HIT
726 else:
727 extracted_params = None
728 compiled_sql = self._compiler(
729 dialect,
730 cache_key=elem_cache_key,
731 column_keys=column_keys,
732 for_executemany=for_executemany,
733 schema_translate_map=schema_translate_map,
734 **kw,
735 )
737 if not dialect._supports_statement_cache:
738 cache_hit = dialect.NO_DIALECT_SUPPORT
739 elif compiled_cache is None:
740 cache_hit = dialect.CACHING_DISABLED
741 else:
742 cache_hit = dialect.NO_CACHE_KEY
744 return compiled_sql, extracted_params, cache_hit
746 def __invert__(self):
747 # undocumented element currently used by the ORM for
748 # relationship.contains()
749 if hasattr(self, "negation_clause"):
750 return self.negation_clause
751 else:
752 return self._negate()
754 def _negate(self) -> ClauseElement:
755 # TODO: this code is uncovered and in all likelihood is not included
756 # in any codepath. So this should raise NotImplementedError in 2.1
757 grouped = self.self_group(against=operators.inv)
758 assert isinstance(grouped, ColumnElement)
759 return UnaryExpression(grouped, operator=operators.inv)
761 def __bool__(self):
762 raise TypeError("Boolean value of this clause is not defined")
764 def __repr__(self):
765 friendly = self.description
766 if friendly is None:
767 return object.__repr__(self)
768 else:
769 return "<%s.%s at 0x%x; %s>" % (
770 self.__module__,
771 self.__class__.__name__,
772 id(self),
773 friendly,
774 )
777class DQLDMLClauseElement(ClauseElement):
778 """represents a :class:`.ClauseElement` that compiles to a DQL or DML
779 expression, not DDL.
781 .. versionadded:: 2.0
783 """
785 if typing.TYPE_CHECKING:
787 def _compiler(self, dialect: Dialect, **kw: Any) -> SQLCompiler:
788 """Return a compiler appropriate for this ClauseElement, given a
789 Dialect."""
790 ...
792 def compile( # noqa: A001
793 self,
794 bind: Optional[_HasDialect] = None,
795 dialect: Optional[Dialect] = None,
796 **kw: Any,
797 ) -> SQLCompiler: ...
800class CompilerColumnElement(
801 roles.DMLColumnRole,
802 roles.DDLConstraintColumnRole,
803 roles.ColumnsClauseRole,
804 CompilerElement,
805):
806 """A compiler-only column element used for ad-hoc string compilations.
808 .. versionadded:: 2.0
810 """
812 __slots__ = ()
814 _propagate_attrs = util.EMPTY_DICT
815 _is_collection_aggregate = False
818# SQLCoreOperations should be suiting the ExpressionElementRole
819# and ColumnsClauseRole. however the MRO issues become too elaborate
820# at the moment.
821class SQLCoreOperations(Generic[_T_co], ColumnOperators, TypingOnly):
822 __slots__ = ()
824 # annotations for comparison methods
825 # these are from operators->Operators / ColumnOperators,
826 # redefined with the specific types returned by ColumnElement hierarchies
827 if typing.TYPE_CHECKING:
829 @util.non_memoized_property
830 def _propagate_attrs(self) -> _PropagateAttrsType: ...
832 def operate(
833 self, op: OperatorType, *other: Any, **kwargs: Any
834 ) -> ColumnElement[Any]: ...
836 def reverse_operate(
837 self, op: OperatorType, other: Any, **kwargs: Any
838 ) -> ColumnElement[Any]: ...
840 @overload
841 def op(
842 self,
843 opstring: str,
844 precedence: int = ...,
845 is_comparison: bool = ...,
846 *,
847 return_type: _TypeEngineArgument[_OPT],
848 python_impl: Optional[Callable[..., Any]] = None,
849 ) -> Callable[[Any], BinaryExpression[_OPT]]: ...
851 @overload
852 def op(
853 self,
854 opstring: str,
855 precedence: int = ...,
856 is_comparison: bool = ...,
857 return_type: Optional[_TypeEngineArgument[Any]] = ...,
858 python_impl: Optional[Callable[..., Any]] = ...,
859 ) -> Callable[[Any], BinaryExpression[Any]]: ...
861 def op(
862 self,
863 opstring: str,
864 precedence: int = 0,
865 is_comparison: bool = False,
866 return_type: Optional[_TypeEngineArgument[Any]] = None,
867 python_impl: Optional[Callable[..., Any]] = None,
868 ) -> Callable[[Any], BinaryExpression[Any]]: ...
870 def bool_op(
871 self,
872 opstring: str,
873 precedence: int = 0,
874 python_impl: Optional[Callable[..., Any]] = None,
875 ) -> Callable[[Any], BinaryExpression[bool]]: ...
877 def __and__(self, other: Any) -> BooleanClauseList: ...
879 def __or__(self, other: Any) -> BooleanClauseList: ...
881 def __invert__(self) -> ColumnElement[_T_co]: ...
883 def __lt__(self, other: Any) -> ColumnElement[bool]: ...
885 def __le__(self, other: Any) -> ColumnElement[bool]: ...
887 # declare also that this class has an hash method otherwise
888 # it may be assumed to be None by type checkers since the
889 # object defines __eq__ and python sets it to None in that case:
890 # https://docs.python.org/3/reference/datamodel.html#object.__hash__
891 def __hash__(self) -> int: ...
893 def __eq__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
894 ...
896 def __ne__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
897 ...
899 def is_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
901 def is_not_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
903 def __gt__(self, other: Any) -> ColumnElement[bool]: ...
905 def __ge__(self, other: Any) -> ColumnElement[bool]: ...
907 def __neg__(self) -> UnaryExpression[_T_co]: ...
909 def __contains__(self, other: Any) -> ColumnElement[bool]: ...
911 def __getitem__(self, index: Any) -> ColumnElement[Any]: ...
913 @overload
914 def __lshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
916 @overload
917 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
919 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
921 @overload
922 def __rshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
924 @overload
925 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
927 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
929 @overload
930 def concat(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
932 @overload
933 def concat(self, other: Any) -> ColumnElement[Any]: ...
935 def concat(self, other: Any) -> ColumnElement[Any]: ...
937 def like(
938 self, other: Any, escape: Optional[str] = None
939 ) -> BinaryExpression[bool]: ...
941 def ilike(
942 self, other: Any, escape: Optional[str] = None
943 ) -> BinaryExpression[bool]: ...
945 def bitwise_xor(self, other: Any) -> BinaryExpression[Any]: ...
947 def bitwise_or(self, other: Any) -> BinaryExpression[Any]: ...
949 def bitwise_and(self, other: Any) -> BinaryExpression[Any]: ...
951 def bitwise_not(self) -> UnaryExpression[_T_co]: ...
953 def bitwise_lshift(self, other: Any) -> BinaryExpression[Any]: ...
955 def bitwise_rshift(self, other: Any) -> BinaryExpression[Any]: ...
957 def in_(
958 self,
959 other: Union[
960 Iterable[Any], BindParameter[Any], roles.InElementRole
961 ],
962 ) -> BinaryExpression[bool]: ...
964 def not_in(
965 self,
966 other: Union[
967 Iterable[Any], BindParameter[Any], roles.InElementRole
968 ],
969 ) -> BinaryExpression[bool]: ...
971 def notin_(
972 self,
973 other: Union[
974 Iterable[Any], BindParameter[Any], roles.InElementRole
975 ],
976 ) -> BinaryExpression[bool]: ...
978 def not_like(
979 self, other: Any, escape: Optional[str] = None
980 ) -> BinaryExpression[bool]: ...
982 def notlike(
983 self, other: Any, escape: Optional[str] = None
984 ) -> BinaryExpression[bool]: ...
986 def not_ilike(
987 self, other: Any, escape: Optional[str] = None
988 ) -> BinaryExpression[bool]: ...
990 def notilike(
991 self, other: Any, escape: Optional[str] = None
992 ) -> BinaryExpression[bool]: ...
994 def is_(self, other: Any) -> BinaryExpression[bool]: ...
996 def is_not(self, other: Any) -> BinaryExpression[bool]: ...
998 def isnot(self, other: Any) -> BinaryExpression[bool]: ...
1000 def startswith(
1001 self,
1002 other: Any,
1003 escape: Optional[str] = None,
1004 autoescape: bool = False,
1005 ) -> ColumnElement[bool]: ...
1007 def istartswith(
1008 self,
1009 other: Any,
1010 escape: Optional[str] = None,
1011 autoescape: bool = False,
1012 ) -> ColumnElement[bool]: ...
1014 def endswith(
1015 self,
1016 other: Any,
1017 escape: Optional[str] = None,
1018 autoescape: bool = False,
1019 ) -> ColumnElement[bool]: ...
1021 def iendswith(
1022 self,
1023 other: Any,
1024 escape: Optional[str] = None,
1025 autoescape: bool = False,
1026 ) -> ColumnElement[bool]: ...
1028 def contains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1030 def icontains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1032 def match(self, other: Any, **kwargs: Any) -> ColumnElement[bool]: ...
1034 def regexp_match(
1035 self, pattern: Any, flags: Optional[str] = None
1036 ) -> ColumnElement[bool]: ...
1038 def regexp_replace(
1039 self, pattern: Any, replacement: Any, flags: Optional[str] = None
1040 ) -> ColumnElement[str]: ...
1042 def desc(self) -> UnaryExpression[_T_co]: ...
1044 def asc(self) -> UnaryExpression[_T_co]: ...
1046 def nulls_first(self) -> UnaryExpression[_T_co]: ...
1048 def nullsfirst(self) -> UnaryExpression[_T_co]: ...
1050 def nulls_last(self) -> UnaryExpression[_T_co]: ...
1052 def nullslast(self) -> UnaryExpression[_T_co]: ...
1054 def collate(self, collation: str) -> CollationClause: ...
1056 def between(
1057 self, cleft: Any, cright: Any, symmetric: bool = False
1058 ) -> BinaryExpression[bool]: ...
1060 def distinct(self: _SQO[_T_co]) -> UnaryExpression[_T_co]: ...
1062 def any_(self) -> CollectionAggregate[Any]: ...
1064 def all_(self) -> CollectionAggregate[Any]: ...
1066 # numeric overloads. These need more tweaking
1067 # in particular they all need to have a variant for Optiona[_T]
1068 # because Optional only applies to the data side, not the expression
1069 # side
1071 @overload
1072 def __add__(
1073 self: _SQO[_NMT],
1074 other: Any,
1075 ) -> ColumnElement[_NMT]: ...
1077 @overload
1078 def __add__(
1079 self: _SQO[str],
1080 other: Any,
1081 ) -> ColumnElement[str]: ...
1083 @overload
1084 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1086 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1088 @overload
1089 def __radd__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1091 @overload
1092 def __radd__(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
1094 def __radd__(self, other: Any) -> ColumnElement[Any]: ...
1096 @overload
1097 def __sub__(
1098 self: _SQO[_NMT],
1099 other: Any,
1100 ) -> ColumnElement[_NMT]: ...
1102 @overload
1103 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1105 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1107 @overload
1108 def __rsub__(
1109 self: _SQO[_NMT],
1110 other: Any,
1111 ) -> ColumnElement[_NMT]: ...
1113 @overload
1114 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1116 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1118 @overload
1119 def __mul__(
1120 self: _SQO[_NMT],
1121 other: Any,
1122 ) -> ColumnElement[_NMT]: ...
1124 @overload
1125 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1127 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1129 @overload
1130 def __rmul__(
1131 self: _SQO[_NMT],
1132 other: Any,
1133 ) -> ColumnElement[_NMT]: ...
1135 @overload
1136 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1138 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1140 @overload
1141 def __mod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1143 @overload
1144 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1146 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1148 @overload
1149 def __rmod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1151 @overload
1152 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1154 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1156 @overload
1157 def __truediv__(
1158 self: _SQO[int], other: Any
1159 ) -> ColumnElement[_NUMERIC]: ...
1161 @overload
1162 def __truediv__(self: _SQO[_NT], other: Any) -> ColumnElement[_NT]: ...
1164 @overload
1165 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1167 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1169 @overload
1170 def __rtruediv__(
1171 self: _SQO[_NMT], other: Any
1172 ) -> ColumnElement[_NUMERIC]: ...
1174 @overload
1175 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1177 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1179 @overload
1180 def __floordiv__(
1181 self: _SQO[_NMT], other: Any
1182 ) -> ColumnElement[_NMT]: ...
1184 @overload
1185 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1187 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1189 @overload
1190 def __rfloordiv__(
1191 self: _SQO[_NMT], other: Any
1192 ) -> ColumnElement[_NMT]: ...
1194 @overload
1195 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1197 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1200class SQLColumnExpression(
1201 SQLCoreOperations[_T_co], roles.ExpressionElementRole[_T_co], TypingOnly
1202):
1203 """A type that may be used to indicate any SQL column element or object
1204 that acts in place of one.
1206 :class:`.SQLColumnExpression` is a base of
1207 :class:`.ColumnElement`, as well as within the bases of ORM elements
1208 such as :class:`.InstrumentedAttribute`, and may be used in :pep:`484`
1209 typing to indicate arguments or return values that should behave
1210 as column expressions.
1212 .. versionadded:: 2.0.0b4
1215 """
1217 __slots__ = ()
1220_SQO = SQLCoreOperations
1223class ColumnElement(
1224 roles.ColumnArgumentOrKeyRole,
1225 roles.StatementOptionRole,
1226 roles.WhereHavingRole,
1227 roles.BinaryElementRole[_T],
1228 roles.OrderByRole,
1229 roles.ColumnsClauseRole,
1230 roles.LimitOffsetRole,
1231 roles.DMLColumnRole,
1232 roles.DDLConstraintColumnRole,
1233 roles.DDLExpressionRole,
1234 SQLColumnExpression[_T],
1235 DQLDMLClauseElement,
1236):
1237 """Represent a column-oriented SQL expression suitable for usage in the
1238 "columns" clause, WHERE clause etc. of a statement.
1240 While the most familiar kind of :class:`_expression.ColumnElement` is the
1241 :class:`_schema.Column` object, :class:`_expression.ColumnElement`
1242 serves as the basis
1243 for any unit that may be present in a SQL expression, including
1244 the expressions themselves, SQL functions, bound parameters,
1245 literal expressions, keywords such as ``NULL``, etc.
1246 :class:`_expression.ColumnElement`
1247 is the ultimate base class for all such elements.
1249 A wide variety of SQLAlchemy Core functions work at the SQL expression
1250 level, and are intended to accept instances of
1251 :class:`_expression.ColumnElement` as
1252 arguments. These functions will typically document that they accept a
1253 "SQL expression" as an argument. What this means in terms of SQLAlchemy
1254 usually refers to an input which is either already in the form of a
1255 :class:`_expression.ColumnElement` object,
1256 or a value which can be **coerced** into
1257 one. The coercion rules followed by most, but not all, SQLAlchemy Core
1258 functions with regards to SQL expressions are as follows:
1260 * a literal Python value, such as a string, integer or floating
1261 point value, boolean, datetime, ``Decimal`` object, or virtually
1262 any other Python object, will be coerced into a "literal bound
1263 value". This generally means that a :func:`.bindparam` will be
1264 produced featuring the given value embedded into the construct; the
1265 resulting :class:`.BindParameter` object is an instance of
1266 :class:`_expression.ColumnElement`.
1267 The Python value will ultimately be sent
1268 to the DBAPI at execution time as a parameterized argument to the
1269 ``execute()`` or ``executemany()`` methods, after SQLAlchemy
1270 type-specific converters (e.g. those provided by any associated
1271 :class:`.TypeEngine` objects) are applied to the value.
1273 * any special object value, typically ORM-level constructs, which
1274 feature an accessor called ``__clause_element__()``. The Core
1275 expression system looks for this method when an object of otherwise
1276 unknown type is passed to a function that is looking to coerce the
1277 argument into a :class:`_expression.ColumnElement` and sometimes a
1278 :class:`_expression.SelectBase` expression.
1279 It is used within the ORM to
1280 convert from ORM-specific objects like mapped classes and
1281 mapped attributes into Core expression objects.
1283 * The Python ``None`` value is typically interpreted as ``NULL``,
1284 which in SQLAlchemy Core produces an instance of :func:`.null`.
1286 A :class:`_expression.ColumnElement` provides the ability to generate new
1287 :class:`_expression.ColumnElement`
1288 objects using Python expressions. This means that Python operators
1289 such as ``==``, ``!=`` and ``<`` are overloaded to mimic SQL operations,
1290 and allow the instantiation of further :class:`_expression.ColumnElement`
1291 instances
1292 which are composed from other, more fundamental
1293 :class:`_expression.ColumnElement`
1294 objects. For example, two :class:`.ColumnClause` objects can be added
1295 together with the addition operator ``+`` to produce
1296 a :class:`.BinaryExpression`.
1297 Both :class:`.ColumnClause` and :class:`.BinaryExpression` are subclasses
1298 of :class:`_expression.ColumnElement`:
1300 .. sourcecode:: pycon+sql
1302 >>> from sqlalchemy.sql import column
1303 >>> column("a") + column("b")
1304 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
1305 >>> print(column("a") + column("b"))
1306 {printsql}a + b
1308 .. seealso::
1310 :class:`_schema.Column`
1312 :func:`_expression.column`
1314 """
1316 __visit_name__ = "column_element"
1318 primary_key: bool = False
1319 _is_clone_of: Optional[ColumnElement[_T]]
1320 _is_column_element = True
1321 _insert_sentinel: bool = False
1322 _omit_from_statements = False
1323 _is_collection_aggregate = False
1325 foreign_keys: AbstractSet[ForeignKey] = frozenset()
1327 @util.memoized_property
1328 def _proxies(self) -> List[ColumnElement[Any]]:
1329 return []
1331 @util.non_memoized_property
1332 def _tq_label(self) -> Optional[str]:
1333 """The named label that can be used to target
1334 this column in a result set in a "table qualified" context.
1336 This label is almost always the label used when
1337 rendering <expr> AS <label> in a SELECT statement when using
1338 the LABEL_STYLE_TABLENAME_PLUS_COL label style, which is what the
1339 legacy ORM ``Query`` object uses as well.
1341 For a regular Column bound to a Table, this is typically the label
1342 <tablename>_<columnname>. For other constructs, different rules
1343 may apply, such as anonymized labels and others.
1345 .. versionchanged:: 1.4.21 renamed from ``._label``
1347 """
1348 return None
1350 key: Optional[str] = None
1351 """The 'key' that in some circumstances refers to this object in a
1352 Python namespace.
1354 This typically refers to the "key" of the column as present in the
1355 ``.c`` collection of a selectable, e.g. ``sometable.c["somekey"]`` would
1356 return a :class:`_schema.Column` with a ``.key`` of "somekey".
1358 """
1360 @HasMemoized.memoized_attribute
1361 def _tq_key_label(self) -> Optional[str]:
1362 """A label-based version of 'key' that in some circumstances refers
1363 to this object in a Python namespace.
1366 _tq_key_label comes into play when a select() statement is constructed
1367 with apply_labels(); in this case, all Column objects in the ``.c``
1368 collection are rendered as <tablename>_<columnname> in SQL; this is
1369 essentially the value of ._label. But to locate those columns in the
1370 ``.c`` collection, the name is along the lines of <tablename>_<key>;
1371 that's the typical value of .key_label.
1373 .. versionchanged:: 1.4.21 renamed from ``._key_label``
1375 """
1376 return self._proxy_key
1378 @property
1379 def _key_label(self) -> Optional[str]:
1380 """legacy; renamed to _tq_key_label"""
1381 return self._tq_key_label
1383 @property
1384 def _label(self) -> Optional[str]:
1385 """legacy; renamed to _tq_label"""
1386 return self._tq_label
1388 @property
1389 def _non_anon_label(self) -> Optional[str]:
1390 """the 'name' that naturally applies this element when rendered in
1391 SQL.
1393 Concretely, this is the "name" of a column or a label in a
1394 SELECT statement; ``<columnname>`` and ``<labelname>`` below:
1396 .. sourcecode:: sql
1398 SELECT <columnmame> FROM table
1400 SELECT column AS <labelname> FROM table
1402 Above, the two names noted will be what's present in the DBAPI
1403 ``cursor.description`` as the names.
1405 If this attribute returns ``None``, it means that the SQL element as
1406 written does not have a 100% fully predictable "name" that would appear
1407 in the ``cursor.description``. Examples include SQL functions, CAST
1408 functions, etc. While such things do return names in
1409 ``cursor.description``, they are only predictable on a
1410 database-specific basis; e.g. an expression like ``MAX(table.col)`` may
1411 appear as the string ``max`` on one database (like PostgreSQL) or may
1412 appear as the whole expression ``max(table.col)`` on SQLite.
1414 The default implementation looks for a ``.name`` attribute on the
1415 object, as has been the precedent established in SQLAlchemy for many
1416 years. An exception is made on the ``FunctionElement`` subclass
1417 so that the return value is always ``None``.
1419 .. versionadded:: 1.4.21
1423 """
1424 return getattr(self, "name", None)
1426 _render_label_in_columns_clause = True
1427 """A flag used by select._columns_plus_names that helps to determine
1428 we are actually going to render in terms of "SELECT <col> AS <label>".
1429 This flag can be returned as False for some Column objects that want
1430 to be rendered as simple "SELECT <col>"; typically columns that don't have
1431 any parent table and are named the same as what the label would be
1432 in any case.
1434 """
1436 _allow_label_resolve = True
1437 """A flag that can be flipped to prevent a column from being resolvable
1438 by string label name.
1440 The joined eager loader strategy in the ORM uses this, for example.
1442 """
1444 _is_implicitly_boolean = False
1446 _alt_names: Sequence[str] = ()
1448 if TYPE_CHECKING:
1450 def _ungroup(self) -> ColumnElement[_T]: ...
1452 @overload
1453 def self_group(self, against: None = None) -> ColumnElement[_T]: ...
1455 @overload
1456 def self_group(
1457 self, against: Optional[OperatorType] = None
1458 ) -> ColumnElement[Any]: ...
1460 def self_group(
1461 self, against: Optional[OperatorType] = None
1462 ) -> ColumnElement[Any]:
1463 if (
1464 against in (operators.and_, operators.or_, operators._asbool)
1465 and self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity
1466 ):
1467 return AsBoolean(self, operators.is_true, operators.is_false)
1468 elif against in (operators.any_op, operators.all_op):
1469 return Grouping(self)
1470 else:
1471 return self
1473 @overload
1474 def _negate(self: ColumnElement[bool]) -> ColumnElement[bool]: ...
1476 @overload
1477 def _negate(self: ColumnElement[_T]) -> ColumnElement[_T]: ...
1479 def _negate(self) -> ColumnElement[Any]:
1480 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
1481 return AsBoolean(self, operators.is_false, operators.is_true)
1482 else:
1483 grouped = self.self_group(against=operators.inv)
1484 assert isinstance(grouped, ColumnElement)
1485 return UnaryExpression(
1486 grouped,
1487 operator=operators.inv,
1488 )
1490 type: TypeEngine[_T]
1492 if not TYPE_CHECKING:
1494 @util.memoized_property
1495 def type(self) -> TypeEngine[_T]: # noqa: A001
1496 # used for delayed setup of
1497 # type_api
1498 return type_api.NULLTYPE
1500 @HasMemoized.memoized_attribute
1501 def comparator(self) -> TypeEngine.Comparator[_T]:
1502 try:
1503 comparator_factory = self.type.comparator_factory
1504 except AttributeError as err:
1505 raise TypeError(
1506 "Object %r associated with '.type' attribute "
1507 "is not a TypeEngine class or object" % self.type
1508 ) from err
1509 else:
1510 return comparator_factory(self)
1512 def __setstate__(self, state):
1513 self.__dict__.update(state)
1515 def __getattr__(self, key: str) -> Any:
1516 try:
1517 return getattr(self.comparator, key)
1518 except AttributeError as err:
1519 raise AttributeError(
1520 "Neither %r object nor %r object has an attribute %r"
1521 % (
1522 type(self).__name__,
1523 type(self.comparator).__name__,
1524 key,
1525 )
1526 ) from err
1528 def operate(
1529 self,
1530 op: operators.OperatorType,
1531 *other: Any,
1532 **kwargs: Any,
1533 ) -> ColumnElement[Any]:
1534 return op(self.comparator, *other, **kwargs) # type: ignore[no-any-return] # noqa: E501
1536 def reverse_operate(
1537 self, op: operators.OperatorType, other: Any, **kwargs: Any
1538 ) -> ColumnElement[Any]:
1539 return op(other, self.comparator, **kwargs) # type: ignore[no-any-return] # noqa: E501
1541 def _bind_param(
1542 self,
1543 operator: operators.OperatorType,
1544 obj: Any,
1545 type_: Optional[TypeEngine[_T]] = None,
1546 expanding: bool = False,
1547 ) -> BindParameter[_T]:
1548 return BindParameter(
1549 None,
1550 obj,
1551 _compared_to_operator=operator,
1552 type_=type_,
1553 _compared_to_type=self.type,
1554 unique=True,
1555 expanding=expanding,
1556 )
1558 @property
1559 def expression(self) -> ColumnElement[Any]:
1560 """Return a column expression.
1562 Part of the inspection interface; returns self.
1564 """
1565 return self
1567 @property
1568 def _select_iterable(self) -> _SelectIterable:
1569 return (self,)
1571 @util.memoized_property
1572 def base_columns(self) -> FrozenSet[ColumnElement[Any]]:
1573 return frozenset(c for c in self.proxy_set if not c._proxies)
1575 @util.memoized_property
1576 def proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1577 """set of all columns we are proxying
1579 as of 2.0 this is explicitly deannotated columns. previously it was
1580 effectively deannotated columns but wasn't enforced. annotated
1581 columns should basically not go into sets if at all possible because
1582 their hashing behavior is very non-performant.
1584 """
1585 return frozenset([self._deannotate()]).union(
1586 itertools.chain(*[c.proxy_set for c in self._proxies])
1587 )
1589 @util.memoized_property
1590 def _expanded_proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1591 return frozenset(_expand_cloned(self.proxy_set))
1593 def _uncached_proxy_list(self) -> List[ColumnElement[Any]]:
1594 """An 'uncached' version of proxy set.
1596 This list includes annotated columns which perform very poorly in
1597 set operations.
1599 """
1601 return [self] + list(
1602 itertools.chain(*[c._uncached_proxy_list() for c in self._proxies])
1603 )
1605 def shares_lineage(self, othercolumn: ColumnElement[Any]) -> bool:
1606 """Return True if the given :class:`_expression.ColumnElement`
1607 has a common ancestor to this :class:`_expression.ColumnElement`."""
1609 return bool(self.proxy_set.intersection(othercolumn.proxy_set))
1611 def _compare_name_for_result(self, other: ColumnElement[Any]) -> bool:
1612 """Return True if the given column element compares to this one
1613 when targeting within a result row."""
1615 return (
1616 hasattr(other, "name")
1617 and hasattr(self, "name")
1618 and other.name == self.name
1619 )
1621 @HasMemoized.memoized_attribute
1622 def _proxy_key(self) -> Optional[str]:
1623 if self._annotations and "proxy_key" in self._annotations:
1624 return cast(str, self._annotations["proxy_key"])
1626 name = self.key
1627 if not name:
1628 # there's a bit of a seeming contradiction which is that the
1629 # "_non_anon_label" of a column can in fact be an
1630 # "_anonymous_label"; this is when it's on a column that is
1631 # proxying for an anonymous expression in a subquery.
1632 name = self._non_anon_label
1634 if isinstance(name, _anonymous_label):
1635 return None
1636 else:
1637 return name
1639 @HasMemoized.memoized_attribute
1640 def _expression_label(self) -> Optional[str]:
1641 """a suggested label to use in the case that the column has no name,
1642 which should be used if possible as the explicit 'AS <label>'
1643 where this expression would normally have an anon label.
1645 this is essentially mostly what _proxy_key does except it returns
1646 None if the column has a normal name that can be used.
1648 """
1650 if getattr(self, "name", None) is not None:
1651 return None
1652 elif self._annotations and "proxy_key" in self._annotations:
1653 return cast(str, self._annotations["proxy_key"])
1654 else:
1655 return None
1657 def _make_proxy(
1658 self,
1659 selectable: FromClause,
1660 *,
1661 primary_key: ColumnSet,
1662 foreign_keys: Set[KeyedColumnElement[Any]],
1663 name: Optional[str] = None,
1664 key: Optional[str] = None,
1665 name_is_truncatable: bool = False,
1666 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
1667 **kw: Any,
1668 ) -> typing_Tuple[str, ColumnClause[_T]]:
1669 """Create a new :class:`_expression.ColumnElement` representing this
1670 :class:`_expression.ColumnElement` as it appears in the select list of
1671 a descending selectable.
1673 """
1674 if name is None:
1675 name = self._anon_name_label
1676 if key is None:
1677 key = self._proxy_key
1678 else:
1679 key = name
1681 assert key is not None
1683 co: ColumnClause[_T] = ColumnClause(
1684 (
1685 coercions.expect(roles.TruncatedLabelRole, name)
1686 if name_is_truncatable
1687 else name
1688 ),
1689 type_=getattr(self, "type", None),
1690 _selectable=selectable,
1691 )
1693 co._propagate_attrs = selectable._propagate_attrs
1694 if compound_select_cols:
1695 co._proxies = list(compound_select_cols)
1696 else:
1697 co._proxies = [self]
1698 if selectable._is_clone_of is not None:
1699 co._is_clone_of = selectable._is_clone_of.columns.get(key)
1700 return key, co
1702 def cast(self, type_: _TypeEngineArgument[_OPT]) -> Cast[_OPT]:
1703 """Produce a type cast, i.e. ``CAST(<expression> AS <type>)``.
1705 This is a shortcut to the :func:`_expression.cast` function.
1707 .. seealso::
1709 :ref:`tutorial_casts`
1711 :func:`_expression.cast`
1713 :func:`_expression.type_coerce`
1715 """
1716 return Cast(self, type_)
1718 def label(self, name: Optional[str]) -> Label[_T]:
1719 """Produce a column label, i.e. ``<columnname> AS <name>``.
1721 This is a shortcut to the :func:`_expression.label` function.
1723 If 'name' is ``None``, an anonymous label name will be generated.
1725 """
1726 return Label(name, self, self.type)
1728 def _anon_label(
1729 self, seed: Optional[str], add_hash: Optional[int] = None
1730 ) -> _anonymous_label:
1731 while self._is_clone_of is not None:
1732 self = self._is_clone_of
1734 # as of 1.4 anonymous label for ColumnElement uses hash(), not id(),
1735 # as the identifier, because a column and its annotated version are
1736 # the same thing in a SQL statement
1737 hash_value = hash(self)
1739 if add_hash:
1740 # this path is used for disambiguating anon labels that would
1741 # otherwise be the same name for the same element repeated.
1742 # an additional numeric value is factored in for each label.
1744 # shift hash(self) (which is id(self), typically 8 byte integer)
1745 # 16 bits leftward. fill extra add_hash on right
1746 assert add_hash < (2 << 15)
1747 assert seed
1748 hash_value = (hash_value << 16) | add_hash
1750 # extra underscore is added for labels with extra hash
1751 # values, to isolate the "deduped anon" namespace from the
1752 # regular namespace. eliminates chance of these
1753 # manufactured hash values overlapping with regular ones for some
1754 # undefined python interpreter
1755 seed = seed + "_"
1757 if isinstance(seed, _anonymous_label):
1758 return _anonymous_label.safe_construct(
1759 hash_value, "", enclosing_label=seed
1760 )
1762 return _anonymous_label.safe_construct(hash_value, seed or "anon")
1764 @util.memoized_property
1765 def _anon_name_label(self) -> str:
1766 """Provides a constant 'anonymous label' for this ColumnElement.
1768 This is a label() expression which will be named at compile time.
1769 The same label() is returned each time ``anon_label`` is called so
1770 that expressions can reference ``anon_label`` multiple times,
1771 producing the same label name at compile time.
1773 The compiler uses this function automatically at compile time
1774 for expressions that are known to be 'unnamed' like binary
1775 expressions and function calls.
1777 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1778 public and is renamed to _anon_name_label. anon_name exists
1779 for backwards compat
1781 """
1782 name = getattr(self, "name", None)
1783 return self._anon_label(name)
1785 @util.memoized_property
1786 def _anon_key_label(self) -> _anonymous_label:
1787 """Provides a constant 'anonymous key label' for this ColumnElement.
1789 Compare to ``anon_label``, except that the "key" of the column,
1790 if available, is used to generate the label.
1792 This is used when a deduplicating key is placed into the columns
1793 collection of a selectable.
1795 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1796 public and is renamed to _anon_key_label. anon_key_label exists
1797 for backwards compat
1799 """
1800 return self._anon_label(self._proxy_key)
1802 @property
1803 @util.deprecated(
1804 "1.4",
1805 "The :attr:`_expression.ColumnElement.anon_label` attribute is now "
1806 "private, and the public accessor is deprecated.",
1807 )
1808 def anon_label(self) -> str:
1809 return self._anon_name_label
1811 @property
1812 @util.deprecated(
1813 "1.4",
1814 "The :attr:`_expression.ColumnElement.anon_key_label` attribute is "
1815 "now private, and the public accessor is deprecated.",
1816 )
1817 def anon_key_label(self) -> str:
1818 return self._anon_key_label
1820 def _dedupe_anon_label_idx(self, idx: int) -> str:
1821 """label to apply to a column that is anon labeled, but repeated
1822 in the SELECT, so that we have to make an "extra anon" label that
1823 disambiguates it from the previous appearance.
1825 these labels come out like "foo_bar_id__1" and have double underscores
1826 in them.
1828 """
1829 label = getattr(self, "name", None)
1831 # current convention is that if the element doesn't have a
1832 # ".name" (usually because it is not NamedColumn), we try to
1833 # use a "table qualified" form for the "dedupe anon" label,
1834 # based on the notion that a label like
1835 # "CAST(casttest.v1 AS DECIMAL) AS casttest_v1__1" looks better than
1836 # "CAST(casttest.v1 AS DECIMAL) AS anon__1"
1838 if label is None:
1839 return self._dedupe_anon_tq_label_idx(idx)
1840 else:
1841 return self._anon_label(label, add_hash=idx)
1843 @util.memoized_property
1844 def _anon_tq_label(self) -> _anonymous_label:
1845 return self._anon_label(getattr(self, "_tq_label", None))
1847 @util.memoized_property
1848 def _anon_tq_key_label(self) -> _anonymous_label:
1849 return self._anon_label(getattr(self, "_tq_key_label", None))
1851 def _dedupe_anon_tq_label_idx(self, idx: int) -> _anonymous_label:
1852 label = getattr(self, "_tq_label", None) or "anon"
1854 return self._anon_label(label, add_hash=idx)
1857class KeyedColumnElement(ColumnElement[_T]):
1858 """ColumnElement where ``.key`` is non-None."""
1860 _is_keyed_column_element = True
1862 key: str
1865class WrapsColumnExpression(ColumnElement[_T]):
1866 """Mixin that defines a :class:`_expression.ColumnElement`
1867 as a wrapper with special
1868 labeling behavior for an expression that already has a name.
1870 .. versionadded:: 1.4
1872 .. seealso::
1874 :ref:`change_4449`
1877 """
1879 @property
1880 def wrapped_column_expression(self) -> ColumnElement[_T]:
1881 raise NotImplementedError()
1883 @util.non_memoized_property
1884 def _tq_label(self) -> Optional[str]:
1885 wce = self.wrapped_column_expression
1886 if hasattr(wce, "_tq_label"):
1887 return wce._tq_label
1888 else:
1889 return None
1891 @property
1892 def _label(self) -> Optional[str]:
1893 return self._tq_label
1895 @property
1896 def _non_anon_label(self) -> Optional[str]:
1897 return None
1899 @util.non_memoized_property
1900 def _anon_name_label(self) -> str:
1901 wce = self.wrapped_column_expression
1903 # this logic tries to get the WrappedColumnExpression to render
1904 # with "<expr> AS <name>", where "<name>" is the natural name
1905 # within the expression itself. e.g. "CAST(table.foo) AS foo".
1906 if not wce._is_text_clause:
1907 nal = wce._non_anon_label
1908 if nal:
1909 return nal
1910 elif hasattr(wce, "_anon_name_label"):
1911 return wce._anon_name_label
1912 return super()._anon_name_label
1914 def _dedupe_anon_label_idx(self, idx: int) -> str:
1915 wce = self.wrapped_column_expression
1916 nal = wce._non_anon_label
1917 if nal:
1918 return self._anon_label(nal + "_")
1919 else:
1920 return self._dedupe_anon_tq_label_idx(idx)
1922 @property
1923 def _proxy_key(self):
1924 wce = self.wrapped_column_expression
1926 if not wce._is_text_clause:
1927 return wce._proxy_key
1928 return super()._proxy_key
1931class BindParameter(roles.InElementRole, KeyedColumnElement[_T]):
1932 r"""Represent a "bound expression".
1934 :class:`.BindParameter` is invoked explicitly using the
1935 :func:`.bindparam` function, as in::
1937 from sqlalchemy import bindparam
1939 stmt = select(users_table).where(
1940 users_table.c.name == bindparam("username")
1941 )
1943 Detailed discussion of how :class:`.BindParameter` is used is
1944 at :func:`.bindparam`.
1946 .. seealso::
1948 :func:`.bindparam`
1950 """
1952 __visit_name__ = "bindparam"
1954 _traverse_internals: _TraverseInternalsType = [
1955 ("key", InternalTraversal.dp_anon_name),
1956 ("type", InternalTraversal.dp_type),
1957 ("callable", InternalTraversal.dp_plain_dict),
1958 ("value", InternalTraversal.dp_plain_obj),
1959 ("literal_execute", InternalTraversal.dp_boolean),
1960 ]
1962 key: str
1963 type: TypeEngine[_T]
1964 value: Optional[_T]
1966 _is_crud = False
1967 _is_bind_parameter = True
1968 _key_is_anon = False
1970 # bindparam implements its own _gen_cache_key() method however
1971 # we check subclasses for this flag, else no cache key is generated
1972 inherit_cache = True
1974 def __init__(
1975 self,
1976 key: Optional[str],
1977 value: Any = _NoArg.NO_ARG,
1978 type_: Optional[_TypeEngineArgument[_T]] = None,
1979 unique: bool = False,
1980 required: Union[bool, Literal[_NoArg.NO_ARG]] = _NoArg.NO_ARG,
1981 quote: Optional[bool] = None,
1982 callable_: Optional[Callable[[], Any]] = None,
1983 expanding: bool = False,
1984 isoutparam: bool = False,
1985 literal_execute: bool = False,
1986 _compared_to_operator: Optional[OperatorType] = None,
1987 _compared_to_type: Optional[TypeEngine[Any]] = None,
1988 _is_crud: bool = False,
1989 ):
1990 if required is _NoArg.NO_ARG:
1991 required = value is _NoArg.NO_ARG and callable_ is None
1992 if value is _NoArg.NO_ARG:
1993 value = None
1995 if quote is not None:
1996 key = quoted_name.construct(key, quote)
1998 if unique:
1999 self.key = _anonymous_label.safe_construct(
2000 id(self),
2001 (
2002 key
2003 if key is not None
2004 and not isinstance(key, _anonymous_label)
2005 else "param"
2006 ),
2007 sanitize_key=True,
2008 )
2009 self._key_is_anon = True
2010 elif key:
2011 self.key = key
2012 else:
2013 self.key = _anonymous_label.safe_construct(id(self), "param")
2014 self._key_is_anon = True
2016 # identifying key that won't change across
2017 # clones, used to identify the bind's logical
2018 # identity
2019 self._identifying_key = self.key
2021 # key that was passed in the first place, used to
2022 # generate new keys
2023 self._orig_key = key or "param"
2025 self.unique = unique
2026 self.value = value
2027 self.callable = callable_
2028 self.isoutparam = isoutparam
2029 self.required = required
2031 # indicate an "expanding" parameter; the compiler sets this
2032 # automatically in the compiler _render_in_expr_w_bindparam method
2033 # for an IN expression
2034 self.expanding = expanding
2036 # this is another hint to help w/ expanding and is typically
2037 # set in the compiler _render_in_expr_w_bindparam method for an
2038 # IN expression
2039 self.expand_op = None
2041 self.literal_execute = literal_execute
2042 if _is_crud:
2043 self._is_crud = True
2045 if type_ is None:
2046 if expanding:
2047 if value:
2048 check_value = value[0]
2049 else:
2050 check_value = type_api._NO_VALUE_IN_LIST
2051 else:
2052 check_value = value
2053 if _compared_to_type is not None:
2054 self.type = _compared_to_type.coerce_compared_value(
2055 _compared_to_operator, check_value
2056 )
2057 else:
2058 self.type = type_api._resolve_value_to_type(check_value)
2059 elif isinstance(type_, type):
2060 self.type = type_()
2061 elif is_tuple_type(type_):
2062 if value:
2063 if expanding:
2064 check_value = value[0]
2065 else:
2066 check_value = value
2067 cast("BindParameter[typing_Tuple[Any, ...]]", self).type = (
2068 type_._resolve_values_to_types(check_value)
2069 )
2070 else:
2071 cast("BindParameter[typing_Tuple[Any, ...]]", self).type = (
2072 type_
2073 )
2074 else:
2075 self.type = type_
2077 def _with_value(self, value, maintain_key=False, required=NO_ARG):
2078 """Return a copy of this :class:`.BindParameter` with the given value
2079 set.
2080 """
2081 cloned = self._clone(maintain_key=maintain_key)
2082 cloned.value = value
2083 cloned.callable = None
2084 cloned.required = required if required is not NO_ARG else self.required
2085 if cloned.type is type_api.NULLTYPE:
2086 cloned.type = type_api._resolve_value_to_type(value)
2087 return cloned
2089 @property
2090 def effective_value(self) -> Optional[_T]:
2091 """Return the value of this bound parameter,
2092 taking into account if the ``callable`` parameter
2093 was set.
2095 The ``callable`` value will be evaluated
2096 and returned if present, else ``value``.
2098 """
2099 if self.callable:
2100 # TODO: set up protocol for bind parameter callable
2101 return self.callable() # type: ignore
2102 else:
2103 return self.value
2105 def render_literal_execute(self) -> BindParameter[_T]:
2106 """Produce a copy of this bound parameter that will enable the
2107 :paramref:`_sql.BindParameter.literal_execute` flag.
2109 The :paramref:`_sql.BindParameter.literal_execute` flag will
2110 have the effect of the parameter rendered in the compiled SQL
2111 string using ``[POSTCOMPILE]`` form, which is a special form that
2112 is converted to be a rendering of the literal value of the parameter
2113 at SQL execution time. The rationale is to support caching
2114 of SQL statement strings that can embed per-statement literal values,
2115 such as LIMIT and OFFSET parameters, in the final SQL string that
2116 is passed to the DBAPI. Dialects in particular may want to use
2117 this method within custom compilation schemes.
2119 .. versionadded:: 1.4.5
2121 .. seealso::
2123 :ref:`engine_thirdparty_caching`
2125 """
2126 c = ClauseElement._clone(self)
2127 c.literal_execute = True
2128 return c
2130 def _negate_in_binary(self, negated_op, original_op):
2131 if self.expand_op is original_op:
2132 bind = self._clone()
2133 bind.expand_op = negated_op
2134 return bind
2135 else:
2136 return self
2138 def _with_binary_element_type(self, type_: TypeEngine[Any]) -> Self:
2139 c: Self = ClauseElement._clone(self)
2140 c.type = type_
2141 return c
2143 def _clone(self, maintain_key: bool = False, **kw: Any) -> Self:
2144 c = ClauseElement._clone(self, **kw)
2145 # ensure all the BindParameter objects stay in cloned set.
2146 # in #7823, we changed "clone" so that a clone only keeps a reference
2147 # to the "original" element, since for column correspondence, that's
2148 # all we need. However, for BindParam, _cloned_set is used by
2149 # the "cache key bind match" lookup, which means if any of those
2150 # interim BindParameter objects became part of a cache key in the
2151 # cache, we need it. So here, make sure all clones keep carrying
2152 # forward.
2153 c._cloned_set.update(self._cloned_set)
2154 if not maintain_key and self.unique:
2155 c.key = _anonymous_label.safe_construct(
2156 id(c), c._orig_key or "param", sanitize_key=True
2157 )
2158 return c
2160 def _gen_cache_key(self, anon_map, bindparams):
2161 _gen_cache_ok = self.__class__.__dict__.get("inherit_cache", False)
2163 if not _gen_cache_ok:
2164 if anon_map is not None:
2165 anon_map[NO_CACHE] = True
2166 return None
2168 id_, found = anon_map.get_anon(self)
2169 if found:
2170 return (id_, self.__class__)
2172 if bindparams is not None:
2173 bindparams.append(self)
2175 return (
2176 id_,
2177 self.__class__,
2178 self.type._static_cache_key,
2179 self.key % anon_map if self._key_is_anon else self.key,
2180 self.literal_execute,
2181 )
2183 def _convert_to_unique(self):
2184 if not self.unique:
2185 self.unique = True
2186 self.key = _anonymous_label.safe_construct(
2187 id(self), self._orig_key or "param", sanitize_key=True
2188 )
2190 def __getstate__(self):
2191 """execute a deferred value for serialization purposes."""
2193 d = self.__dict__.copy()
2194 v = self.value
2195 if self.callable:
2196 v = self.callable()
2197 d["callable"] = None
2198 d["value"] = v
2199 return d
2201 def __setstate__(self, state):
2202 if state.get("unique", False):
2203 state["key"] = _anonymous_label.safe_construct(
2204 id(self), state.get("_orig_key", "param"), sanitize_key=True
2205 )
2206 self.__dict__.update(state)
2208 def __repr__(self):
2209 return "%s(%r, %r, type_=%r)" % (
2210 self.__class__.__name__,
2211 self.key,
2212 self.value,
2213 self.type,
2214 )
2217class TypeClause(DQLDMLClauseElement):
2218 """Handle a type keyword in a SQL statement.
2220 Used by the ``Case`` statement.
2222 """
2224 __visit_name__ = "typeclause"
2226 _traverse_internals: _TraverseInternalsType = [
2227 ("type", InternalTraversal.dp_type)
2228 ]
2229 type: TypeEngine[Any]
2231 def __init__(self, type_: TypeEngine[Any]):
2232 self.type = type_
2235class TextClause(
2236 roles.DDLConstraintColumnRole,
2237 roles.DDLExpressionRole,
2238 roles.StatementOptionRole,
2239 roles.WhereHavingRole,
2240 roles.OrderByRole,
2241 roles.FromClauseRole,
2242 roles.SelectStatementRole,
2243 roles.InElementRole,
2244 Generative,
2245 Executable,
2246 DQLDMLClauseElement,
2247 roles.BinaryElementRole[Any],
2248 inspection.Inspectable["TextClause"],
2249):
2250 """Represent a literal SQL text fragment.
2252 E.g.::
2254 from sqlalchemy import text
2256 t = text("SELECT * FROM users")
2257 result = connection.execute(t)
2259 The :class:`_expression.TextClause` construct is produced using the
2260 :func:`_expression.text`
2261 function; see that function for full documentation.
2263 .. seealso::
2265 :func:`_expression.text`
2267 """
2269 __visit_name__ = "textclause"
2271 _traverse_internals: _TraverseInternalsType = [
2272 ("_bindparams", InternalTraversal.dp_string_clauseelement_dict),
2273 ("text", InternalTraversal.dp_string),
2274 ]
2276 _is_text_clause = True
2278 _is_textual = True
2280 _bind_params_regex = re.compile(r"(?<![:\w\x5c]):(\w+)(?!:)", re.UNICODE)
2281 _is_implicitly_boolean = False
2283 _render_label_in_columns_clause = False
2285 _omit_from_statements = False
2287 _is_collection_aggregate = False
2289 @property
2290 def _hide_froms(self) -> Iterable[FromClause]:
2291 return ()
2293 def __and__(self, other):
2294 # support use in select.where(), query.filter()
2295 return and_(self, other)
2297 @property
2298 def _select_iterable(self) -> _SelectIterable:
2299 return (self,)
2301 # help in those cases where text() is
2302 # interpreted in a column expression situation
2303 key: Optional[str] = None
2304 _label: Optional[str] = None
2306 _allow_label_resolve = False
2308 @property
2309 def _is_star(self): # type: ignore[override]
2310 return self.text == "*"
2312 def __init__(self, text: str):
2313 self._bindparams: Dict[str, BindParameter[Any]] = {}
2315 def repl(m):
2316 self._bindparams[m.group(1)] = BindParameter(m.group(1))
2317 return ":%s" % m.group(1)
2319 # scan the string and search for bind parameter names, add them
2320 # to the list of bindparams
2321 self.text = self._bind_params_regex.sub(repl, text)
2323 @_generative
2324 def bindparams(
2325 self,
2326 *binds: BindParameter[Any],
2327 **names_to_values: Any,
2328 ) -> Self:
2329 """Establish the values and/or types of bound parameters within
2330 this :class:`_expression.TextClause` construct.
2332 Given a text construct such as::
2334 from sqlalchemy import text
2336 stmt = text(
2337 "SELECT id, name FROM user WHERE name=:name AND timestamp=:timestamp"
2338 )
2340 the :meth:`_expression.TextClause.bindparams`
2341 method can be used to establish
2342 the initial value of ``:name`` and ``:timestamp``,
2343 using simple keyword arguments::
2345 stmt = stmt.bindparams(
2346 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2347 )
2349 Where above, new :class:`.BindParameter` objects
2350 will be generated with the names ``name`` and ``timestamp``, and
2351 values of ``jack`` and ``datetime.datetime(2012, 10, 8, 15, 12, 5)``,
2352 respectively. The types will be
2353 inferred from the values given, in this case :class:`.String` and
2354 :class:`.DateTime`.
2356 When specific typing behavior is needed, the positional ``*binds``
2357 argument can be used in which to specify :func:`.bindparam` constructs
2358 directly. These constructs must include at least the ``key``
2359 argument, then an optional value and type::
2361 from sqlalchemy import bindparam
2363 stmt = stmt.bindparams(
2364 bindparam("name", value="jack", type_=String),
2365 bindparam("timestamp", type_=DateTime),
2366 )
2368 Above, we specified the type of :class:`.DateTime` for the
2369 ``timestamp`` bind, and the type of :class:`.String` for the ``name``
2370 bind. In the case of ``name`` we also set the default value of
2371 ``"jack"``.
2373 Additional bound parameters can be supplied at statement execution
2374 time, e.g.::
2376 result = connection.execute(
2377 stmt, timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2378 )
2380 The :meth:`_expression.TextClause.bindparams`
2381 method can be called repeatedly,
2382 where it will re-use existing :class:`.BindParameter` objects to add
2383 new information. For example, we can call
2384 :meth:`_expression.TextClause.bindparams`
2385 first with typing information, and a
2386 second time with value information, and it will be combined::
2388 stmt = text(
2389 "SELECT id, name FROM user WHERE name=:name "
2390 "AND timestamp=:timestamp"
2391 )
2392 stmt = stmt.bindparams(
2393 bindparam("name", type_=String), bindparam("timestamp", type_=DateTime)
2394 )
2395 stmt = stmt.bindparams(
2396 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2397 )
2399 The :meth:`_expression.TextClause.bindparams`
2400 method also supports the concept of
2401 **unique** bound parameters. These are parameters that are
2402 "uniquified" on name at statement compilation time, so that multiple
2403 :func:`_expression.text`
2404 constructs may be combined together without the names
2405 conflicting. To use this feature, specify the
2406 :paramref:`.BindParameter.unique` flag on each :func:`.bindparam`
2407 object::
2409 stmt1 = text("select id from table where name=:name").bindparams(
2410 bindparam("name", value="name1", unique=True)
2411 )
2412 stmt2 = text("select id from table where name=:name").bindparams(
2413 bindparam("name", value="name2", unique=True)
2414 )
2416 union = union_all(stmt1.columns(column("id")), stmt2.columns(column("id")))
2418 The above statement will render as:
2420 .. sourcecode:: sql
2422 select id from table where name=:name_1
2423 UNION ALL select id from table where name=:name_2
2425 .. versionadded:: 1.3.11 Added support for the
2426 :paramref:`.BindParameter.unique` flag to work with
2427 :func:`_expression.text`
2428 constructs.
2430 """ # noqa: E501
2431 self._bindparams = new_params = self._bindparams.copy()
2433 for bind in binds:
2434 try:
2435 # the regex used for text() currently will not match
2436 # a unique/anonymous key in any case, so use the _orig_key
2437 # so that a text() construct can support unique parameters
2438 existing = new_params[bind._orig_key]
2439 except KeyError as err:
2440 raise exc.ArgumentError(
2441 "This text() construct doesn't define a "
2442 "bound parameter named %r" % bind._orig_key
2443 ) from err
2444 else:
2445 new_params[existing._orig_key] = bind
2447 for key, value in names_to_values.items():
2448 try:
2449 existing = new_params[key]
2450 except KeyError as err:
2451 raise exc.ArgumentError(
2452 "This text() construct doesn't define a "
2453 "bound parameter named %r" % key
2454 ) from err
2455 else:
2456 new_params[key] = existing._with_value(value, required=False)
2457 return self
2459 @util.preload_module("sqlalchemy.sql.selectable")
2460 def columns(
2461 self,
2462 *cols: _ColumnExpressionArgument[Any],
2463 **types: _TypeEngineArgument[Any],
2464 ) -> TextualSelect:
2465 r"""Turn this :class:`_expression.TextClause` object into a
2466 :class:`_expression.TextualSelect`
2467 object that serves the same role as a SELECT
2468 statement.
2470 The :class:`_expression.TextualSelect` is part of the
2471 :class:`_expression.SelectBase`
2472 hierarchy and can be embedded into another statement by using the
2473 :meth:`_expression.TextualSelect.subquery` method to produce a
2474 :class:`.Subquery`
2475 object, which can then be SELECTed from.
2477 This function essentially bridges the gap between an entirely
2478 textual SELECT statement and the SQL expression language concept
2479 of a "selectable"::
2481 from sqlalchemy.sql import column, text
2483 stmt = text("SELECT id, name FROM some_table")
2484 stmt = stmt.columns(column("id"), column("name")).subquery("st")
2486 stmt = (
2487 select(mytable)
2488 .select_from(mytable.join(stmt, mytable.c.name == stmt.c.name))
2489 .where(stmt.c.id > 5)
2490 )
2492 Above, we pass a series of :func:`_expression.column` elements to the
2493 :meth:`_expression.TextClause.columns` method positionally. These
2494 :func:`_expression.column`
2495 elements now become first class elements upon the
2496 :attr:`_expression.TextualSelect.selected_columns` column collection,
2497 which then
2498 become part of the :attr:`.Subquery.c` collection after
2499 :meth:`_expression.TextualSelect.subquery` is invoked.
2501 The column expressions we pass to
2502 :meth:`_expression.TextClause.columns` may
2503 also be typed; when we do so, these :class:`.TypeEngine` objects become
2504 the effective return type of the column, so that SQLAlchemy's
2505 result-set-processing systems may be used on the return values.
2506 This is often needed for types such as date or boolean types, as well
2507 as for unicode processing on some dialect configurations::
2509 stmt = text("SELECT id, name, timestamp FROM some_table")
2510 stmt = stmt.columns(
2511 column("id", Integer),
2512 column("name", Unicode),
2513 column("timestamp", DateTime),
2514 )
2516 for id, name, timestamp in connection.execute(stmt):
2517 print(id, name, timestamp)
2519 As a shortcut to the above syntax, keyword arguments referring to
2520 types alone may be used, if only type conversion is needed::
2522 stmt = text("SELECT id, name, timestamp FROM some_table")
2523 stmt = stmt.columns(id=Integer, name=Unicode, timestamp=DateTime)
2525 for id, name, timestamp in connection.execute(stmt):
2526 print(id, name, timestamp)
2528 The positional form of :meth:`_expression.TextClause.columns`
2529 also provides the
2530 unique feature of **positional column targeting**, which is
2531 particularly useful when using the ORM with complex textual queries. If
2532 we specify the columns from our model to
2533 :meth:`_expression.TextClause.columns`,
2534 the result set will match to those columns positionally, meaning the
2535 name or origin of the column in the textual SQL doesn't matter::
2537 stmt = text(
2538 "SELECT users.id, addresses.id, users.id, "
2539 "users.name, addresses.email_address AS email "
2540 "FROM users JOIN addresses ON users.id=addresses.user_id "
2541 "WHERE users.id = 1"
2542 ).columns(
2543 User.id,
2544 Address.id,
2545 Address.user_id,
2546 User.name,
2547 Address.email_address,
2548 )
2550 query = (
2551 session.query(User)
2552 .from_statement(stmt)
2553 .options(contains_eager(User.addresses))
2554 )
2556 The :meth:`_expression.TextClause.columns` method provides a direct
2557 route to calling :meth:`_expression.FromClause.subquery` as well as
2558 :meth:`_expression.SelectBase.cte`
2559 against a textual SELECT statement::
2561 stmt = stmt.columns(id=Integer, name=String).cte("st")
2563 stmt = select(sometable).where(sometable.c.id == stmt.c.id)
2565 :param \*cols: A series of :class:`_expression.ColumnElement` objects,
2566 typically
2567 :class:`_schema.Column` objects from a :class:`_schema.Table`
2568 or ORM level
2569 column-mapped attributes, representing a set of columns that this
2570 textual string will SELECT from.
2572 :param \**types: A mapping of string names to :class:`.TypeEngine`
2573 type objects indicating the datatypes to use for names that are
2574 SELECTed from the textual string. Prefer to use the ``*cols``
2575 argument as it also indicates positional ordering.
2577 """
2578 selectable = util.preloaded.sql_selectable
2580 input_cols: List[NamedColumn[Any]] = [
2581 coercions.expect(roles.LabeledColumnExprRole, col) for col in cols
2582 ]
2584 positional_input_cols = [
2585 (
2586 ColumnClause(col.key, types.pop(col.key))
2587 if col.key in types
2588 else col
2589 )
2590 for col in input_cols
2591 ]
2592 keyed_input_cols: List[NamedColumn[Any]] = [
2593 ColumnClause(key, type_) for key, type_ in types.items()
2594 ]
2596 elem = selectable.TextualSelect.__new__(selectable.TextualSelect)
2597 elem._init(
2598 self,
2599 positional_input_cols + keyed_input_cols,
2600 positional=bool(positional_input_cols) and not keyed_input_cols,
2601 )
2602 return elem
2604 @property
2605 def type(self) -> TypeEngine[Any]:
2606 return type_api.NULLTYPE
2608 @property
2609 def comparator(self):
2610 # TODO: this seems wrong, it seems like we might not
2611 # be using this method.
2612 return self.type.comparator_factory(self) # type: ignore
2614 def self_group(
2615 self, against: Optional[OperatorType] = None
2616 ) -> Union[Self, Grouping[Any]]:
2617 if against is operators.in_op:
2618 return Grouping(self)
2619 else:
2620 return self
2623class Null(SingletonConstant, roles.ConstExprRole[None], ColumnElement[None]):
2624 """Represent the NULL keyword in a SQL statement.
2626 :class:`.Null` is accessed as a constant via the
2627 :func:`.null` function.
2629 """
2631 __visit_name__ = "null"
2633 _traverse_internals: _TraverseInternalsType = []
2634 _singleton: Null
2636 if not TYPE_CHECKING:
2638 @util.memoized_property
2639 def type(self) -> TypeEngine[_T]: # noqa: A001
2640 return type_api.NULLTYPE
2642 @classmethod
2643 def _instance(cls) -> Null:
2644 """Return a constant :class:`.Null` construct."""
2646 return Null._singleton
2649Null._create_singleton()
2652class False_(
2653 SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]
2654):
2655 """Represent the ``false`` keyword, or equivalent, in a SQL statement.
2657 :class:`.False_` is accessed as a constant via the
2658 :func:`.false` function.
2660 """
2662 __visit_name__ = "false"
2663 _traverse_internals: _TraverseInternalsType = []
2664 _singleton: False_
2666 if not TYPE_CHECKING:
2668 @util.memoized_property
2669 def type(self) -> TypeEngine[_T]: # noqa: A001
2670 return type_api.BOOLEANTYPE
2672 def _negate(self) -> True_:
2673 return True_._singleton
2675 @classmethod
2676 def _instance(cls) -> False_:
2677 return False_._singleton
2680False_._create_singleton()
2683class True_(SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]):
2684 """Represent the ``true`` keyword, or equivalent, in a SQL statement.
2686 :class:`.True_` is accessed as a constant via the
2687 :func:`.true` function.
2689 """
2691 __visit_name__ = "true"
2693 _traverse_internals: _TraverseInternalsType = []
2694 _singleton: True_
2696 if not TYPE_CHECKING:
2698 @util.memoized_property
2699 def type(self) -> TypeEngine[_T]: # noqa: A001
2700 return type_api.BOOLEANTYPE
2702 def _negate(self) -> False_:
2703 return False_._singleton
2705 @classmethod
2706 def _ifnone(
2707 cls, other: Optional[ColumnElement[Any]]
2708 ) -> ColumnElement[Any]:
2709 if other is None:
2710 return cls._instance()
2711 else:
2712 return other
2714 @classmethod
2715 def _instance(cls) -> True_:
2716 return True_._singleton
2719True_._create_singleton()
2722class ClauseList(
2723 roles.InElementRole,
2724 roles.OrderByRole,
2725 roles.ColumnsClauseRole,
2726 roles.DMLColumnRole,
2727 DQLDMLClauseElement,
2728):
2729 """Describe a list of clauses, separated by an operator.
2731 By default, is comma-separated, such as a column listing.
2733 """
2735 __visit_name__ = "clauselist"
2737 # this is used only by the ORM in a legacy use case for
2738 # composite attributes
2739 _is_clause_list = True
2741 _traverse_internals: _TraverseInternalsType = [
2742 ("clauses", InternalTraversal.dp_clauseelement_list),
2743 ("operator", InternalTraversal.dp_operator),
2744 ]
2746 clauses: List[ColumnElement[Any]]
2748 def __init__(
2749 self,
2750 *clauses: _ColumnExpressionArgument[Any],
2751 operator: OperatorType = operators.comma_op,
2752 group: bool = True,
2753 group_contents: bool = True,
2754 _literal_as_text_role: Type[roles.SQLRole] = roles.WhereHavingRole,
2755 ):
2756 self.operator = operator
2757 self.group = group
2758 self.group_contents = group_contents
2759 clauses_iterator: Iterable[_ColumnExpressionArgument[Any]] = clauses
2760 text_converter_role: Type[roles.SQLRole] = _literal_as_text_role
2761 self._text_converter_role = text_converter_role
2763 if self.group_contents:
2764 self.clauses = [
2765 coercions.expect(
2766 text_converter_role, clause, apply_propagate_attrs=self
2767 ).self_group(against=self.operator)
2768 for clause in clauses_iterator
2769 ]
2770 else:
2771 self.clauses = [
2772 coercions.expect(
2773 text_converter_role, clause, apply_propagate_attrs=self
2774 )
2775 for clause in clauses_iterator
2776 ]
2777 self._is_implicitly_boolean = operators.is_boolean(self.operator)
2779 @classmethod
2780 def _construct_raw(
2781 cls,
2782 operator: OperatorType,
2783 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
2784 ) -> ClauseList:
2785 self = cls.__new__(cls)
2786 self.clauses = list(clauses) if clauses else []
2787 self.group = True
2788 self.operator = operator
2789 self.group_contents = True
2790 self._is_implicitly_boolean = False
2791 return self
2793 def __iter__(self) -> Iterator[ColumnElement[Any]]:
2794 return iter(self.clauses)
2796 def __len__(self) -> int:
2797 return len(self.clauses)
2799 @property
2800 def _select_iterable(self) -> _SelectIterable:
2801 return itertools.chain.from_iterable(
2802 [elem._select_iterable for elem in self.clauses]
2803 )
2805 def append(self, clause):
2806 if self.group_contents:
2807 self.clauses.append(
2808 coercions.expect(self._text_converter_role, clause).self_group(
2809 against=self.operator
2810 )
2811 )
2812 else:
2813 self.clauses.append(
2814 coercions.expect(self._text_converter_role, clause)
2815 )
2817 @util.ro_non_memoized_property
2818 def _from_objects(self) -> List[FromClause]:
2819 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
2821 def self_group(
2822 self, against: Optional[OperatorType] = None
2823 ) -> Union[Self, Grouping[Any]]:
2824 if self.group and operators.is_precedent(self.operator, against):
2825 return Grouping(self)
2826 else:
2827 return self
2830class OperatorExpression(ColumnElement[_T]):
2831 """base for expressions that contain an operator and operands
2833 .. versionadded:: 2.0
2835 """
2837 operator: OperatorType
2838 type: TypeEngine[_T]
2840 group: bool = True
2842 @property
2843 def is_comparison(self):
2844 return operators.is_comparison(self.operator)
2846 def self_group(
2847 self, against: Optional[OperatorType] = None
2848 ) -> Union[Self, Grouping[_T]]:
2849 if (
2850 self.group
2851 and operators.is_precedent(self.operator, against)
2852 or (
2853 # a negate against a non-boolean operator
2854 # doesn't make too much sense but we should
2855 # group for that
2856 against is operators.inv
2857 and not operators.is_boolean(self.operator)
2858 )
2859 ):
2860 return Grouping(self)
2861 else:
2862 return self
2864 @property
2865 def _flattened_operator_clauses(
2866 self,
2867 ) -> typing_Tuple[ColumnElement[Any], ...]:
2868 raise NotImplementedError()
2870 @classmethod
2871 def _construct_for_op(
2872 cls,
2873 left: ColumnElement[Any],
2874 right: ColumnElement[Any],
2875 op: OperatorType,
2876 *,
2877 type_: TypeEngine[_T],
2878 negate: Optional[OperatorType] = None,
2879 modifiers: Optional[Mapping[str, Any]] = None,
2880 ) -> OperatorExpression[_T]:
2881 if operators.is_associative(op):
2882 assert (
2883 negate is None
2884 ), f"negate not supported for associative operator {op}"
2886 multi = False
2887 if getattr(
2888 left, "operator", None
2889 ) is op and type_._compare_type_affinity(left.type):
2890 multi = True
2891 left_flattened = left._flattened_operator_clauses
2892 else:
2893 left_flattened = (left,)
2895 if getattr(
2896 right, "operator", None
2897 ) is op and type_._compare_type_affinity(right.type):
2898 multi = True
2899 right_flattened = right._flattened_operator_clauses
2900 else:
2901 right_flattened = (right,)
2903 if multi:
2904 return ExpressionClauseList._construct_for_list(
2905 op,
2906 type_,
2907 *(left_flattened + right_flattened),
2908 )
2910 if right._is_collection_aggregate:
2911 negate = None
2913 return BinaryExpression(
2914 left, right, op, type_=type_, negate=negate, modifiers=modifiers
2915 )
2918class ExpressionClauseList(OperatorExpression[_T]):
2919 """Describe a list of clauses, separated by an operator,
2920 in a column expression context.
2922 :class:`.ExpressionClauseList` differs from :class:`.ClauseList` in that
2923 it represents a column-oriented DQL expression only, not an open ended
2924 list of anything comma separated.
2926 .. versionadded:: 2.0
2928 """
2930 __visit_name__ = "expression_clauselist"
2932 _traverse_internals: _TraverseInternalsType = [
2933 ("clauses", InternalTraversal.dp_clauseelement_tuple),
2934 ("operator", InternalTraversal.dp_operator),
2935 ]
2937 clauses: typing_Tuple[ColumnElement[Any], ...]
2939 group: bool
2941 def __init__(
2942 self,
2943 operator: OperatorType,
2944 *clauses: _ColumnExpressionArgument[Any],
2945 type_: Optional[_TypeEngineArgument[_T]] = None,
2946 ):
2947 self.operator = operator
2949 self.clauses = tuple(
2950 coercions.expect(
2951 roles.ExpressionElementRole, clause, apply_propagate_attrs=self
2952 )
2953 for clause in clauses
2954 )
2955 self._is_implicitly_boolean = operators.is_boolean(self.operator)
2956 self.type = type_api.to_instance(type_) # type: ignore
2958 @property
2959 def _flattened_operator_clauses(
2960 self,
2961 ) -> typing_Tuple[ColumnElement[Any], ...]:
2962 return self.clauses
2964 def __iter__(self) -> Iterator[ColumnElement[Any]]:
2965 return iter(self.clauses)
2967 def __len__(self) -> int:
2968 return len(self.clauses)
2970 @property
2971 def _select_iterable(self) -> _SelectIterable:
2972 return (self,)
2974 @util.ro_non_memoized_property
2975 def _from_objects(self) -> List[FromClause]:
2976 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
2978 def _append_inplace(self, clause: ColumnElement[Any]) -> None:
2979 self.clauses += (clause,)
2981 @classmethod
2982 def _construct_for_list(
2983 cls,
2984 operator: OperatorType,
2985 type_: TypeEngine[_T],
2986 *clauses: ColumnElement[Any],
2987 group: bool = True,
2988 ) -> ExpressionClauseList[_T]:
2989 self = cls.__new__(cls)
2990 self.group = group
2991 if group:
2992 self.clauses = tuple(
2993 c.self_group(against=operator) for c in clauses
2994 )
2995 else:
2996 self.clauses = clauses
2997 self.operator = operator
2998 self.type = type_
2999 for c in clauses:
3000 if c._propagate_attrs:
3001 self._propagate_attrs = c._propagate_attrs
3002 break
3003 return self
3005 def _negate(self) -> Any:
3006 grouped = self.self_group(against=operators.inv)
3007 assert isinstance(grouped, ColumnElement)
3008 return UnaryExpression(grouped, operator=operators.inv)
3011class BooleanClauseList(ExpressionClauseList[bool]):
3012 __visit_name__ = "expression_clauselist"
3013 inherit_cache = True
3015 def __init__(self, *arg, **kw):
3016 raise NotImplementedError(
3017 "BooleanClauseList has a private constructor"
3018 )
3020 @classmethod
3021 def _process_clauses_for_boolean(
3022 cls,
3023 operator: OperatorType,
3024 continue_on: Any,
3025 skip_on: Any,
3026 clauses: Iterable[ColumnElement[Any]],
3027 ) -> typing_Tuple[int, List[ColumnElement[Any]]]:
3028 has_continue_on = None
3030 convert_clauses = []
3032 against = operators._asbool
3033 lcc = 0
3035 for clause in clauses:
3036 if clause is continue_on:
3037 # instance of continue_on, like and_(x, y, True, z), store it
3038 # if we didn't find one already, we will use it if there
3039 # are no other expressions here.
3040 has_continue_on = clause
3041 elif clause is skip_on:
3042 # instance of skip_on, e.g. and_(x, y, False, z), cancels
3043 # the rest out
3044 convert_clauses = [clause]
3045 lcc = 1
3046 break
3047 else:
3048 if not lcc:
3049 lcc = 1
3050 else:
3051 against = operator
3052 # technically this would be len(convert_clauses) + 1
3053 # however this only needs to indicate "greater than one"
3054 lcc = 2
3055 convert_clauses.append(clause)
3057 if not convert_clauses and has_continue_on is not None:
3058 convert_clauses = [has_continue_on]
3059 lcc = 1
3061 return lcc, [c.self_group(against=against) for c in convert_clauses]
3063 @classmethod
3064 def _construct(
3065 cls,
3066 operator: OperatorType,
3067 continue_on: Any,
3068 skip_on: Any,
3069 initial_clause: Any = _NoArg.NO_ARG,
3070 *clauses: Any,
3071 **kw: Any,
3072 ) -> ColumnElement[Any]:
3073 if initial_clause is _NoArg.NO_ARG:
3074 # no elements period. deprecated use case. return an empty
3075 # ClauseList construct that generates nothing unless it has
3076 # elements added to it.
3077 name = operator.__name__
3079 util.warn_deprecated(
3080 f"Invoking {name}() without arguments is deprecated, and "
3081 f"will be disallowed in a future release. For an empty "
3082 f"""{name}() construct, use '{name}({
3083 'true()' if continue_on is True_._singleton else 'false()'
3084 }, *args)' """
3085 f"""or '{name}({
3086 'True' if continue_on is True_._singleton else 'False'
3087 }, *args)'.""",
3088 version="1.4",
3089 )
3090 return cls._construct_raw(operator)
3092 lcc, convert_clauses = cls._process_clauses_for_boolean(
3093 operator,
3094 continue_on,
3095 skip_on,
3096 [
3097 coercions.expect(roles.WhereHavingRole, clause)
3098 for clause in util.coerce_generator_arg(
3099 (initial_clause,) + clauses
3100 )
3101 ],
3102 )
3104 if lcc > 1:
3105 # multiple elements. Return regular BooleanClauseList
3106 # which will link elements against the operator.
3108 flattened_clauses = itertools.chain.from_iterable(
3109 (
3110 (c for c in to_flat._flattened_operator_clauses)
3111 if getattr(to_flat, "operator", None) is operator
3112 else (to_flat,)
3113 )
3114 for to_flat in convert_clauses
3115 )
3117 return cls._construct_raw(operator, flattened_clauses) # type: ignore # noqa: E501
3118 else:
3119 assert lcc
3120 # just one element. return it as a single boolean element,
3121 # not a list and discard the operator.
3122 return convert_clauses[0]
3124 @classmethod
3125 def _construct_for_whereclause(
3126 cls, clauses: Iterable[ColumnElement[Any]]
3127 ) -> Optional[ColumnElement[bool]]:
3128 operator, continue_on, skip_on = (
3129 operators.and_,
3130 True_._singleton,
3131 False_._singleton,
3132 )
3134 lcc, convert_clauses = cls._process_clauses_for_boolean(
3135 operator,
3136 continue_on,
3137 skip_on,
3138 clauses, # these are assumed to be coerced already
3139 )
3141 if lcc > 1:
3142 # multiple elements. Return regular BooleanClauseList
3143 # which will link elements against the operator.
3144 return cls._construct_raw(operator, convert_clauses)
3145 elif lcc == 1:
3146 # just one element. return it as a single boolean element,
3147 # not a list and discard the operator.
3148 return convert_clauses[0]
3149 else:
3150 return None
3152 @classmethod
3153 def _construct_raw(
3154 cls,
3155 operator: OperatorType,
3156 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
3157 ) -> BooleanClauseList:
3158 self = cls.__new__(cls)
3159 self.clauses = tuple(clauses) if clauses else ()
3160 self.group = True
3161 self.operator = operator
3162 self.type = type_api.BOOLEANTYPE
3163 self._is_implicitly_boolean = True
3164 return self
3166 @classmethod
3167 def and_(
3168 cls,
3169 initial_clause: Union[
3170 Literal[True], _ColumnExpressionArgument[bool], _NoArg
3171 ] = _NoArg.NO_ARG,
3172 *clauses: _ColumnExpressionArgument[bool],
3173 ) -> ColumnElement[bool]:
3174 r"""Produce a conjunction of expressions joined by ``AND``.
3176 See :func:`_sql.and_` for full documentation.
3177 """
3178 return cls._construct(
3179 operators.and_,
3180 True_._singleton,
3181 False_._singleton,
3182 initial_clause,
3183 *clauses,
3184 )
3186 @classmethod
3187 def or_(
3188 cls,
3189 initial_clause: Union[
3190 Literal[False], _ColumnExpressionArgument[bool], _NoArg
3191 ] = _NoArg.NO_ARG,
3192 *clauses: _ColumnExpressionArgument[bool],
3193 ) -> ColumnElement[bool]:
3194 """Produce a conjunction of expressions joined by ``OR``.
3196 See :func:`_sql.or_` for full documentation.
3197 """
3198 return cls._construct(
3199 operators.or_,
3200 False_._singleton,
3201 True_._singleton,
3202 initial_clause,
3203 *clauses,
3204 )
3206 @property
3207 def _select_iterable(self) -> _SelectIterable:
3208 return (self,)
3210 def self_group(
3211 self, against: Optional[OperatorType] = None
3212 ) -> Union[Self, Grouping[bool]]:
3213 if not self.clauses:
3214 return self
3215 else:
3216 return super().self_group(against=against)
3219and_ = BooleanClauseList.and_
3220or_ = BooleanClauseList.or_
3223class Tuple(ClauseList, ColumnElement[typing_Tuple[Any, ...]]):
3224 """Represent a SQL tuple."""
3226 __visit_name__ = "tuple"
3228 _traverse_internals: _TraverseInternalsType = (
3229 ClauseList._traverse_internals + []
3230 )
3232 type: TupleType
3234 @util.preload_module("sqlalchemy.sql.sqltypes")
3235 def __init__(
3236 self,
3237 *clauses: _ColumnExpressionArgument[Any],
3238 types: Optional[Sequence[_TypeEngineArgument[Any]]] = None,
3239 ):
3240 sqltypes = util.preloaded.sql_sqltypes
3242 if types is None:
3243 init_clauses: List[ColumnElement[Any]] = [
3244 coercions.expect(roles.ExpressionElementRole, c)
3245 for c in clauses
3246 ]
3247 else:
3248 if len(types) != len(clauses):
3249 raise exc.ArgumentError(
3250 "Wrong number of elements for %d-tuple: %r "
3251 % (len(types), clauses)
3252 )
3253 init_clauses = [
3254 coercions.expect(
3255 roles.ExpressionElementRole,
3256 c,
3257 type_=typ if not typ._isnull else None,
3258 )
3259 for typ, c in zip(types, clauses)
3260 ]
3262 self.type = sqltypes.TupleType(*[arg.type for arg in init_clauses])
3263 super().__init__(*init_clauses)
3265 @property
3266 def _select_iterable(self) -> _SelectIterable:
3267 return (self,)
3269 def _bind_param(self, operator, obj, type_=None, expanding=False):
3270 if expanding:
3271 return BindParameter(
3272 None,
3273 value=obj,
3274 _compared_to_operator=operator,
3275 unique=True,
3276 expanding=True,
3277 type_=type_,
3278 _compared_to_type=self.type,
3279 )
3280 else:
3281 return Tuple(
3282 *[
3283 BindParameter(
3284 None,
3285 o,
3286 _compared_to_operator=operator,
3287 _compared_to_type=compared_to_type,
3288 unique=True,
3289 type_=type_,
3290 )
3291 for o, compared_to_type in zip(obj, self.type.types)
3292 ]
3293 )
3295 def self_group(self, against: Optional[OperatorType] = None) -> Self:
3296 # Tuple is parenthesized by definition.
3297 return self
3300class Case(ColumnElement[_T]):
3301 """Represent a ``CASE`` expression.
3303 :class:`.Case` is produced using the :func:`.case` factory function,
3304 as in::
3306 from sqlalchemy import case
3308 stmt = select(users_table).where(
3309 case(
3310 (users_table.c.name == "wendy", "W"),
3311 (users_table.c.name == "jack", "J"),
3312 else_="E",
3313 )
3314 )
3316 Details on :class:`.Case` usage is at :func:`.case`.
3318 .. seealso::
3320 :func:`.case`
3322 """
3324 __visit_name__ = "case"
3326 _traverse_internals: _TraverseInternalsType = [
3327 ("value", InternalTraversal.dp_clauseelement),
3328 ("whens", InternalTraversal.dp_clauseelement_tuples),
3329 ("else_", InternalTraversal.dp_clauseelement),
3330 ]
3332 # for case(), the type is derived from the whens. so for the moment
3333 # users would have to cast() the case to get a specific type
3335 whens: List[typing_Tuple[ColumnElement[bool], ColumnElement[_T]]]
3336 else_: Optional[ColumnElement[_T]]
3337 value: Optional[ColumnElement[Any]]
3339 def __init__(
3340 self,
3341 *whens: Union[
3342 typing_Tuple[_ColumnExpressionArgument[bool], Any],
3343 Mapping[Any, Any],
3344 ],
3345 value: Optional[Any] = None,
3346 else_: Optional[Any] = None,
3347 ):
3348 new_whens: Iterable[Any] = coercions._expression_collection_was_a_list(
3349 "whens", "case", whens
3350 )
3351 try:
3352 new_whens = util.dictlike_iteritems(new_whens)
3353 except TypeError:
3354 pass
3356 self.whens = [
3357 (
3358 coercions.expect(
3359 roles.ExpressionElementRole,
3360 c,
3361 apply_propagate_attrs=self,
3362 ).self_group(),
3363 coercions.expect(roles.ExpressionElementRole, r),
3364 )
3365 for (c, r) in new_whens
3366 ]
3368 if value is None:
3369 self.value = None
3370 else:
3371 self.value = coercions.expect(roles.ExpressionElementRole, value)
3373 if else_ is not None:
3374 self.else_ = coercions.expect(roles.ExpressionElementRole, else_)
3375 else:
3376 self.else_ = None
3378 type_ = next(
3379 (
3380 then.type
3381 # Iterate `whens` in reverse to match previous behaviour
3382 # where type of final element took priority
3383 for *_, then in reversed(self.whens)
3384 if not then.type._isnull
3385 ),
3386 self.else_.type if self.else_ is not None else type_api.NULLTYPE,
3387 )
3388 self.type = cast(_T, type_)
3390 @util.ro_non_memoized_property
3391 def _from_objects(self) -> List[FromClause]:
3392 return list(
3393 itertools.chain(*[x._from_objects for x in self.get_children()])
3394 )
3397class Cast(WrapsColumnExpression[_T]):
3398 """Represent a ``CAST`` expression.
3400 :class:`.Cast` is produced using the :func:`.cast` factory function,
3401 as in::
3403 from sqlalchemy import cast, Numeric
3405 stmt = select(cast(product_table.c.unit_price, Numeric(10, 4)))
3407 Details on :class:`.Cast` usage is at :func:`.cast`.
3409 .. seealso::
3411 :ref:`tutorial_casts`
3413 :func:`.cast`
3415 :func:`.try_cast`
3417 :func:`.type_coerce` - an alternative to CAST that coerces the type
3418 on the Python side only, which is often sufficient to generate the
3419 correct SQL and data coercion.
3421 """
3423 __visit_name__ = "cast"
3425 _traverse_internals: _TraverseInternalsType = [
3426 ("clause", InternalTraversal.dp_clauseelement),
3427 ("type", InternalTraversal.dp_type),
3428 ]
3430 clause: ColumnElement[Any]
3431 type: TypeEngine[_T]
3432 typeclause: TypeClause
3434 def __init__(
3435 self,
3436 expression: _ColumnExpressionArgument[Any],
3437 type_: _TypeEngineArgument[_T],
3438 ):
3439 self.type = type_api.to_instance(type_)
3440 self.clause = coercions.expect(
3441 roles.ExpressionElementRole,
3442 expression,
3443 type_=self.type,
3444 apply_propagate_attrs=self,
3445 )
3446 self.typeclause = TypeClause(self.type)
3448 @util.ro_non_memoized_property
3449 def _from_objects(self) -> List[FromClause]:
3450 return self.clause._from_objects
3452 @property
3453 def wrapped_column_expression(self):
3454 return self.clause
3457class TryCast(Cast[_T]):
3458 """Represent a TRY_CAST expression.
3460 Details on :class:`.TryCast` usage is at :func:`.try_cast`.
3462 .. seealso::
3464 :func:`.try_cast`
3466 :ref:`tutorial_casts`
3467 """
3469 __visit_name__ = "try_cast"
3470 inherit_cache = True
3473class TypeCoerce(WrapsColumnExpression[_T]):
3474 """Represent a Python-side type-coercion wrapper.
3476 :class:`.TypeCoerce` supplies the :func:`_expression.type_coerce`
3477 function; see that function for usage details.
3479 .. seealso::
3481 :func:`_expression.type_coerce`
3483 :func:`.cast`
3485 """
3487 __visit_name__ = "type_coerce"
3489 _traverse_internals: _TraverseInternalsType = [
3490 ("clause", InternalTraversal.dp_clauseelement),
3491 ("type", InternalTraversal.dp_type),
3492 ]
3494 clause: ColumnElement[Any]
3495 type: TypeEngine[_T]
3497 def __init__(
3498 self,
3499 expression: _ColumnExpressionArgument[Any],
3500 type_: _TypeEngineArgument[_T],
3501 ):
3502 self.type = type_api.to_instance(type_)
3503 self.clause = coercions.expect(
3504 roles.ExpressionElementRole,
3505 expression,
3506 type_=self.type,
3507 apply_propagate_attrs=self,
3508 )
3510 @util.ro_non_memoized_property
3511 def _from_objects(self) -> List[FromClause]:
3512 return self.clause._from_objects
3514 @HasMemoized.memoized_attribute
3515 def typed_expression(self):
3516 if isinstance(self.clause, BindParameter):
3517 bp = self.clause._clone()
3518 bp.type = self.type
3519 return bp
3520 else:
3521 return self.clause
3523 @property
3524 def wrapped_column_expression(self):
3525 return self.clause
3527 def self_group(
3528 self, against: Optional[OperatorType] = None
3529 ) -> TypeCoerce[_T]:
3530 grouped = self.clause.self_group(against=against)
3531 if grouped is not self.clause:
3532 return TypeCoerce(grouped, self.type)
3533 else:
3534 return self
3537class Extract(ColumnElement[int]):
3538 """Represent a SQL EXTRACT clause, ``extract(field FROM expr)``."""
3540 __visit_name__ = "extract"
3542 _traverse_internals: _TraverseInternalsType = [
3543 ("expr", InternalTraversal.dp_clauseelement),
3544 ("field", InternalTraversal.dp_string),
3545 ]
3547 expr: ColumnElement[Any]
3548 field: str
3550 def __init__(self, field: str, expr: _ColumnExpressionArgument[Any]):
3551 self.type = type_api.INTEGERTYPE
3552 self.field = field
3553 self.expr = coercions.expect(roles.ExpressionElementRole, expr)
3555 @util.ro_non_memoized_property
3556 def _from_objects(self) -> List[FromClause]:
3557 return self.expr._from_objects
3560class _label_reference(ColumnElement[_T]):
3561 """Wrap a column expression as it appears in a 'reference' context.
3563 This expression is any that includes an _order_by_label_element,
3564 which is a Label, or a DESC / ASC construct wrapping a Label.
3566 The production of _label_reference() should occur when an expression
3567 is added to this context; this includes the ORDER BY or GROUP BY of a
3568 SELECT statement, as well as a few other places, such as the ORDER BY
3569 within an OVER clause.
3571 """
3573 __visit_name__ = "label_reference"
3575 _traverse_internals: _TraverseInternalsType = [
3576 ("element", InternalTraversal.dp_clauseelement)
3577 ]
3579 element: ColumnElement[_T]
3581 def __init__(self, element: ColumnElement[_T]):
3582 self.element = element
3584 @util.ro_non_memoized_property
3585 def _from_objects(self) -> List[FromClause]:
3586 return []
3589class _textual_label_reference(ColumnElement[Any]):
3590 __visit_name__ = "textual_label_reference"
3592 _traverse_internals: _TraverseInternalsType = [
3593 ("element", InternalTraversal.dp_string)
3594 ]
3596 def __init__(self, element: str):
3597 self.element = element
3599 @util.memoized_property
3600 def _text_clause(self) -> TextClause:
3601 return TextClause(self.element)
3604class UnaryExpression(ColumnElement[_T]):
3605 """Define a 'unary' expression.
3607 A unary expression has a single column expression
3608 and an operator. The operator can be placed on the left
3609 (where it is called the 'operator') or right (where it is called the
3610 'modifier') of the column expression.
3612 :class:`.UnaryExpression` is the basis for several unary operators
3613 including those used by :func:`.desc`, :func:`.asc`, :func:`.distinct`,
3614 :func:`.nulls_first` and :func:`.nulls_last`.
3616 """
3618 __visit_name__ = "unary"
3620 _traverse_internals: _TraverseInternalsType = [
3621 ("element", InternalTraversal.dp_clauseelement),
3622 ("operator", InternalTraversal.dp_operator),
3623 ("modifier", InternalTraversal.dp_operator),
3624 ]
3626 element: ColumnElement[Any]
3627 operator: Optional[OperatorType]
3628 modifier: Optional[OperatorType]
3630 def __init__(
3631 self,
3632 element: ColumnElement[Any],
3633 *,
3634 operator: Optional[OperatorType] = None,
3635 modifier: Optional[OperatorType] = None,
3636 type_: Optional[_TypeEngineArgument[_T]] = None,
3637 wraps_column_expression: bool = False, # legacy, not used as of 2.0.42
3638 ):
3639 self.operator = operator
3640 self.modifier = modifier
3641 self._propagate_attrs = element._propagate_attrs
3642 self.element = element.self_group(
3643 against=self.operator or self.modifier
3644 )
3646 # if type is None, we get NULLTYPE, which is our _T. But I don't
3647 # know how to get the overloads to express that correctly
3648 self.type = type_api.to_instance(type_) # type: ignore
3650 def _wraps_unnamed_column(self):
3651 ungrouped = self.element._ungroup()
3652 return (
3653 not isinstance(ungrouped, NamedColumn)
3654 or ungrouped._non_anon_label is None
3655 )
3657 @classmethod
3658 def _create_nulls_first(
3659 cls,
3660 column: _ColumnExpressionArgument[_T],
3661 ) -> UnaryExpression[_T]:
3662 return UnaryExpression(
3663 coercions.expect(roles.ByOfRole, column),
3664 modifier=operators.nulls_first_op,
3665 )
3667 @classmethod
3668 def _create_nulls_last(
3669 cls,
3670 column: _ColumnExpressionArgument[_T],
3671 ) -> UnaryExpression[_T]:
3672 return UnaryExpression(
3673 coercions.expect(roles.ByOfRole, column),
3674 modifier=operators.nulls_last_op,
3675 )
3677 @classmethod
3678 def _create_desc(
3679 cls, column: _ColumnExpressionOrStrLabelArgument[_T]
3680 ) -> UnaryExpression[_T]:
3681 return UnaryExpression(
3682 coercions.expect(roles.ByOfRole, column),
3683 modifier=operators.desc_op,
3684 )
3686 @classmethod
3687 def _create_asc(
3688 cls,
3689 column: _ColumnExpressionOrStrLabelArgument[_T],
3690 ) -> UnaryExpression[_T]:
3691 return UnaryExpression(
3692 coercions.expect(roles.ByOfRole, column),
3693 modifier=operators.asc_op,
3694 )
3696 @classmethod
3697 def _create_distinct(
3698 cls,
3699 expr: _ColumnExpressionArgument[_T],
3700 ) -> UnaryExpression[_T]:
3701 col_expr: ColumnElement[_T] = coercions.expect(
3702 roles.ExpressionElementRole, expr
3703 )
3704 return UnaryExpression(
3705 col_expr,
3706 operator=operators.distinct_op,
3707 type_=col_expr.type,
3708 )
3710 @classmethod
3711 def _create_bitwise_not(
3712 cls,
3713 expr: _ColumnExpressionArgument[_T],
3714 ) -> UnaryExpression[_T]:
3715 col_expr: ColumnElement[_T] = coercions.expect(
3716 roles.ExpressionElementRole, expr
3717 )
3718 return UnaryExpression(
3719 col_expr,
3720 operator=operators.bitwise_not_op,
3721 type_=col_expr.type,
3722 )
3724 @property
3725 def _order_by_label_element(self) -> Optional[Label[Any]]:
3726 if operators.is_order_by_modifier(self.modifier):
3727 return self.element._order_by_label_element
3728 else:
3729 return None
3731 @util.ro_non_memoized_property
3732 def _from_objects(self) -> List[FromClause]:
3733 return self.element._from_objects
3735 def _negate(self) -> ColumnElement[Any]:
3736 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
3737 return UnaryExpression(
3738 self.self_group(against=operators.inv),
3739 operator=operators.inv,
3740 type_=type_api.BOOLEANTYPE,
3741 )
3742 else:
3743 return ColumnElement._negate(self)
3745 def self_group(
3746 self, against: Optional[OperatorType] = None
3747 ) -> Union[Self, Grouping[_T]]:
3748 if self.operator and operators.is_precedent(self.operator, against):
3749 return Grouping(self)
3750 else:
3751 return self
3754class CollectionAggregate(UnaryExpression[_T]):
3755 """Forms the basis for right-hand collection operator modifiers
3756 ANY and ALL.
3758 The ANY and ALL keywords are available in different ways on different
3759 backends. On PostgreSQL, they only work for an ARRAY type. On
3760 MySQL, they only work for subqueries.
3762 """
3764 inherit_cache = True
3765 _is_collection_aggregate = True
3767 @classmethod
3768 def _create_any(
3769 cls, expr: _ColumnExpressionArgument[_T]
3770 ) -> CollectionAggregate[bool]:
3771 col_expr: ColumnElement[_T] = coercions.expect(
3772 roles.ExpressionElementRole,
3773 expr,
3774 )
3775 col_expr = col_expr.self_group()
3776 return CollectionAggregate(
3777 col_expr,
3778 operator=operators.any_op,
3779 type_=type_api.BOOLEANTYPE,
3780 )
3782 @classmethod
3783 def _create_all(
3784 cls, expr: _ColumnExpressionArgument[_T]
3785 ) -> CollectionAggregate[bool]:
3786 col_expr: ColumnElement[_T] = coercions.expect(
3787 roles.ExpressionElementRole,
3788 expr,
3789 )
3790 col_expr = col_expr.self_group()
3791 return CollectionAggregate(
3792 col_expr,
3793 operator=operators.all_op,
3794 type_=type_api.BOOLEANTYPE,
3795 )
3797 # operate and reverse_operate are hardwired to
3798 # dispatch onto the type comparator directly, so that we can
3799 # ensure "reversed" behavior.
3800 def operate(
3801 self, op: OperatorType, *other: Any, **kwargs: Any
3802 ) -> ColumnElement[_T]:
3803 if not operators.is_comparison(op):
3804 raise exc.ArgumentError(
3805 "Only comparison operators may be used with ANY/ALL"
3806 )
3807 kwargs["reverse"] = True
3808 return self.comparator.operate(operators.mirror(op), *other, **kwargs)
3810 def reverse_operate(
3811 self, op: OperatorType, other: Any, **kwargs: Any
3812 ) -> ColumnElement[_T]:
3813 # comparison operators should never call reverse_operate
3814 assert not operators.is_comparison(op)
3815 raise exc.ArgumentError(
3816 "Only comparison operators may be used with ANY/ALL"
3817 )
3820class AsBoolean(WrapsColumnExpression[bool], UnaryExpression[bool]):
3821 inherit_cache = True
3823 def __init__(self, element, operator, negate):
3824 self.element = element
3825 self.type = type_api.BOOLEANTYPE
3826 self.operator = operator
3827 self.negate = negate
3828 self.modifier = None
3829 self._is_implicitly_boolean = element._is_implicitly_boolean
3831 @property
3832 def wrapped_column_expression(self):
3833 return self.element
3835 def self_group(self, against: Optional[OperatorType] = None) -> Self:
3836 return self
3838 def _negate(self):
3839 if isinstance(self.element, (True_, False_)):
3840 return self.element._negate()
3841 else:
3842 return AsBoolean(self.element, self.negate, self.operator)
3845class BinaryExpression(OperatorExpression[_T]):
3846 """Represent an expression that is ``LEFT <operator> RIGHT``.
3848 A :class:`.BinaryExpression` is generated automatically
3849 whenever two column expressions are used in a Python binary expression:
3851 .. sourcecode:: pycon+sql
3853 >>> from sqlalchemy.sql import column
3854 >>> column("a") + column("b")
3855 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
3856 >>> print(column("a") + column("b"))
3857 {printsql}a + b
3859 """
3861 __visit_name__ = "binary"
3863 _traverse_internals: _TraverseInternalsType = [
3864 ("left", InternalTraversal.dp_clauseelement),
3865 ("right", InternalTraversal.dp_clauseelement),
3866 ("operator", InternalTraversal.dp_operator),
3867 ("negate", InternalTraversal.dp_operator),
3868 ("modifiers", InternalTraversal.dp_plain_dict),
3869 (
3870 "type",
3871 InternalTraversal.dp_type,
3872 ),
3873 ]
3875 _cache_key_traversal = [
3876 ("left", InternalTraversal.dp_clauseelement),
3877 ("right", InternalTraversal.dp_clauseelement),
3878 ("operator", InternalTraversal.dp_operator),
3879 ("modifiers", InternalTraversal.dp_plain_dict),
3880 # "type" affects JSON CAST operators, so while redundant in most cases,
3881 # is needed for that one
3882 (
3883 "type",
3884 InternalTraversal.dp_type,
3885 ),
3886 ]
3888 _is_implicitly_boolean = True
3889 """Indicates that any database will know this is a boolean expression
3890 even if the database does not have an explicit boolean datatype.
3892 """
3894 left: ColumnElement[Any]
3895 right: ColumnElement[Any]
3896 modifiers: Mapping[str, Any]
3898 def __init__(
3899 self,
3900 left: ColumnElement[Any],
3901 right: ColumnElement[Any],
3902 operator: OperatorType,
3903 type_: Optional[_TypeEngineArgument[_T]] = None,
3904 negate: Optional[OperatorType] = None,
3905 modifiers: Optional[Mapping[str, Any]] = None,
3906 ):
3907 # allow compatibility with libraries that
3908 # refer to BinaryExpression directly and pass strings
3909 if isinstance(operator, str):
3910 operator = operators.custom_op(operator)
3911 self._orig = (left.__hash__(), right.__hash__())
3912 self._propagate_attrs = left._propagate_attrs or right._propagate_attrs
3913 self.left = left.self_group(against=operator)
3914 self.right = right.self_group(against=operator)
3915 self.operator = operator
3917 # if type is None, we get NULLTYPE, which is our _T. But I don't
3918 # know how to get the overloads to express that correctly
3919 self.type = type_api.to_instance(type_) # type: ignore
3921 self.negate = negate
3922 self._is_implicitly_boolean = operators.is_boolean(operator)
3924 if modifiers is None:
3925 self.modifiers = {}
3926 else:
3927 self.modifiers = modifiers
3929 @property
3930 def _flattened_operator_clauses(
3931 self,
3932 ) -> typing_Tuple[ColumnElement[Any], ...]:
3933 return (self.left, self.right)
3935 def __bool__(self):
3936 """Implement Python-side "bool" for BinaryExpression as a
3937 simple "identity" check for the left and right attributes,
3938 if the operator is "eq" or "ne". Otherwise the expression
3939 continues to not support "bool" like all other column expressions.
3941 The rationale here is so that ColumnElement objects can be hashable.
3942 What? Well, suppose you do this::
3944 c1, c2 = column("x"), column("y")
3945 s1 = set([c1, c2])
3947 We do that **a lot**, columns inside of sets is an extremely basic
3948 thing all over the ORM for example.
3950 So what happens if we do this? ::
3952 c1 in s1
3954 Hashing means it will normally use ``__hash__()`` of the object,
3955 but in case of hash collision, it's going to also do ``c1 == c1``
3956 and/or ``c1 == c2`` inside. Those operations need to return a
3957 True/False value. But because we override ``==`` and ``!=``, they're
3958 going to get a BinaryExpression. Hence we implement ``__bool__`` here
3959 so that these comparisons behave in this particular context mostly
3960 like regular object comparisons. Thankfully Python is OK with
3961 that! Otherwise we'd have to use special set classes for columns
3962 (which we used to do, decades ago).
3964 """
3965 if self.operator in (operators.eq, operators.ne):
3966 # this is using the eq/ne operator given int hash values,
3967 # rather than Operator, so that "bool" can be based on
3968 # identity
3969 return self.operator(*self._orig) # type: ignore
3970 else:
3971 raise TypeError("Boolean value of this clause is not defined")
3973 if typing.TYPE_CHECKING:
3975 def __invert__(
3976 self: BinaryExpression[_T],
3977 ) -> BinaryExpression[_T]: ...
3979 @util.ro_non_memoized_property
3980 def _from_objects(self) -> List[FromClause]:
3981 return self.left._from_objects + self.right._from_objects
3983 def _negate(self):
3984 if self.negate is not None:
3985 return BinaryExpression(
3986 self.left,
3987 self.right._negate_in_binary(self.negate, self.operator),
3988 self.negate,
3989 negate=self.operator,
3990 type_=self.type,
3991 modifiers=self.modifiers,
3992 )
3993 else:
3994 return self.self_group()._negate()
3997class Slice(ColumnElement[Any]):
3998 """Represent SQL for a Python array-slice object.
4000 This is not a specific SQL construct at this level, but
4001 may be interpreted by specific dialects, e.g. PostgreSQL.
4003 """
4005 __visit_name__ = "slice"
4007 _traverse_internals: _TraverseInternalsType = [
4008 ("start", InternalTraversal.dp_clauseelement),
4009 ("stop", InternalTraversal.dp_clauseelement),
4010 ("step", InternalTraversal.dp_clauseelement),
4011 ]
4013 def __init__(self, start, stop, step, _name=None):
4014 self.start = coercions.expect(
4015 roles.ExpressionElementRole,
4016 start,
4017 name=_name,
4018 type_=type_api.INTEGERTYPE,
4019 )
4020 self.stop = coercions.expect(
4021 roles.ExpressionElementRole,
4022 stop,
4023 name=_name,
4024 type_=type_api.INTEGERTYPE,
4025 )
4026 self.step = coercions.expect(
4027 roles.ExpressionElementRole,
4028 step,
4029 name=_name,
4030 type_=type_api.INTEGERTYPE,
4031 )
4032 self.type = type_api.NULLTYPE
4034 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4035 assert against is operator.getitem
4036 return self
4039class IndexExpression(BinaryExpression[Any]):
4040 """Represent the class of expressions that are like an "index"
4041 operation."""
4043 inherit_cache = True
4046class GroupedElement(DQLDMLClauseElement):
4047 """Represent any parenthesized expression"""
4049 __visit_name__ = "grouping"
4051 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4052 return self
4054 def _ungroup(self) -> ClauseElement:
4055 raise NotImplementedError()
4058class Grouping(GroupedElement, ColumnElement[_T]):
4059 """Represent a grouping within a column expression"""
4061 _traverse_internals: _TraverseInternalsType = [
4062 ("element", InternalTraversal.dp_clauseelement),
4063 ("type", InternalTraversal.dp_type),
4064 ]
4066 _cache_key_traversal = [
4067 ("element", InternalTraversal.dp_clauseelement),
4068 ]
4070 element: Union[TextClause, ClauseList, ColumnElement[_T]]
4072 def __init__(
4073 self, element: Union[TextClause, ClauseList, ColumnElement[_T]]
4074 ):
4075 self.element = element
4077 # nulltype assignment issue
4078 self.type = getattr(element, "type", type_api.NULLTYPE) # type: ignore
4079 self._propagate_attrs = element._propagate_attrs
4081 def _with_binary_element_type(self, type_):
4082 return self.__class__(self.element._with_binary_element_type(type_))
4084 def _ungroup(self) -> ColumnElement[_T]:
4085 assert isinstance(self.element, ColumnElement)
4086 return self.element._ungroup()
4088 @util.memoized_property
4089 def _is_implicitly_boolean(self):
4090 return self.element._is_implicitly_boolean
4092 @util.non_memoized_property
4093 def _tq_label(self) -> Optional[str]:
4094 return (
4095 getattr(self.element, "_tq_label", None) or self._anon_name_label
4096 )
4098 @util.non_memoized_property
4099 def _proxies(self) -> List[ColumnElement[Any]]:
4100 if isinstance(self.element, ColumnElement):
4101 return [self.element]
4102 else:
4103 return []
4105 @util.ro_non_memoized_property
4106 def _from_objects(self) -> List[FromClause]:
4107 return self.element._from_objects
4109 def __getattr__(self, attr):
4110 return getattr(self.element, attr)
4112 def __getstate__(self):
4113 return {"element": self.element, "type": self.type}
4115 def __setstate__(self, state):
4116 self.element = state["element"]
4117 self.type = state["type"]
4119 if TYPE_CHECKING:
4121 def self_group(
4122 self, against: Optional[OperatorType] = None
4123 ) -> Self: ...
4126class _OverrideBinds(Grouping[_T]):
4127 """used by cache_key->_apply_params_to_element to allow compilation /
4128 execution of a SQL element that's been cached, using an alternate set of
4129 bound parameter values.
4131 This is used by the ORM to swap new parameter values into expressions
4132 that are embedded into loader options like with_expression(),
4133 selectinload(). Previously, this task was accomplished using the
4134 .params() method which would perform a deep-copy instead. This deep
4135 copy proved to be too expensive for more complex expressions.
4137 See #11085
4139 """
4141 __visit_name__ = "override_binds"
4143 def __init__(
4144 self,
4145 element: ColumnElement[_T],
4146 bindparams: Sequence[BindParameter[Any]],
4147 replaces_params: Sequence[BindParameter[Any]],
4148 ):
4149 self.element = element
4150 self.translate = {
4151 k.key: v.value for k, v in zip(replaces_params, bindparams)
4152 }
4154 def _gen_cache_key(
4155 self, anon_map: anon_map, bindparams: List[BindParameter[Any]]
4156 ) -> Optional[typing_Tuple[Any, ...]]:
4157 """generate a cache key for the given element, substituting its bind
4158 values for the translation values present."""
4160 existing_bps: List[BindParameter[Any]] = []
4161 ck = self.element._gen_cache_key(anon_map, existing_bps)
4163 bindparams.extend(
4164 (
4165 bp._with_value(
4166 self.translate[bp.key], maintain_key=True, required=False
4167 )
4168 if bp.key in self.translate
4169 else bp
4170 )
4171 for bp in existing_bps
4172 )
4174 return ck
4177class _OverRange(Enum):
4178 RANGE_UNBOUNDED = 0
4179 RANGE_CURRENT = 1
4182RANGE_UNBOUNDED = _OverRange.RANGE_UNBOUNDED
4183RANGE_CURRENT = _OverRange.RANGE_CURRENT
4185_IntOrRange = Union[int, _OverRange]
4188class Over(ColumnElement[_T]):
4189 """Represent an OVER clause.
4191 This is a special operator against a so-called
4192 "window" function, as well as any aggregate function,
4193 which produces results relative to the result set
4194 itself. Most modern SQL backends now support window functions.
4196 """
4198 __visit_name__ = "over"
4200 _traverse_internals: _TraverseInternalsType = [
4201 ("element", InternalTraversal.dp_clauseelement),
4202 ("order_by", InternalTraversal.dp_clauseelement),
4203 ("partition_by", InternalTraversal.dp_clauseelement),
4204 ("range_", InternalTraversal.dp_plain_obj),
4205 ("rows", InternalTraversal.dp_plain_obj),
4206 ("groups", InternalTraversal.dp_plain_obj),
4207 ]
4209 order_by: Optional[ClauseList] = None
4210 partition_by: Optional[ClauseList] = None
4212 element: ColumnElement[_T]
4213 """The underlying expression object to which this :class:`.Over`
4214 object refers."""
4216 range_: Optional[typing_Tuple[_IntOrRange, _IntOrRange]]
4217 rows: Optional[typing_Tuple[_IntOrRange, _IntOrRange]]
4218 groups: Optional[typing_Tuple[_IntOrRange, _IntOrRange]]
4220 def __init__(
4221 self,
4222 element: ColumnElement[_T],
4223 partition_by: Optional[_ByArgument] = None,
4224 order_by: Optional[_ByArgument] = None,
4225 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4226 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4227 groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4228 ):
4229 self.element = element
4230 if order_by is not None:
4231 self.order_by = ClauseList(
4232 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4233 )
4234 if partition_by is not None:
4235 self.partition_by = ClauseList(
4236 *util.to_list(partition_by),
4237 _literal_as_text_role=roles.ByOfRole,
4238 )
4240 if sum(bool(item) for item in (range_, rows, groups)) > 1:
4241 raise exc.ArgumentError(
4242 "only one of 'rows', 'range_', or 'groups' may be provided"
4243 )
4244 else:
4245 self.range_ = self._interpret_range(range_) if range_ else None
4246 self.rows = self._interpret_range(rows) if rows else None
4247 self.groups = self._interpret_range(groups) if groups else None
4249 def __reduce__(self):
4250 return self.__class__, (
4251 self.element,
4252 self.partition_by,
4253 self.order_by,
4254 self.range_,
4255 self.rows,
4256 self.groups,
4257 )
4259 def _interpret_range(
4260 self,
4261 range_: typing_Tuple[Optional[_IntOrRange], Optional[_IntOrRange]],
4262 ) -> typing_Tuple[_IntOrRange, _IntOrRange]:
4263 if not isinstance(range_, tuple) or len(range_) != 2:
4264 raise exc.ArgumentError("2-tuple expected for range/rows")
4266 r0, r1 = range_
4268 lower: _IntOrRange
4269 upper: _IntOrRange
4271 if r0 is None:
4272 lower = RANGE_UNBOUNDED
4273 elif isinstance(r0, _OverRange):
4274 lower = r0
4275 else:
4276 try:
4277 lower = int(r0)
4278 except ValueError as err:
4279 raise exc.ArgumentError(
4280 "Integer or None expected for range value"
4281 ) from err
4282 else:
4283 if lower == 0:
4284 lower = RANGE_CURRENT
4286 if r1 is None:
4287 upper = RANGE_UNBOUNDED
4288 elif isinstance(r1, _OverRange):
4289 upper = r1
4290 else:
4291 try:
4292 upper = int(r1)
4293 except ValueError as err:
4294 raise exc.ArgumentError(
4295 "Integer or None expected for range value"
4296 ) from err
4297 else:
4298 if upper == 0:
4299 upper = RANGE_CURRENT
4301 return lower, upper
4303 if not TYPE_CHECKING:
4305 @util.memoized_property
4306 def type(self) -> TypeEngine[_T]: # noqa: A001
4307 return self.element.type
4309 @util.ro_non_memoized_property
4310 def _from_objects(self) -> List[FromClause]:
4311 return list(
4312 itertools.chain(
4313 *[
4314 c._from_objects
4315 for c in (self.element, self.partition_by, self.order_by)
4316 if c is not None
4317 ]
4318 )
4319 )
4322class WithinGroup(ColumnElement[_T]):
4323 """Represent a WITHIN GROUP (ORDER BY) clause.
4325 This is a special operator against so-called
4326 "ordered set aggregate" and "hypothetical
4327 set aggregate" functions, including ``percentile_cont()``,
4328 ``rank()``, ``dense_rank()``, etc.
4330 It's supported only by certain database backends, such as PostgreSQL,
4331 Oracle Database and MS SQL Server.
4333 The :class:`.WithinGroup` construct extracts its type from the
4334 method :meth:`.FunctionElement.within_group_type`. If this returns
4335 ``None``, the function's ``.type`` is used.
4337 """
4339 __visit_name__ = "withingroup"
4341 _traverse_internals: _TraverseInternalsType = [
4342 ("element", InternalTraversal.dp_clauseelement),
4343 ("order_by", InternalTraversal.dp_clauseelement),
4344 ]
4346 order_by: Optional[ClauseList] = None
4348 def __init__(
4349 self,
4350 element: Union[FunctionElement[_T], FunctionFilter[_T]],
4351 *order_by: _ColumnExpressionArgument[Any],
4352 ):
4353 self.element = element
4354 if order_by is not None:
4355 self.order_by = ClauseList(
4356 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4357 )
4359 def __reduce__(self):
4360 return self.__class__, (self.element,) + (
4361 tuple(self.order_by) if self.order_by is not None else ()
4362 )
4364 def over(
4365 self,
4366 *,
4367 partition_by: Optional[_ByArgument] = None,
4368 order_by: Optional[_ByArgument] = None,
4369 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4370 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4371 groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4372 ) -> Over[_T]:
4373 """Produce an OVER clause against this :class:`.WithinGroup`
4374 construct.
4376 This function has the same signature as that of
4377 :meth:`.FunctionElement.over`.
4379 """
4380 return Over(
4381 self,
4382 partition_by=partition_by,
4383 order_by=order_by,
4384 range_=range_,
4385 rows=rows,
4386 groups=groups,
4387 )
4389 @overload
4390 def filter(self) -> Self: ...
4392 @overload
4393 def filter(
4394 self,
4395 __criterion0: _ColumnExpressionArgument[bool],
4396 *criterion: _ColumnExpressionArgument[bool],
4397 ) -> FunctionFilter[_T]: ...
4399 def filter(
4400 self, *criterion: _ColumnExpressionArgument[bool]
4401 ) -> Union[Self, FunctionFilter[_T]]:
4402 """Produce a FILTER clause against this function."""
4403 if not criterion:
4404 return self
4405 return FunctionFilter(self, *criterion)
4407 if not TYPE_CHECKING:
4409 @util.memoized_property
4410 def type(self) -> TypeEngine[_T]: # noqa: A001
4411 wgt = self.element.within_group_type(self)
4412 if wgt is not None:
4413 return wgt
4414 else:
4415 return self.element.type
4417 @util.ro_non_memoized_property
4418 def _from_objects(self) -> List[FromClause]:
4419 return list(
4420 itertools.chain(
4421 *[
4422 c._from_objects
4423 for c in (self.element, self.order_by)
4424 if c is not None
4425 ]
4426 )
4427 )
4430class FunctionFilter(Generative, ColumnElement[_T]):
4431 """Represent a function FILTER clause.
4433 This is a special operator against aggregate and window functions,
4434 which controls which rows are passed to it.
4435 It's supported only by certain database backends.
4437 Invocation of :class:`.FunctionFilter` is via
4438 :meth:`.FunctionElement.filter`::
4440 func.count(1).filter(True)
4442 .. seealso::
4444 :meth:`.FunctionElement.filter`
4446 """
4448 __visit_name__ = "funcfilter"
4450 _traverse_internals: _TraverseInternalsType = [
4451 ("func", InternalTraversal.dp_clauseelement),
4452 ("criterion", InternalTraversal.dp_clauseelement),
4453 ]
4455 criterion: Optional[ColumnElement[bool]] = None
4457 def __init__(
4458 self,
4459 func: Union[FunctionElement[_T], WithinGroup[_T]],
4460 *criterion: _ColumnExpressionArgument[bool],
4461 ):
4462 self.func = func
4463 self.filter.non_generative(self, *criterion) # type: ignore
4465 @_generative
4466 def filter(self, *criterion: _ColumnExpressionArgument[bool]) -> Self:
4467 """Produce an additional FILTER against the function.
4469 This method adds additional criteria to the initial criteria
4470 set up by :meth:`.FunctionElement.filter`.
4472 Multiple criteria are joined together at SQL render time
4473 via ``AND``.
4476 """
4478 for crit in list(criterion):
4479 crit = coercions.expect(roles.WhereHavingRole, crit)
4481 if self.criterion is not None:
4482 self.criterion = self.criterion & crit
4483 else:
4484 self.criterion = crit
4486 return self
4488 def over(
4489 self,
4490 partition_by: Optional[
4491 Union[
4492 Iterable[_ColumnExpressionArgument[Any]],
4493 _ColumnExpressionArgument[Any],
4494 ]
4495 ] = None,
4496 order_by: Optional[
4497 Union[
4498 Iterable[_ColumnExpressionArgument[Any]],
4499 _ColumnExpressionArgument[Any],
4500 ]
4501 ] = None,
4502 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4503 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4504 groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4505 ) -> Over[_T]:
4506 """Produce an OVER clause against this filtered function.
4508 Used against aggregate or so-called "window" functions,
4509 for database backends that support window functions.
4511 The expression::
4513 func.rank().filter(MyClass.y > 5).over(order_by="x")
4515 is shorthand for::
4517 from sqlalchemy import over, funcfilter
4519 over(funcfilter(func.rank(), MyClass.y > 5), order_by="x")
4521 See :func:`_expression.over` for a full description.
4523 """
4524 return Over(
4525 self,
4526 partition_by=partition_by,
4527 order_by=order_by,
4528 range_=range_,
4529 rows=rows,
4530 groups=groups,
4531 )
4533 def within_group(
4534 self, *order_by: _ColumnExpressionArgument[Any]
4535 ) -> WithinGroup[_T]:
4536 """Produce a WITHIN GROUP (ORDER BY expr) clause against
4537 this function.
4538 """
4539 return WithinGroup(self, *order_by)
4541 def within_group_type(
4542 self, within_group: WithinGroup[_T]
4543 ) -> Optional[TypeEngine[_T]]:
4544 return None
4546 def self_group(
4547 self, against: Optional[OperatorType] = None
4548 ) -> Union[Self, Grouping[_T]]:
4549 if operators.is_precedent(operators.filter_op, against):
4550 return Grouping(self)
4551 else:
4552 return self
4554 if not TYPE_CHECKING:
4556 @util.memoized_property
4557 def type(self) -> TypeEngine[_T]: # noqa: A001
4558 return self.func.type
4560 @util.ro_non_memoized_property
4561 def _from_objects(self) -> List[FromClause]:
4562 return list(
4563 itertools.chain(
4564 *[
4565 c._from_objects
4566 for c in (self.func, self.criterion)
4567 if c is not None
4568 ]
4569 )
4570 )
4573class NamedColumn(KeyedColumnElement[_T]):
4574 is_literal = False
4575 table: Optional[FromClause] = None
4576 name: str
4577 key: str
4579 def _compare_name_for_result(self, other):
4580 return (hasattr(other, "name") and self.name == other.name) or (
4581 hasattr(other, "_label") and self._label == other._label
4582 )
4584 @util.ro_memoized_property
4585 def description(self) -> str:
4586 return self.name
4588 @HasMemoized.memoized_attribute
4589 def _tq_key_label(self) -> Optional[str]:
4590 """table qualified label based on column key.
4592 for table-bound columns this is <tablename>_<column key/proxy key>;
4594 all other expressions it resolves to key/proxy key.
4596 """
4597 proxy_key = self._proxy_key
4598 if proxy_key and proxy_key != self.name:
4599 return self._gen_tq_label(proxy_key)
4600 else:
4601 return self._tq_label
4603 @HasMemoized.memoized_attribute
4604 def _tq_label(self) -> Optional[str]:
4605 """table qualified label based on column name.
4607 for table-bound columns this is <tablename>_<columnname>; all other
4608 expressions it resolves to .name.
4610 """
4611 return self._gen_tq_label(self.name)
4613 @HasMemoized.memoized_attribute
4614 def _render_label_in_columns_clause(self):
4615 return True
4617 @HasMemoized.memoized_attribute
4618 def _non_anon_label(self):
4619 return self.name
4621 def _gen_tq_label(
4622 self, name: str, dedupe_on_key: bool = True
4623 ) -> Optional[str]:
4624 return name
4626 def _bind_param(
4627 self,
4628 operator: OperatorType,
4629 obj: Any,
4630 type_: Optional[TypeEngine[_T]] = None,
4631 expanding: bool = False,
4632 ) -> BindParameter[_T]:
4633 return BindParameter(
4634 self.key,
4635 obj,
4636 _compared_to_operator=operator,
4637 _compared_to_type=self.type,
4638 type_=type_,
4639 unique=True,
4640 expanding=expanding,
4641 )
4643 def _make_proxy(
4644 self,
4645 selectable: FromClause,
4646 *,
4647 primary_key: ColumnSet,
4648 foreign_keys: Set[KeyedColumnElement[Any]],
4649 name: Optional[str] = None,
4650 key: Optional[str] = None,
4651 name_is_truncatable: bool = False,
4652 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
4653 disallow_is_literal: bool = False,
4654 **kw: Any,
4655 ) -> typing_Tuple[str, ColumnClause[_T]]:
4656 c = ColumnClause(
4657 (
4658 coercions.expect(roles.TruncatedLabelRole, name or self.name)
4659 if name_is_truncatable
4660 else (name or self.name)
4661 ),
4662 type_=self.type,
4663 _selectable=selectable,
4664 is_literal=False,
4665 )
4667 c._propagate_attrs = selectable._propagate_attrs
4668 if name is None:
4669 c.key = self.key
4670 if compound_select_cols:
4671 c._proxies = list(compound_select_cols)
4672 else:
4673 c._proxies = [self]
4675 if selectable._is_clone_of is not None:
4676 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
4677 return c.key, c
4680_PS = ParamSpec("_PS")
4683class Label(roles.LabeledColumnExprRole[_T], NamedColumn[_T]):
4684 """Represents a column label (AS).
4686 Represent a label, as typically applied to any column-level
4687 element using the ``AS`` sql keyword.
4689 """
4691 __visit_name__ = "label"
4693 _traverse_internals: _TraverseInternalsType = [
4694 ("name", InternalTraversal.dp_anon_name),
4695 ("type", InternalTraversal.dp_type),
4696 ("_element", InternalTraversal.dp_clauseelement),
4697 ]
4699 _cache_key_traversal = [
4700 ("name", InternalTraversal.dp_anon_name),
4701 ("_element", InternalTraversal.dp_clauseelement),
4702 ]
4704 _element: ColumnElement[_T]
4705 name: str
4707 def __init__(
4708 self,
4709 name: Optional[str],
4710 element: _ColumnExpressionArgument[_T],
4711 type_: Optional[_TypeEngineArgument[_T]] = None,
4712 ):
4713 orig_element = element
4714 element = coercions.expect(
4715 roles.ExpressionElementRole,
4716 element,
4717 apply_propagate_attrs=self,
4718 )
4719 while isinstance(element, Label):
4720 # TODO: this is only covered in test_text.py, but nothing
4721 # fails if it's removed. determine rationale
4722 element = element.element
4724 if name:
4725 self.name = name
4726 else:
4727 self.name = _anonymous_label.safe_construct(
4728 id(self), getattr(element, "name", "anon")
4729 )
4730 if isinstance(orig_element, Label):
4731 # TODO: no coverage for this block, again would be in
4732 # test_text.py where the resolve_label concept is important
4733 self._resolve_label = orig_element._label
4735 self.key = self._tq_label = self._tq_key_label = self.name
4736 self._element = element
4738 self.type = (
4739 type_api.to_instance(type_)
4740 if type_ is not None
4741 else self._element.type
4742 )
4744 self._proxies = [element]
4746 def __reduce__(self):
4747 return self.__class__, (self.name, self._element, self.type)
4749 @HasMemoized.memoized_attribute
4750 def _render_label_in_columns_clause(self):
4751 return True
4753 def _bind_param(self, operator, obj, type_=None, expanding=False):
4754 return BindParameter(
4755 None,
4756 obj,
4757 _compared_to_operator=operator,
4758 type_=type_,
4759 _compared_to_type=self.type,
4760 unique=True,
4761 expanding=expanding,
4762 )
4764 @util.memoized_property
4765 def _is_implicitly_boolean(self):
4766 return self.element._is_implicitly_boolean
4768 @HasMemoized.memoized_attribute
4769 def _allow_label_resolve(self):
4770 return self.element._allow_label_resolve
4772 @property
4773 def _order_by_label_element(self):
4774 return self
4776 @HasMemoized.memoized_attribute
4777 def element(self) -> ColumnElement[_T]:
4778 return self._element.self_group(against=operators.as_)
4780 def self_group(self, against: Optional[OperatorType] = None) -> Label[_T]:
4781 return self._apply_to_inner(self._element.self_group, against=against)
4783 def _negate(self):
4784 return self._apply_to_inner(self._element._negate)
4786 def _apply_to_inner(
4787 self,
4788 fn: Callable[_PS, ColumnElement[_T]],
4789 *arg: _PS.args,
4790 **kw: _PS.kwargs,
4791 ) -> Label[_T]:
4792 sub_element = fn(*arg, **kw)
4793 if sub_element is not self._element:
4794 return Label(self.name, sub_element, type_=self.type)
4795 else:
4796 return self
4798 @property
4799 def primary_key(self): # type: ignore[override]
4800 return self.element.primary_key
4802 @property
4803 def foreign_keys(self): # type: ignore[override]
4804 return self.element.foreign_keys
4806 def _copy_internals(
4807 self,
4808 *,
4809 clone: _CloneCallableType = _clone,
4810 anonymize_labels: bool = False,
4811 **kw: Any,
4812 ) -> None:
4813 self._reset_memoizations()
4814 self._element = clone(self._element, **kw)
4815 if anonymize_labels:
4816 self.name = _anonymous_label.safe_construct(
4817 id(self), getattr(self.element, "name", "anon")
4818 )
4819 self.key = self._tq_label = self._tq_key_label = self.name
4821 @util.ro_non_memoized_property
4822 def _from_objects(self) -> List[FromClause]:
4823 return self.element._from_objects
4825 def _make_proxy(
4826 self,
4827 selectable: FromClause,
4828 *,
4829 primary_key: ColumnSet,
4830 foreign_keys: Set[KeyedColumnElement[Any]],
4831 name: Optional[str] = None,
4832 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
4833 **kw: Any,
4834 ) -> typing_Tuple[str, ColumnClause[_T]]:
4835 name = self.name if not name else name
4837 key, e = self.element._make_proxy(
4838 selectable,
4839 name=name,
4840 disallow_is_literal=True,
4841 name_is_truncatable=isinstance(name, _truncated_label),
4842 compound_select_cols=compound_select_cols,
4843 primary_key=primary_key,
4844 foreign_keys=foreign_keys,
4845 )
4847 # there was a note here to remove this assertion, which was here
4848 # to determine if we later could support a use case where
4849 # the key and name of a label are separate. But I don't know what
4850 # that case was. For now, this is an unexpected case that occurs
4851 # when a label name conflicts with other columns and select()
4852 # is attempting to disambiguate an explicit label, which is not what
4853 # the user would want. See issue #6090.
4854 if key != self.name and not isinstance(self.name, _anonymous_label):
4855 raise exc.InvalidRequestError(
4856 "Label name %s is being renamed to an anonymous label due "
4857 "to disambiguation "
4858 "which is not supported right now. Please use unique names "
4859 "for explicit labels." % (self.name)
4860 )
4862 e._propagate_attrs = selectable._propagate_attrs
4863 e._proxies.append(self)
4864 if self.type is not None:
4865 e.type = self.type
4867 return self.key, e
4870class ColumnClause(
4871 roles.DDLReferredColumnRole,
4872 roles.LabeledColumnExprRole[_T],
4873 roles.StrAsPlainColumnRole,
4874 Immutable,
4875 NamedColumn[_T],
4876):
4877 """Represents a column expression from any textual string.
4879 The :class:`.ColumnClause`, a lightweight analogue to the
4880 :class:`_schema.Column` class, is typically invoked using the
4881 :func:`_expression.column` function, as in::
4883 from sqlalchemy import column
4885 id, name = column("id"), column("name")
4886 stmt = select(id, name).select_from("user")
4888 The above statement would produce SQL like:
4890 .. sourcecode:: sql
4892 SELECT id, name FROM user
4894 :class:`.ColumnClause` is the immediate superclass of the schema-specific
4895 :class:`_schema.Column` object. While the :class:`_schema.Column`
4896 class has all the
4897 same capabilities as :class:`.ColumnClause`, the :class:`.ColumnClause`
4898 class is usable by itself in those cases where behavioral requirements
4899 are limited to simple SQL expression generation. The object has none of
4900 the associations with schema-level metadata or with execution-time
4901 behavior that :class:`_schema.Column` does,
4902 so in that sense is a "lightweight"
4903 version of :class:`_schema.Column`.
4905 Full details on :class:`.ColumnClause` usage is at
4906 :func:`_expression.column`.
4908 .. seealso::
4910 :func:`_expression.column`
4912 :class:`_schema.Column`
4914 """
4916 table: Optional[FromClause]
4917 is_literal: bool
4919 __visit_name__ = "column"
4921 _traverse_internals: _TraverseInternalsType = [
4922 ("name", InternalTraversal.dp_anon_name),
4923 ("type", InternalTraversal.dp_type),
4924 ("table", InternalTraversal.dp_clauseelement),
4925 ("is_literal", InternalTraversal.dp_boolean),
4926 ]
4928 onupdate: Optional[DefaultGenerator] = None
4929 default: Optional[DefaultGenerator] = None
4930 server_default: Optional[FetchedValue] = None
4931 server_onupdate: Optional[FetchedValue] = None
4933 _is_multiparam_column = False
4935 @property
4936 def _is_star(self): # type: ignore[override]
4937 return self.is_literal and self.name == "*"
4939 def __init__(
4940 self,
4941 text: str,
4942 type_: Optional[_TypeEngineArgument[_T]] = None,
4943 is_literal: bool = False,
4944 _selectable: Optional[FromClause] = None,
4945 ):
4946 self.key = self.name = text
4947 self.table = _selectable
4949 # if type is None, we get NULLTYPE, which is our _T. But I don't
4950 # know how to get the overloads to express that correctly
4951 self.type = type_api.to_instance(type_) # type: ignore
4953 self.is_literal = is_literal
4955 def get_children(self, *, column_tables=False, **kw):
4956 # override base get_children() to not return the Table
4957 # or selectable that is parent to this column. Traversals
4958 # expect the columns of tables and subqueries to be leaf nodes.
4959 return []
4961 @property
4962 def entity_namespace(self):
4963 if self.table is not None:
4964 return self.table.entity_namespace
4965 else:
4966 return super().entity_namespace
4968 def _clone(self, detect_subquery_cols=False, **kw):
4969 if (
4970 detect_subquery_cols
4971 and self.table is not None
4972 and self.table._is_subquery
4973 ):
4974 clone = kw.pop("clone")
4975 table = clone(self.table, **kw)
4976 new = table.c.corresponding_column(self)
4977 return new
4979 return super()._clone(**kw)
4981 @HasMemoized_ro_memoized_attribute
4982 def _from_objects(self) -> List[FromClause]:
4983 t = self.table
4984 if t is not None:
4985 return [t]
4986 else:
4987 return []
4989 @HasMemoized.memoized_attribute
4990 def _render_label_in_columns_clause(self):
4991 return self.table is not None
4993 @property
4994 def _ddl_label(self):
4995 return self._gen_tq_label(self.name, dedupe_on_key=False)
4997 def _compare_name_for_result(self, other):
4998 if (
4999 self.is_literal
5000 or self.table is None
5001 or self.table._is_textual
5002 or not hasattr(other, "proxy_set")
5003 or (
5004 isinstance(other, ColumnClause)
5005 and (
5006 other.is_literal
5007 or other.table is None
5008 or other.table._is_textual
5009 )
5010 )
5011 ):
5012 return (hasattr(other, "name") and self.name == other.name) or (
5013 hasattr(other, "_tq_label")
5014 and self._tq_label == other._tq_label
5015 )
5016 else:
5017 return other.proxy_set.intersection(self.proxy_set)
5019 def _gen_tq_label(
5020 self, name: str, dedupe_on_key: bool = True
5021 ) -> Optional[str]:
5022 """generate table-qualified label
5024 for a table-bound column this is <tablename>_<columnname>.
5026 used primarily for LABEL_STYLE_TABLENAME_PLUS_COL
5027 as well as the .columns collection on a Join object.
5029 """
5030 label: str
5031 t = self.table
5032 if self.is_literal:
5033 return None
5034 elif t is not None and is_named_from_clause(t):
5035 if has_schema_attr(t) and t.schema:
5036 label = t.schema.replace(".", "_") + "_" + t.name + "_" + name
5037 else:
5038 assert not TYPE_CHECKING or isinstance(t, NamedFromClause)
5039 label = t.name + "_" + name
5041 # propagate name quoting rules for labels.
5042 if is_quoted_name(name) and name.quote is not None:
5043 if is_quoted_name(label):
5044 label.quote = name.quote
5045 else:
5046 label = quoted_name(label, name.quote)
5047 elif is_quoted_name(t.name) and t.name.quote is not None:
5048 # can't get this situation to occur, so let's
5049 # assert false on it for now
5050 assert not isinstance(label, quoted_name)
5051 label = quoted_name(label, t.name.quote)
5053 if dedupe_on_key:
5054 # ensure the label name doesn't conflict with that of an
5055 # existing column. note that this implies that any Column
5056 # must **not** set up its _label before its parent table has
5057 # all of its other Column objects set up. There are several
5058 # tables in the test suite which will fail otherwise; example:
5059 # table "owner" has columns "name" and "owner_name". Therefore
5060 # column owner.name cannot use the label "owner_name", it has
5061 # to be "owner_name_1".
5062 if label in t.c:
5063 _label = label
5064 counter = 1
5065 while _label in t.c:
5066 _label = label + "_" + str(counter)
5067 counter += 1
5068 label = _label
5070 return coercions.expect(roles.TruncatedLabelRole, label)
5072 else:
5073 return name
5075 def _make_proxy(
5076 self,
5077 selectable: FromClause,
5078 *,
5079 primary_key: ColumnSet,
5080 foreign_keys: Set[KeyedColumnElement[Any]],
5081 name: Optional[str] = None,
5082 key: Optional[str] = None,
5083 name_is_truncatable: bool = False,
5084 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5085 disallow_is_literal: bool = False,
5086 **kw: Any,
5087 ) -> typing_Tuple[str, ColumnClause[_T]]:
5088 # the "is_literal" flag normally should never be propagated; a proxied
5089 # column is always a SQL identifier and never the actual expression
5090 # being evaluated. however, there is a case where the "is_literal" flag
5091 # might be used to allow the given identifier to have a fixed quoting
5092 # pattern already, so maintain the flag for the proxy unless a
5093 # :class:`.Label` object is creating the proxy. See [ticket:4730].
5094 is_literal = (
5095 not disallow_is_literal
5096 and self.is_literal
5097 and (
5098 # note this does not accommodate for quoted_name differences
5099 # right now
5100 name is None
5101 or name == self.name
5102 )
5103 )
5104 c = self._constructor(
5105 (
5106 coercions.expect(roles.TruncatedLabelRole, name or self.name)
5107 if name_is_truncatable
5108 else (name or self.name)
5109 ),
5110 type_=self.type,
5111 _selectable=selectable,
5112 is_literal=is_literal,
5113 )
5114 c._propagate_attrs = selectable._propagate_attrs
5115 if name is None:
5116 c.key = self.key
5117 if compound_select_cols:
5118 c._proxies = list(compound_select_cols)
5119 else:
5120 c._proxies = [self]
5122 if selectable._is_clone_of is not None:
5123 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
5124 return c.key, c
5127class TableValuedColumn(NamedColumn[_T]):
5128 __visit_name__ = "table_valued_column"
5130 _traverse_internals: _TraverseInternalsType = [
5131 ("name", InternalTraversal.dp_anon_name),
5132 ("type", InternalTraversal.dp_type),
5133 ("scalar_alias", InternalTraversal.dp_clauseelement),
5134 ]
5136 def __init__(self, scalar_alias: NamedFromClause, type_: TypeEngine[_T]):
5137 self.scalar_alias = scalar_alias
5138 self.key = self.name = scalar_alias.name
5139 self.type = type_
5141 def _copy_internals(
5142 self, clone: _CloneCallableType = _clone, **kw: Any
5143 ) -> None:
5144 self.scalar_alias = clone(self.scalar_alias, **kw)
5145 self.key = self.name = self.scalar_alias.name
5147 @util.ro_non_memoized_property
5148 def _from_objects(self) -> List[FromClause]:
5149 return [self.scalar_alias]
5152class CollationClause(ColumnElement[str]):
5153 __visit_name__ = "collation"
5155 _traverse_internals: _TraverseInternalsType = [
5156 ("collation", InternalTraversal.dp_string)
5157 ]
5159 @classmethod
5160 @util.preload_module("sqlalchemy.sql.sqltypes")
5161 def _create_collation_expression(
5162 cls, expression: _ColumnExpressionArgument[str], collation: str
5163 ) -> BinaryExpression[str]:
5165 sqltypes = util.preloaded.sql_sqltypes
5167 expr = coercions.expect(roles.ExpressionElementRole[str], expression)
5169 if expr.type._type_affinity is sqltypes.String:
5170 collate_type = expr.type._with_collation(collation)
5171 else:
5172 collate_type = expr.type
5174 return BinaryExpression(
5175 expr,
5176 CollationClause(collation),
5177 operators.collate,
5178 type_=collate_type,
5179 )
5181 def __init__(self, collation):
5182 self.collation = collation
5185class _IdentifiedClause(Executable, ClauseElement):
5186 __visit_name__ = "identified"
5188 def __init__(self, ident):
5189 self.ident = ident
5192class SavepointClause(_IdentifiedClause):
5193 __visit_name__ = "savepoint"
5194 inherit_cache = False
5197class RollbackToSavepointClause(_IdentifiedClause):
5198 __visit_name__ = "rollback_to_savepoint"
5199 inherit_cache = False
5202class ReleaseSavepointClause(_IdentifiedClause):
5203 __visit_name__ = "release_savepoint"
5204 inherit_cache = False
5207class quoted_name(util.MemoizedSlots, str):
5208 """Represent a SQL identifier combined with quoting preferences.
5210 :class:`.quoted_name` is a Python unicode/str subclass which
5211 represents a particular identifier name along with a
5212 ``quote`` flag. This ``quote`` flag, when set to
5213 ``True`` or ``False``, overrides automatic quoting behavior
5214 for this identifier in order to either unconditionally quote
5215 or to not quote the name. If left at its default of ``None``,
5216 quoting behavior is applied to the identifier on a per-backend basis
5217 based on an examination of the token itself.
5219 A :class:`.quoted_name` object with ``quote=True`` is also
5220 prevented from being modified in the case of a so-called
5221 "name normalize" option. Certain database backends, such as
5222 Oracle Database, Firebird, and DB2 "normalize" case-insensitive names
5223 as uppercase. The SQLAlchemy dialects for these backends
5224 convert from SQLAlchemy's lower-case-means-insensitive convention
5225 to the upper-case-means-insensitive conventions of those backends.
5226 The ``quote=True`` flag here will prevent this conversion from occurring
5227 to support an identifier that's quoted as all lower case against
5228 such a backend.
5230 The :class:`.quoted_name` object is normally created automatically
5231 when specifying the name for key schema constructs such as
5232 :class:`_schema.Table`, :class:`_schema.Column`, and others.
5233 The class can also be
5234 passed explicitly as the name to any function that receives a name which
5235 can be quoted. Such as to use the :meth:`_engine.Engine.has_table`
5236 method with
5237 an unconditionally quoted name::
5239 from sqlalchemy import create_engine
5240 from sqlalchemy import inspect
5241 from sqlalchemy.sql import quoted_name
5243 engine = create_engine("oracle+oracledb://some_dsn")
5244 print(inspect(engine).has_table(quoted_name("some_table", True)))
5246 The above logic will run the "has table" logic against the Oracle Database
5247 backend, passing the name exactly as ``"some_table"`` without converting to
5248 upper case.
5250 .. versionchanged:: 1.2 The :class:`.quoted_name` construct is now
5251 importable from ``sqlalchemy.sql``, in addition to the previous
5252 location of ``sqlalchemy.sql.elements``.
5254 """
5256 __slots__ = "quote", "lower", "upper"
5258 quote: Optional[bool]
5260 @overload
5261 @classmethod
5262 def construct(cls, value: str, quote: Optional[bool]) -> quoted_name: ...
5264 @overload
5265 @classmethod
5266 def construct(cls, value: None, quote: Optional[bool]) -> None: ...
5268 @classmethod
5269 def construct(
5270 cls, value: Optional[str], quote: Optional[bool]
5271 ) -> Optional[quoted_name]:
5272 if value is None:
5273 return None
5274 else:
5275 return quoted_name(value, quote)
5277 def __new__(cls, value: str, quote: Optional[bool]) -> quoted_name:
5278 assert (
5279 value is not None
5280 ), "use quoted_name.construct() for None passthrough"
5281 if isinstance(value, cls) and (quote is None or value.quote == quote):
5282 return value
5283 self = super().__new__(cls, value)
5285 self.quote = quote
5286 return self
5288 def __reduce__(self):
5289 return quoted_name, (str(self), self.quote)
5291 def _memoized_method_lower(self):
5292 if self.quote:
5293 return self
5294 else:
5295 return str(self).lower()
5297 def _memoized_method_upper(self):
5298 if self.quote:
5299 return self
5300 else:
5301 return str(self).upper()
5304def _find_columns(clause: ClauseElement) -> Set[ColumnClause[Any]]:
5305 """locate Column objects within the given expression."""
5307 cols: Set[ColumnClause[Any]] = set()
5308 traverse(clause, {}, {"column": cols.add})
5309 return cols
5312def _type_from_args(args: Sequence[ColumnElement[_T]]) -> TypeEngine[_T]:
5313 for a in args:
5314 if not a.type._isnull:
5315 return a.type
5316 else:
5317 return type_api.NULLTYPE # type: ignore
5320def _corresponding_column_or_error(fromclause, column, require_embedded=False):
5321 c = fromclause.corresponding_column(
5322 column, require_embedded=require_embedded
5323 )
5324 if c is None:
5325 raise exc.InvalidRequestError(
5326 "Given column '%s', attached to table '%s', "
5327 "failed to locate a corresponding column from table '%s'"
5328 % (column, getattr(column, "table", None), fromclause.description)
5329 )
5330 return c
5333class _memoized_property_but_not_nulltype(
5334 util.memoized_property["TypeEngine[_T]"]
5335):
5336 """memoized property, but dont memoize NullType"""
5338 def __get__(self, obj, cls):
5339 if obj is None:
5340 return self
5341 result = self.fget(obj)
5342 if not result._isnull:
5343 obj.__dict__[self.__name__] = result
5344 return result
5347class AnnotatedColumnElement(Annotated):
5348 _Annotated__element: ColumnElement[Any]
5350 def __init__(self, element, values):
5351 Annotated.__init__(self, element, values)
5352 for attr in (
5353 "comparator",
5354 "_proxy_key",
5355 "_tq_key_label",
5356 "_tq_label",
5357 "_non_anon_label",
5358 "type",
5359 ):
5360 self.__dict__.pop(attr, None)
5361 for attr in ("name", "key", "table"):
5362 if self.__dict__.get(attr, False) is None:
5363 self.__dict__.pop(attr)
5365 def _with_annotations(self, values):
5366 clone = super()._with_annotations(values)
5367 for attr in (
5368 "comparator",
5369 "_proxy_key",
5370 "_tq_key_label",
5371 "_tq_label",
5372 "_non_anon_label",
5373 ):
5374 clone.__dict__.pop(attr, None)
5375 return clone
5377 @util.memoized_property
5378 def name(self):
5379 """pull 'name' from parent, if not present"""
5380 return self._Annotated__element.name
5382 @_memoized_property_but_not_nulltype
5383 def type(self):
5384 """pull 'type' from parent and don't cache if null.
5386 type is routinely changed on existing columns within the
5387 mapped_column() initialization process, and "type" is also consulted
5388 during the creation of SQL expressions. Therefore it can change after
5389 it was already retrieved. At the same time we don't want annotated
5390 objects having overhead when expressions are produced, so continue
5391 to memoize, but only when we have a non-null type.
5393 """
5394 return self._Annotated__element.type
5396 @util.memoized_property
5397 def table(self):
5398 """pull 'table' from parent, if not present"""
5399 return self._Annotated__element.table
5401 @util.memoized_property
5402 def key(self):
5403 """pull 'key' from parent, if not present"""
5404 return self._Annotated__element.key
5406 @util.memoized_property
5407 def info(self) -> _InfoType:
5408 if TYPE_CHECKING:
5409 assert isinstance(self._Annotated__element, Column)
5410 return self._Annotated__element.info
5412 @util.memoized_property
5413 def _anon_name_label(self) -> str:
5414 return self._Annotated__element._anon_name_label
5417class _truncated_label(quoted_name):
5418 """A unicode subclass used to identify symbolic "
5419 "names that may require truncation."""
5421 __slots__ = ()
5423 def __new__(cls, value: str, quote: Optional[bool] = None) -> Any:
5424 quote = getattr(value, "quote", quote)
5425 # return super(_truncated_label, cls).__new__(cls, value, quote, True)
5426 return super().__new__(cls, value, quote)
5428 def __reduce__(self) -> Any:
5429 return self.__class__, (str(self), self.quote)
5431 def apply_map(self, map_: Mapping[str, Any]) -> str:
5432 return self
5435class conv(_truncated_label):
5436 """Mark a string indicating that a name has already been converted
5437 by a naming convention.
5439 This is a string subclass that indicates a name that should not be
5440 subject to any further naming conventions.
5442 E.g. when we create a :class:`.Constraint` using a naming convention
5443 as follows::
5445 m = MetaData(
5446 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5447 )
5448 t = Table(
5449 "t", m, Column("x", Integer), CheckConstraint("x > 5", name="x5")
5450 )
5452 The name of the above constraint will be rendered as ``"ck_t_x5"``.
5453 That is, the existing name ``x5`` is used in the naming convention as the
5454 ``constraint_name`` token.
5456 In some situations, such as in migration scripts, we may be rendering
5457 the above :class:`.CheckConstraint` with a name that's already been
5458 converted. In order to make sure the name isn't double-modified, the
5459 new name is applied using the :func:`_schema.conv` marker. We can
5460 use this explicitly as follows::
5463 m = MetaData(
5464 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5465 )
5466 t = Table(
5467 "t",
5468 m,
5469 Column("x", Integer),
5470 CheckConstraint("x > 5", name=conv("ck_t_x5")),
5471 )
5473 Where above, the :func:`_schema.conv` marker indicates that the constraint
5474 name here is final, and the name will render as ``"ck_t_x5"`` and not
5475 ``"ck_t_ck_t_x5"``
5477 .. seealso::
5479 :ref:`constraint_naming_conventions`
5481 """
5483 __slots__ = ()
5486# for backwards compatibility in case
5487# someone is re-implementing the
5488# _truncated_identifier() sequence in a custom
5489# compiler
5490_generated_label = _truncated_label
5493class _anonymous_label(_truncated_label):
5494 """A unicode subclass used to identify anonymously
5495 generated names."""
5497 __slots__ = ()
5499 @classmethod
5500 def safe_construct(
5501 cls,
5502 seed: int,
5503 body: str,
5504 enclosing_label: Optional[str] = None,
5505 sanitize_key: bool = False,
5506 ) -> _anonymous_label:
5507 # need to escape chars that interfere with format
5508 # strings in any case, issue #8724
5509 body = re.sub(r"[%\(\) \$]+", "_", body)
5511 if sanitize_key:
5512 # sanitize_key is then an extra step used by BindParameter
5513 body = body.strip("_")
5515 label = "%%(%d %s)s" % (seed, body.replace("%", "%%"))
5516 if enclosing_label:
5517 label = "%s%s" % (enclosing_label, label)
5519 return _anonymous_label(label)
5521 def __add__(self, other):
5522 if "%" in other and not isinstance(other, _anonymous_label):
5523 other = str(other).replace("%", "%%")
5524 else:
5525 other = str(other)
5527 return _anonymous_label(
5528 quoted_name(
5529 str.__add__(self, other),
5530 self.quote,
5531 )
5532 )
5534 def __radd__(self, other):
5535 if "%" in other and not isinstance(other, _anonymous_label):
5536 other = str(other).replace("%", "%%")
5537 else:
5538 other = str(other)
5540 return _anonymous_label(
5541 quoted_name(
5542 str.__add__(other, self),
5543 self.quote,
5544 )
5545 )
5547 def apply_map(self, map_):
5548 if self.quote is not None:
5549 # preserve quoting only if necessary
5550 return quoted_name(self % map_, self.quote)
5551 else:
5552 # else skip the constructor call
5553 return self % map_