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-2024 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
82from ..util.typing import TupleAny
83from ..util.typing import Unpack
85if typing.TYPE_CHECKING:
86 from ._typing import _ByArgument
87 from ._typing import _ColumnExpressionArgument
88 from ._typing import _ColumnExpressionOrStrLabelArgument
89 from ._typing import _HasDialect
90 from ._typing import _InfoType
91 from ._typing import _PropagateAttrsType
92 from ._typing import _TypeEngineArgument
93 from .cache_key import _CacheKeyTraversalType
94 from .cache_key import CacheKey
95 from .compiler import Compiled
96 from .compiler import SQLCompiler
97 from .functions import FunctionElement
98 from .operators import OperatorType
99 from .schema import Column
100 from .schema import DefaultGenerator
101 from .schema import FetchedValue
102 from .schema import ForeignKey
103 from .selectable import _SelectIterable
104 from .selectable import FromClause
105 from .selectable import NamedFromClause
106 from .selectable import TextualSelect
107 from .sqltypes import TupleType
108 from .type_api import TypeEngine
109 from .visitors import _CloneCallableType
110 from .visitors import _TraverseInternalsType
111 from .visitors import anon_map
112 from ..engine import Connection
113 from ..engine import Dialect
114 from ..engine.interfaces import _CoreMultiExecuteParams
115 from ..engine.interfaces import CacheStats
116 from ..engine.interfaces import CompiledCacheType
117 from ..engine.interfaces import CoreExecuteOptionsParameter
118 from ..engine.interfaces import SchemaTranslateMapType
119 from ..engine.result import Result
121_NUMERIC = Union[float, Decimal]
122_NUMBER = Union[float, int, Decimal]
124_T = TypeVar("_T", bound="Any")
125_T_co = TypeVar("_T_co", bound=Any, covariant=True)
126_OPT = TypeVar("_OPT", bound="Any")
127_NT = TypeVar("_NT", bound="_NUMERIC")
129_NMT = TypeVar("_NMT", bound="_NUMBER")
132@overload
133def literal(
134 value: Any,
135 type_: _TypeEngineArgument[_T],
136 literal_execute: bool = False,
137) -> BindParameter[_T]: ...
140@overload
141def literal(
142 value: _T,
143 type_: None = None,
144 literal_execute: bool = False,
145) -> BindParameter[_T]: ...
148@overload
149def literal(
150 value: Any,
151 type_: Optional[_TypeEngineArgument[Any]] = None,
152 literal_execute: bool = False,
153) -> BindParameter[Any]: ...
156def literal(
157 value: Any,
158 type_: Optional[_TypeEngineArgument[Any]] = None,
159 literal_execute: bool = False,
160) -> BindParameter[Any]:
161 r"""Return a literal clause, bound to a bind parameter.
163 Literal clauses are created automatically when non-
164 :class:`_expression.ClauseElement` objects (such as strings, ints, dates,
165 etc.) are
166 used in a comparison operation with a :class:`_expression.ColumnElement`
167 subclass,
168 such as a :class:`~sqlalchemy.schema.Column` object. Use this function
169 to force the generation of a literal clause, which will be created as a
170 :class:`BindParameter` with a bound value.
172 :param value: the value to be bound. Can be any Python object supported by
173 the underlying DB-API, or is translatable via the given type argument.
175 :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine` which will
176 provide bind-parameter translation for this literal.
178 :param literal_execute: optional bool, when True, the SQL engine will
179 attempt to render the bound value directly in the SQL statement at
180 execution time rather than providing as a parameter value.
182 .. versionadded:: 2.0
184 """
185 return coercions.expect(
186 roles.LiteralValueRole,
187 value,
188 type_=type_,
189 literal_execute=literal_execute,
190 )
193def literal_column(
194 text: str, type_: Optional[_TypeEngineArgument[_T]] = None
195) -> ColumnClause[_T]:
196 r"""Produce a :class:`.ColumnClause` object that has the
197 :paramref:`_expression.column.is_literal` flag set to True.
199 :func:`_expression.literal_column` is similar to
200 :func:`_expression.column`, except that
201 it is more often used as a "standalone" column expression that renders
202 exactly as stated; while :func:`_expression.column`
203 stores a string name that
204 will be assumed to be part of a table and may be quoted as such,
205 :func:`_expression.literal_column` can be that,
206 or any other arbitrary column-oriented
207 expression.
209 :param text: the text of the expression; can be any SQL expression.
210 Quoting rules will not be applied. To specify a column-name expression
211 which should be subject to quoting rules, use the :func:`column`
212 function.
214 :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine`
215 object which will
216 provide result-set translation and additional expression semantics for
217 this column. If left as ``None`` the type will be :class:`.NullType`.
219 .. seealso::
221 :func:`_expression.column`
223 :func:`_expression.text`
225 :ref:`tutorial_select_arbitrary_text`
227 """
228 return ColumnClause(text, type_=type_, is_literal=True)
231class CompilerElement(Visitable):
232 """base class for SQL elements that can be compiled to produce a
233 SQL string.
235 .. versionadded:: 2.0
237 """
239 __slots__ = ()
240 __visit_name__ = "compiler_element"
242 supports_execution = False
244 stringify_dialect = "default"
246 @util.preload_module("sqlalchemy.engine.default")
247 @util.preload_module("sqlalchemy.engine.url")
248 def compile(
249 self,
250 bind: Optional[_HasDialect] = None,
251 dialect: Optional[Dialect] = None,
252 **kw: Any,
253 ) -> Compiled:
254 """Compile this SQL expression.
256 The return value is a :class:`~.Compiled` object.
257 Calling ``str()`` or ``unicode()`` on the returned value will yield a
258 string representation of the result. The
259 :class:`~.Compiled` object also can return a
260 dictionary of bind parameter names and values
261 using the ``params`` accessor.
263 :param bind: An :class:`.Connection` or :class:`.Engine` which
264 can provide a :class:`.Dialect` in order to generate a
265 :class:`.Compiled` object. If the ``bind`` and
266 ``dialect`` parameters are both omitted, a default SQL compiler
267 is used.
269 :param column_keys: Used for INSERT and UPDATE statements, a list of
270 column names which should be present in the VALUES clause of the
271 compiled statement. If ``None``, all columns from the target table
272 object are rendered.
274 :param dialect: A :class:`.Dialect` instance which can generate
275 a :class:`.Compiled` object. This argument takes precedence over
276 the ``bind`` argument.
278 :param compile_kwargs: optional dictionary of additional parameters
279 that will be passed through to the compiler within all "visit"
280 methods. This allows any custom flag to be passed through to
281 a custom compilation construct, for example. It is also used
282 for the case of passing the ``literal_binds`` flag through::
284 from sqlalchemy.sql import table, column, select
286 t = table('t', column('x'))
288 s = select(t).where(t.c.x == 5)
290 print(s.compile(compile_kwargs={"literal_binds": True}))
292 .. seealso::
294 :ref:`faq_sql_expression_string`
296 """
298 if dialect is None:
299 if bind:
300 dialect = bind.dialect
301 elif self.stringify_dialect == "default":
302 default = util.preloaded.engine_default
303 dialect = default.StrCompileDialect()
304 else:
305 url = util.preloaded.engine_url
306 dialect = url.URL.create(
307 self.stringify_dialect
308 ).get_dialect()()
310 return self._compiler(dialect, **kw)
312 def _compiler(self, dialect: Dialect, **kw: Any) -> Compiled:
313 """Return a compiler appropriate for this ClauseElement, given a
314 Dialect."""
316 if TYPE_CHECKING:
317 assert isinstance(self, ClauseElement)
318 return dialect.statement_compiler(dialect, self, **kw)
320 def __str__(self) -> str:
321 return str(self.compile())
324@inspection._self_inspects
325class ClauseElement(
326 SupportsWrappingAnnotations,
327 MemoizedHasCacheKey,
328 HasCopyInternals,
329 ExternallyTraversible,
330 CompilerElement,
331):
332 """Base class for elements of a programmatically constructed SQL
333 expression.
335 """
337 __visit_name__ = "clause"
339 if TYPE_CHECKING:
341 @util.memoized_property
342 def _propagate_attrs(self) -> _PropagateAttrsType:
343 """like annotations, however these propagate outwards liberally
344 as SQL constructs are built, and are set up at construction time.
346 """
347 ...
349 else:
350 _propagate_attrs = util.EMPTY_DICT
352 @util.ro_memoized_property
353 def description(self) -> Optional[str]:
354 return None
356 _is_clone_of: Optional[Self] = None
358 is_clause_element = True
359 is_selectable = False
360 is_dml = False
361 _is_column_element = False
362 _is_keyed_column_element = False
363 _is_table = False
364 _gen_static_annotations_cache_key = False
365 _is_textual = False
366 _is_from_clause = False
367 _is_returns_rows = False
368 _is_text_clause = False
369 _is_from_container = False
370 _is_select_container = False
371 _is_select_base = False
372 _is_select_statement = False
373 _is_bind_parameter = False
374 _is_clause_list = False
375 _is_lambda_element = False
376 _is_singleton_constant = False
377 _is_immutable = False
378 _is_star = False
380 @property
381 def _order_by_label_element(self) -> Optional[Label[Any]]:
382 return None
384 _cache_key_traversal: _CacheKeyTraversalType = None
386 negation_clause: ColumnElement[bool]
388 if typing.TYPE_CHECKING:
390 def get_children(
391 self, *, omit_attrs: typing_Tuple[str, ...] = ..., **kw: Any
392 ) -> Iterable[ClauseElement]: ...
394 @util.ro_non_memoized_property
395 def _from_objects(self) -> List[FromClause]:
396 return []
398 def _set_propagate_attrs(self, values: Mapping[str, Any]) -> Self:
399 # usually, self._propagate_attrs is empty here. one case where it's
400 # not is a subquery against ORM select, that is then pulled as a
401 # property of an aliased class. should all be good
403 # assert not self._propagate_attrs
405 self._propagate_attrs = util.immutabledict(values)
406 return self
408 def _clone(self, **kw: Any) -> Self:
409 """Create a shallow copy of this ClauseElement.
411 This method may be used by a generative API. Its also used as
412 part of the "deep" copy afforded by a traversal that combines
413 the _copy_internals() method.
415 """
417 skip = self._memoized_keys
418 c = self.__class__.__new__(self.__class__)
420 if skip:
421 # ensure this iteration remains atomic
422 c.__dict__ = {
423 k: v for k, v in self.__dict__.copy().items() if k not in skip
424 }
425 else:
426 c.__dict__ = self.__dict__.copy()
428 # this is a marker that helps to "equate" clauses to each other
429 # when a Select returns its list of FROM clauses. the cloning
430 # process leaves around a lot of remnants of the previous clause
431 # typically in the form of column expressions still attached to the
432 # old table.
433 cc = self._is_clone_of
434 c._is_clone_of = cc if cc is not None else self
435 return c
437 def _negate_in_binary(self, negated_op, original_op):
438 """a hook to allow the right side of a binary expression to respond
439 to a negation of the binary expression.
441 Used for the special case of expanding bind parameter with IN.
443 """
444 return self
446 def _with_binary_element_type(self, type_):
447 """in the context of binary expression, convert the type of this
448 object to the one given.
450 applies only to :class:`_expression.ColumnElement` classes.
452 """
453 return self
455 @property
456 def _constructor(self):
457 """return the 'constructor' for this ClauseElement.
459 This is for the purposes for creating a new object of
460 this type. Usually, its just the element's __class__.
461 However, the "Annotated" version of the object overrides
462 to return the class of its proxied element.
464 """
465 return self.__class__
467 @HasMemoized.memoized_attribute
468 def _cloned_set(self):
469 """Return the set consisting all cloned ancestors of this
470 ClauseElement.
472 Includes this ClauseElement. This accessor tends to be used for
473 FromClause objects to identify 'equivalent' FROM clauses, regardless
474 of transformative operations.
476 """
477 s = util.column_set()
478 f: Optional[ClauseElement] = self
480 # note this creates a cycle, asserted in test_memusage. however,
481 # turning this into a plain @property adds tends of thousands of method
482 # calls to Core / ORM performance tests, so the small overhead
483 # introduced by the relatively small amount of short term cycles
484 # produced here is preferable
485 while f is not None:
486 s.add(f)
487 f = f._is_clone_of
488 return s
490 def _de_clone(self):
491 while self._is_clone_of is not None:
492 self = self._is_clone_of
493 return self
495 @property
496 def entity_namespace(self):
497 raise AttributeError(
498 "This SQL expression has no entity namespace "
499 "with which to filter from."
500 )
502 def __getstate__(self):
503 d = self.__dict__.copy()
504 d.pop("_is_clone_of", None)
505 d.pop("_generate_cache_key", None)
506 return d
508 def _execute_on_connection(
509 self,
510 connection: Connection,
511 distilled_params: _CoreMultiExecuteParams,
512 execution_options: CoreExecuteOptionsParameter,
513 ) -> Result[Unpack[TupleAny]]:
514 if self.supports_execution:
515 if TYPE_CHECKING:
516 assert isinstance(self, Executable)
517 return connection._execute_clauseelement(
518 self, distilled_params, execution_options
519 )
520 else:
521 raise exc.ObjectNotExecutableError(self)
523 def _execute_on_scalar(
524 self,
525 connection: Connection,
526 distilled_params: _CoreMultiExecuteParams,
527 execution_options: CoreExecuteOptionsParameter,
528 ) -> Any:
529 """an additional hook for subclasses to provide a different
530 implementation for connection.scalar() vs. connection.execute().
532 .. versionadded:: 2.0
534 """
535 return self._execute_on_connection(
536 connection, distilled_params, execution_options
537 ).scalar()
539 def _get_embedded_bindparams(self) -> Sequence[BindParameter[Any]]:
540 """Return the list of :class:`.BindParameter` objects embedded in the
541 object.
543 This accomplishes the same purpose as ``visitors.traverse()`` or
544 similar would provide, however by making use of the cache key
545 it takes advantage of memoization of the key to result in fewer
546 net method calls, assuming the statement is also going to be
547 executed.
549 """
551 key = self._generate_cache_key()
552 if key is None:
553 bindparams: List[BindParameter[Any]] = []
555 traverse(self, {}, {"bindparam": bindparams.append})
556 return bindparams
558 else:
559 return key.bindparams
561 def unique_params(
562 self,
563 __optionaldict: Optional[Dict[str, Any]] = None,
564 /,
565 **kwargs: Any,
566 ) -> Self:
567 """Return a copy with :func:`_expression.bindparam` elements
568 replaced.
570 Same functionality as :meth:`_expression.ClauseElement.params`,
571 except adds `unique=True`
572 to affected bind parameters so that multiple statements can be
573 used.
575 """
576 return self._replace_params(True, __optionaldict, kwargs)
578 def params(
579 self,
580 __optionaldict: Optional[Mapping[str, Any]] = None,
581 /,
582 **kwargs: Any,
583 ) -> Self:
584 """Return a copy with :func:`_expression.bindparam` elements
585 replaced.
587 Returns a copy of this ClauseElement with
588 :func:`_expression.bindparam`
589 elements replaced with values taken from the given dictionary::
591 >>> clause = column('x') + bindparam('foo')
592 >>> print(clause.compile().params)
593 {'foo':None}
594 >>> print(clause.params({'foo':7}).compile().params)
595 {'foo':7}
597 """
598 return self._replace_params(False, __optionaldict, kwargs)
600 def _replace_params(
601 self,
602 unique: bool,
603 optionaldict: Optional[Mapping[str, Any]],
604 kwargs: Dict[str, Any],
605 ) -> Self:
606 if optionaldict:
607 kwargs.update(optionaldict)
609 def visit_bindparam(bind: BindParameter[Any]) -> None:
610 if bind.key in kwargs:
611 bind.value = kwargs[bind.key]
612 bind.required = False
613 if unique:
614 bind._convert_to_unique()
616 return cloned_traverse(
617 self,
618 {"maintain_key": True, "detect_subquery_cols": True},
619 {"bindparam": visit_bindparam},
620 )
622 def compare(self, other: ClauseElement, **kw: Any) -> bool:
623 r"""Compare this :class:`_expression.ClauseElement` to
624 the given :class:`_expression.ClauseElement`.
626 Subclasses should override the default behavior, which is a
627 straight identity comparison.
629 \**kw are arguments consumed by subclass ``compare()`` methods and
630 may be used to modify the criteria for comparison
631 (see :class:`_expression.ColumnElement`).
633 """
634 return traversals.compare(self, other, **kw)
636 def self_group(
637 self, against: Optional[OperatorType] = None
638 ) -> ClauseElement:
639 """Apply a 'grouping' to this :class:`_expression.ClauseElement`.
641 This method is overridden by subclasses to return a "grouping"
642 construct, i.e. parenthesis. In particular it's used by "binary"
643 expressions to provide a grouping around themselves when placed into a
644 larger expression, as well as by :func:`_expression.select`
645 constructs when placed into the FROM clause of another
646 :func:`_expression.select`. (Note that subqueries should be
647 normally created using the :meth:`_expression.Select.alias` method,
648 as many
649 platforms require nested SELECT statements to be named).
651 As expressions are composed together, the application of
652 :meth:`self_group` is automatic - end-user code should never
653 need to use this method directly. Note that SQLAlchemy's
654 clause constructs take operator precedence into account -
655 so parenthesis might not be needed, for example, in
656 an expression like ``x OR (y AND z)`` - AND takes precedence
657 over OR.
659 The base :meth:`self_group` method of
660 :class:`_expression.ClauseElement`
661 just returns self.
662 """
663 return self
665 def _ungroup(self) -> ClauseElement:
666 """Return this :class:`_expression.ClauseElement`
667 without any groupings.
668 """
670 return self
672 def _compile_w_cache(
673 self,
674 dialect: Dialect,
675 *,
676 compiled_cache: Optional[CompiledCacheType],
677 column_keys: List[str],
678 for_executemany: bool = False,
679 schema_translate_map: Optional[SchemaTranslateMapType] = None,
680 **kw: Any,
681 ) -> typing_Tuple[
682 Compiled, Optional[Sequence[BindParameter[Any]]], CacheStats
683 ]:
684 elem_cache_key: Optional[CacheKey]
686 if compiled_cache is not None and dialect._supports_statement_cache:
687 elem_cache_key = self._generate_cache_key()
688 else:
689 elem_cache_key = None
691 if elem_cache_key is not None:
692 if TYPE_CHECKING:
693 assert compiled_cache is not None
695 cache_key, extracted_params = elem_cache_key
696 key = (
697 dialect,
698 cache_key,
699 tuple(column_keys),
700 bool(schema_translate_map),
701 for_executemany,
702 )
703 compiled_sql = compiled_cache.get(key)
705 if compiled_sql is None:
706 cache_hit = dialect.CACHE_MISS
707 compiled_sql = self._compiler(
708 dialect,
709 cache_key=elem_cache_key,
710 column_keys=column_keys,
711 for_executemany=for_executemany,
712 schema_translate_map=schema_translate_map,
713 **kw,
714 )
715 compiled_cache[key] = compiled_sql
716 else:
717 cache_hit = dialect.CACHE_HIT
718 else:
719 extracted_params = None
720 compiled_sql = self._compiler(
721 dialect,
722 cache_key=elem_cache_key,
723 column_keys=column_keys,
724 for_executemany=for_executemany,
725 schema_translate_map=schema_translate_map,
726 **kw,
727 )
729 if not dialect._supports_statement_cache:
730 cache_hit = dialect.NO_DIALECT_SUPPORT
731 elif compiled_cache is None:
732 cache_hit = dialect.CACHING_DISABLED
733 else:
734 cache_hit = dialect.NO_CACHE_KEY
736 return compiled_sql, extracted_params, cache_hit
738 def __invert__(self):
739 # undocumented element currently used by the ORM for
740 # relationship.contains()
741 if hasattr(self, "negation_clause"):
742 return self.negation_clause
743 else:
744 return self._negate()
746 def _negate(self) -> ClauseElement:
747 grouped = self.self_group(against=operators.inv)
748 assert isinstance(grouped, ColumnElement)
749 return UnaryExpression(grouped, operator=operators.inv)
751 def __bool__(self):
752 raise TypeError("Boolean value of this clause is not defined")
754 def __repr__(self):
755 friendly = self.description
756 if friendly is None:
757 return object.__repr__(self)
758 else:
759 return "<%s.%s at 0x%x; %s>" % (
760 self.__module__,
761 self.__class__.__name__,
762 id(self),
763 friendly,
764 )
767class DQLDMLClauseElement(ClauseElement):
768 """represents a :class:`.ClauseElement` that compiles to a DQL or DML
769 expression, not DDL.
771 .. versionadded:: 2.0
773 """
775 if typing.TYPE_CHECKING:
777 def _compiler(self, dialect: Dialect, **kw: Any) -> SQLCompiler:
778 """Return a compiler appropriate for this ClauseElement, given a
779 Dialect."""
780 ...
782 def compile( # noqa: A001
783 self,
784 bind: Optional[_HasDialect] = None,
785 dialect: Optional[Dialect] = None,
786 **kw: Any,
787 ) -> SQLCompiler: ...
790class CompilerColumnElement(
791 roles.DMLColumnRole,
792 roles.DDLConstraintColumnRole,
793 roles.ColumnsClauseRole,
794 CompilerElement,
795):
796 """A compiler-only column element used for ad-hoc string compilations.
798 .. versionadded:: 2.0
800 """
802 __slots__ = ()
804 _propagate_attrs = util.EMPTY_DICT
805 _is_collection_aggregate = False
808# SQLCoreOperations should be suiting the ExpressionElementRole
809# and ColumnsClauseRole. however the MRO issues become too elaborate
810# at the moment.
811class SQLCoreOperations(Generic[_T_co], ColumnOperators, TypingOnly):
812 __slots__ = ()
814 # annotations for comparison methods
815 # these are from operators->Operators / ColumnOperators,
816 # redefined with the specific types returned by ColumnElement hierarchies
817 if typing.TYPE_CHECKING:
819 @util.non_memoized_property
820 def _propagate_attrs(self) -> _PropagateAttrsType: ...
822 def operate(
823 self, op: OperatorType, *other: Any, **kwargs: Any
824 ) -> ColumnElement[Any]: ...
826 def reverse_operate(
827 self, op: OperatorType, other: Any, **kwargs: Any
828 ) -> ColumnElement[Any]: ...
830 @overload
831 def op(
832 self,
833 opstring: str,
834 precedence: int = ...,
835 is_comparison: bool = ...,
836 *,
837 return_type: _TypeEngineArgument[_OPT],
838 python_impl: Optional[Callable[..., Any]] = None,
839 ) -> Callable[[Any], BinaryExpression[_OPT]]: ...
841 @overload
842 def op(
843 self,
844 opstring: str,
845 precedence: int = ...,
846 is_comparison: bool = ...,
847 return_type: Optional[_TypeEngineArgument[Any]] = ...,
848 python_impl: Optional[Callable[..., Any]] = ...,
849 ) -> Callable[[Any], BinaryExpression[Any]]: ...
851 def op(
852 self,
853 opstring: str,
854 precedence: int = 0,
855 is_comparison: bool = False,
856 return_type: Optional[_TypeEngineArgument[Any]] = None,
857 python_impl: Optional[Callable[..., Any]] = None,
858 ) -> Callable[[Any], BinaryExpression[Any]]: ...
860 def bool_op(
861 self,
862 opstring: str,
863 precedence: int = 0,
864 python_impl: Optional[Callable[..., Any]] = None,
865 ) -> Callable[[Any], BinaryExpression[bool]]: ...
867 def __and__(self, other: Any) -> BooleanClauseList: ...
869 def __or__(self, other: Any) -> BooleanClauseList: ...
871 def __invert__(self) -> ColumnElement[_T_co]: ...
873 def __lt__(self, other: Any) -> ColumnElement[bool]: ...
875 def __le__(self, other: Any) -> ColumnElement[bool]: ...
877 # declare also that this class has an hash method otherwise
878 # it may be assumed to be None by type checkers since the
879 # object defines __eq__ and python sets it to None in that case:
880 # https://docs.python.org/3/reference/datamodel.html#object.__hash__
881 def __hash__(self) -> int: ...
883 def __eq__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
884 ...
886 def __ne__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
887 ...
889 def is_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
891 def is_not_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
893 def __gt__(self, other: Any) -> ColumnElement[bool]: ...
895 def __ge__(self, other: Any) -> ColumnElement[bool]: ...
897 def __neg__(self) -> UnaryExpression[_T_co]: ...
899 def __contains__(self, other: Any) -> ColumnElement[bool]: ...
901 def __getitem__(self, index: Any) -> ColumnElement[Any]: ...
903 @overload
904 def __lshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
906 @overload
907 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
909 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
911 @overload
912 def __rshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
914 @overload
915 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
917 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
919 @overload
920 def concat(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
922 @overload
923 def concat(self, other: Any) -> ColumnElement[Any]: ...
925 def concat(self, other: Any) -> ColumnElement[Any]: ...
927 def like(
928 self, other: Any, escape: Optional[str] = None
929 ) -> BinaryExpression[bool]: ...
931 def ilike(
932 self, other: Any, escape: Optional[str] = None
933 ) -> BinaryExpression[bool]: ...
935 def bitwise_xor(self, other: Any) -> BinaryExpression[Any]: ...
937 def bitwise_or(self, other: Any) -> BinaryExpression[Any]: ...
939 def bitwise_and(self, other: Any) -> BinaryExpression[Any]: ...
941 def bitwise_not(self) -> UnaryExpression[_T_co]: ...
943 def bitwise_lshift(self, other: Any) -> BinaryExpression[Any]: ...
945 def bitwise_rshift(self, other: Any) -> BinaryExpression[Any]: ...
947 def in_(
948 self,
949 other: Union[
950 Iterable[Any], BindParameter[Any], roles.InElementRole
951 ],
952 ) -> BinaryExpression[bool]: ...
954 def not_in(
955 self,
956 other: Union[
957 Iterable[Any], BindParameter[Any], roles.InElementRole
958 ],
959 ) -> BinaryExpression[bool]: ...
961 def notin_(
962 self,
963 other: Union[
964 Iterable[Any], BindParameter[Any], roles.InElementRole
965 ],
966 ) -> BinaryExpression[bool]: ...
968 def not_like(
969 self, other: Any, escape: Optional[str] = None
970 ) -> BinaryExpression[bool]: ...
972 def notlike(
973 self, other: Any, escape: Optional[str] = None
974 ) -> BinaryExpression[bool]: ...
976 def not_ilike(
977 self, other: Any, escape: Optional[str] = None
978 ) -> BinaryExpression[bool]: ...
980 def notilike(
981 self, other: Any, escape: Optional[str] = None
982 ) -> BinaryExpression[bool]: ...
984 def is_(self, other: Any) -> BinaryExpression[bool]: ...
986 def is_not(self, other: Any) -> BinaryExpression[bool]: ...
988 def isnot(self, other: Any) -> BinaryExpression[bool]: ...
990 def startswith(
991 self,
992 other: Any,
993 escape: Optional[str] = None,
994 autoescape: bool = False,
995 ) -> ColumnElement[bool]: ...
997 def istartswith(
998 self,
999 other: Any,
1000 escape: Optional[str] = None,
1001 autoescape: bool = False,
1002 ) -> ColumnElement[bool]: ...
1004 def endswith(
1005 self,
1006 other: Any,
1007 escape: Optional[str] = None,
1008 autoescape: bool = False,
1009 ) -> ColumnElement[bool]: ...
1011 def iendswith(
1012 self,
1013 other: Any,
1014 escape: Optional[str] = None,
1015 autoescape: bool = False,
1016 ) -> ColumnElement[bool]: ...
1018 def contains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1020 def icontains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1022 def match(self, other: Any, **kwargs: Any) -> ColumnElement[bool]: ...
1024 def regexp_match(
1025 self, pattern: Any, flags: Optional[str] = None
1026 ) -> ColumnElement[bool]: ...
1028 def regexp_replace(
1029 self, pattern: Any, replacement: Any, flags: Optional[str] = None
1030 ) -> ColumnElement[str]: ...
1032 def desc(self) -> UnaryExpression[_T_co]: ...
1034 def asc(self) -> UnaryExpression[_T_co]: ...
1036 def nulls_first(self) -> UnaryExpression[_T_co]: ...
1038 def nullsfirst(self) -> UnaryExpression[_T_co]: ...
1040 def nulls_last(self) -> UnaryExpression[_T_co]: ...
1042 def nullslast(self) -> UnaryExpression[_T_co]: ...
1044 def collate(self, collation: str) -> CollationClause: ...
1046 def between(
1047 self, cleft: Any, cright: Any, symmetric: bool = False
1048 ) -> BinaryExpression[bool]: ...
1050 def distinct(self: _SQO[_T_co]) -> UnaryExpression[_T_co]: ...
1052 def any_(self) -> CollectionAggregate[Any]: ...
1054 def all_(self) -> CollectionAggregate[Any]: ...
1056 # numeric overloads. These need more tweaking
1057 # in particular they all need to have a variant for Optiona[_T]
1058 # because Optional only applies to the data side, not the expression
1059 # side
1061 @overload
1062 def __add__(
1063 self: _SQO[_NMT],
1064 other: Any,
1065 ) -> ColumnElement[_NMT]: ...
1067 @overload
1068 def __add__(
1069 self: _SQO[str],
1070 other: Any,
1071 ) -> ColumnElement[str]: ...
1073 @overload
1074 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1076 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1078 @overload
1079 def __radd__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1081 @overload
1082 def __radd__(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
1084 def __radd__(self, other: Any) -> ColumnElement[Any]: ...
1086 @overload
1087 def __sub__(
1088 self: _SQO[_NMT],
1089 other: Any,
1090 ) -> ColumnElement[_NMT]: ...
1092 @overload
1093 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1095 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1097 @overload
1098 def __rsub__(
1099 self: _SQO[_NMT],
1100 other: Any,
1101 ) -> ColumnElement[_NMT]: ...
1103 @overload
1104 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1106 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1108 @overload
1109 def __mul__(
1110 self: _SQO[_NMT],
1111 other: Any,
1112 ) -> ColumnElement[_NMT]: ...
1114 @overload
1115 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1117 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1119 @overload
1120 def __rmul__(
1121 self: _SQO[_NMT],
1122 other: Any,
1123 ) -> ColumnElement[_NMT]: ...
1125 @overload
1126 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1128 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1130 @overload
1131 def __mod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1133 @overload
1134 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1136 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1138 @overload
1139 def __rmod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1141 @overload
1142 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1144 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1146 @overload
1147 def __truediv__(
1148 self: _SQO[int], other: Any
1149 ) -> ColumnElement[_NUMERIC]: ...
1151 @overload
1152 def __truediv__(self: _SQO[_NT], other: Any) -> ColumnElement[_NT]: ...
1154 @overload
1155 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1157 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1159 @overload
1160 def __rtruediv__(
1161 self: _SQO[_NMT], other: Any
1162 ) -> ColumnElement[_NUMERIC]: ...
1164 @overload
1165 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1167 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1169 @overload
1170 def __floordiv__(
1171 self: _SQO[_NMT], other: Any
1172 ) -> ColumnElement[_NMT]: ...
1174 @overload
1175 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1177 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1179 @overload
1180 def __rfloordiv__(
1181 self: _SQO[_NMT], other: Any
1182 ) -> ColumnElement[_NMT]: ...
1184 @overload
1185 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1187 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1190class SQLColumnExpression(
1191 SQLCoreOperations[_T_co], roles.ExpressionElementRole[_T_co], TypingOnly
1192):
1193 """A type that may be used to indicate any SQL column element or object
1194 that acts in place of one.
1196 :class:`.SQLColumnExpression` is a base of
1197 :class:`.ColumnElement`, as well as within the bases of ORM elements
1198 such as :class:`.InstrumentedAttribute`, and may be used in :pep:`484`
1199 typing to indicate arguments or return values that should behave
1200 as column expressions.
1202 .. versionadded:: 2.0.0b4
1205 """
1207 __slots__ = ()
1210_SQO = SQLCoreOperations
1213class ColumnElement(
1214 roles.ColumnArgumentOrKeyRole,
1215 roles.StatementOptionRole,
1216 roles.WhereHavingRole,
1217 roles.BinaryElementRole[_T],
1218 roles.OrderByRole,
1219 roles.ColumnsClauseRole,
1220 roles.LimitOffsetRole,
1221 roles.DMLColumnRole,
1222 roles.DDLConstraintColumnRole,
1223 roles.DDLExpressionRole,
1224 SQLColumnExpression[_T],
1225 DQLDMLClauseElement,
1226):
1227 """Represent a column-oriented SQL expression suitable for usage in the
1228 "columns" clause, WHERE clause etc. of a statement.
1230 While the most familiar kind of :class:`_expression.ColumnElement` is the
1231 :class:`_schema.Column` object, :class:`_expression.ColumnElement`
1232 serves as the basis
1233 for any unit that may be present in a SQL expression, including
1234 the expressions themselves, SQL functions, bound parameters,
1235 literal expressions, keywords such as ``NULL``, etc.
1236 :class:`_expression.ColumnElement`
1237 is the ultimate base class for all such elements.
1239 A wide variety of SQLAlchemy Core functions work at the SQL expression
1240 level, and are intended to accept instances of
1241 :class:`_expression.ColumnElement` as
1242 arguments. These functions will typically document that they accept a
1243 "SQL expression" as an argument. What this means in terms of SQLAlchemy
1244 usually refers to an input which is either already in the form of a
1245 :class:`_expression.ColumnElement` object,
1246 or a value which can be **coerced** into
1247 one. The coercion rules followed by most, but not all, SQLAlchemy Core
1248 functions with regards to SQL expressions are as follows:
1250 * a literal Python value, such as a string, integer or floating
1251 point value, boolean, datetime, ``Decimal`` object, or virtually
1252 any other Python object, will be coerced into a "literal bound
1253 value". This generally means that a :func:`.bindparam` will be
1254 produced featuring the given value embedded into the construct; the
1255 resulting :class:`.BindParameter` object is an instance of
1256 :class:`_expression.ColumnElement`.
1257 The Python value will ultimately be sent
1258 to the DBAPI at execution time as a parameterized argument to the
1259 ``execute()`` or ``executemany()`` methods, after SQLAlchemy
1260 type-specific converters (e.g. those provided by any associated
1261 :class:`.TypeEngine` objects) are applied to the value.
1263 * any special object value, typically ORM-level constructs, which
1264 feature an accessor called ``__clause_element__()``. The Core
1265 expression system looks for this method when an object of otherwise
1266 unknown type is passed to a function that is looking to coerce the
1267 argument into a :class:`_expression.ColumnElement` and sometimes a
1268 :class:`_expression.SelectBase` expression.
1269 It is used within the ORM to
1270 convert from ORM-specific objects like mapped classes and
1271 mapped attributes into Core expression objects.
1273 * The Python ``None`` value is typically interpreted as ``NULL``,
1274 which in SQLAlchemy Core produces an instance of :func:`.null`.
1276 A :class:`_expression.ColumnElement` provides the ability to generate new
1277 :class:`_expression.ColumnElement`
1278 objects using Python expressions. This means that Python operators
1279 such as ``==``, ``!=`` and ``<`` are overloaded to mimic SQL operations,
1280 and allow the instantiation of further :class:`_expression.ColumnElement`
1281 instances
1282 which are composed from other, more fundamental
1283 :class:`_expression.ColumnElement`
1284 objects. For example, two :class:`.ColumnClause` objects can be added
1285 together with the addition operator ``+`` to produce
1286 a :class:`.BinaryExpression`.
1287 Both :class:`.ColumnClause` and :class:`.BinaryExpression` are subclasses
1288 of :class:`_expression.ColumnElement`:
1290 .. sourcecode:: pycon+sql
1292 >>> from sqlalchemy.sql import column
1293 >>> column('a') + column('b')
1294 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
1295 >>> print(column('a') + column('b'))
1296 {printsql}a + b
1298 .. seealso::
1300 :class:`_schema.Column`
1302 :func:`_expression.column`
1304 """
1306 __visit_name__ = "column_element"
1308 primary_key: bool = False
1309 _is_clone_of: Optional[ColumnElement[_T]]
1310 _is_column_element = True
1311 _insert_sentinel: bool = False
1312 _omit_from_statements = False
1313 _is_collection_aggregate = False
1315 foreign_keys: AbstractSet[ForeignKey] = frozenset()
1317 @util.memoized_property
1318 def _proxies(self) -> List[ColumnElement[Any]]:
1319 return []
1321 @util.non_memoized_property
1322 def _tq_label(self) -> Optional[str]:
1323 """The named label that can be used to target
1324 this column in a result set in a "table qualified" context.
1326 This label is almost always the label used when
1327 rendering <expr> AS <label> in a SELECT statement when using
1328 the LABEL_STYLE_TABLENAME_PLUS_COL label style, which is what the
1329 legacy ORM ``Query`` object uses as well.
1331 For a regular Column bound to a Table, this is typically the label
1332 <tablename>_<columnname>. For other constructs, different rules
1333 may apply, such as anonymized labels and others.
1335 .. versionchanged:: 1.4.21 renamed from ``._label``
1337 """
1338 return None
1340 key: Optional[str] = None
1341 """The 'key' that in some circumstances refers to this object in a
1342 Python namespace.
1344 This typically refers to the "key" of the column as present in the
1345 ``.c`` collection of a selectable, e.g. ``sometable.c["somekey"]`` would
1346 return a :class:`_schema.Column` with a ``.key`` of "somekey".
1348 """
1350 @HasMemoized.memoized_attribute
1351 def _tq_key_label(self) -> Optional[str]:
1352 """A label-based version of 'key' that in some circumstances refers
1353 to this object in a Python namespace.
1356 _tq_key_label comes into play when a select() statement is constructed
1357 with apply_labels(); in this case, all Column objects in the ``.c``
1358 collection are rendered as <tablename>_<columnname> in SQL; this is
1359 essentially the value of ._label. But to locate those columns in the
1360 ``.c`` collection, the name is along the lines of <tablename>_<key>;
1361 that's the typical value of .key_label.
1363 .. versionchanged:: 1.4.21 renamed from ``._key_label``
1365 """
1366 return self._proxy_key
1368 @property
1369 def _key_label(self) -> Optional[str]:
1370 """legacy; renamed to _tq_key_label"""
1371 return self._tq_key_label
1373 @property
1374 def _label(self) -> Optional[str]:
1375 """legacy; renamed to _tq_label"""
1376 return self._tq_label
1378 @property
1379 def _non_anon_label(self) -> Optional[str]:
1380 """the 'name' that naturally applies this element when rendered in
1381 SQL.
1383 Concretely, this is the "name" of a column or a label in a
1384 SELECT statement; ``<columnname>`` and ``<labelname>`` below::
1386 SELECT <columnmame> FROM table
1388 SELECT column AS <labelname> FROM table
1390 Above, the two names noted will be what's present in the DBAPI
1391 ``cursor.description`` as the names.
1393 If this attribute returns ``None``, it means that the SQL element as
1394 written does not have a 100% fully predictable "name" that would appear
1395 in the ``cursor.description``. Examples include SQL functions, CAST
1396 functions, etc. While such things do return names in
1397 ``cursor.description``, they are only predictable on a
1398 database-specific basis; e.g. an expression like ``MAX(table.col)`` may
1399 appear as the string ``max`` on one database (like PostgreSQL) or may
1400 appear as the whole expression ``max(table.col)`` on SQLite.
1402 The default implementation looks for a ``.name`` attribute on the
1403 object, as has been the precedent established in SQLAlchemy for many
1404 years. An exception is made on the ``FunctionElement`` subclass
1405 so that the return value is always ``None``.
1407 .. versionadded:: 1.4.21
1411 """
1412 return getattr(self, "name", None)
1414 _render_label_in_columns_clause = True
1415 """A flag used by select._columns_plus_names that helps to determine
1416 we are actually going to render in terms of "SELECT <col> AS <label>".
1417 This flag can be returned as False for some Column objects that want
1418 to be rendered as simple "SELECT <col>"; typically columns that don't have
1419 any parent table and are named the same as what the label would be
1420 in any case.
1422 """
1424 _allow_label_resolve = True
1425 """A flag that can be flipped to prevent a column from being resolvable
1426 by string label name.
1428 The joined eager loader strategy in the ORM uses this, for example.
1430 """
1432 _is_implicitly_boolean = False
1434 _alt_names: Sequence[str] = ()
1436 @overload
1437 def self_group(self, against: None = None) -> ColumnElement[_T]: ...
1439 @overload
1440 def self_group(
1441 self, against: Optional[OperatorType] = None
1442 ) -> ColumnElement[Any]: ...
1444 def self_group(
1445 self, against: Optional[OperatorType] = None
1446 ) -> ColumnElement[Any]:
1447 if (
1448 against in (operators.and_, operators.or_, operators._asbool)
1449 and self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity
1450 ):
1451 return AsBoolean(self, operators.is_true, operators.is_false)
1452 elif against in (operators.any_op, operators.all_op):
1453 return Grouping(self)
1454 else:
1455 return self
1457 @overload
1458 def _negate(self: ColumnElement[bool]) -> ColumnElement[bool]: ...
1460 @overload
1461 def _negate(self: ColumnElement[_T]) -> ColumnElement[_T]: ...
1463 def _negate(self) -> ColumnElement[Any]:
1464 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
1465 return AsBoolean(self, operators.is_false, operators.is_true)
1466 else:
1467 grouped = self.self_group(against=operators.inv)
1468 assert isinstance(grouped, ColumnElement)
1469 return UnaryExpression(
1470 grouped, operator=operators.inv, wraps_column_expression=True
1471 )
1473 type: TypeEngine[_T]
1475 if not TYPE_CHECKING:
1477 @util.memoized_property
1478 def type(self) -> TypeEngine[_T]: # noqa: A001
1479 # used for delayed setup of
1480 # type_api
1481 return type_api.NULLTYPE
1483 @HasMemoized.memoized_attribute
1484 def comparator(self) -> TypeEngine.Comparator[_T]:
1485 try:
1486 comparator_factory = self.type.comparator_factory
1487 except AttributeError as err:
1488 raise TypeError(
1489 "Object %r associated with '.type' attribute "
1490 "is not a TypeEngine class or object" % self.type
1491 ) from err
1492 else:
1493 return comparator_factory(self)
1495 def __setstate__(self, state):
1496 self.__dict__.update(state)
1498 def __getattr__(self, key: str) -> Any:
1499 try:
1500 return getattr(self.comparator, key)
1501 except AttributeError as err:
1502 raise AttributeError(
1503 "Neither %r object nor %r object has an attribute %r"
1504 % (
1505 type(self).__name__,
1506 type(self.comparator).__name__,
1507 key,
1508 )
1509 ) from err
1511 def operate(
1512 self,
1513 op: operators.OperatorType,
1514 *other: Any,
1515 **kwargs: Any,
1516 ) -> ColumnElement[Any]:
1517 return op(self.comparator, *other, **kwargs) # type: ignore[no-any-return] # noqa: E501
1519 def reverse_operate(
1520 self, op: operators.OperatorType, other: Any, **kwargs: Any
1521 ) -> ColumnElement[Any]:
1522 return op(other, self.comparator, **kwargs) # type: ignore[no-any-return] # noqa: E501
1524 def _bind_param(
1525 self,
1526 operator: operators.OperatorType,
1527 obj: Any,
1528 type_: Optional[TypeEngine[_T]] = None,
1529 expanding: bool = False,
1530 ) -> BindParameter[_T]:
1531 return BindParameter(
1532 None,
1533 obj,
1534 _compared_to_operator=operator,
1535 type_=type_,
1536 _compared_to_type=self.type,
1537 unique=True,
1538 expanding=expanding,
1539 )
1541 @property
1542 def expression(self) -> ColumnElement[Any]:
1543 """Return a column expression.
1545 Part of the inspection interface; returns self.
1547 """
1548 return self
1550 @property
1551 def _select_iterable(self) -> _SelectIterable:
1552 return (self,)
1554 @util.memoized_property
1555 def base_columns(self) -> FrozenSet[ColumnElement[Any]]:
1556 return frozenset(c for c in self.proxy_set if not c._proxies)
1558 @util.memoized_property
1559 def proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1560 """set of all columns we are proxying
1562 as of 2.0 this is explicitly deannotated columns. previously it was
1563 effectively deannotated columns but wasn't enforced. annotated
1564 columns should basically not go into sets if at all possible because
1565 their hashing behavior is very non-performant.
1567 """
1568 return frozenset([self._deannotate()]).union(
1569 itertools.chain(*[c.proxy_set for c in self._proxies])
1570 )
1572 @util.memoized_property
1573 def _expanded_proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1574 return frozenset(_expand_cloned(self.proxy_set))
1576 def _uncached_proxy_list(self) -> List[ColumnElement[Any]]:
1577 """An 'uncached' version of proxy set.
1579 This list includes annotated columns which perform very poorly in
1580 set operations.
1582 """
1584 return [self] + list(
1585 itertools.chain(*[c._uncached_proxy_list() for c in self._proxies])
1586 )
1588 def shares_lineage(self, othercolumn: ColumnElement[Any]) -> bool:
1589 """Return True if the given :class:`_expression.ColumnElement`
1590 has a common ancestor to this :class:`_expression.ColumnElement`."""
1592 return bool(self.proxy_set.intersection(othercolumn.proxy_set))
1594 def _compare_name_for_result(self, other: ColumnElement[Any]) -> bool:
1595 """Return True if the given column element compares to this one
1596 when targeting within a result row."""
1598 return (
1599 hasattr(other, "name")
1600 and hasattr(self, "name")
1601 and other.name == self.name
1602 )
1604 @HasMemoized.memoized_attribute
1605 def _proxy_key(self) -> Optional[str]:
1606 if self._annotations and "proxy_key" in self._annotations:
1607 return cast(str, self._annotations["proxy_key"])
1609 name = self.key
1610 if not name:
1611 # there's a bit of a seeming contradiction which is that the
1612 # "_non_anon_label" of a column can in fact be an
1613 # "_anonymous_label"; this is when it's on a column that is
1614 # proxying for an anonymous expression in a subquery.
1615 name = self._non_anon_label
1617 if isinstance(name, _anonymous_label):
1618 return None
1619 else:
1620 return name
1622 @HasMemoized.memoized_attribute
1623 def _expression_label(self) -> Optional[str]:
1624 """a suggested label to use in the case that the column has no name,
1625 which should be used if possible as the explicit 'AS <label>'
1626 where this expression would normally have an anon label.
1628 this is essentially mostly what _proxy_key does except it returns
1629 None if the column has a normal name that can be used.
1631 """
1633 if getattr(self, "name", None) is not None:
1634 return None
1635 elif self._annotations and "proxy_key" in self._annotations:
1636 return cast(str, self._annotations["proxy_key"])
1637 else:
1638 return None
1640 def _make_proxy(
1641 self,
1642 selectable: FromClause,
1643 *,
1644 name: Optional[str] = None,
1645 key: Optional[str] = None,
1646 name_is_truncatable: bool = False,
1647 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
1648 **kw: Any,
1649 ) -> typing_Tuple[str, ColumnClause[_T]]:
1650 """Create a new :class:`_expression.ColumnElement` representing this
1651 :class:`_expression.ColumnElement` as it appears in the select list of
1652 a descending selectable.
1654 """
1655 if name is None:
1656 name = self._anon_name_label
1657 if key is None:
1658 key = self._proxy_key
1659 else:
1660 key = name
1662 assert key is not None
1664 co: ColumnClause[_T] = ColumnClause(
1665 (
1666 coercions.expect(roles.TruncatedLabelRole, name)
1667 if name_is_truncatable
1668 else name
1669 ),
1670 type_=getattr(self, "type", None),
1671 _selectable=selectable,
1672 )
1674 co._propagate_attrs = selectable._propagate_attrs
1675 if compound_select_cols:
1676 co._proxies = list(compound_select_cols)
1677 else:
1678 co._proxies = [self]
1679 if selectable._is_clone_of is not None:
1680 co._is_clone_of = selectable._is_clone_of.columns.get(key)
1681 return key, co
1683 def cast(self, type_: _TypeEngineArgument[_OPT]) -> Cast[_OPT]:
1684 """Produce a type cast, i.e. ``CAST(<expression> AS <type>)``.
1686 This is a shortcut to the :func:`_expression.cast` function.
1688 .. seealso::
1690 :ref:`tutorial_casts`
1692 :func:`_expression.cast`
1694 :func:`_expression.type_coerce`
1696 """
1697 return Cast(self, type_)
1699 def label(self, name: Optional[str]) -> Label[_T]:
1700 """Produce a column label, i.e. ``<columnname> AS <name>``.
1702 This is a shortcut to the :func:`_expression.label` function.
1704 If 'name' is ``None``, an anonymous label name will be generated.
1706 """
1707 return Label(name, self, self.type)
1709 def _anon_label(
1710 self, seed: Optional[str], add_hash: Optional[int] = None
1711 ) -> _anonymous_label:
1712 while self._is_clone_of is not None:
1713 self = self._is_clone_of
1715 # as of 1.4 anonymous label for ColumnElement uses hash(), not id(),
1716 # as the identifier, because a column and its annotated version are
1717 # the same thing in a SQL statement
1718 hash_value = hash(self)
1720 if add_hash:
1721 # this path is used for disambiguating anon labels that would
1722 # otherwise be the same name for the same element repeated.
1723 # an additional numeric value is factored in for each label.
1725 # shift hash(self) (which is id(self), typically 8 byte integer)
1726 # 16 bits leftward. fill extra add_hash on right
1727 assert add_hash < (2 << 15)
1728 assert seed
1729 hash_value = (hash_value << 16) | add_hash
1731 # extra underscore is added for labels with extra hash
1732 # values, to isolate the "deduped anon" namespace from the
1733 # regular namespace. eliminates chance of these
1734 # manufactured hash values overlapping with regular ones for some
1735 # undefined python interpreter
1736 seed = seed + "_"
1738 if isinstance(seed, _anonymous_label):
1739 # NOTE: the space after the hash is required
1740 return _anonymous_label(f"{seed}%({hash_value} )s")
1742 return _anonymous_label.safe_construct(hash_value, seed or "anon")
1744 @util.memoized_property
1745 def _anon_name_label(self) -> str:
1746 """Provides a constant 'anonymous label' for this ColumnElement.
1748 This is a label() expression which will be named at compile time.
1749 The same label() is returned each time ``anon_label`` is called so
1750 that expressions can reference ``anon_label`` multiple times,
1751 producing the same label name at compile time.
1753 The compiler uses this function automatically at compile time
1754 for expressions that are known to be 'unnamed' like binary
1755 expressions and function calls.
1757 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1758 public and is renamed to _anon_name_label. anon_name exists
1759 for backwards compat
1761 """
1762 name = getattr(self, "name", None)
1763 return self._anon_label(name)
1765 @util.memoized_property
1766 def _anon_key_label(self) -> _anonymous_label:
1767 """Provides a constant 'anonymous key label' for this ColumnElement.
1769 Compare to ``anon_label``, except that the "key" of the column,
1770 if available, is used to generate the label.
1772 This is used when a deduplicating key is placed into the columns
1773 collection of a selectable.
1775 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1776 public and is renamed to _anon_key_label. anon_key_label exists
1777 for backwards compat
1779 """
1780 return self._anon_label(self._proxy_key)
1782 @property
1783 @util.deprecated(
1784 "1.4",
1785 "The :attr:`_expression.ColumnElement.anon_label` attribute is now "
1786 "private, and the public accessor is deprecated.",
1787 )
1788 def anon_label(self) -> str:
1789 return self._anon_name_label
1791 @property
1792 @util.deprecated(
1793 "1.4",
1794 "The :attr:`_expression.ColumnElement.anon_key_label` attribute is "
1795 "now private, and the public accessor is deprecated.",
1796 )
1797 def anon_key_label(self) -> str:
1798 return self._anon_key_label
1800 def _dedupe_anon_label_idx(self, idx: int) -> str:
1801 """label to apply to a column that is anon labeled, but repeated
1802 in the SELECT, so that we have to make an "extra anon" label that
1803 disambiguates it from the previous appearance.
1805 these labels come out like "foo_bar_id__1" and have double underscores
1806 in them.
1808 """
1809 label = getattr(self, "name", None)
1811 # current convention is that if the element doesn't have a
1812 # ".name" (usually because it is not NamedColumn), we try to
1813 # use a "table qualified" form for the "dedupe anon" label,
1814 # based on the notion that a label like
1815 # "CAST(casttest.v1 AS DECIMAL) AS casttest_v1__1" looks better than
1816 # "CAST(casttest.v1 AS DECIMAL) AS anon__1"
1818 if label is None:
1819 return self._dedupe_anon_tq_label_idx(idx)
1820 else:
1821 return self._anon_label(label, add_hash=idx)
1823 @util.memoized_property
1824 def _anon_tq_label(self) -> _anonymous_label:
1825 return self._anon_label(getattr(self, "_tq_label", None))
1827 @util.memoized_property
1828 def _anon_tq_key_label(self) -> _anonymous_label:
1829 return self._anon_label(getattr(self, "_tq_key_label", None))
1831 def _dedupe_anon_tq_label_idx(self, idx: int) -> _anonymous_label:
1832 label = getattr(self, "_tq_label", None) or "anon"
1834 return self._anon_label(label, add_hash=idx)
1837class KeyedColumnElement(ColumnElement[_T]):
1838 """ColumnElement where ``.key`` is non-None."""
1840 _is_keyed_column_element = True
1842 key: str
1845class WrapsColumnExpression(ColumnElement[_T]):
1846 """Mixin that defines a :class:`_expression.ColumnElement`
1847 as a wrapper with special
1848 labeling behavior for an expression that already has a name.
1850 .. versionadded:: 1.4
1852 .. seealso::
1854 :ref:`change_4449`
1857 """
1859 @property
1860 def wrapped_column_expression(self) -> ColumnElement[_T]:
1861 raise NotImplementedError()
1863 @util.non_memoized_property
1864 def _tq_label(self) -> Optional[str]:
1865 wce = self.wrapped_column_expression
1866 if hasattr(wce, "_tq_label"):
1867 return wce._tq_label
1868 else:
1869 return None
1871 @property
1872 def _label(self) -> Optional[str]:
1873 return self._tq_label
1875 @property
1876 def _non_anon_label(self) -> Optional[str]:
1877 return None
1879 @util.non_memoized_property
1880 def _anon_name_label(self) -> str:
1881 wce = self.wrapped_column_expression
1883 # this logic tries to get the WrappedColumnExpression to render
1884 # with "<expr> AS <name>", where "<name>" is the natural name
1885 # within the expression itself. e.g. "CAST(table.foo) AS foo".
1886 if not wce._is_text_clause:
1887 nal = wce._non_anon_label
1888 if nal:
1889 return nal
1890 elif hasattr(wce, "_anon_name_label"):
1891 return wce._anon_name_label
1892 return super()._anon_name_label
1894 def _dedupe_anon_label_idx(self, idx: int) -> str:
1895 wce = self.wrapped_column_expression
1896 nal = wce._non_anon_label
1897 if nal:
1898 return self._anon_label(nal + "_")
1899 else:
1900 return self._dedupe_anon_tq_label_idx(idx)
1902 @property
1903 def _proxy_key(self):
1904 wce = self.wrapped_column_expression
1906 if not wce._is_text_clause:
1907 return wce._proxy_key
1908 return super()._proxy_key
1911class BindParameter(roles.InElementRole, KeyedColumnElement[_T]):
1912 r"""Represent a "bound expression".
1914 :class:`.BindParameter` is invoked explicitly using the
1915 :func:`.bindparam` function, as in::
1917 from sqlalchemy import bindparam
1919 stmt = select(users_table).where(
1920 users_table.c.name == bindparam("username")
1921 )
1923 Detailed discussion of how :class:`.BindParameter` is used is
1924 at :func:`.bindparam`.
1926 .. seealso::
1928 :func:`.bindparam`
1930 """
1932 __visit_name__ = "bindparam"
1934 _traverse_internals: _TraverseInternalsType = [
1935 ("key", InternalTraversal.dp_anon_name),
1936 ("type", InternalTraversal.dp_type),
1937 ("callable", InternalTraversal.dp_plain_dict),
1938 ("value", InternalTraversal.dp_plain_obj),
1939 ("literal_execute", InternalTraversal.dp_boolean),
1940 ]
1942 key: str
1943 _anon_map_key: Optional[str] = None
1944 type: TypeEngine[_T]
1945 value: Optional[_T]
1947 _is_crud = False
1948 _is_bind_parameter = True
1950 # bindparam implements its own _gen_cache_key() method however
1951 # we check subclasses for this flag, else no cache key is generated
1952 inherit_cache = True
1954 def __init__(
1955 self,
1956 key: Optional[str],
1957 value: Any = _NoArg.NO_ARG,
1958 type_: Optional[_TypeEngineArgument[_T]] = None,
1959 unique: bool = False,
1960 required: Union[bool, Literal[_NoArg.NO_ARG]] = _NoArg.NO_ARG,
1961 quote: Optional[bool] = None,
1962 callable_: Optional[Callable[[], Any]] = None,
1963 expanding: bool = False,
1964 isoutparam: bool = False,
1965 literal_execute: bool = False,
1966 _compared_to_operator: Optional[OperatorType] = None,
1967 _compared_to_type: Optional[TypeEngine[Any]] = None,
1968 _is_crud: bool = False,
1969 ):
1970 if required is _NoArg.NO_ARG:
1971 required = value is _NoArg.NO_ARG and callable_ is None
1972 if value is _NoArg.NO_ARG:
1973 value = None
1975 if quote is not None:
1976 key = quoted_name.construct(key, quote)
1978 if unique:
1979 self.key, self._anon_map_key = (
1980 _anonymous_label.safe_construct_with_key(
1981 id(self),
1982 (
1983 key
1984 if key is not None
1985 and not isinstance(key, _anonymous_label)
1986 else "param"
1987 ),
1988 sanitize_key=True,
1989 )
1990 )
1991 elif key:
1992 self.key = key
1993 else:
1994 self.key, self._anon_map_key = (
1995 _anonymous_label.safe_construct_with_key(id(self), "param")
1996 )
1998 # identifying key that won't change across
1999 # clones, used to identify the bind's logical
2000 # identity
2001 self._identifying_key = self.key
2003 # key that was passed in the first place, used to
2004 # generate new keys
2005 self._orig_key = key or "param"
2007 self.unique = unique
2008 self.value = value
2009 self.callable = callable_
2010 self.isoutparam = isoutparam
2011 self.required = required
2013 # indicate an "expanding" parameter; the compiler sets this
2014 # automatically in the compiler _render_in_expr_w_bindparam method
2015 # for an IN expression
2016 self.expanding = expanding
2018 # this is another hint to help w/ expanding and is typically
2019 # set in the compiler _render_in_expr_w_bindparam method for an
2020 # IN expression
2021 self.expand_op = None
2023 self.literal_execute = literal_execute
2024 if _is_crud:
2025 self._is_crud = True
2027 if type_ is None:
2028 if expanding:
2029 if value:
2030 check_value = value[0]
2031 else:
2032 check_value = type_api._NO_VALUE_IN_LIST
2033 else:
2034 check_value = value
2035 if _compared_to_type is not None:
2036 self.type = _compared_to_type.coerce_compared_value(
2037 _compared_to_operator, check_value
2038 )
2039 else:
2040 self.type = type_api._resolve_value_to_type(check_value)
2041 elif isinstance(type_, type):
2042 self.type = type_()
2043 elif is_tuple_type(type_):
2044 if value:
2045 if expanding:
2046 check_value = value[0]
2047 else:
2048 check_value = value
2049 cast("BindParameter[TupleAny]", self).type = (
2050 type_._resolve_values_to_types(check_value)
2051 )
2052 else:
2053 cast("BindParameter[TupleAny]", self).type = type_
2054 else:
2055 self.type = type_
2057 def _with_value(self, value, maintain_key=False, required=NO_ARG):
2058 """Return a copy of this :class:`.BindParameter` with the given value
2059 set.
2060 """
2061 cloned = self._clone(maintain_key=maintain_key)
2062 cloned.value = value
2063 cloned.callable = None
2064 cloned.required = required if required is not NO_ARG else self.required
2065 if cloned.type is type_api.NULLTYPE:
2066 cloned.type = type_api._resolve_value_to_type(value)
2067 return cloned
2069 @property
2070 def effective_value(self) -> Optional[_T]:
2071 """Return the value of this bound parameter,
2072 taking into account if the ``callable`` parameter
2073 was set.
2075 The ``callable`` value will be evaluated
2076 and returned if present, else ``value``.
2078 """
2079 if self.callable:
2080 # TODO: set up protocol for bind parameter callable
2081 return self.callable() # type: ignore
2082 else:
2083 return self.value
2085 def render_literal_execute(self) -> Self:
2086 """Produce a copy of this bound parameter that will enable the
2087 :paramref:`_sql.BindParameter.literal_execute` flag.
2089 The :paramref:`_sql.BindParameter.literal_execute` flag will
2090 have the effect of the parameter rendered in the compiled SQL
2091 string using ``[POSTCOMPILE]`` form, which is a special form that
2092 is converted to be a rendering of the literal value of the parameter
2093 at SQL execution time. The rationale is to support caching
2094 of SQL statement strings that can embed per-statement literal values,
2095 such as LIMIT and OFFSET parameters, in the final SQL string that
2096 is passed to the DBAPI. Dialects in particular may want to use
2097 this method within custom compilation schemes.
2099 .. versionadded:: 1.4.5
2101 .. seealso::
2103 :ref:`engine_thirdparty_caching`
2105 """
2106 c: Self = ClauseElement._clone(self)
2107 c.literal_execute = True
2108 return c
2110 def _negate_in_binary(self, negated_op, original_op):
2111 if self.expand_op is original_op:
2112 bind = self._clone()
2113 bind.expand_op = negated_op
2114 return bind
2115 else:
2116 return self
2118 def _with_binary_element_type(self, type_):
2119 c: Self = ClauseElement._clone(self) # type: ignore[assignment]
2120 c.type = type_
2121 return c
2123 def _clone(self, maintain_key: bool = False, **kw: Any) -> Self:
2124 c: Self = ClauseElement._clone(self, **kw)
2125 # ensure all the BindParameter objects stay in cloned set.
2126 # in #7823, we changed "clone" so that a clone only keeps a reference
2127 # to the "original" element, since for column correspondence, that's
2128 # all we need. However, for BindParam, _cloned_set is used by
2129 # the "cache key bind match" lookup, which means if any of those
2130 # interim BindParameter objects became part of a cache key in the
2131 # cache, we need it. So here, make sure all clones keep carrying
2132 # forward.
2133 c._cloned_set.update(self._cloned_set)
2134 if not maintain_key and self.unique:
2135 c.key, c._anon_map_key = _anonymous_label.safe_construct_with_key(
2136 id(c), c._orig_key or "param", sanitize_key=True
2137 )
2138 return c
2140 def _gen_cache_key(self, anon_map, bindparams):
2141 _gen_cache_ok = self.__class__.__dict__.get("inherit_cache", False)
2143 if not _gen_cache_ok:
2144 if anon_map is not None:
2145 anon_map[NO_CACHE] = True
2146 return None
2148 id_, found = anon_map.get_anon(self)
2149 if found:
2150 return (id_, self.__class__)
2152 if bindparams is not None:
2153 bindparams.append(self)
2155 return (
2156 id_,
2157 self.__class__,
2158 self.type._static_cache_key,
2159 (
2160 anon_map[self._anon_map_key]
2161 if self._anon_map_key is not None
2162 else self.key
2163 ),
2164 self.literal_execute,
2165 )
2167 def _convert_to_unique(self):
2168 if not self.unique:
2169 self.unique = True
2170 self.key, self._anon_map_key = (
2171 _anonymous_label.safe_construct_with_key(
2172 id(self), self._orig_key or "param", sanitize_key=True
2173 )
2174 )
2176 def __getstate__(self):
2177 """execute a deferred value for serialization purposes."""
2179 d = self.__dict__.copy()
2180 v = self.value
2181 if self.callable:
2182 v = self.callable()
2183 d["callable"] = None
2184 d["value"] = v
2185 return d
2187 def __setstate__(self, state):
2188 if state.get("unique", False):
2189 anon_and_key = _anonymous_label.safe_construct_with_key(
2190 id(self), state.get("_orig_key", "param"), sanitize_key=True
2191 )
2192 state["key"], state["_anon_map_key"] = anon_and_key
2193 self.__dict__.update(state)
2195 def __repr__(self):
2196 return "%s(%r, %r, type_=%r)" % (
2197 self.__class__.__name__,
2198 self.key,
2199 self.value,
2200 self.type,
2201 )
2204class TypeClause(DQLDMLClauseElement):
2205 """Handle a type keyword in a SQL statement.
2207 Used by the ``Case`` statement.
2209 """
2211 __visit_name__ = "typeclause"
2213 _traverse_internals: _TraverseInternalsType = [
2214 ("type", InternalTraversal.dp_type)
2215 ]
2217 def __init__(self, type_):
2218 self.type = type_
2221class TextClause(
2222 roles.DDLConstraintColumnRole,
2223 roles.DDLExpressionRole,
2224 roles.StatementOptionRole,
2225 roles.WhereHavingRole,
2226 roles.OrderByRole,
2227 roles.FromClauseRole,
2228 roles.SelectStatementRole,
2229 roles.InElementRole,
2230 Generative,
2231 Executable,
2232 DQLDMLClauseElement,
2233 roles.BinaryElementRole[Any],
2234 inspection.Inspectable["TextClause"],
2235):
2236 """Represent a literal SQL text fragment.
2238 E.g.::
2240 from sqlalchemy import text
2242 t = text("SELECT * FROM users")
2243 result = connection.execute(t)
2246 The :class:`_expression.TextClause` construct is produced using the
2247 :func:`_expression.text`
2248 function; see that function for full documentation.
2250 .. seealso::
2252 :func:`_expression.text`
2254 """
2256 __visit_name__ = "textclause"
2258 _traverse_internals: _TraverseInternalsType = [
2259 ("_bindparams", InternalTraversal.dp_string_clauseelement_dict),
2260 ("text", InternalTraversal.dp_string),
2261 ]
2263 _is_text_clause = True
2265 _is_textual = True
2267 _bind_params_regex = re.compile(r"(?<![:\w\x5c]):(\w+)(?!:)", re.UNICODE)
2268 _is_implicitly_boolean = False
2270 _render_label_in_columns_clause = False
2272 _omit_from_statements = False
2274 _is_collection_aggregate = False
2276 @property
2277 def _hide_froms(self) -> Iterable[FromClause]:
2278 return ()
2280 def __and__(self, other):
2281 # support use in select.where(), query.filter()
2282 return and_(self, other)
2284 @property
2285 def _select_iterable(self) -> _SelectIterable:
2286 return (self,)
2288 # help in those cases where text() is
2289 # interpreted in a column expression situation
2290 key: Optional[str] = None
2291 _label: Optional[str] = None
2293 _allow_label_resolve = False
2295 @property
2296 def _is_star(self):
2297 return self.text == "*"
2299 def __init__(self, text: str):
2300 self._bindparams: Dict[str, BindParameter[Any]] = {}
2302 def repl(m):
2303 self._bindparams[m.group(1)] = BindParameter(m.group(1))
2304 return ":%s" % m.group(1)
2306 # scan the string and search for bind parameter names, add them
2307 # to the list of bindparams
2308 self.text = self._bind_params_regex.sub(repl, text)
2310 @_generative
2311 def bindparams(
2312 self,
2313 *binds: BindParameter[Any],
2314 **names_to_values: Any,
2315 ) -> Self:
2316 """Establish the values and/or types of bound parameters within
2317 this :class:`_expression.TextClause` construct.
2319 Given a text construct such as::
2321 from sqlalchemy import text
2322 stmt = text("SELECT id, name FROM user WHERE name=:name "
2323 "AND timestamp=:timestamp")
2325 the :meth:`_expression.TextClause.bindparams`
2326 method can be used to establish
2327 the initial value of ``:name`` and ``:timestamp``,
2328 using simple keyword arguments::
2330 stmt = stmt.bindparams(name='jack',
2331 timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5))
2333 Where above, new :class:`.BindParameter` objects
2334 will be generated with the names ``name`` and ``timestamp``, and
2335 values of ``jack`` and ``datetime.datetime(2012, 10, 8, 15, 12, 5)``,
2336 respectively. The types will be
2337 inferred from the values given, in this case :class:`.String` and
2338 :class:`.DateTime`.
2340 When specific typing behavior is needed, the positional ``*binds``
2341 argument can be used in which to specify :func:`.bindparam` constructs
2342 directly. These constructs must include at least the ``key``
2343 argument, then an optional value and type::
2345 from sqlalchemy import bindparam
2346 stmt = stmt.bindparams(
2347 bindparam('name', value='jack', type_=String),
2348 bindparam('timestamp', type_=DateTime)
2349 )
2351 Above, we specified the type of :class:`.DateTime` for the
2352 ``timestamp`` bind, and the type of :class:`.String` for the ``name``
2353 bind. In the case of ``name`` we also set the default value of
2354 ``"jack"``.
2356 Additional bound parameters can be supplied at statement execution
2357 time, e.g.::
2359 result = connection.execute(stmt,
2360 timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5))
2362 The :meth:`_expression.TextClause.bindparams`
2363 method can be called repeatedly,
2364 where it will re-use existing :class:`.BindParameter` objects to add
2365 new information. For example, we can call
2366 :meth:`_expression.TextClause.bindparams`
2367 first with typing information, and a
2368 second time with value information, and it will be combined::
2370 stmt = text("SELECT id, name FROM user WHERE name=:name "
2371 "AND timestamp=:timestamp")
2372 stmt = stmt.bindparams(
2373 bindparam('name', type_=String),
2374 bindparam('timestamp', type_=DateTime)
2375 )
2376 stmt = stmt.bindparams(
2377 name='jack',
2378 timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2379 )
2381 The :meth:`_expression.TextClause.bindparams`
2382 method also supports the concept of
2383 **unique** bound parameters. These are parameters that are
2384 "uniquified" on name at statement compilation time, so that multiple
2385 :func:`_expression.text`
2386 constructs may be combined together without the names
2387 conflicting. To use this feature, specify the
2388 :paramref:`.BindParameter.unique` flag on each :func:`.bindparam`
2389 object::
2391 stmt1 = text("select id from table where name=:name").bindparams(
2392 bindparam("name", value='name1', unique=True)
2393 )
2394 stmt2 = text("select id from table where name=:name").bindparams(
2395 bindparam("name", value='name2', unique=True)
2396 )
2398 union = union_all(
2399 stmt1.columns(column("id")),
2400 stmt2.columns(column("id"))
2401 )
2403 The above statement will render as::
2405 select id from table where name=:name_1
2406 UNION ALL select id from table where name=:name_2
2408 .. versionadded:: 1.3.11 Added support for the
2409 :paramref:`.BindParameter.unique` flag to work with
2410 :func:`_expression.text`
2411 constructs.
2413 """
2414 self._bindparams = new_params = self._bindparams.copy()
2416 for bind in binds:
2417 try:
2418 # the regex used for text() currently will not match
2419 # a unique/anonymous key in any case, so use the _orig_key
2420 # so that a text() construct can support unique parameters
2421 existing = new_params[bind._orig_key]
2422 except KeyError as err:
2423 raise exc.ArgumentError(
2424 "This text() construct doesn't define a "
2425 "bound parameter named %r" % bind._orig_key
2426 ) from err
2427 else:
2428 new_params[existing._orig_key] = bind
2430 for key, value in names_to_values.items():
2431 try:
2432 existing = new_params[key]
2433 except KeyError as err:
2434 raise exc.ArgumentError(
2435 "This text() construct doesn't define a "
2436 "bound parameter named %r" % key
2437 ) from err
2438 else:
2439 new_params[key] = existing._with_value(value, required=False)
2440 return self
2442 @util.preload_module("sqlalchemy.sql.selectable")
2443 def columns(
2444 self, *cols: _ColumnExpressionArgument[Any], **types: TypeEngine[Any]
2445 ) -> TextualSelect:
2446 r"""Turn this :class:`_expression.TextClause` object into a
2447 :class:`_expression.TextualSelect`
2448 object that serves the same role as a SELECT
2449 statement.
2451 The :class:`_expression.TextualSelect` is part of the
2452 :class:`_expression.SelectBase`
2453 hierarchy and can be embedded into another statement by using the
2454 :meth:`_expression.TextualSelect.subquery` method to produce a
2455 :class:`.Subquery`
2456 object, which can then be SELECTed from.
2458 This function essentially bridges the gap between an entirely
2459 textual SELECT statement and the SQL expression language concept
2460 of a "selectable"::
2462 from sqlalchemy.sql import column, text
2464 stmt = text("SELECT id, name FROM some_table")
2465 stmt = stmt.columns(column('id'), column('name')).subquery('st')
2467 stmt = select(mytable).\
2468 select_from(
2469 mytable.join(stmt, mytable.c.name == stmt.c.name)
2470 ).where(stmt.c.id > 5)
2472 Above, we pass a series of :func:`_expression.column` elements to the
2473 :meth:`_expression.TextClause.columns` method positionally. These
2474 :func:`_expression.column`
2475 elements now become first class elements upon the
2476 :attr:`_expression.TextualSelect.selected_columns` column collection,
2477 which then
2478 become part of the :attr:`.Subquery.c` collection after
2479 :meth:`_expression.TextualSelect.subquery` is invoked.
2481 The column expressions we pass to
2482 :meth:`_expression.TextClause.columns` may
2483 also be typed; when we do so, these :class:`.TypeEngine` objects become
2484 the effective return type of the column, so that SQLAlchemy's
2485 result-set-processing systems may be used on the return values.
2486 This is often needed for types such as date or boolean types, as well
2487 as for unicode processing on some dialect configurations::
2489 stmt = text("SELECT id, name, timestamp FROM some_table")
2490 stmt = stmt.columns(
2491 column('id', Integer),
2492 column('name', Unicode),
2493 column('timestamp', DateTime)
2494 )
2496 for id, name, timestamp in connection.execute(stmt):
2497 print(id, name, timestamp)
2499 As a shortcut to the above syntax, keyword arguments referring to
2500 types alone may be used, if only type conversion is needed::
2502 stmt = text("SELECT id, name, timestamp FROM some_table")
2503 stmt = stmt.columns(
2504 id=Integer,
2505 name=Unicode,
2506 timestamp=DateTime
2507 )
2509 for id, name, timestamp in connection.execute(stmt):
2510 print(id, name, timestamp)
2512 The positional form of :meth:`_expression.TextClause.columns`
2513 also provides the
2514 unique feature of **positional column targeting**, which is
2515 particularly useful when using the ORM with complex textual queries. If
2516 we specify the columns from our model to
2517 :meth:`_expression.TextClause.columns`,
2518 the result set will match to those columns positionally, meaning the
2519 name or origin of the column in the textual SQL doesn't matter::
2521 stmt = text("SELECT users.id, addresses.id, users.id, "
2522 "users.name, addresses.email_address AS email "
2523 "FROM users JOIN addresses ON users.id=addresses.user_id "
2524 "WHERE users.id = 1").columns(
2525 User.id,
2526 Address.id,
2527 Address.user_id,
2528 User.name,
2529 Address.email_address
2530 )
2532 query = session.query(User).from_statement(stmt).options(
2533 contains_eager(User.addresses))
2535 The :meth:`_expression.TextClause.columns` method provides a direct
2536 route to calling :meth:`_expression.FromClause.subquery` as well as
2537 :meth:`_expression.SelectBase.cte`
2538 against a textual SELECT statement::
2540 stmt = stmt.columns(id=Integer, name=String).cte('st')
2542 stmt = select(sometable).where(sometable.c.id == stmt.c.id)
2544 :param \*cols: A series of :class:`_expression.ColumnElement` objects,
2545 typically
2546 :class:`_schema.Column` objects from a :class:`_schema.Table`
2547 or ORM level
2548 column-mapped attributes, representing a set of columns that this
2549 textual string will SELECT from.
2551 :param \**types: A mapping of string names to :class:`.TypeEngine`
2552 type objects indicating the datatypes to use for names that are
2553 SELECTed from the textual string. Prefer to use the ``*cols``
2554 argument as it also indicates positional ordering.
2556 """
2557 selectable = util.preloaded.sql_selectable
2559 input_cols: List[NamedColumn[Any]] = [
2560 coercions.expect(roles.LabeledColumnExprRole, col) for col in cols
2561 ]
2563 positional_input_cols = [
2564 (
2565 ColumnClause(col.key, types.pop(col.key))
2566 if col.key in types
2567 else col
2568 )
2569 for col in input_cols
2570 ]
2571 keyed_input_cols: List[NamedColumn[Any]] = [
2572 ColumnClause(key, type_) for key, type_ in types.items()
2573 ]
2575 elem = selectable.TextualSelect.__new__(selectable.TextualSelect)
2576 elem._init(
2577 self,
2578 positional_input_cols + keyed_input_cols,
2579 positional=bool(positional_input_cols) and not keyed_input_cols,
2580 )
2581 return elem
2583 @property
2584 def type(self) -> TypeEngine[Any]:
2585 return type_api.NULLTYPE
2587 @property
2588 def comparator(self):
2589 # TODO: this seems wrong, it seems like we might not
2590 # be using this method.
2591 return self.type.comparator_factory(self) # type: ignore
2593 def self_group(
2594 self, against: Optional[OperatorType] = None
2595 ) -> Union[Self, Grouping[Any]]:
2596 if against is operators.in_op:
2597 return Grouping(self)
2598 else:
2599 return self
2602class Null(SingletonConstant, roles.ConstExprRole[None], ColumnElement[None]):
2603 """Represent the NULL keyword in a SQL statement.
2605 :class:`.Null` is accessed as a constant via the
2606 :func:`.null` function.
2608 """
2610 __visit_name__ = "null"
2612 _traverse_internals: _TraverseInternalsType = []
2613 _singleton: Null
2615 if not TYPE_CHECKING:
2617 @util.memoized_property
2618 def type(self) -> TypeEngine[_T]: # noqa: A001
2619 return type_api.NULLTYPE
2621 @classmethod
2622 def _instance(cls) -> Null:
2623 """Return a constant :class:`.Null` construct."""
2625 return Null._singleton
2628Null._create_singleton()
2631class False_(
2632 SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]
2633):
2634 """Represent the ``false`` keyword, or equivalent, in a SQL statement.
2636 :class:`.False_` is accessed as a constant via the
2637 :func:`.false` function.
2639 """
2641 __visit_name__ = "false"
2642 _traverse_internals: _TraverseInternalsType = []
2643 _singleton: False_
2645 if not TYPE_CHECKING:
2647 @util.memoized_property
2648 def type(self) -> TypeEngine[_T]: # noqa: A001
2649 return type_api.BOOLEANTYPE
2651 def _negate(self) -> True_:
2652 return True_._singleton
2654 @classmethod
2655 def _instance(cls) -> False_:
2656 return False_._singleton
2659False_._create_singleton()
2662class True_(SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]):
2663 """Represent the ``true`` keyword, or equivalent, in a SQL statement.
2665 :class:`.True_` is accessed as a constant via the
2666 :func:`.true` function.
2668 """
2670 __visit_name__ = "true"
2672 _traverse_internals: _TraverseInternalsType = []
2673 _singleton: True_
2675 if not TYPE_CHECKING:
2677 @util.memoized_property
2678 def type(self) -> TypeEngine[_T]: # noqa: A001
2679 return type_api.BOOLEANTYPE
2681 def _negate(self) -> False_:
2682 return False_._singleton
2684 @classmethod
2685 def _ifnone(
2686 cls, other: Optional[ColumnElement[Any]]
2687 ) -> ColumnElement[Any]:
2688 if other is None:
2689 return cls._instance()
2690 else:
2691 return other
2693 @classmethod
2694 def _instance(cls) -> True_:
2695 return True_._singleton
2698True_._create_singleton()
2701class ClauseList(
2702 roles.InElementRole,
2703 roles.OrderByRole,
2704 roles.ColumnsClauseRole,
2705 roles.DMLColumnRole,
2706 DQLDMLClauseElement,
2707):
2708 """Describe a list of clauses, separated by an operator.
2710 By default, is comma-separated, such as a column listing.
2712 """
2714 __visit_name__ = "clauselist"
2716 # this is used only by the ORM in a legacy use case for
2717 # composite attributes
2718 _is_clause_list = True
2720 _traverse_internals: _TraverseInternalsType = [
2721 ("clauses", InternalTraversal.dp_clauseelement_list),
2722 ("operator", InternalTraversal.dp_operator),
2723 ]
2725 clauses: List[ColumnElement[Any]]
2727 def __init__(
2728 self,
2729 *clauses: _ColumnExpressionArgument[Any],
2730 operator: OperatorType = operators.comma_op,
2731 group: bool = True,
2732 group_contents: bool = True,
2733 _literal_as_text_role: Type[roles.SQLRole] = roles.WhereHavingRole,
2734 ):
2735 self.operator = operator
2736 self.group = group
2737 self.group_contents = group_contents
2738 clauses_iterator: Iterable[_ColumnExpressionArgument[Any]] = clauses
2739 text_converter_role: Type[roles.SQLRole] = _literal_as_text_role
2740 self._text_converter_role = text_converter_role
2742 if self.group_contents:
2743 self.clauses = [
2744 coercions.expect(
2745 text_converter_role, clause, apply_propagate_attrs=self
2746 ).self_group(against=self.operator)
2747 for clause in clauses_iterator
2748 ]
2749 else:
2750 self.clauses = [
2751 coercions.expect(
2752 text_converter_role, clause, apply_propagate_attrs=self
2753 )
2754 for clause in clauses_iterator
2755 ]
2756 self._is_implicitly_boolean = operators.is_boolean(self.operator)
2758 @classmethod
2759 def _construct_raw(
2760 cls,
2761 operator: OperatorType,
2762 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
2763 ) -> ClauseList:
2764 self = cls.__new__(cls)
2765 self.clauses = list(clauses) if clauses else []
2766 self.group = True
2767 self.operator = operator
2768 self.group_contents = True
2769 self._is_implicitly_boolean = False
2770 return self
2772 def __iter__(self) -> Iterator[ColumnElement[Any]]:
2773 return iter(self.clauses)
2775 def __len__(self) -> int:
2776 return len(self.clauses)
2778 @property
2779 def _select_iterable(self) -> _SelectIterable:
2780 return itertools.chain.from_iterable(
2781 [elem._select_iterable for elem in self.clauses]
2782 )
2784 def append(self, clause):
2785 if self.group_contents:
2786 self.clauses.append(
2787 coercions.expect(self._text_converter_role, clause).self_group(
2788 against=self.operator
2789 )
2790 )
2791 else:
2792 self.clauses.append(
2793 coercions.expect(self._text_converter_role, clause)
2794 )
2796 @util.ro_non_memoized_property
2797 def _from_objects(self) -> List[FromClause]:
2798 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
2800 def self_group(
2801 self, against: Optional[OperatorType] = None
2802 ) -> Union[Self, Grouping[Any]]:
2803 if self.group and operators.is_precedent(self.operator, against):
2804 return Grouping(self)
2805 else:
2806 return self
2809class OperatorExpression(ColumnElement[_T]):
2810 """base for expressions that contain an operator and operands
2812 .. versionadded:: 2.0
2814 """
2816 operator: OperatorType
2817 type: TypeEngine[_T]
2819 group: bool = True
2821 @property
2822 def is_comparison(self):
2823 return operators.is_comparison(self.operator)
2825 def self_group(
2826 self, against: Optional[OperatorType] = None
2827 ) -> Union[Self, Grouping[_T]]:
2828 if (
2829 self.group
2830 and operators.is_precedent(self.operator, against)
2831 or (
2832 # a negate against a non-boolean operator
2833 # doesn't make too much sense but we should
2834 # group for that
2835 against is operators.inv
2836 and not operators.is_boolean(self.operator)
2837 )
2838 ):
2839 return Grouping(self)
2840 else:
2841 return self
2843 @property
2844 def _flattened_operator_clauses(
2845 self,
2846 ) -> typing_Tuple[ColumnElement[Any], ...]:
2847 raise NotImplementedError()
2849 @classmethod
2850 def _construct_for_op(
2851 cls,
2852 left: ColumnElement[Any],
2853 right: ColumnElement[Any],
2854 op: OperatorType,
2855 *,
2856 type_: TypeEngine[_T],
2857 negate: Optional[OperatorType] = None,
2858 modifiers: Optional[Mapping[str, Any]] = None,
2859 ) -> OperatorExpression[_T]:
2860 if operators.is_associative(op):
2861 assert (
2862 negate is None
2863 ), f"negate not supported for associative operator {op}"
2865 multi = False
2866 if getattr(
2867 left, "operator", None
2868 ) is op and type_._compare_type_affinity(left.type):
2869 multi = True
2870 left_flattened = left._flattened_operator_clauses
2871 else:
2872 left_flattened = (left,)
2874 if getattr(
2875 right, "operator", None
2876 ) is op and type_._compare_type_affinity(right.type):
2877 multi = True
2878 right_flattened = right._flattened_operator_clauses
2879 else:
2880 right_flattened = (right,)
2882 if multi:
2883 return ExpressionClauseList._construct_for_list(
2884 op,
2885 type_,
2886 *(left_flattened + right_flattened),
2887 )
2889 if right._is_collection_aggregate:
2890 negate = None
2892 return BinaryExpression(
2893 left, right, op, type_=type_, negate=negate, modifiers=modifiers
2894 )
2897class ExpressionClauseList(OperatorExpression[_T]):
2898 """Describe a list of clauses, separated by an operator,
2899 in a column expression context.
2901 :class:`.ExpressionClauseList` differs from :class:`.ClauseList` in that
2902 it represents a column-oriented DQL expression only, not an open ended
2903 list of anything comma separated.
2905 .. versionadded:: 2.0
2907 """
2909 __visit_name__ = "expression_clauselist"
2911 _traverse_internals: _TraverseInternalsType = [
2912 ("clauses", InternalTraversal.dp_clauseelement_tuple),
2913 ("operator", InternalTraversal.dp_operator),
2914 ]
2916 clauses: typing_Tuple[ColumnElement[Any], ...]
2918 group: bool
2920 def __init__(
2921 self,
2922 operator: OperatorType,
2923 *clauses: _ColumnExpressionArgument[Any],
2924 type_: Optional[_TypeEngineArgument[_T]] = None,
2925 ):
2926 self.operator = operator
2928 self.clauses = tuple(
2929 coercions.expect(
2930 roles.ExpressionElementRole, clause, apply_propagate_attrs=self
2931 )
2932 for clause in clauses
2933 )
2934 self._is_implicitly_boolean = operators.is_boolean(self.operator)
2935 self.type = type_api.to_instance(type_) # type: ignore
2937 @property
2938 def _flattened_operator_clauses(
2939 self,
2940 ) -> typing_Tuple[ColumnElement[Any], ...]:
2941 return self.clauses
2943 def __iter__(self) -> Iterator[ColumnElement[Any]]:
2944 return iter(self.clauses)
2946 def __len__(self) -> int:
2947 return len(self.clauses)
2949 @property
2950 def _select_iterable(self) -> _SelectIterable:
2951 return (self,)
2953 @util.ro_non_memoized_property
2954 def _from_objects(self) -> List[FromClause]:
2955 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
2957 def _append_inplace(self, clause: ColumnElement[Any]) -> None:
2958 self.clauses += (clause,)
2960 @classmethod
2961 def _construct_for_list(
2962 cls,
2963 operator: OperatorType,
2964 type_: TypeEngine[_T],
2965 *clauses: ColumnElement[Any],
2966 group: bool = True,
2967 ) -> ExpressionClauseList[_T]:
2968 self = cls.__new__(cls)
2969 self.group = group
2970 if group:
2971 self.clauses = tuple(
2972 c.self_group(against=operator) for c in clauses
2973 )
2974 else:
2975 self.clauses = clauses
2976 self.operator = operator
2977 self.type = type_
2978 return self
2980 def _negate(self) -> Any:
2981 grouped = self.self_group(against=operators.inv)
2982 assert isinstance(grouped, ColumnElement)
2983 return UnaryExpression(
2984 grouped, operator=operators.inv, wraps_column_expression=True
2985 )
2988class BooleanClauseList(ExpressionClauseList[bool]):
2989 __visit_name__ = "expression_clauselist"
2990 inherit_cache = True
2992 def __init__(self, *arg, **kw):
2993 raise NotImplementedError(
2994 "BooleanClauseList has a private constructor"
2995 )
2997 @classmethod
2998 def _process_clauses_for_boolean(
2999 cls,
3000 operator: OperatorType,
3001 continue_on: Any,
3002 skip_on: Any,
3003 clauses: Iterable[ColumnElement[Any]],
3004 ) -> typing_Tuple[int, List[ColumnElement[Any]]]:
3005 has_continue_on = None
3007 convert_clauses = []
3009 against = operators._asbool
3010 lcc = 0
3012 for clause in clauses:
3013 if clause is continue_on:
3014 # instance of continue_on, like and_(x, y, True, z), store it
3015 # if we didn't find one already, we will use it if there
3016 # are no other expressions here.
3017 has_continue_on = clause
3018 elif clause is skip_on:
3019 # instance of skip_on, e.g. and_(x, y, False, z), cancels
3020 # the rest out
3021 convert_clauses = [clause]
3022 lcc = 1
3023 break
3024 else:
3025 if not lcc:
3026 lcc = 1
3027 else:
3028 against = operator
3029 # technically this would be len(convert_clauses) + 1
3030 # however this only needs to indicate "greater than one"
3031 lcc = 2
3032 convert_clauses.append(clause)
3034 if not convert_clauses and has_continue_on is not None:
3035 convert_clauses = [has_continue_on]
3036 lcc = 1
3038 return lcc, [c.self_group(against=against) for c in convert_clauses]
3040 @classmethod
3041 def _construct(
3042 cls,
3043 operator: OperatorType,
3044 continue_on: Any,
3045 skip_on: Any,
3046 initial_clause: Any = _NoArg.NO_ARG,
3047 *clauses: Any,
3048 **kw: Any,
3049 ) -> ColumnElement[Any]:
3050 if initial_clause is _NoArg.NO_ARG:
3051 # no elements period. deprecated use case. return an empty
3052 # ClauseList construct that generates nothing unless it has
3053 # elements added to it.
3054 name = operator.__name__
3056 util.warn_deprecated(
3057 f"Invoking {name}() without arguments is deprecated, and "
3058 f"will be disallowed in a future release. For an empty "
3059 f"""{name}() construct, use '{name}({
3060 'true()' if continue_on is True_._singleton else 'false()'
3061 }, *args)' """
3062 f"""or '{name}({
3063 'True' if continue_on is True_._singleton else 'False'
3064 }, *args)'.""",
3065 version="1.4",
3066 )
3067 return cls._construct_raw(operator)
3069 lcc, convert_clauses = cls._process_clauses_for_boolean(
3070 operator,
3071 continue_on,
3072 skip_on,
3073 [
3074 coercions.expect(roles.WhereHavingRole, clause)
3075 for clause in util.coerce_generator_arg(
3076 (initial_clause,) + clauses
3077 )
3078 ],
3079 )
3081 if lcc > 1:
3082 # multiple elements. Return regular BooleanClauseList
3083 # which will link elements against the operator.
3085 flattened_clauses = itertools.chain.from_iterable(
3086 (
3087 (c for c in to_flat._flattened_operator_clauses)
3088 if getattr(to_flat, "operator", None) is operator
3089 else (to_flat,)
3090 )
3091 for to_flat in convert_clauses
3092 )
3094 return cls._construct_raw(operator, flattened_clauses) # type: ignore # noqa: E501
3095 else:
3096 assert lcc
3097 # just one element. return it as a single boolean element,
3098 # not a list and discard the operator.
3099 return convert_clauses[0]
3101 @classmethod
3102 def _construct_for_whereclause(
3103 cls, clauses: Iterable[ColumnElement[Any]]
3104 ) -> Optional[ColumnElement[bool]]:
3105 operator, continue_on, skip_on = (
3106 operators.and_,
3107 True_._singleton,
3108 False_._singleton,
3109 )
3111 lcc, convert_clauses = cls._process_clauses_for_boolean(
3112 operator,
3113 continue_on,
3114 skip_on,
3115 clauses, # these are assumed to be coerced already
3116 )
3118 if lcc > 1:
3119 # multiple elements. Return regular BooleanClauseList
3120 # which will link elements against the operator.
3121 return cls._construct_raw(operator, convert_clauses)
3122 elif lcc == 1:
3123 # just one element. return it as a single boolean element,
3124 # not a list and discard the operator.
3125 return convert_clauses[0]
3126 else:
3127 return None
3129 @classmethod
3130 def _construct_raw(
3131 cls,
3132 operator: OperatorType,
3133 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
3134 ) -> BooleanClauseList:
3135 self = cls.__new__(cls)
3136 self.clauses = tuple(clauses) if clauses else ()
3137 self.group = True
3138 self.operator = operator
3139 self.type = type_api.BOOLEANTYPE
3140 self._is_implicitly_boolean = True
3141 return self
3143 @classmethod
3144 def and_(
3145 cls,
3146 initial_clause: Union[
3147 Literal[True], _ColumnExpressionArgument[bool], _NoArg
3148 ] = _NoArg.NO_ARG,
3149 *clauses: _ColumnExpressionArgument[bool],
3150 ) -> ColumnElement[bool]:
3151 r"""Produce a conjunction of expressions joined by ``AND``.
3153 See :func:`_sql.and_` for full documentation.
3154 """
3155 return cls._construct(
3156 operators.and_,
3157 True_._singleton,
3158 False_._singleton,
3159 initial_clause,
3160 *clauses,
3161 )
3163 @classmethod
3164 def or_(
3165 cls,
3166 initial_clause: Union[
3167 Literal[False], _ColumnExpressionArgument[bool], _NoArg
3168 ] = _NoArg.NO_ARG,
3169 *clauses: _ColumnExpressionArgument[bool],
3170 ) -> ColumnElement[bool]:
3171 """Produce a conjunction of expressions joined by ``OR``.
3173 See :func:`_sql.or_` for full documentation.
3174 """
3175 return cls._construct(
3176 operators.or_,
3177 False_._singleton,
3178 True_._singleton,
3179 initial_clause,
3180 *clauses,
3181 )
3183 @property
3184 def _select_iterable(self) -> _SelectIterable:
3185 return (self,)
3187 def self_group(
3188 self, against: Optional[OperatorType] = None
3189 ) -> Union[Self, Grouping[bool]]:
3190 if not self.clauses:
3191 return self
3192 else:
3193 return super().self_group(against=against)
3196and_ = BooleanClauseList.and_
3197or_ = BooleanClauseList.or_
3200class Tuple(ClauseList, ColumnElement[TupleAny]):
3201 """Represent a SQL tuple."""
3203 __visit_name__ = "tuple"
3205 _traverse_internals: _TraverseInternalsType = (
3206 ClauseList._traverse_internals + []
3207 )
3209 type: TupleType
3211 @util.preload_module("sqlalchemy.sql.sqltypes")
3212 def __init__(
3213 self,
3214 *clauses: _ColumnExpressionArgument[Any],
3215 types: Optional[Sequence[_TypeEngineArgument[Any]]] = None,
3216 ):
3217 sqltypes = util.preloaded.sql_sqltypes
3219 if types is None:
3220 init_clauses: List[ColumnElement[Any]] = [
3221 coercions.expect(roles.ExpressionElementRole, c)
3222 for c in clauses
3223 ]
3224 else:
3225 if len(types) != len(clauses):
3226 raise exc.ArgumentError(
3227 "Wrong number of elements for %d-tuple: %r "
3228 % (len(types), clauses)
3229 )
3230 init_clauses = [
3231 coercions.expect(
3232 roles.ExpressionElementRole,
3233 c,
3234 type_=typ if not typ._isnull else None,
3235 )
3236 for typ, c in zip(types, clauses)
3237 ]
3239 self.type = sqltypes.TupleType(*[arg.type for arg in init_clauses])
3240 super().__init__(*init_clauses)
3242 @property
3243 def _select_iterable(self) -> _SelectIterable:
3244 return (self,)
3246 def _bind_param(self, operator, obj, type_=None, expanding=False):
3247 if expanding:
3248 return BindParameter(
3249 None,
3250 value=obj,
3251 _compared_to_operator=operator,
3252 unique=True,
3253 expanding=True,
3254 type_=type_,
3255 _compared_to_type=self.type,
3256 )
3257 else:
3258 return Tuple(
3259 *[
3260 BindParameter(
3261 None,
3262 o,
3263 _compared_to_operator=operator,
3264 _compared_to_type=compared_to_type,
3265 unique=True,
3266 type_=type_,
3267 )
3268 for o, compared_to_type in zip(obj, self.type.types)
3269 ]
3270 )
3272 def self_group(self, against: Optional[OperatorType] = None) -> Self:
3273 # Tuple is parenthesized by definition.
3274 return self
3277class Case(ColumnElement[_T]):
3278 """Represent a ``CASE`` expression.
3280 :class:`.Case` is produced using the :func:`.case` factory function,
3281 as in::
3283 from sqlalchemy import case
3285 stmt = select(users_table).\
3286 where(
3287 case(
3288 (users_table.c.name == 'wendy', 'W'),
3289 (users_table.c.name == 'jack', 'J'),
3290 else_='E'
3291 )
3292 )
3294 Details on :class:`.Case` usage is at :func:`.case`.
3296 .. seealso::
3298 :func:`.case`
3300 """
3302 __visit_name__ = "case"
3304 _traverse_internals: _TraverseInternalsType = [
3305 ("value", InternalTraversal.dp_clauseelement),
3306 ("whens", InternalTraversal.dp_clauseelement_tuples),
3307 ("else_", InternalTraversal.dp_clauseelement),
3308 ]
3310 # for case(), the type is derived from the whens. so for the moment
3311 # users would have to cast() the case to get a specific type
3313 whens: List[typing_Tuple[ColumnElement[bool], ColumnElement[_T]]]
3314 else_: Optional[ColumnElement[_T]]
3315 value: Optional[ColumnElement[Any]]
3317 def __init__(
3318 self,
3319 *whens: Union[
3320 typing_Tuple[_ColumnExpressionArgument[bool], Any],
3321 Mapping[Any, Any],
3322 ],
3323 value: Optional[Any] = None,
3324 else_: Optional[Any] = None,
3325 ):
3326 new_whens: Iterable[Any] = coercions._expression_collection_was_a_list(
3327 "whens", "case", whens
3328 )
3329 try:
3330 new_whens = util.dictlike_iteritems(new_whens)
3331 except TypeError:
3332 pass
3334 self.whens = [
3335 (
3336 coercions.expect(
3337 roles.ExpressionElementRole,
3338 c,
3339 apply_propagate_attrs=self,
3340 ).self_group(),
3341 coercions.expect(roles.ExpressionElementRole, r),
3342 )
3343 for (c, r) in new_whens
3344 ]
3346 if value is None:
3347 self.value = None
3348 else:
3349 self.value = coercions.expect(roles.ExpressionElementRole, value)
3351 if else_ is not None:
3352 self.else_ = coercions.expect(roles.ExpressionElementRole, else_)
3353 else:
3354 self.else_ = None
3356 type_ = next(
3357 (
3358 then.type
3359 # Iterate `whens` in reverse to match previous behaviour
3360 # where type of final element took priority
3361 for *_, then in reversed(self.whens)
3362 if not then.type._isnull
3363 ),
3364 self.else_.type if self.else_ is not None else type_api.NULLTYPE,
3365 )
3366 self.type = cast(_T, type_)
3368 @util.ro_non_memoized_property
3369 def _from_objects(self) -> List[FromClause]:
3370 return list(
3371 itertools.chain(*[x._from_objects for x in self.get_children()])
3372 )
3375class Cast(WrapsColumnExpression[_T]):
3376 """Represent a ``CAST`` expression.
3378 :class:`.Cast` is produced using the :func:`.cast` factory function,
3379 as in::
3381 from sqlalchemy import cast, Numeric
3383 stmt = select(cast(product_table.c.unit_price, Numeric(10, 4)))
3385 Details on :class:`.Cast` usage is at :func:`.cast`.
3387 .. seealso::
3389 :ref:`tutorial_casts`
3391 :func:`.cast`
3393 :func:`.try_cast`
3395 :func:`.type_coerce` - an alternative to CAST that coerces the type
3396 on the Python side only, which is often sufficient to generate the
3397 correct SQL and data coercion.
3399 """
3401 __visit_name__ = "cast"
3403 _traverse_internals: _TraverseInternalsType = [
3404 ("clause", InternalTraversal.dp_clauseelement),
3405 ("type", InternalTraversal.dp_type),
3406 ]
3408 clause: ColumnElement[Any]
3409 type: TypeEngine[_T]
3410 typeclause: TypeClause
3412 def __init__(
3413 self,
3414 expression: _ColumnExpressionArgument[Any],
3415 type_: _TypeEngineArgument[_T],
3416 ):
3417 self.type = type_api.to_instance(type_)
3418 self.clause = coercions.expect(
3419 roles.ExpressionElementRole,
3420 expression,
3421 type_=self.type,
3422 apply_propagate_attrs=self,
3423 )
3424 self.typeclause = TypeClause(self.type)
3426 @util.ro_non_memoized_property
3427 def _from_objects(self) -> List[FromClause]:
3428 return self.clause._from_objects
3430 @property
3431 def wrapped_column_expression(self):
3432 return self.clause
3435class TryCast(Cast[_T]):
3436 """Represent a TRY_CAST expression.
3438 Details on :class:`.TryCast` usage is at :func:`.try_cast`.
3440 .. seealso::
3442 :func:`.try_cast`
3444 :ref:`tutorial_casts`
3445 """
3447 __visit_name__ = "try_cast"
3448 inherit_cache = True
3451class TypeCoerce(WrapsColumnExpression[_T]):
3452 """Represent a Python-side type-coercion wrapper.
3454 :class:`.TypeCoerce` supplies the :func:`_expression.type_coerce`
3455 function; see that function for usage details.
3457 .. seealso::
3459 :func:`_expression.type_coerce`
3461 :func:`.cast`
3463 """
3465 __visit_name__ = "type_coerce"
3467 _traverse_internals: _TraverseInternalsType = [
3468 ("clause", InternalTraversal.dp_clauseelement),
3469 ("type", InternalTraversal.dp_type),
3470 ]
3472 clause: ColumnElement[Any]
3473 type: TypeEngine[_T]
3475 def __init__(
3476 self,
3477 expression: _ColumnExpressionArgument[Any],
3478 type_: _TypeEngineArgument[_T],
3479 ):
3480 self.type = type_api.to_instance(type_)
3481 self.clause = coercions.expect(
3482 roles.ExpressionElementRole,
3483 expression,
3484 type_=self.type,
3485 apply_propagate_attrs=self,
3486 )
3488 @util.ro_non_memoized_property
3489 def _from_objects(self) -> List[FromClause]:
3490 return self.clause._from_objects
3492 @HasMemoized.memoized_attribute
3493 def typed_expression(self):
3494 if isinstance(self.clause, BindParameter):
3495 bp = self.clause._clone()
3496 bp.type = self.type
3497 return bp
3498 else:
3499 return self.clause
3501 @property
3502 def wrapped_column_expression(self):
3503 return self.clause
3505 def self_group(
3506 self, against: Optional[OperatorType] = None
3507 ) -> TypeCoerce[_T]:
3508 grouped = self.clause.self_group(against=against)
3509 if grouped is not self.clause:
3510 return TypeCoerce(grouped, self.type)
3511 else:
3512 return self
3515class Extract(ColumnElement[int]):
3516 """Represent a SQL EXTRACT clause, ``extract(field FROM expr)``."""
3518 __visit_name__ = "extract"
3520 _traverse_internals: _TraverseInternalsType = [
3521 ("expr", InternalTraversal.dp_clauseelement),
3522 ("field", InternalTraversal.dp_string),
3523 ]
3525 expr: ColumnElement[Any]
3526 field: str
3528 def __init__(self, field: str, expr: _ColumnExpressionArgument[Any]):
3529 self.type = type_api.INTEGERTYPE
3530 self.field = field
3531 self.expr = coercions.expect(roles.ExpressionElementRole, expr)
3533 @util.ro_non_memoized_property
3534 def _from_objects(self) -> List[FromClause]:
3535 return self.expr._from_objects
3538class _label_reference(ColumnElement[_T]):
3539 """Wrap a column expression as it appears in a 'reference' context.
3541 This expression is any that includes an _order_by_label_element,
3542 which is a Label, or a DESC / ASC construct wrapping a Label.
3544 The production of _label_reference() should occur when an expression
3545 is added to this context; this includes the ORDER BY or GROUP BY of a
3546 SELECT statement, as well as a few other places, such as the ORDER BY
3547 within an OVER clause.
3549 """
3551 __visit_name__ = "label_reference"
3553 _traverse_internals: _TraverseInternalsType = [
3554 ("element", InternalTraversal.dp_clauseelement)
3555 ]
3557 element: ColumnElement[_T]
3559 def __init__(self, element: ColumnElement[_T]):
3560 self.element = element
3562 @util.ro_non_memoized_property
3563 def _from_objects(self) -> List[FromClause]:
3564 return []
3567class _textual_label_reference(ColumnElement[Any]):
3568 __visit_name__ = "textual_label_reference"
3570 _traverse_internals: _TraverseInternalsType = [
3571 ("element", InternalTraversal.dp_string)
3572 ]
3574 def __init__(self, element: str):
3575 self.element = element
3577 @util.memoized_property
3578 def _text_clause(self) -> TextClause:
3579 return TextClause(self.element)
3582class UnaryExpression(ColumnElement[_T]):
3583 """Define a 'unary' expression.
3585 A unary expression has a single column expression
3586 and an operator. The operator can be placed on the left
3587 (where it is called the 'operator') or right (where it is called the
3588 'modifier') of the column expression.
3590 :class:`.UnaryExpression` is the basis for several unary operators
3591 including those used by :func:`.desc`, :func:`.asc`, :func:`.distinct`,
3592 :func:`.nulls_first` and :func:`.nulls_last`.
3594 """
3596 __visit_name__ = "unary"
3598 _traverse_internals: _TraverseInternalsType = [
3599 ("element", InternalTraversal.dp_clauseelement),
3600 ("operator", InternalTraversal.dp_operator),
3601 ("modifier", InternalTraversal.dp_operator),
3602 ]
3604 element: ClauseElement
3606 def __init__(
3607 self,
3608 element: ColumnElement[Any],
3609 operator: Optional[OperatorType] = None,
3610 modifier: Optional[OperatorType] = None,
3611 type_: Optional[_TypeEngineArgument[_T]] = None,
3612 wraps_column_expression: bool = False,
3613 ):
3614 self.operator = operator
3615 self.modifier = modifier
3616 self._propagate_attrs = element._propagate_attrs
3617 self.element = element.self_group(
3618 against=self.operator or self.modifier
3619 )
3621 # if type is None, we get NULLTYPE, which is our _T. But I don't
3622 # know how to get the overloads to express that correctly
3623 self.type = type_api.to_instance(type_) # type: ignore
3625 self.wraps_column_expression = wraps_column_expression
3627 @classmethod
3628 def _create_nulls_first(
3629 cls,
3630 column: _ColumnExpressionArgument[_T],
3631 ) -> UnaryExpression[_T]:
3632 return UnaryExpression(
3633 coercions.expect(roles.ByOfRole, column),
3634 modifier=operators.nulls_first_op,
3635 wraps_column_expression=False,
3636 )
3638 @classmethod
3639 def _create_nulls_last(
3640 cls,
3641 column: _ColumnExpressionArgument[_T],
3642 ) -> UnaryExpression[_T]:
3643 return UnaryExpression(
3644 coercions.expect(roles.ByOfRole, column),
3645 modifier=operators.nulls_last_op,
3646 wraps_column_expression=False,
3647 )
3649 @classmethod
3650 def _create_desc(
3651 cls, column: _ColumnExpressionOrStrLabelArgument[_T]
3652 ) -> UnaryExpression[_T]:
3653 return UnaryExpression(
3654 coercions.expect(roles.ByOfRole, column),
3655 modifier=operators.desc_op,
3656 wraps_column_expression=False,
3657 )
3659 @classmethod
3660 def _create_asc(
3661 cls,
3662 column: _ColumnExpressionOrStrLabelArgument[_T],
3663 ) -> UnaryExpression[_T]:
3664 return UnaryExpression(
3665 coercions.expect(roles.ByOfRole, column),
3666 modifier=operators.asc_op,
3667 wraps_column_expression=False,
3668 )
3670 @classmethod
3671 def _create_distinct(
3672 cls,
3673 expr: _ColumnExpressionArgument[_T],
3674 ) -> UnaryExpression[_T]:
3675 col_expr: ColumnElement[_T] = coercions.expect(
3676 roles.ExpressionElementRole, expr
3677 )
3678 return UnaryExpression(
3679 col_expr,
3680 operator=operators.distinct_op,
3681 type_=col_expr.type,
3682 wraps_column_expression=False,
3683 )
3685 @classmethod
3686 def _create_bitwise_not(
3687 cls,
3688 expr: _ColumnExpressionArgument[_T],
3689 ) -> UnaryExpression[_T]:
3690 col_expr: ColumnElement[_T] = coercions.expect(
3691 roles.ExpressionElementRole, expr
3692 )
3693 return UnaryExpression(
3694 col_expr,
3695 operator=operators.bitwise_not_op,
3696 type_=col_expr.type,
3697 wraps_column_expression=False,
3698 )
3700 @property
3701 def _order_by_label_element(self) -> Optional[Label[Any]]:
3702 if operators.is_order_by_modifier(self.modifier):
3703 return self.element._order_by_label_element
3704 else:
3705 return None
3707 @util.ro_non_memoized_property
3708 def _from_objects(self) -> List[FromClause]:
3709 return self.element._from_objects
3711 def _negate(self):
3712 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
3713 return UnaryExpression(
3714 self.self_group(against=operators.inv),
3715 operator=operators.inv,
3716 type_=type_api.BOOLEANTYPE,
3717 wraps_column_expression=self.wraps_column_expression,
3718 )
3719 else:
3720 return ClauseElement._negate(self)
3722 def self_group(
3723 self, against: Optional[OperatorType] = None
3724 ) -> Union[Self, Grouping[_T]]:
3725 if self.operator and operators.is_precedent(self.operator, against):
3726 return Grouping(self)
3727 else:
3728 return self
3731class CollectionAggregate(UnaryExpression[_T]):
3732 """Forms the basis for right-hand collection operator modifiers
3733 ANY and ALL.
3735 The ANY and ALL keywords are available in different ways on different
3736 backends. On PostgreSQL, they only work for an ARRAY type. On
3737 MySQL, they only work for subqueries.
3739 """
3741 inherit_cache = True
3742 _is_collection_aggregate = True
3744 @classmethod
3745 def _create_any(
3746 cls, expr: _ColumnExpressionArgument[_T]
3747 ) -> CollectionAggregate[bool]:
3748 col_expr: ColumnElement[_T] = coercions.expect(
3749 roles.ExpressionElementRole,
3750 expr,
3751 )
3752 col_expr = col_expr.self_group()
3753 return CollectionAggregate(
3754 col_expr,
3755 operator=operators.any_op,
3756 type_=type_api.BOOLEANTYPE,
3757 wraps_column_expression=False,
3758 )
3760 @classmethod
3761 def _create_all(
3762 cls, expr: _ColumnExpressionArgument[_T]
3763 ) -> CollectionAggregate[bool]:
3764 col_expr: ColumnElement[_T] = coercions.expect(
3765 roles.ExpressionElementRole,
3766 expr,
3767 )
3768 col_expr = col_expr.self_group()
3769 return CollectionAggregate(
3770 col_expr,
3771 operator=operators.all_op,
3772 type_=type_api.BOOLEANTYPE,
3773 wraps_column_expression=False,
3774 )
3776 # operate and reverse_operate are hardwired to
3777 # dispatch onto the type comparator directly, so that we can
3778 # ensure "reversed" behavior.
3779 def operate(self, op, *other, **kwargs):
3780 if not operators.is_comparison(op):
3781 raise exc.ArgumentError(
3782 "Only comparison operators may be used with ANY/ALL"
3783 )
3784 kwargs["reverse"] = True
3785 return self.comparator.operate(operators.mirror(op), *other, **kwargs)
3787 def reverse_operate(self, op, other, **kwargs):
3788 # comparison operators should never call reverse_operate
3789 assert not operators.is_comparison(op)
3790 raise exc.ArgumentError(
3791 "Only comparison operators may be used with ANY/ALL"
3792 )
3795class AsBoolean(WrapsColumnExpression[bool], UnaryExpression[bool]):
3796 inherit_cache = True
3798 def __init__(self, element, operator, negate):
3799 self.element = element
3800 self.type = type_api.BOOLEANTYPE
3801 self.operator = operator
3802 self.negate = negate
3803 self.modifier = None
3804 self.wraps_column_expression = True
3805 self._is_implicitly_boolean = element._is_implicitly_boolean
3807 @property
3808 def wrapped_column_expression(self):
3809 return self.element
3811 def self_group(self, against: Optional[OperatorType] = None) -> Self:
3812 return self
3814 def _negate(self):
3815 if isinstance(self.element, (True_, False_)):
3816 return self.element._negate()
3817 else:
3818 return AsBoolean(self.element, self.negate, self.operator)
3821class BinaryExpression(OperatorExpression[_T]):
3822 """Represent an expression that is ``LEFT <operator> RIGHT``.
3824 A :class:`.BinaryExpression` is generated automatically
3825 whenever two column expressions are used in a Python binary expression:
3827 .. sourcecode:: pycon+sql
3829 >>> from sqlalchemy.sql import column
3830 >>> column('a') + column('b')
3831 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
3832 >>> print(column('a') + column('b'))
3833 {printsql}a + b
3835 """
3837 __visit_name__ = "binary"
3839 _traverse_internals: _TraverseInternalsType = [
3840 ("left", InternalTraversal.dp_clauseelement),
3841 ("right", InternalTraversal.dp_clauseelement),
3842 ("operator", InternalTraversal.dp_operator),
3843 ("negate", InternalTraversal.dp_operator),
3844 ("modifiers", InternalTraversal.dp_plain_dict),
3845 (
3846 "type",
3847 InternalTraversal.dp_type,
3848 ),
3849 ]
3851 _cache_key_traversal = [
3852 ("left", InternalTraversal.dp_clauseelement),
3853 ("right", InternalTraversal.dp_clauseelement),
3854 ("operator", InternalTraversal.dp_operator),
3855 ("modifiers", InternalTraversal.dp_plain_dict),
3856 # "type" affects JSON CAST operators, so while redundant in most cases,
3857 # is needed for that one
3858 (
3859 "type",
3860 InternalTraversal.dp_type,
3861 ),
3862 ]
3864 _is_implicitly_boolean = True
3865 """Indicates that any database will know this is a boolean expression
3866 even if the database does not have an explicit boolean datatype.
3868 """
3870 modifiers: Optional[Mapping[str, Any]]
3872 left: ColumnElement[Any]
3873 right: ColumnElement[Any]
3875 def __init__(
3876 self,
3877 left: ColumnElement[Any],
3878 right: ColumnElement[Any],
3879 operator: OperatorType,
3880 type_: Optional[_TypeEngineArgument[_T]] = None,
3881 negate: Optional[OperatorType] = None,
3882 modifiers: Optional[Mapping[str, Any]] = None,
3883 ):
3884 # allow compatibility with libraries that
3885 # refer to BinaryExpression directly and pass strings
3886 if isinstance(operator, str):
3887 operator = operators.custom_op(operator)
3888 self._orig = (left.__hash__(), right.__hash__())
3889 self._propagate_attrs = left._propagate_attrs or right._propagate_attrs
3890 self.left = left.self_group(against=operator)
3891 self.right = right.self_group(against=operator)
3892 self.operator = operator
3894 # if type is None, we get NULLTYPE, which is our _T. But I don't
3895 # know how to get the overloads to express that correctly
3896 self.type = type_api.to_instance(type_) # type: ignore
3898 self.negate = negate
3899 self._is_implicitly_boolean = operators.is_boolean(operator)
3901 if modifiers is None:
3902 self.modifiers = {}
3903 else:
3904 self.modifiers = modifiers
3906 @property
3907 def _flattened_operator_clauses(
3908 self,
3909 ) -> typing_Tuple[ColumnElement[Any], ...]:
3910 return (self.left, self.right)
3912 def __bool__(self):
3913 """Implement Python-side "bool" for BinaryExpression as a
3914 simple "identity" check for the left and right attributes,
3915 if the operator is "eq" or "ne". Otherwise the expression
3916 continues to not support "bool" like all other column expressions.
3918 The rationale here is so that ColumnElement objects can be hashable.
3919 What? Well, suppose you do this::
3921 c1, c2 = column('x'), column('y')
3922 s1 = set([c1, c2])
3924 We do that **a lot**, columns inside of sets is an extremely basic
3925 thing all over the ORM for example.
3927 So what happens if we do this? ::
3929 c1 in s1
3931 Hashing means it will normally use ``__hash__()`` of the object,
3932 but in case of hash collision, it's going to also do ``c1 == c1``
3933 and/or ``c1 == c2`` inside. Those operations need to return a
3934 True/False value. But because we override ``==`` and ``!=``, they're
3935 going to get a BinaryExpression. Hence we implement ``__bool__`` here
3936 so that these comparisons behave in this particular context mostly
3937 like regular object comparisons. Thankfully Python is OK with
3938 that! Otherwise we'd have to use special set classes for columns
3939 (which we used to do, decades ago).
3941 """
3942 if self.operator in (operators.eq, operators.ne):
3943 # this is using the eq/ne operator given int hash values,
3944 # rather than Operator, so that "bool" can be based on
3945 # identity
3946 return self.operator(*self._orig) # type: ignore
3947 else:
3948 raise TypeError("Boolean value of this clause is not defined")
3950 if typing.TYPE_CHECKING:
3952 def __invert__(
3953 self: BinaryExpression[_T],
3954 ) -> BinaryExpression[_T]: ...
3956 @util.ro_non_memoized_property
3957 def _from_objects(self) -> List[FromClause]:
3958 return self.left._from_objects + self.right._from_objects
3960 def _negate(self):
3961 if self.negate is not None:
3962 return BinaryExpression(
3963 self.left,
3964 self.right._negate_in_binary(self.negate, self.operator),
3965 self.negate,
3966 negate=self.operator,
3967 type_=self.type,
3968 modifiers=self.modifiers,
3969 )
3970 else:
3971 return self.self_group()._negate()
3974class Slice(ColumnElement[Any]):
3975 """Represent SQL for a Python array-slice object.
3977 This is not a specific SQL construct at this level, but
3978 may be interpreted by specific dialects, e.g. PostgreSQL.
3980 """
3982 __visit_name__ = "slice"
3984 _traverse_internals: _TraverseInternalsType = [
3985 ("start", InternalTraversal.dp_clauseelement),
3986 ("stop", InternalTraversal.dp_clauseelement),
3987 ("step", InternalTraversal.dp_clauseelement),
3988 ]
3990 def __init__(self, start, stop, step, _name=None):
3991 self.start = coercions.expect(
3992 roles.ExpressionElementRole,
3993 start,
3994 name=_name,
3995 type_=type_api.INTEGERTYPE,
3996 )
3997 self.stop = coercions.expect(
3998 roles.ExpressionElementRole,
3999 stop,
4000 name=_name,
4001 type_=type_api.INTEGERTYPE,
4002 )
4003 self.step = coercions.expect(
4004 roles.ExpressionElementRole,
4005 step,
4006 name=_name,
4007 type_=type_api.INTEGERTYPE,
4008 )
4009 self.type = type_api.NULLTYPE
4011 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4012 assert against is operator.getitem
4013 return self
4016class IndexExpression(BinaryExpression[Any]):
4017 """Represent the class of expressions that are like an "index"
4018 operation."""
4020 inherit_cache = True
4023class GroupedElement(DQLDMLClauseElement):
4024 """Represent any parenthesized expression"""
4026 __visit_name__ = "grouping"
4028 element: ClauseElement
4030 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4031 return self
4033 def _ungroup(self):
4034 return self.element._ungroup()
4037class Grouping(GroupedElement, ColumnElement[_T]):
4038 """Represent a grouping within a column expression"""
4040 _traverse_internals: _TraverseInternalsType = [
4041 ("element", InternalTraversal.dp_clauseelement),
4042 ("type", InternalTraversal.dp_type),
4043 ]
4045 _cache_key_traversal = [
4046 ("element", InternalTraversal.dp_clauseelement),
4047 ]
4049 element: Union[TextClause, ClauseList, ColumnElement[_T]]
4051 def __init__(
4052 self, element: Union[TextClause, ClauseList, ColumnElement[_T]]
4053 ):
4054 self.element = element
4056 # nulltype assignment issue
4057 self.type = getattr(element, "type", type_api.NULLTYPE) # type: ignore
4058 self._propagate_attrs = element._propagate_attrs
4060 def _with_binary_element_type(self, type_):
4061 return self.__class__(self.element._with_binary_element_type(type_))
4063 @util.memoized_property
4064 def _is_implicitly_boolean(self):
4065 return self.element._is_implicitly_boolean
4067 @util.non_memoized_property
4068 def _tq_label(self) -> Optional[str]:
4069 return (
4070 getattr(self.element, "_tq_label", None) or self._anon_name_label
4071 )
4073 @util.non_memoized_property
4074 def _proxies(self) -> List[ColumnElement[Any]]:
4075 if isinstance(self.element, ColumnElement):
4076 return [self.element]
4077 else:
4078 return []
4080 @util.ro_non_memoized_property
4081 def _from_objects(self) -> List[FromClause]:
4082 return self.element._from_objects
4084 def __getattr__(self, attr):
4085 return getattr(self.element, attr)
4087 def __getstate__(self):
4088 return {"element": self.element, "type": self.type}
4090 def __setstate__(self, state):
4091 self.element = state["element"]
4092 self.type = state["type"]
4094 if TYPE_CHECKING:
4096 def self_group(
4097 self, against: Optional[OperatorType] = None
4098 ) -> Self: ...
4101class _OverrideBinds(Grouping[_T]):
4102 """used by cache_key->_apply_params_to_element to allow compilation /
4103 execution of a SQL element that's been cached, using an alternate set of
4104 bound parameter values.
4106 This is used by the ORM to swap new parameter values into expressions
4107 that are embedded into loader options like with_expression(),
4108 selectinload(). Previously, this task was accomplished using the
4109 .params() method which would perform a deep-copy instead. This deep
4110 copy proved to be too expensive for more complex expressions.
4112 See #11085
4114 """
4116 __visit_name__ = "override_binds"
4118 def __init__(
4119 self,
4120 element: ColumnElement[_T],
4121 bindparams: Sequence[BindParameter[Any]],
4122 replaces_params: Sequence[BindParameter[Any]],
4123 ):
4124 self.element = element
4125 self.translate = {
4126 k.key: v.value for k, v in zip(replaces_params, bindparams)
4127 }
4129 def _gen_cache_key(
4130 self, anon_map: anon_map, bindparams: List[BindParameter[Any]]
4131 ) -> Optional[typing_Tuple[Any, ...]]:
4132 """generate a cache key for the given element, substituting its bind
4133 values for the translation values present."""
4135 existing_bps: List[BindParameter[Any]] = []
4136 ck = self.element._gen_cache_key(anon_map, existing_bps)
4138 bindparams.extend(
4139 (
4140 bp._with_value(
4141 self.translate[bp.key], maintain_key=True, required=False
4142 )
4143 if bp.key in self.translate
4144 else bp
4145 )
4146 for bp in existing_bps
4147 )
4149 return ck
4152class Over(ColumnElement[_T]):
4153 """Represent an OVER clause.
4155 This is a special operator against a so-called
4156 "window" function, as well as any aggregate function,
4157 which produces results relative to the result set
4158 itself. Most modern SQL backends now support window functions.
4160 """
4162 __visit_name__ = "over"
4164 _traverse_internals: _TraverseInternalsType = [
4165 ("element", InternalTraversal.dp_clauseelement),
4166 ("order_by", InternalTraversal.dp_clauseelement),
4167 ("partition_by", InternalTraversal.dp_clauseelement),
4168 ("range_", InternalTraversal.dp_clauseelement),
4169 ("rows", InternalTraversal.dp_clauseelement),
4170 ]
4172 order_by: Optional[ClauseList] = None
4173 partition_by: Optional[ClauseList] = None
4175 element: ColumnElement[_T]
4176 """The underlying expression object to which this :class:`.Over`
4177 object refers."""
4179 range_: Optional[_FrameClause]
4180 rows: Optional[_FrameClause]
4182 def __init__(
4183 self,
4184 element: ColumnElement[_T],
4185 partition_by: Optional[_ByArgument] = None,
4186 order_by: Optional[_ByArgument] = None,
4187 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4188 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4189 ):
4190 self.element = element
4191 if order_by is not None:
4192 self.order_by = ClauseList(
4193 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4194 )
4195 if partition_by is not None:
4196 self.partition_by = ClauseList(
4197 *util.to_list(partition_by),
4198 _literal_as_text_role=roles.ByOfRole,
4199 )
4201 if range_:
4202 self.range_ = _FrameClause(range_)
4203 if rows:
4204 raise exc.ArgumentError(
4205 "'range_' and 'rows' are mutually exclusive"
4206 )
4207 else:
4208 self.rows = None
4209 elif rows:
4210 self.rows = _FrameClause(rows)
4211 self.range_ = None
4212 else:
4213 self.rows = self.range_ = None
4215 if not TYPE_CHECKING:
4217 @util.memoized_property
4218 def type(self) -> TypeEngine[_T]: # noqa: A001
4219 return self.element.type
4221 @util.ro_non_memoized_property
4222 def _from_objects(self) -> List[FromClause]:
4223 return list(
4224 itertools.chain(
4225 *[
4226 c._from_objects
4227 for c in (self.element, self.partition_by, self.order_by)
4228 if c is not None
4229 ]
4230 )
4231 )
4234class _FrameClauseType(Enum):
4235 RANGE_UNBOUNDED = 0
4236 RANGE_CURRENT = 1
4237 RANGE_PRECEDING = 2
4238 RANGE_FOLLOWING = 3
4241class _FrameClause(ClauseElement):
4242 """indicate the 'rows' or 'range' field of a window function, e.g. using
4243 :class:`.Over`.
4245 .. versionadded:: 2.1
4247 """
4249 __visit_name__ = "frame_clause"
4251 _traverse_internals: _TraverseInternalsType = [
4252 ("lower_integer_bind", InternalTraversal.dp_clauseelement),
4253 ("upper_integer_bind", InternalTraversal.dp_clauseelement),
4254 ("lower_type", InternalTraversal.dp_plain_obj),
4255 ("upper_type", InternalTraversal.dp_plain_obj),
4256 ]
4258 def __init__(
4259 self,
4260 range_: typing_Tuple[Optional[int], Optional[int]],
4261 ):
4262 try:
4263 r0, r1 = range_
4264 except (ValueError, TypeError) as ve:
4265 raise exc.ArgumentError("2-tuple expected for range/rows") from ve
4267 if r0 is None:
4268 self.lower_type = _FrameClauseType.RANGE_UNBOUNDED
4269 self.lower_integer_bind = None
4270 else:
4271 try:
4272 lower_integer = int(r0)
4273 except ValueError as err:
4274 raise exc.ArgumentError(
4275 "Integer or None expected for range value"
4276 ) from err
4277 else:
4278 if lower_integer == 0:
4279 self.lower_type = _FrameClauseType.RANGE_CURRENT
4280 self.lower_integer_bind = None
4281 elif lower_integer < 0:
4282 self.lower_type = _FrameClauseType.RANGE_PRECEDING
4283 self.lower_integer_bind = literal(
4284 abs(lower_integer), type_api.INTEGERTYPE
4285 )
4286 else:
4287 self.lower_type = _FrameClauseType.RANGE_FOLLOWING
4288 self.lower_integer_bind = literal(
4289 lower_integer, type_api.INTEGERTYPE
4290 )
4292 if r1 is None:
4293 self.upper_type = _FrameClauseType.RANGE_UNBOUNDED
4294 self.upper_integer_bind = None
4295 else:
4296 try:
4297 upper_integer = int(r1)
4298 except ValueError as err:
4299 raise exc.ArgumentError(
4300 "Integer or None expected for range value"
4301 ) from err
4302 else:
4303 if upper_integer == 0:
4304 self.upper_type = _FrameClauseType.RANGE_CURRENT
4305 self.upper_integer_bind = None
4306 elif upper_integer < 0:
4307 self.upper_type = _FrameClauseType.RANGE_PRECEDING
4308 self.upper_integer_bind = literal(
4309 abs(upper_integer), type_api.INTEGERTYPE
4310 )
4311 else:
4312 self.upper_type = _FrameClauseType.RANGE_FOLLOWING
4313 self.upper_integer_bind = literal(
4314 upper_integer, type_api.INTEGERTYPE
4315 )
4318class WithinGroup(ColumnElement[_T]):
4319 """Represent a WITHIN GROUP (ORDER BY) clause.
4321 This is a special operator against so-called
4322 "ordered set aggregate" and "hypothetical
4323 set aggregate" functions, including ``percentile_cont()``,
4324 ``rank()``, ``dense_rank()``, etc.
4326 It's supported only by certain database backends, such as PostgreSQL,
4327 Oracle and MS SQL Server.
4329 The :class:`.WithinGroup` construct extracts its type from the
4330 method :meth:`.FunctionElement.within_group_type`. If this returns
4331 ``None``, the function's ``.type`` is used.
4333 """
4335 __visit_name__ = "withingroup"
4337 _traverse_internals: _TraverseInternalsType = [
4338 ("element", InternalTraversal.dp_clauseelement),
4339 ("order_by", InternalTraversal.dp_clauseelement),
4340 ]
4342 order_by: Optional[ClauseList] = None
4344 def __init__(
4345 self,
4346 element: Union[FunctionElement[_T], FunctionFilter[_T]],
4347 *order_by: _ColumnExpressionArgument[Any],
4348 ):
4349 self.element = element
4350 if order_by is not None:
4351 self.order_by = ClauseList(
4352 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4353 )
4355 def __reduce__(self):
4356 return self.__class__, (self.element,) + (
4357 tuple(self.order_by) if self.order_by is not None else ()
4358 )
4360 def over(
4361 self,
4362 *,
4363 partition_by: Optional[_ByArgument] = None,
4364 order_by: Optional[_ByArgument] = None,
4365 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4366 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4367 ) -> Over[_T]:
4368 """Produce an OVER clause against this :class:`.WithinGroup`
4369 construct.
4371 This function has the same signature as that of
4372 :meth:`.FunctionElement.over`.
4374 """
4375 return Over(
4376 self,
4377 partition_by=partition_by,
4378 order_by=order_by,
4379 range_=range_,
4380 rows=rows,
4381 )
4383 @overload
4384 def filter(self) -> Self: ...
4386 @overload
4387 def filter(
4388 self,
4389 __criterion0: _ColumnExpressionArgument[bool],
4390 *criterion: _ColumnExpressionArgument[bool],
4391 ) -> FunctionFilter[_T]: ...
4393 def filter(
4394 self, *criterion: _ColumnExpressionArgument[bool]
4395 ) -> Union[Self, FunctionFilter[_T]]:
4396 """Produce a FILTER clause against this function."""
4397 if not criterion:
4398 return self
4399 return FunctionFilter(self, *criterion)
4401 if not TYPE_CHECKING:
4403 @util.memoized_property
4404 def type(self) -> TypeEngine[_T]: # noqa: A001
4405 wgt = self.element.within_group_type(self)
4406 if wgt is not None:
4407 return wgt
4408 else:
4409 return self.element.type
4411 @util.ro_non_memoized_property
4412 def _from_objects(self) -> List[FromClause]:
4413 return list(
4414 itertools.chain(
4415 *[
4416 c._from_objects
4417 for c in (self.element, self.order_by)
4418 if c is not None
4419 ]
4420 )
4421 )
4424class FunctionFilter(Generative, ColumnElement[_T]):
4425 """Represent a function FILTER clause.
4427 This is a special operator against aggregate and window functions,
4428 which controls which rows are passed to it.
4429 It's supported only by certain database backends.
4431 Invocation of :class:`.FunctionFilter` is via
4432 :meth:`.FunctionElement.filter`::
4434 func.count(1).filter(True)
4436 .. seealso::
4438 :meth:`.FunctionElement.filter`
4440 """
4442 __visit_name__ = "funcfilter"
4444 _traverse_internals: _TraverseInternalsType = [
4445 ("func", InternalTraversal.dp_clauseelement),
4446 ("criterion", InternalTraversal.dp_clauseelement),
4447 ]
4449 criterion: Optional[ColumnElement[bool]] = None
4451 def __init__(
4452 self,
4453 func: Union[FunctionElement[_T], WithinGroup[_T]],
4454 *criterion: _ColumnExpressionArgument[bool],
4455 ):
4456 self.func = func
4457 self.filter.non_generative(self, *criterion) # type: ignore
4459 @_generative
4460 def filter(self, *criterion: _ColumnExpressionArgument[bool]) -> Self:
4461 """Produce an additional FILTER against the function.
4463 This method adds additional criteria to the initial criteria
4464 set up by :meth:`.FunctionElement.filter`.
4466 Multiple criteria are joined together at SQL render time
4467 via ``AND``.
4470 """
4472 for crit in list(criterion):
4473 crit = coercions.expect(roles.WhereHavingRole, crit)
4475 if self.criterion is not None:
4476 self.criterion = self.criterion & crit
4477 else:
4478 self.criterion = crit
4480 return self
4482 def over(
4483 self,
4484 partition_by: Optional[
4485 Union[
4486 Iterable[_ColumnExpressionArgument[Any]],
4487 _ColumnExpressionArgument[Any],
4488 ]
4489 ] = None,
4490 order_by: Optional[
4491 Union[
4492 Iterable[_ColumnExpressionArgument[Any]],
4493 _ColumnExpressionArgument[Any],
4494 ]
4495 ] = None,
4496 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4497 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4498 ) -> Over[_T]:
4499 """Produce an OVER clause against this filtered function.
4501 Used against aggregate or so-called "window" functions,
4502 for database backends that support window functions.
4504 The expression::
4506 func.rank().filter(MyClass.y > 5).over(order_by='x')
4508 is shorthand for::
4510 from sqlalchemy import over, funcfilter
4511 over(funcfilter(func.rank(), MyClass.y > 5), order_by='x')
4513 See :func:`_expression.over` for a full description.
4515 """
4516 return Over(
4517 self,
4518 partition_by=partition_by,
4519 order_by=order_by,
4520 range_=range_,
4521 rows=rows,
4522 )
4524 def within_group(
4525 self, *order_by: _ColumnExpressionArgument[Any]
4526 ) -> WithinGroup[_T]:
4527 """Produce a WITHIN GROUP (ORDER BY expr) clause against
4528 this function.
4529 """
4530 return WithinGroup(self, *order_by)
4532 def within_group_type(
4533 self, within_group: WithinGroup[_T]
4534 ) -> Optional[TypeEngine[_T]]:
4535 return None
4537 def self_group(
4538 self, against: Optional[OperatorType] = None
4539 ) -> Union[Self, Grouping[_T]]:
4540 if operators.is_precedent(operators.filter_op, against):
4541 return Grouping(self)
4542 else:
4543 return self
4545 if not TYPE_CHECKING:
4547 @util.memoized_property
4548 def type(self) -> TypeEngine[_T]: # noqa: A001
4549 return self.func.type
4551 @util.ro_non_memoized_property
4552 def _from_objects(self) -> List[FromClause]:
4553 return list(
4554 itertools.chain(
4555 *[
4556 c._from_objects
4557 for c in (self.func, self.criterion)
4558 if c is not None
4559 ]
4560 )
4561 )
4564class NamedColumn(KeyedColumnElement[_T]):
4565 is_literal = False
4566 table: Optional[FromClause] = None
4567 name: str
4568 key: str
4570 def _compare_name_for_result(self, other):
4571 return (hasattr(other, "name") and self.name == other.name) or (
4572 hasattr(other, "_label") and self._label == other._label
4573 )
4575 @util.ro_memoized_property
4576 def description(self) -> str:
4577 return self.name
4579 @HasMemoized.memoized_attribute
4580 def _tq_key_label(self):
4581 """table qualified label based on column key.
4583 for table-bound columns this is <tablename>_<column key/proxy key>;
4585 all other expressions it resolves to key/proxy key.
4587 """
4588 proxy_key = self._proxy_key
4589 if proxy_key and proxy_key != self.name:
4590 return self._gen_tq_label(proxy_key)
4591 else:
4592 return self._tq_label
4594 @HasMemoized.memoized_attribute
4595 def _tq_label(self) -> Optional[str]:
4596 """table qualified label based on column name.
4598 for table-bound columns this is <tablename>_<columnname>; all other
4599 expressions it resolves to .name.
4601 """
4602 return self._gen_tq_label(self.name)
4604 @HasMemoized.memoized_attribute
4605 def _render_label_in_columns_clause(self):
4606 return True
4608 @HasMemoized.memoized_attribute
4609 def _non_anon_label(self):
4610 return self.name
4612 def _gen_tq_label(
4613 self, name: str, dedupe_on_key: bool = True
4614 ) -> Optional[str]:
4615 return name
4617 def _bind_param(
4618 self,
4619 operator: OperatorType,
4620 obj: Any,
4621 type_: Optional[TypeEngine[_T]] = None,
4622 expanding: bool = False,
4623 ) -> BindParameter[_T]:
4624 return BindParameter(
4625 self.key,
4626 obj,
4627 _compared_to_operator=operator,
4628 _compared_to_type=self.type,
4629 type_=type_,
4630 unique=True,
4631 expanding=expanding,
4632 )
4634 def _make_proxy(
4635 self,
4636 selectable: FromClause,
4637 *,
4638 name: Optional[str] = None,
4639 key: Optional[str] = None,
4640 name_is_truncatable: bool = False,
4641 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
4642 disallow_is_literal: bool = False,
4643 **kw: Any,
4644 ) -> typing_Tuple[str, ColumnClause[_T]]:
4645 c = ColumnClause(
4646 (
4647 coercions.expect(roles.TruncatedLabelRole, name or self.name)
4648 if name_is_truncatable
4649 else (name or self.name)
4650 ),
4651 type_=self.type,
4652 _selectable=selectable,
4653 is_literal=False,
4654 )
4656 c._propagate_attrs = selectable._propagate_attrs
4657 if name is None:
4658 c.key = self.key
4659 if compound_select_cols:
4660 c._proxies = list(compound_select_cols)
4661 else:
4662 c._proxies = [self]
4664 if selectable._is_clone_of is not None:
4665 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
4666 return c.key, c
4669_PS = ParamSpec("_PS")
4672class Label(roles.LabeledColumnExprRole[_T], NamedColumn[_T]):
4673 """Represents a column label (AS).
4675 Represent a label, as typically applied to any column-level
4676 element using the ``AS`` sql keyword.
4678 """
4680 __visit_name__ = "label"
4682 _traverse_internals: _TraverseInternalsType = [
4683 ("name", InternalTraversal.dp_anon_name),
4684 ("type", InternalTraversal.dp_type),
4685 ("_element", InternalTraversal.dp_clauseelement),
4686 ]
4688 _cache_key_traversal = [
4689 ("name", InternalTraversal.dp_anon_name),
4690 ("_element", InternalTraversal.dp_clauseelement),
4691 ]
4693 _element: ColumnElement[_T]
4694 name: str
4696 def __init__(
4697 self,
4698 name: Optional[str],
4699 element: _ColumnExpressionArgument[_T],
4700 type_: Optional[_TypeEngineArgument[_T]] = None,
4701 ):
4702 orig_element = element
4703 element = coercions.expect(
4704 roles.ExpressionElementRole,
4705 element,
4706 apply_propagate_attrs=self,
4707 )
4708 while isinstance(element, Label):
4709 # TODO: this is only covered in test_text.py, but nothing
4710 # fails if it's removed. determine rationale
4711 element = element.element
4713 if name:
4714 self.name = name
4715 else:
4716 self.name = _anonymous_label.safe_construct(
4717 id(self), getattr(element, "name", "anon")
4718 )
4719 if isinstance(orig_element, Label):
4720 # TODO: no coverage for this block, again would be in
4721 # test_text.py where the resolve_label concept is important
4722 self._resolve_label = orig_element._label
4724 self.key = self._tq_label = self._tq_key_label = self.name
4725 self._element = element
4727 self.type = (
4728 type_api.to_instance(type_)
4729 if type_ is not None
4730 else self._element.type
4731 )
4733 self._proxies = [element]
4735 def __reduce__(self):
4736 return self.__class__, (self.name, self._element, self.type)
4738 @HasMemoized.memoized_attribute
4739 def _render_label_in_columns_clause(self):
4740 return True
4742 def _bind_param(self, operator, obj, type_=None, expanding=False):
4743 return BindParameter(
4744 None,
4745 obj,
4746 _compared_to_operator=operator,
4747 type_=type_,
4748 _compared_to_type=self.type,
4749 unique=True,
4750 expanding=expanding,
4751 )
4753 @util.memoized_property
4754 def _is_implicitly_boolean(self):
4755 return self.element._is_implicitly_boolean
4757 @HasMemoized.memoized_attribute
4758 def _allow_label_resolve(self):
4759 return self.element._allow_label_resolve
4761 @property
4762 def _order_by_label_element(self):
4763 return self
4765 @HasMemoized.memoized_attribute
4766 def element(self) -> ColumnElement[_T]:
4767 return self._element.self_group(against=operators.as_)
4769 def self_group(self, against: Optional[OperatorType] = None) -> Label[_T]:
4770 return self._apply_to_inner(self._element.self_group, against=against)
4772 def _negate(self):
4773 return self._apply_to_inner(self._element._negate)
4775 def _apply_to_inner(
4776 self,
4777 fn: Callable[_PS, ColumnElement[_T]],
4778 *arg: _PS.args,
4779 **kw: _PS.kwargs,
4780 ) -> Label[_T]:
4781 sub_element = fn(*arg, **kw)
4782 if sub_element is not self._element:
4783 return Label(self.name, sub_element, type_=self.type)
4784 else:
4785 return self
4787 @property
4788 def primary_key(self):
4789 return self.element.primary_key
4791 @property
4792 def foreign_keys(self):
4793 return self.element.foreign_keys
4795 def _copy_internals(
4796 self,
4797 *,
4798 clone: _CloneCallableType = _clone,
4799 anonymize_labels: bool = False,
4800 **kw: Any,
4801 ) -> None:
4802 self._reset_memoizations()
4803 self._element = clone(self._element, **kw)
4804 if anonymize_labels:
4805 self.name = _anonymous_label.safe_construct(
4806 id(self), getattr(self.element, "name", "anon")
4807 )
4808 self.key = self._tq_label = self._tq_key_label = self.name
4810 @util.ro_non_memoized_property
4811 def _from_objects(self) -> List[FromClause]:
4812 return self.element._from_objects
4814 def _make_proxy(
4815 self,
4816 selectable: FromClause,
4817 *,
4818 name: Optional[str] = None,
4819 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
4820 **kw: Any,
4821 ) -> typing_Tuple[str, ColumnClause[_T]]:
4822 name = self.name if not name else name
4824 key, e = self.element._make_proxy(
4825 selectable,
4826 name=name,
4827 disallow_is_literal=True,
4828 name_is_truncatable=isinstance(name, _truncated_label),
4829 compound_select_cols=compound_select_cols,
4830 )
4832 # there was a note here to remove this assertion, which was here
4833 # to determine if we later could support a use case where
4834 # the key and name of a label are separate. But I don't know what
4835 # that case was. For now, this is an unexpected case that occurs
4836 # when a label name conflicts with other columns and select()
4837 # is attempting to disambiguate an explicit label, which is not what
4838 # the user would want. See issue #6090.
4839 if key != self.name and not isinstance(self.name, _anonymous_label):
4840 raise exc.InvalidRequestError(
4841 "Label name %s is being renamed to an anonymous label due "
4842 "to disambiguation "
4843 "which is not supported right now. Please use unique names "
4844 "for explicit labels." % (self.name)
4845 )
4847 e._propagate_attrs = selectable._propagate_attrs
4848 e._proxies.append(self)
4849 if self.type is not None:
4850 e.type = self.type
4852 return self.key, e
4855class ColumnClause(
4856 roles.DDLReferredColumnRole,
4857 roles.LabeledColumnExprRole[_T],
4858 roles.StrAsPlainColumnRole,
4859 Immutable,
4860 NamedColumn[_T],
4861):
4862 """Represents a column expression from any textual string.
4864 The :class:`.ColumnClause`, a lightweight analogue to the
4865 :class:`_schema.Column` class, is typically invoked using the
4866 :func:`_expression.column` function, as in::
4868 from sqlalchemy import column
4870 id, name = column("id"), column("name")
4871 stmt = select(id, name).select_from("user")
4873 The above statement would produce SQL like::
4875 SELECT id, name FROM user
4877 :class:`.ColumnClause` is the immediate superclass of the schema-specific
4878 :class:`_schema.Column` object. While the :class:`_schema.Column`
4879 class has all the
4880 same capabilities as :class:`.ColumnClause`, the :class:`.ColumnClause`
4881 class is usable by itself in those cases where behavioral requirements
4882 are limited to simple SQL expression generation. The object has none of
4883 the associations with schema-level metadata or with execution-time
4884 behavior that :class:`_schema.Column` does,
4885 so in that sense is a "lightweight"
4886 version of :class:`_schema.Column`.
4888 Full details on :class:`.ColumnClause` usage is at
4889 :func:`_expression.column`.
4891 .. seealso::
4893 :func:`_expression.column`
4895 :class:`_schema.Column`
4897 """
4899 table: Optional[FromClause]
4900 is_literal: bool
4902 __visit_name__ = "column"
4904 _traverse_internals: _TraverseInternalsType = [
4905 ("name", InternalTraversal.dp_anon_name),
4906 ("type", InternalTraversal.dp_type),
4907 ("table", InternalTraversal.dp_clauseelement),
4908 ("is_literal", InternalTraversal.dp_boolean),
4909 ]
4911 onupdate: Optional[DefaultGenerator] = None
4912 default: Optional[DefaultGenerator] = None
4913 server_default: Optional[FetchedValue] = None
4914 server_onupdate: Optional[FetchedValue] = None
4916 _is_multiparam_column = False
4918 @property
4919 def _is_star(self):
4920 return self.is_literal and self.name == "*"
4922 def __init__(
4923 self,
4924 text: str,
4925 type_: Optional[_TypeEngineArgument[_T]] = None,
4926 is_literal: bool = False,
4927 _selectable: Optional[FromClause] = None,
4928 ):
4929 self.key = self.name = text
4930 self.table = _selectable
4932 # if type is None, we get NULLTYPE, which is our _T. But I don't
4933 # know how to get the overloads to express that correctly
4934 self.type = type_api.to_instance(type_) # type: ignore
4936 self.is_literal = is_literal
4938 def get_children(self, *, column_tables=False, **kw):
4939 # override base get_children() to not return the Table
4940 # or selectable that is parent to this column. Traversals
4941 # expect the columns of tables and subqueries to be leaf nodes.
4942 return []
4944 @property
4945 def entity_namespace(self):
4946 if self.table is not None:
4947 return self.table.entity_namespace
4948 else:
4949 return super().entity_namespace
4951 def _clone(self, detect_subquery_cols=False, **kw):
4952 if (
4953 detect_subquery_cols
4954 and self.table is not None
4955 and self.table._is_subquery
4956 ):
4957 clone = kw.pop("clone")
4958 table = clone(self.table, **kw)
4959 new = table.c.corresponding_column(self)
4960 return new
4962 return super()._clone(**kw)
4964 @HasMemoized_ro_memoized_attribute
4965 def _from_objects(self) -> List[FromClause]:
4966 t = self.table
4967 if t is not None:
4968 return [t]
4969 else:
4970 return []
4972 @HasMemoized.memoized_attribute
4973 def _render_label_in_columns_clause(self):
4974 return self.table is not None
4976 @property
4977 def _ddl_label(self):
4978 return self._gen_tq_label(self.name, dedupe_on_key=False)
4980 def _compare_name_for_result(self, other):
4981 if (
4982 self.is_literal
4983 or self.table is None
4984 or self.table._is_textual
4985 or not hasattr(other, "proxy_set")
4986 or (
4987 isinstance(other, ColumnClause)
4988 and (
4989 other.is_literal
4990 or other.table is None
4991 or other.table._is_textual
4992 )
4993 )
4994 ):
4995 return (hasattr(other, "name") and self.name == other.name) or (
4996 hasattr(other, "_tq_label")
4997 and self._tq_label == other._tq_label
4998 )
4999 else:
5000 return other.proxy_set.intersection(self.proxy_set)
5002 def _gen_tq_label(
5003 self, name: str, dedupe_on_key: bool = True
5004 ) -> Optional[str]:
5005 """generate table-qualified label
5007 for a table-bound column this is <tablename>_<columnname>.
5009 used primarily for LABEL_STYLE_TABLENAME_PLUS_COL
5010 as well as the .columns collection on a Join object.
5012 """
5013 label: str
5014 t = self.table
5015 if self.is_literal:
5016 return None
5017 elif t is not None and is_named_from_clause(t):
5018 if has_schema_attr(t) and t.schema:
5019 label = (
5020 t.schema.replace(".", "_") + "_" + t.name + ("_" + name)
5021 )
5022 else:
5023 assert not TYPE_CHECKING or isinstance(t, NamedFromClause)
5024 label = t.name + ("_" + name)
5026 # propagate name quoting rules for labels.
5027 if is_quoted_name(name) and name.quote is not None:
5028 if is_quoted_name(label):
5029 label.quote = name.quote
5030 else:
5031 label = quoted_name(label, name.quote)
5032 elif is_quoted_name(t.name) and t.name.quote is not None:
5033 # can't get this situation to occur, so let's
5034 # assert false on it for now
5035 assert not isinstance(label, quoted_name)
5036 label = quoted_name(label, t.name.quote)
5038 if dedupe_on_key:
5039 # ensure the label name doesn't conflict with that of an
5040 # existing column. note that this implies that any Column
5041 # must **not** set up its _label before its parent table has
5042 # all of its other Column objects set up. There are several
5043 # tables in the test suite which will fail otherwise; example:
5044 # table "owner" has columns "name" and "owner_name". Therefore
5045 # column owner.name cannot use the label "owner_name", it has
5046 # to be "owner_name_1".
5047 if label in t.c:
5048 _label = label
5049 counter = 1
5050 while _label in t.c:
5051 _label = label + f"_{counter}"
5052 counter += 1
5053 label = _label
5055 return coercions.expect(roles.TruncatedLabelRole, label)
5057 else:
5058 return name
5060 def _make_proxy(
5061 self,
5062 selectable: FromClause,
5063 *,
5064 name: Optional[str] = None,
5065 key: Optional[str] = None,
5066 name_is_truncatable: bool = False,
5067 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5068 disallow_is_literal: bool = False,
5069 **kw: Any,
5070 ) -> typing_Tuple[str, ColumnClause[_T]]:
5071 # the "is_literal" flag normally should never be propagated; a proxied
5072 # column is always a SQL identifier and never the actual expression
5073 # being evaluated. however, there is a case where the "is_literal" flag
5074 # might be used to allow the given identifier to have a fixed quoting
5075 # pattern already, so maintain the flag for the proxy unless a
5076 # :class:`.Label` object is creating the proxy. See [ticket:4730].
5077 is_literal = (
5078 not disallow_is_literal
5079 and self.is_literal
5080 and (
5081 # note this does not accommodate for quoted_name differences
5082 # right now
5083 name is None
5084 or name == self.name
5085 )
5086 )
5087 c = self._constructor(
5088 (
5089 coercions.expect(roles.TruncatedLabelRole, name or self.name)
5090 if name_is_truncatable
5091 else (name or self.name)
5092 ),
5093 type_=self.type,
5094 _selectable=selectable,
5095 is_literal=is_literal,
5096 )
5097 c._propagate_attrs = selectable._propagate_attrs
5098 if name is None:
5099 c.key = self.key
5100 if compound_select_cols:
5101 c._proxies = list(compound_select_cols)
5102 else:
5103 c._proxies = [self]
5105 if selectable._is_clone_of is not None:
5106 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
5107 return c.key, c
5110class TableValuedColumn(NamedColumn[_T]):
5111 __visit_name__ = "table_valued_column"
5113 _traverse_internals: _TraverseInternalsType = [
5114 ("name", InternalTraversal.dp_anon_name),
5115 ("type", InternalTraversal.dp_type),
5116 ("scalar_alias", InternalTraversal.dp_clauseelement),
5117 ]
5119 def __init__(self, scalar_alias: NamedFromClause, type_: TypeEngine[_T]):
5120 self.scalar_alias = scalar_alias
5121 self.key = self.name = scalar_alias.name
5122 self.type = type_
5124 def _copy_internals(
5125 self, clone: _CloneCallableType = _clone, **kw: Any
5126 ) -> None:
5127 self.scalar_alias = clone(self.scalar_alias, **kw)
5128 self.key = self.name = self.scalar_alias.name
5130 @util.ro_non_memoized_property
5131 def _from_objects(self) -> List[FromClause]:
5132 return [self.scalar_alias]
5135class CollationClause(ColumnElement[str]):
5136 __visit_name__ = "collation"
5138 _traverse_internals: _TraverseInternalsType = [
5139 ("collation", InternalTraversal.dp_string)
5140 ]
5142 @classmethod
5143 @util.preload_module("sqlalchemy.sql.sqltypes")
5144 def _create_collation_expression(
5145 cls, expression: _ColumnExpressionArgument[str], collation: str
5146 ) -> BinaryExpression[str]:
5148 sqltypes = util.preloaded.sql_sqltypes
5150 expr = coercions.expect(roles.ExpressionElementRole[str], expression)
5152 if expr.type._type_affinity is sqltypes.String:
5153 collate_type = expr.type._with_collation(collation)
5154 else:
5155 collate_type = expr.type
5157 return BinaryExpression(
5158 expr,
5159 CollationClause(collation),
5160 operators.collate,
5161 type_=collate_type,
5162 )
5164 def __init__(self, collation):
5165 self.collation = collation
5168class _IdentifiedClause(Executable, ClauseElement):
5169 __visit_name__ = "identified"
5171 def __init__(self, ident):
5172 self.ident = ident
5175class SavepointClause(_IdentifiedClause):
5176 __visit_name__ = "savepoint"
5177 inherit_cache = False
5180class RollbackToSavepointClause(_IdentifiedClause):
5181 __visit_name__ = "rollback_to_savepoint"
5182 inherit_cache = False
5185class ReleaseSavepointClause(_IdentifiedClause):
5186 __visit_name__ = "release_savepoint"
5187 inherit_cache = False
5190class quoted_name(util.MemoizedSlots, str):
5191 """Represent a SQL identifier combined with quoting preferences.
5193 :class:`.quoted_name` is a Python unicode/str subclass which
5194 represents a particular identifier name along with a
5195 ``quote`` flag. This ``quote`` flag, when set to
5196 ``True`` or ``False``, overrides automatic quoting behavior
5197 for this identifier in order to either unconditionally quote
5198 or to not quote the name. If left at its default of ``None``,
5199 quoting behavior is applied to the identifier on a per-backend basis
5200 based on an examination of the token itself.
5202 A :class:`.quoted_name` object with ``quote=True`` is also
5203 prevented from being modified in the case of a so-called
5204 "name normalize" option. Certain database backends, such as
5205 Oracle, Firebird, and DB2 "normalize" case-insensitive names
5206 as uppercase. The SQLAlchemy dialects for these backends
5207 convert from SQLAlchemy's lower-case-means-insensitive convention
5208 to the upper-case-means-insensitive conventions of those backends.
5209 The ``quote=True`` flag here will prevent this conversion from occurring
5210 to support an identifier that's quoted as all lower case against
5211 such a backend.
5213 The :class:`.quoted_name` object is normally created automatically
5214 when specifying the name for key schema constructs such as
5215 :class:`_schema.Table`, :class:`_schema.Column`, and others.
5216 The class can also be
5217 passed explicitly as the name to any function that receives a name which
5218 can be quoted. Such as to use the :meth:`_engine.Engine.has_table`
5219 method with
5220 an unconditionally quoted name::
5222 from sqlalchemy import create_engine
5223 from sqlalchemy import inspect
5224 from sqlalchemy.sql import quoted_name
5226 engine = create_engine("oracle+cx_oracle://some_dsn")
5227 print(inspect(engine).has_table(quoted_name("some_table", True)))
5229 The above logic will run the "has table" logic against the Oracle backend,
5230 passing the name exactly as ``"some_table"`` without converting to
5231 upper case.
5233 .. versionchanged:: 1.2 The :class:`.quoted_name` construct is now
5234 importable from ``sqlalchemy.sql``, in addition to the previous
5235 location of ``sqlalchemy.sql.elements``.
5237 """
5239 __slots__ = "quote", "lower", "upper"
5241 quote: Optional[bool]
5243 @overload
5244 @classmethod
5245 def construct(cls, value: str, quote: Optional[bool]) -> quoted_name: ...
5247 @overload
5248 @classmethod
5249 def construct(cls, value: None, quote: Optional[bool]) -> None: ...
5251 @classmethod
5252 def construct(
5253 cls, value: Optional[str], quote: Optional[bool]
5254 ) -> Optional[quoted_name]:
5255 if value is None:
5256 return None
5257 else:
5258 return quoted_name(value, quote)
5260 def __new__(cls, value: str, quote: Optional[bool]) -> quoted_name:
5261 assert (
5262 value is not None
5263 ), "use quoted_name.construct() for None passthrough"
5264 if isinstance(value, cls) and (quote is None or value.quote == quote):
5265 return value
5266 self = super().__new__(cls, value)
5268 self.quote = quote
5269 return self
5271 def __reduce__(self):
5272 return quoted_name, (str(self), self.quote)
5274 def _memoized_method_lower(self):
5275 if self.quote:
5276 return self
5277 else:
5278 return str(self).lower()
5280 def _memoized_method_upper(self):
5281 if self.quote:
5282 return self
5283 else:
5284 return str(self).upper()
5287def _find_columns(clause: ClauseElement) -> Set[ColumnClause[Any]]:
5288 """locate Column objects within the given expression."""
5290 cols: Set[ColumnClause[Any]] = set()
5291 traverse(clause, {}, {"column": cols.add})
5292 return cols
5295def _type_from_args(args: Sequence[ColumnElement[_T]]) -> TypeEngine[_T]:
5296 for a in args:
5297 if not a.type._isnull:
5298 return a.type
5299 else:
5300 return type_api.NULLTYPE # type: ignore
5303def _corresponding_column_or_error(fromclause, column, require_embedded=False):
5304 c = fromclause.corresponding_column(
5305 column, require_embedded=require_embedded
5306 )
5307 if c is None:
5308 raise exc.InvalidRequestError(
5309 "Given column '%s', attached to table '%s', "
5310 "failed to locate a corresponding column from table '%s'"
5311 % (column, getattr(column, "table", None), fromclause.description)
5312 )
5313 return c
5316class _memoized_property_but_not_nulltype(
5317 util.memoized_property["TypeEngine[_T]"]
5318):
5319 """memoized property, but dont memoize NullType"""
5321 def __get__(self, obj, cls):
5322 if obj is None:
5323 return self
5324 result = self.fget(obj)
5325 if not result._isnull:
5326 obj.__dict__[self.__name__] = result
5327 return result
5330class AnnotatedColumnElement(Annotated):
5331 _Annotated__element: ColumnElement[Any]
5333 def __init__(self, element, values):
5334 Annotated.__init__(self, element, values)
5335 for attr in (
5336 "comparator",
5337 "_proxy_key",
5338 "_tq_key_label",
5339 "_tq_label",
5340 "_non_anon_label",
5341 "type",
5342 ):
5343 self.__dict__.pop(attr, None)
5344 for attr in ("name", "key", "table"):
5345 if self.__dict__.get(attr, False) is None:
5346 self.__dict__.pop(attr)
5348 def _with_annotations(self, values):
5349 clone = super()._with_annotations(values)
5350 for attr in (
5351 "comparator",
5352 "_proxy_key",
5353 "_tq_key_label",
5354 "_tq_label",
5355 "_non_anon_label",
5356 ):
5357 clone.__dict__.pop(attr, None)
5358 return clone
5360 @util.memoized_property
5361 def name(self):
5362 """pull 'name' from parent, if not present"""
5363 return self._Annotated__element.name
5365 @_memoized_property_but_not_nulltype
5366 def type(self):
5367 """pull 'type' from parent and don't cache if null.
5369 type is routinely changed on existing columns within the
5370 mapped_column() initialization process, and "type" is also consulted
5371 during the creation of SQL expressions. Therefore it can change after
5372 it was already retrieved. At the same time we don't want annotated
5373 objects having overhead when expressions are produced, so continue
5374 to memoize, but only when we have a non-null type.
5376 """
5377 return self._Annotated__element.type
5379 @util.memoized_property
5380 def table(self):
5381 """pull 'table' from parent, if not present"""
5382 return self._Annotated__element.table
5384 @util.memoized_property
5385 def key(self):
5386 """pull 'key' from parent, if not present"""
5387 return self._Annotated__element.key
5389 @util.memoized_property
5390 def info(self) -> _InfoType:
5391 if TYPE_CHECKING:
5392 assert isinstance(self._Annotated__element, Column)
5393 return self._Annotated__element.info
5395 @util.memoized_property
5396 def _anon_name_label(self) -> str:
5397 return self._Annotated__element._anon_name_label
5400class _truncated_label(quoted_name):
5401 """A unicode subclass used to identify symbolic "
5402 "names that may require truncation."""
5404 __slots__ = ()
5406 def __new__(cls, value: str, quote: Optional[bool] = None) -> Any:
5407 quote = getattr(value, "quote", quote)
5408 # return super(_truncated_label, cls).__new__(cls, value, quote, True)
5409 return super().__new__(cls, value, quote)
5411 def __reduce__(self) -> Any:
5412 return self.__class__, (str(self), self.quote)
5414 def apply_map(self, map_: Mapping[str, Any]) -> str:
5415 return self
5418class conv(_truncated_label):
5419 """Mark a string indicating that a name has already been converted
5420 by a naming convention.
5422 This is a string subclass that indicates a name that should not be
5423 subject to any further naming conventions.
5425 E.g. when we create a :class:`.Constraint` using a naming convention
5426 as follows::
5428 m = MetaData(naming_convention={
5429 "ck": "ck_%(table_name)s_%(constraint_name)s"
5430 })
5431 t = Table('t', m, Column('x', Integer),
5432 CheckConstraint('x > 5', name='x5'))
5434 The name of the above constraint will be rendered as ``"ck_t_x5"``.
5435 That is, the existing name ``x5`` is used in the naming convention as the
5436 ``constraint_name`` token.
5438 In some situations, such as in migration scripts, we may be rendering
5439 the above :class:`.CheckConstraint` with a name that's already been
5440 converted. In order to make sure the name isn't double-modified, the
5441 new name is applied using the :func:`_schema.conv` marker. We can
5442 use this explicitly as follows::
5445 m = MetaData(naming_convention={
5446 "ck": "ck_%(table_name)s_%(constraint_name)s"
5447 })
5448 t = Table('t', m, Column('x', Integer),
5449 CheckConstraint('x > 5', name=conv('ck_t_x5')))
5451 Where above, the :func:`_schema.conv` marker indicates that the constraint
5452 name here is final, and the name will render as ``"ck_t_x5"`` and not
5453 ``"ck_t_ck_t_x5"``
5455 .. seealso::
5457 :ref:`constraint_naming_conventions`
5459 """
5461 __slots__ = ()
5464# for backwards compatibility in case
5465# someone is re-implementing the
5466# _truncated_identifier() sequence in a custom
5467# compiler
5468_generated_label = _truncated_label
5469_anonymous_label_escape = re.compile(r"[%\(\) \$]+")
5472class _anonymous_label(_truncated_label):
5473 """A unicode subclass used to identify anonymously
5474 generated names."""
5476 __slots__ = ()
5478 @classmethod
5479 def safe_construct_with_key(
5480 cls, seed: int, body: str, sanitize_key: bool = False
5481 ) -> typing_Tuple[_anonymous_label, str]:
5482 # need to escape chars that interfere with format
5483 # strings in any case, issue #8724
5484 body = _anonymous_label_escape.sub("_", body)
5486 if sanitize_key:
5487 # sanitize_key is then an extra step used by BindParameter
5488 body = body.strip("_")
5490 key = f"{seed} {body.replace('%', '%%')}"
5491 label = _anonymous_label(f"%({key})s")
5492 return label, key
5494 @classmethod
5495 def safe_construct(
5496 cls, seed: int, body: str, sanitize_key: bool = False
5497 ) -> _anonymous_label:
5498 # need to escape chars that interfere with format
5499 # strings in any case, issue #8724
5500 body = _anonymous_label_escape.sub("_", body)
5502 if sanitize_key:
5503 # sanitize_key is then an extra step used by BindParameter
5504 body = body.strip("_")
5506 return _anonymous_label(f"%({seed} {body.replace('%', '%%')})s")
5508 def __add__(self, other: str) -> _anonymous_label:
5509 if "%" in other and not isinstance(other, _anonymous_label):
5510 other = str(other).replace("%", "%%")
5511 else:
5512 other = str(other)
5514 return _anonymous_label(
5515 quoted_name(
5516 str.__add__(self, other),
5517 self.quote,
5518 )
5519 )
5521 def __radd__(self, other: str) -> _anonymous_label:
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__(other, self),
5530 self.quote,
5531 )
5532 )
5534 def apply_map(self, map_: Mapping[str, Any]) -> str:
5535 if self.quote is not None:
5536 # preserve quoting only if necessary
5537 return quoted_name(self % map_, self.quote)
5538 else:
5539 # else skip the constructor call
5540 return self % map_