Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# sql/elements.py
2# Copyright (C) 2005-2025 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7# mypy: allow-untyped-defs, allow-untyped-calls
9"""Core SQL expression elements, including :class:`_expression.ClauseElement`,
10:class:`_expression.ColumnElement`, and derived classes.
12"""
14from __future__ import annotations
16from decimal import Decimal
17from enum import Enum
18import itertools
19import operator
20import re
21import typing
22from typing import AbstractSet
23from typing import Any
24from typing import Callable
25from typing import cast
26from typing import Dict
27from typing import FrozenSet
28from typing import Generic
29from typing import Iterable
30from typing import Iterator
31from typing import List
32from typing import Mapping
33from typing import Optional
34from typing import overload
35from typing import Sequence
36from typing import Set
37from typing import Tuple as typing_Tuple
38from typing import Type
39from typing import TYPE_CHECKING
40from typing import TypeVar
41from typing import Union
43from . import coercions
44from . import operators
45from . import roles
46from . import traversals
47from . import type_api
48from ._typing import has_schema_attr
49from ._typing import is_named_from_clause
50from ._typing import is_quoted_name
51from ._typing import is_tuple_type
52from .annotation import Annotated
53from .annotation import SupportsWrappingAnnotations
54from .base import _clone
55from .base import _expand_cloned
56from .base import _generative
57from .base import _NoArg
58from .base import Executable
59from .base import Generative
60from .base import HasMemoized
61from .base import Immutable
62from .base import NO_ARG
63from .base import SingletonConstant
64from .cache_key import MemoizedHasCacheKey
65from .cache_key import NO_CACHE
66from .coercions import _document_text_coercion # noqa
67from .operators import ColumnOperators
68from .traversals import HasCopyInternals
69from .visitors import cloned_traverse
70from .visitors import ExternallyTraversible
71from .visitors import InternalTraversal
72from .visitors import traverse
73from .visitors import Visitable
74from .. import exc
75from .. import inspection
76from .. import util
77from ..util import HasMemoized_ro_memoized_attribute
78from ..util import TypingOnly
79from ..util.typing import Literal
80from ..util.typing import ParamSpec
81from ..util.typing import Self
83if typing.TYPE_CHECKING:
84 from ._typing import _ByArgument
85 from ._typing import _ColumnExpressionArgument
86 from ._typing import _ColumnExpressionOrStrLabelArgument
87 from ._typing import _HasDialect
88 from ._typing import _InfoType
89 from ._typing import _PropagateAttrsType
90 from ._typing import _TypeEngineArgument
91 from .base import ColumnSet
92 from .cache_key import _CacheKeyTraversalType
93 from .cache_key import CacheKey
94 from .compiler import Compiled
95 from .compiler import SQLCompiler
96 from .functions import FunctionElement
97 from .operators import OperatorType
98 from .schema import Column
99 from .schema import DefaultGenerator
100 from .schema import FetchedValue
101 from .schema import ForeignKey
102 from .selectable import _SelectIterable
103 from .selectable import FromClause
104 from .selectable import NamedFromClause
105 from .selectable import TextualSelect
106 from .sqltypes import TupleType
107 from .type_api import TypeEngine
108 from .visitors import _CloneCallableType
109 from .visitors import _TraverseInternalsType
110 from .visitors import anon_map
111 from ..engine import Connection
112 from ..engine import Dialect
113 from ..engine.interfaces import _CoreMultiExecuteParams
114 from ..engine.interfaces import CacheStats
115 from ..engine.interfaces import CompiledCacheType
116 from ..engine.interfaces import CoreExecuteOptionsParameter
117 from ..engine.interfaces import SchemaTranslateMapType
118 from ..engine.result import Result
120_NUMERIC = Union[float, Decimal]
121_NUMBER = Union[float, int, Decimal]
123_T = TypeVar("_T", bound="Any")
124_T_co = TypeVar("_T_co", bound=Any, covariant=True)
125_OPT = TypeVar("_OPT", bound="Any")
126_NT = TypeVar("_NT", bound="_NUMERIC")
128_NMT = TypeVar("_NMT", bound="_NUMBER")
131@overload
132def literal(
133 value: Any,
134 type_: _TypeEngineArgument[_T],
135 literal_execute: bool = False,
136) -> BindParameter[_T]: ...
139@overload
140def literal(
141 value: _T,
142 type_: None = None,
143 literal_execute: bool = False,
144) -> BindParameter[_T]: ...
147@overload
148def literal(
149 value: Any,
150 type_: Optional[_TypeEngineArgument[Any]] = None,
151 literal_execute: bool = False,
152) -> BindParameter[Any]: ...
155def literal(
156 value: Any,
157 type_: Optional[_TypeEngineArgument[Any]] = None,
158 literal_execute: bool = False,
159) -> BindParameter[Any]:
160 r"""Return a literal clause, bound to a bind parameter.
162 Literal clauses are created automatically when non-
163 :class:`_expression.ClauseElement` objects (such as strings, ints, dates,
164 etc.) are
165 used in a comparison operation with a :class:`_expression.ColumnElement`
166 subclass,
167 such as a :class:`~sqlalchemy.schema.Column` object. Use this function
168 to force the generation of a literal clause, which will be created as a
169 :class:`BindParameter` with a bound value.
171 :param value: the value to be bound. Can be any Python object supported by
172 the underlying DB-API, or is translatable via the given type argument.
174 :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine` which will
175 provide bind-parameter translation for this literal.
177 :param literal_execute: optional bool, when True, the SQL engine will
178 attempt to render the bound value directly in the SQL statement at
179 execution time rather than providing as a parameter value.
181 .. versionadded:: 2.0
183 """
184 return coercions.expect(
185 roles.LiteralValueRole,
186 value,
187 type_=type_,
188 literal_execute=literal_execute,
189 )
192def literal_column(
193 text: str, type_: Optional[_TypeEngineArgument[_T]] = None
194) -> ColumnClause[_T]:
195 r"""Produce a :class:`.ColumnClause` object that has the
196 :paramref:`_expression.column.is_literal` flag set to True.
198 :func:`_expression.literal_column` is similar to
199 :func:`_expression.column`, except that
200 it is more often used as a "standalone" column expression that renders
201 exactly as stated; while :func:`_expression.column`
202 stores a string name that
203 will be assumed to be part of a table and may be quoted as such,
204 :func:`_expression.literal_column` can be that,
205 or any other arbitrary column-oriented
206 expression.
208 :param text: the text of the expression; can be any SQL expression.
209 Quoting rules will not be applied. To specify a column-name expression
210 which should be subject to quoting rules, use the :func:`column`
211 function.
213 :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine`
214 object which will
215 provide result-set translation and additional expression semantics for
216 this column. If left as ``None`` the type will be :class:`.NullType`.
218 .. seealso::
220 :func:`_expression.column`
222 :func:`_expression.text`
224 :ref:`tutorial_select_arbitrary_text`
226 """
227 return ColumnClause(text, type_=type_, is_literal=True)
230class CompilerElement(Visitable):
231 """base class for SQL elements that can be compiled to produce a
232 SQL string.
234 .. versionadded:: 2.0
236 """
238 __slots__ = ()
239 __visit_name__ = "compiler_element"
241 supports_execution = False
243 stringify_dialect = "default"
245 @util.preload_module("sqlalchemy.engine.default")
246 @util.preload_module("sqlalchemy.engine.url")
247 def compile(
248 self,
249 bind: Optional[_HasDialect] = None,
250 dialect: Optional[Dialect] = None,
251 **kw: Any,
252 ) -> Compiled:
253 """Compile this SQL expression.
255 The return value is a :class:`~.Compiled` object.
256 Calling ``str()`` or ``unicode()`` on the returned value will yield a
257 string representation of the result. The
258 :class:`~.Compiled` object also can return a
259 dictionary of bind parameter names and values
260 using the ``params`` accessor.
262 :param bind: An :class:`.Connection` or :class:`.Engine` which
263 can provide a :class:`.Dialect` in order to generate a
264 :class:`.Compiled` object. If the ``bind`` and
265 ``dialect`` parameters are both omitted, a default SQL compiler
266 is used.
268 :param column_keys: Used for INSERT and UPDATE statements, a list of
269 column names which should be present in the VALUES clause of the
270 compiled statement. If ``None``, all columns from the target table
271 object are rendered.
273 :param dialect: A :class:`.Dialect` instance which can generate
274 a :class:`.Compiled` object. This argument takes precedence over
275 the ``bind`` argument.
277 :param compile_kwargs: optional dictionary of additional parameters
278 that will be passed through to the compiler within all "visit"
279 methods. This allows any custom flag to be passed through to
280 a custom compilation construct, for example. It is also used
281 for the case of passing the ``literal_binds`` flag through::
283 from sqlalchemy.sql import table, column, select
285 t = table("t", column("x"))
287 s = select(t).where(t.c.x == 5)
289 print(s.compile(compile_kwargs={"literal_binds": True}))
291 .. seealso::
293 :ref:`faq_sql_expression_string`
295 """
297 if dialect is None:
298 if bind:
299 dialect = bind.dialect
300 elif self.stringify_dialect == "default":
301 dialect = self._default_dialect()
302 else:
303 url = util.preloaded.engine_url
304 dialect = url.URL.create(
305 self.stringify_dialect
306 ).get_dialect()()
308 return self._compiler(dialect, **kw)
310 def _default_dialect(self):
311 default = util.preloaded.engine_default
312 return default.StrCompileDialect()
314 def _compiler(self, dialect: Dialect, **kw: Any) -> Compiled:
315 """Return a compiler appropriate for this ClauseElement, given a
316 Dialect."""
318 if TYPE_CHECKING:
319 assert isinstance(self, ClauseElement)
320 return dialect.statement_compiler(dialect, self, **kw)
322 def __str__(self) -> str:
323 return str(self.compile())
326@inspection._self_inspects
327class ClauseElement(
328 SupportsWrappingAnnotations,
329 MemoizedHasCacheKey,
330 HasCopyInternals,
331 ExternallyTraversible,
332 CompilerElement,
333):
334 """Base class for elements of a programmatically constructed SQL
335 expression.
337 """
339 __visit_name__ = "clause"
341 if TYPE_CHECKING:
343 @util.memoized_property
344 def _propagate_attrs(self) -> _PropagateAttrsType:
345 """like annotations, however these propagate outwards liberally
346 as SQL constructs are built, and are set up at construction time.
348 """
349 ...
351 else:
352 _propagate_attrs = util.EMPTY_DICT
354 @util.ro_memoized_property
355 def description(self) -> Optional[str]:
356 return None
358 _is_clone_of: Optional[Self] = None
360 is_clause_element = True
361 is_selectable = False
362 is_dml = False
363 _is_column_element = False
364 _is_keyed_column_element = False
365 _is_table = False
366 _gen_static_annotations_cache_key = False
367 _is_textual = False
368 _is_from_clause = False
369 _is_returns_rows = False
370 _is_text_clause = False
371 _is_from_container = False
372 _is_select_container = False
373 _is_select_base = False
374 _is_select_statement = False
375 _is_bind_parameter = False
376 _is_clause_list = False
377 _is_lambda_element = False
378 _is_singleton_constant = False
379 _is_immutable = False
380 _is_star = False
382 @property
383 def _order_by_label_element(self) -> Optional[Label[Any]]:
384 return None
386 _cache_key_traversal: _CacheKeyTraversalType = None
388 negation_clause: ColumnElement[bool]
390 if typing.TYPE_CHECKING:
392 def get_children(
393 self, *, omit_attrs: typing_Tuple[str, ...] = ..., **kw: Any
394 ) -> Iterable[ClauseElement]: ...
396 @util.ro_non_memoized_property
397 def _from_objects(self) -> List[FromClause]:
398 return []
400 def _set_propagate_attrs(self, values: Mapping[str, Any]) -> Self:
401 # usually, self._propagate_attrs is empty here. one case where it's
402 # not is a subquery against ORM select, that is then pulled as a
403 # property of an aliased class. should all be good
405 # assert not self._propagate_attrs
407 self._propagate_attrs = util.immutabledict(values)
408 return self
410 def _default_compiler(self) -> SQLCompiler:
411 dialect = self._default_dialect()
412 return dialect.statement_compiler(dialect, self) # type: ignore
414 def _clone(self, **kw: Any) -> Self:
415 """Create a shallow copy of this ClauseElement.
417 This method may be used by a generative API. Its also used as
418 part of the "deep" copy afforded by a traversal that combines
419 the _copy_internals() method.
421 """
423 skip = self._memoized_keys
424 c = self.__class__.__new__(self.__class__)
426 if skip:
427 # ensure this iteration remains atomic
428 c.__dict__ = {
429 k: v for k, v in self.__dict__.copy().items() if k not in skip
430 }
431 else:
432 c.__dict__ = self.__dict__.copy()
434 # this is a marker that helps to "equate" clauses to each other
435 # when a Select returns its list of FROM clauses. the cloning
436 # process leaves around a lot of remnants of the previous clause
437 # typically in the form of column expressions still attached to the
438 # old table.
439 cc = self._is_clone_of
440 c._is_clone_of = cc if cc is not None else self
441 return c
443 def _negate_in_binary(self, negated_op, original_op):
444 """a hook to allow the right side of a binary expression to respond
445 to a negation of the binary expression.
447 Used for the special case of expanding bind parameter with IN.
449 """
450 return self
452 def _with_binary_element_type(self, type_):
453 """in the context of binary expression, convert the type of this
454 object to the one given.
456 applies only to :class:`_expression.ColumnElement` classes.
458 """
459 return self
461 @property
462 def _constructor(self):
463 """return the 'constructor' for this ClauseElement.
465 This is for the purposes for creating a new object of
466 this type. Usually, its just the element's __class__.
467 However, the "Annotated" version of the object overrides
468 to return the class of its proxied element.
470 """
471 return self.__class__
473 @HasMemoized.memoized_attribute
474 def _cloned_set(self):
475 """Return the set consisting all cloned ancestors of this
476 ClauseElement.
478 Includes this ClauseElement. This accessor tends to be used for
479 FromClause objects to identify 'equivalent' FROM clauses, regardless
480 of transformative operations.
482 """
483 s = util.column_set()
484 f: Optional[ClauseElement] = self
486 # note this creates a cycle, asserted in test_memusage. however,
487 # turning this into a plain @property adds tends of thousands of method
488 # calls to Core / ORM performance tests, so the small overhead
489 # introduced by the relatively small amount of short term cycles
490 # produced here is preferable
491 while f is not None:
492 s.add(f)
493 f = f._is_clone_of
494 return s
496 def _de_clone(self):
497 while self._is_clone_of is not None:
498 self = self._is_clone_of
499 return self
501 @property
502 def entity_namespace(self):
503 raise AttributeError(
504 "This SQL expression has no entity namespace "
505 "with which to filter from."
506 )
508 def __getstate__(self):
509 d = self.__dict__.copy()
510 d.pop("_is_clone_of", None)
511 d.pop("_generate_cache_key", None)
512 return d
514 def _execute_on_connection(
515 self,
516 connection: Connection,
517 distilled_params: _CoreMultiExecuteParams,
518 execution_options: CoreExecuteOptionsParameter,
519 ) -> Result[Any]:
520 if self.supports_execution:
521 if TYPE_CHECKING:
522 assert isinstance(self, Executable)
523 return connection._execute_clauseelement(
524 self, distilled_params, execution_options
525 )
526 else:
527 raise exc.ObjectNotExecutableError(self)
529 def _execute_on_scalar(
530 self,
531 connection: Connection,
532 distilled_params: _CoreMultiExecuteParams,
533 execution_options: CoreExecuteOptionsParameter,
534 ) -> Any:
535 """an additional hook for subclasses to provide a different
536 implementation for connection.scalar() vs. connection.execute().
538 .. versionadded:: 2.0
540 """
541 return self._execute_on_connection(
542 connection, distilled_params, execution_options
543 ).scalar()
545 def _get_embedded_bindparams(self) -> Sequence[BindParameter[Any]]:
546 """Return the list of :class:`.BindParameter` objects embedded in the
547 object.
549 This accomplishes the same purpose as ``visitors.traverse()`` or
550 similar would provide, however by making use of the cache key
551 it takes advantage of memoization of the key to result in fewer
552 net method calls, assuming the statement is also going to be
553 executed.
555 """
557 key = self._generate_cache_key()
558 if key is None:
559 bindparams: List[BindParameter[Any]] = []
561 traverse(self, {}, {"bindparam": bindparams.append})
562 return bindparams
564 else:
565 return key.bindparams
567 def unique_params(
568 self,
569 __optionaldict: Optional[Dict[str, Any]] = None,
570 **kwargs: Any,
571 ) -> Self:
572 """Return a copy with :func:`_expression.bindparam` elements
573 replaced.
575 Same functionality as :meth:`_expression.ClauseElement.params`,
576 except adds `unique=True`
577 to affected bind parameters so that multiple statements can be
578 used.
580 """
581 return self._replace_params(True, __optionaldict, kwargs)
583 def params(
584 self,
585 __optionaldict: Optional[Mapping[str, Any]] = None,
586 **kwargs: Any,
587 ) -> Self:
588 """Return a copy with :func:`_expression.bindparam` elements
589 replaced.
591 Returns a copy of this ClauseElement with
592 :func:`_expression.bindparam`
593 elements replaced with values taken from the given dictionary::
595 >>> clause = column("x") + bindparam("foo")
596 >>> print(clause.compile().params)
597 {'foo':None}
598 >>> print(clause.params({"foo": 7}).compile().params)
599 {'foo':7}
601 """
602 return self._replace_params(False, __optionaldict, kwargs)
604 def _replace_params(
605 self,
606 unique: bool,
607 optionaldict: Optional[Mapping[str, Any]],
608 kwargs: Dict[str, Any],
609 ) -> Self:
610 if optionaldict:
611 kwargs.update(optionaldict)
613 def visit_bindparam(bind: BindParameter[Any]) -> None:
614 if bind.key in kwargs:
615 bind.value = kwargs[bind.key]
616 bind.required = False
617 if unique:
618 bind._convert_to_unique()
620 return cloned_traverse(
621 self,
622 {"maintain_key": True, "detect_subquery_cols": True},
623 {"bindparam": visit_bindparam},
624 )
626 def compare(self, other: ClauseElement, **kw: Any) -> bool:
627 r"""Compare this :class:`_expression.ClauseElement` to
628 the given :class:`_expression.ClauseElement`.
630 Subclasses should override the default behavior, which is a
631 straight identity comparison.
633 \**kw are arguments consumed by subclass ``compare()`` methods and
634 may be used to modify the criteria for comparison
635 (see :class:`_expression.ColumnElement`).
637 """
638 return traversals.compare(self, other, **kw)
640 def self_group(
641 self, against: Optional[OperatorType] = None
642 ) -> ClauseElement:
643 """Apply a 'grouping' to this :class:`_expression.ClauseElement`.
645 This method is overridden by subclasses to return a "grouping"
646 construct, i.e. parenthesis. In particular it's used by "binary"
647 expressions to provide a grouping around themselves when placed into a
648 larger expression, as well as by :func:`_expression.select`
649 constructs when placed into the FROM clause of another
650 :func:`_expression.select`. (Note that subqueries should be
651 normally created using the :meth:`_expression.Select.alias` method,
652 as many
653 platforms require nested SELECT statements to be named).
655 As expressions are composed together, the application of
656 :meth:`self_group` is automatic - end-user code should never
657 need to use this method directly. Note that SQLAlchemy's
658 clause constructs take operator precedence into account -
659 so parenthesis might not be needed, for example, in
660 an expression like ``x OR (y AND z)`` - AND takes precedence
661 over OR.
663 The base :meth:`self_group` method of
664 :class:`_expression.ClauseElement`
665 just returns self.
666 """
667 return self
669 def _ungroup(self) -> ClauseElement:
670 """Return this :class:`_expression.ClauseElement`
671 without any groupings.
672 """
674 return self
676 def _compile_w_cache(
677 self,
678 dialect: Dialect,
679 *,
680 compiled_cache: Optional[CompiledCacheType],
681 column_keys: List[str],
682 for_executemany: bool = False,
683 schema_translate_map: Optional[SchemaTranslateMapType] = None,
684 **kw: Any,
685 ) -> typing_Tuple[
686 Compiled, Optional[Sequence[BindParameter[Any]]], CacheStats
687 ]:
688 elem_cache_key: Optional[CacheKey]
690 if compiled_cache is not None and dialect._supports_statement_cache:
691 elem_cache_key = self._generate_cache_key()
692 else:
693 elem_cache_key = None
695 if elem_cache_key is not None:
696 if TYPE_CHECKING:
697 assert compiled_cache is not None
699 cache_key, extracted_params = elem_cache_key
700 key = (
701 dialect,
702 cache_key,
703 tuple(column_keys),
704 bool(schema_translate_map),
705 for_executemany,
706 )
707 compiled_sql = compiled_cache.get(key)
709 if compiled_sql is None:
710 cache_hit = dialect.CACHE_MISS
711 compiled_sql = self._compiler(
712 dialect,
713 cache_key=elem_cache_key,
714 column_keys=column_keys,
715 for_executemany=for_executemany,
716 schema_translate_map=schema_translate_map,
717 **kw,
718 )
719 compiled_cache[key] = compiled_sql
720 else:
721 cache_hit = dialect.CACHE_HIT
722 else:
723 extracted_params = None
724 compiled_sql = self._compiler(
725 dialect,
726 cache_key=elem_cache_key,
727 column_keys=column_keys,
728 for_executemany=for_executemany,
729 schema_translate_map=schema_translate_map,
730 **kw,
731 )
733 if not dialect._supports_statement_cache:
734 cache_hit = dialect.NO_DIALECT_SUPPORT
735 elif compiled_cache is None:
736 cache_hit = dialect.CACHING_DISABLED
737 else:
738 cache_hit = dialect.NO_CACHE_KEY
740 return compiled_sql, extracted_params, cache_hit
742 def __invert__(self):
743 # undocumented element currently used by the ORM for
744 # relationship.contains()
745 if hasattr(self, "negation_clause"):
746 return self.negation_clause
747 else:
748 return self._negate()
750 def _negate(self) -> ClauseElement:
751 grouped = self.self_group(against=operators.inv)
752 assert isinstance(grouped, ColumnElement)
753 return UnaryExpression(grouped, operator=operators.inv)
755 def __bool__(self):
756 raise TypeError("Boolean value of this clause is not defined")
758 def __repr__(self):
759 friendly = self.description
760 if friendly is None:
761 return object.__repr__(self)
762 else:
763 return "<%s.%s at 0x%x; %s>" % (
764 self.__module__,
765 self.__class__.__name__,
766 id(self),
767 friendly,
768 )
771class DQLDMLClauseElement(ClauseElement):
772 """represents a :class:`.ClauseElement` that compiles to a DQL or DML
773 expression, not DDL.
775 .. versionadded:: 2.0
777 """
779 if typing.TYPE_CHECKING:
781 def _compiler(self, dialect: Dialect, **kw: Any) -> SQLCompiler:
782 """Return a compiler appropriate for this ClauseElement, given a
783 Dialect."""
784 ...
786 def compile( # noqa: A001
787 self,
788 bind: Optional[_HasDialect] = None,
789 dialect: Optional[Dialect] = None,
790 **kw: Any,
791 ) -> SQLCompiler: ...
794class CompilerColumnElement(
795 roles.DMLColumnRole,
796 roles.DDLConstraintColumnRole,
797 roles.ColumnsClauseRole,
798 CompilerElement,
799):
800 """A compiler-only column element used for ad-hoc string compilations.
802 .. versionadded:: 2.0
804 """
806 __slots__ = ()
808 _propagate_attrs = util.EMPTY_DICT
809 _is_collection_aggregate = False
812# SQLCoreOperations should be suiting the ExpressionElementRole
813# and ColumnsClauseRole. however the MRO issues become too elaborate
814# at the moment.
815class SQLCoreOperations(Generic[_T_co], ColumnOperators, TypingOnly):
816 __slots__ = ()
818 # annotations for comparison methods
819 # these are from operators->Operators / ColumnOperators,
820 # redefined with the specific types returned by ColumnElement hierarchies
821 if typing.TYPE_CHECKING:
823 @util.non_memoized_property
824 def _propagate_attrs(self) -> _PropagateAttrsType: ...
826 def operate(
827 self, op: OperatorType, *other: Any, **kwargs: Any
828 ) -> ColumnElement[Any]: ...
830 def reverse_operate(
831 self, op: OperatorType, other: Any, **kwargs: Any
832 ) -> ColumnElement[Any]: ...
834 @overload
835 def op(
836 self,
837 opstring: str,
838 precedence: int = ...,
839 is_comparison: bool = ...,
840 *,
841 return_type: _TypeEngineArgument[_OPT],
842 python_impl: Optional[Callable[..., Any]] = None,
843 ) -> Callable[[Any], BinaryExpression[_OPT]]: ...
845 @overload
846 def op(
847 self,
848 opstring: str,
849 precedence: int = ...,
850 is_comparison: bool = ...,
851 return_type: Optional[_TypeEngineArgument[Any]] = ...,
852 python_impl: Optional[Callable[..., Any]] = ...,
853 ) -> Callable[[Any], BinaryExpression[Any]]: ...
855 def op(
856 self,
857 opstring: str,
858 precedence: int = 0,
859 is_comparison: bool = False,
860 return_type: Optional[_TypeEngineArgument[Any]] = None,
861 python_impl: Optional[Callable[..., Any]] = None,
862 ) -> Callable[[Any], BinaryExpression[Any]]: ...
864 def bool_op(
865 self,
866 opstring: str,
867 precedence: int = 0,
868 python_impl: Optional[Callable[..., Any]] = None,
869 ) -> Callable[[Any], BinaryExpression[bool]]: ...
871 def __and__(self, other: Any) -> BooleanClauseList: ...
873 def __or__(self, other: Any) -> BooleanClauseList: ...
875 def __invert__(self) -> ColumnElement[_T_co]: ...
877 def __lt__(self, other: Any) -> ColumnElement[bool]: ...
879 def __le__(self, other: Any) -> ColumnElement[bool]: ...
881 # declare also that this class has an hash method otherwise
882 # it may be assumed to be None by type checkers since the
883 # object defines __eq__ and python sets it to None in that case:
884 # https://docs.python.org/3/reference/datamodel.html#object.__hash__
885 def __hash__(self) -> int: ...
887 def __eq__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
888 ...
890 def __ne__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
891 ...
893 def is_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
895 def is_not_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
897 def __gt__(self, other: Any) -> ColumnElement[bool]: ...
899 def __ge__(self, other: Any) -> ColumnElement[bool]: ...
901 def __neg__(self) -> UnaryExpression[_T_co]: ...
903 def __contains__(self, other: Any) -> ColumnElement[bool]: ...
905 def __getitem__(self, index: Any) -> ColumnElement[Any]: ...
907 @overload
908 def __lshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
910 @overload
911 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
913 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
915 @overload
916 def __rshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
918 @overload
919 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
921 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
923 @overload
924 def concat(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
926 @overload
927 def concat(self, other: Any) -> ColumnElement[Any]: ...
929 def concat(self, other: Any) -> ColumnElement[Any]: ...
931 def like(
932 self, other: Any, escape: Optional[str] = None
933 ) -> BinaryExpression[bool]: ...
935 def ilike(
936 self, other: Any, escape: Optional[str] = None
937 ) -> BinaryExpression[bool]: ...
939 def bitwise_xor(self, other: Any) -> BinaryExpression[Any]: ...
941 def bitwise_or(self, other: Any) -> BinaryExpression[Any]: ...
943 def bitwise_and(self, other: Any) -> BinaryExpression[Any]: ...
945 def bitwise_not(self) -> UnaryExpression[_T_co]: ...
947 def bitwise_lshift(self, other: Any) -> BinaryExpression[Any]: ...
949 def bitwise_rshift(self, other: Any) -> BinaryExpression[Any]: ...
951 def in_(
952 self,
953 other: Union[
954 Iterable[Any], BindParameter[Any], roles.InElementRole
955 ],
956 ) -> BinaryExpression[bool]: ...
958 def not_in(
959 self,
960 other: Union[
961 Iterable[Any], BindParameter[Any], roles.InElementRole
962 ],
963 ) -> BinaryExpression[bool]: ...
965 def notin_(
966 self,
967 other: Union[
968 Iterable[Any], BindParameter[Any], roles.InElementRole
969 ],
970 ) -> BinaryExpression[bool]: ...
972 def not_like(
973 self, other: Any, escape: Optional[str] = None
974 ) -> BinaryExpression[bool]: ...
976 def notlike(
977 self, other: Any, escape: Optional[str] = None
978 ) -> BinaryExpression[bool]: ...
980 def not_ilike(
981 self, other: Any, escape: Optional[str] = None
982 ) -> BinaryExpression[bool]: ...
984 def notilike(
985 self, other: Any, escape: Optional[str] = None
986 ) -> BinaryExpression[bool]: ...
988 def is_(self, other: Any) -> BinaryExpression[bool]: ...
990 def is_not(self, other: Any) -> BinaryExpression[bool]: ...
992 def isnot(self, other: Any) -> BinaryExpression[bool]: ...
994 def startswith(
995 self,
996 other: Any,
997 escape: Optional[str] = None,
998 autoescape: bool = False,
999 ) -> ColumnElement[bool]: ...
1001 def istartswith(
1002 self,
1003 other: Any,
1004 escape: Optional[str] = None,
1005 autoescape: bool = False,
1006 ) -> ColumnElement[bool]: ...
1008 def endswith(
1009 self,
1010 other: Any,
1011 escape: Optional[str] = None,
1012 autoescape: bool = False,
1013 ) -> ColumnElement[bool]: ...
1015 def iendswith(
1016 self,
1017 other: Any,
1018 escape: Optional[str] = None,
1019 autoescape: bool = False,
1020 ) -> ColumnElement[bool]: ...
1022 def contains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1024 def icontains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1026 def match(self, other: Any, **kwargs: Any) -> ColumnElement[bool]: ...
1028 def regexp_match(
1029 self, pattern: Any, flags: Optional[str] = None
1030 ) -> ColumnElement[bool]: ...
1032 def regexp_replace(
1033 self, pattern: Any, replacement: Any, flags: Optional[str] = None
1034 ) -> ColumnElement[str]: ...
1036 def desc(self) -> UnaryExpression[_T_co]: ...
1038 def asc(self) -> UnaryExpression[_T_co]: ...
1040 def nulls_first(self) -> UnaryExpression[_T_co]: ...
1042 def nullsfirst(self) -> UnaryExpression[_T_co]: ...
1044 def nulls_last(self) -> UnaryExpression[_T_co]: ...
1046 def nullslast(self) -> UnaryExpression[_T_co]: ...
1048 def collate(self, collation: str) -> CollationClause: ...
1050 def between(
1051 self, cleft: Any, cright: Any, symmetric: bool = False
1052 ) -> BinaryExpression[bool]: ...
1054 def distinct(self: _SQO[_T_co]) -> UnaryExpression[_T_co]: ...
1056 def any_(self) -> CollectionAggregate[Any]: ...
1058 def all_(self) -> CollectionAggregate[Any]: ...
1060 # numeric overloads. These need more tweaking
1061 # in particular they all need to have a variant for Optiona[_T]
1062 # because Optional only applies to the data side, not the expression
1063 # side
1065 @overload
1066 def __add__(
1067 self: _SQO[_NMT],
1068 other: Any,
1069 ) -> ColumnElement[_NMT]: ...
1071 @overload
1072 def __add__(
1073 self: _SQO[str],
1074 other: Any,
1075 ) -> ColumnElement[str]: ...
1077 @overload
1078 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1080 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1082 @overload
1083 def __radd__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1085 @overload
1086 def __radd__(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
1088 def __radd__(self, other: Any) -> ColumnElement[Any]: ...
1090 @overload
1091 def __sub__(
1092 self: _SQO[_NMT],
1093 other: Any,
1094 ) -> ColumnElement[_NMT]: ...
1096 @overload
1097 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1099 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1101 @overload
1102 def __rsub__(
1103 self: _SQO[_NMT],
1104 other: Any,
1105 ) -> ColumnElement[_NMT]: ...
1107 @overload
1108 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1110 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1112 @overload
1113 def __mul__(
1114 self: _SQO[_NMT],
1115 other: Any,
1116 ) -> ColumnElement[_NMT]: ...
1118 @overload
1119 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1121 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1123 @overload
1124 def __rmul__(
1125 self: _SQO[_NMT],
1126 other: Any,
1127 ) -> ColumnElement[_NMT]: ...
1129 @overload
1130 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1132 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1134 @overload
1135 def __mod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1137 @overload
1138 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1140 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1142 @overload
1143 def __rmod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1145 @overload
1146 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1148 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1150 @overload
1151 def __truediv__(
1152 self: _SQO[int], other: Any
1153 ) -> ColumnElement[_NUMERIC]: ...
1155 @overload
1156 def __truediv__(self: _SQO[_NT], other: Any) -> ColumnElement[_NT]: ...
1158 @overload
1159 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1161 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1163 @overload
1164 def __rtruediv__(
1165 self: _SQO[_NMT], other: Any
1166 ) -> ColumnElement[_NUMERIC]: ...
1168 @overload
1169 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1171 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1173 @overload
1174 def __floordiv__(
1175 self: _SQO[_NMT], other: Any
1176 ) -> ColumnElement[_NMT]: ...
1178 @overload
1179 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1181 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1183 @overload
1184 def __rfloordiv__(
1185 self: _SQO[_NMT], other: Any
1186 ) -> ColumnElement[_NMT]: ...
1188 @overload
1189 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1191 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1194class SQLColumnExpression(
1195 SQLCoreOperations[_T_co], roles.ExpressionElementRole[_T_co], TypingOnly
1196):
1197 """A type that may be used to indicate any SQL column element or object
1198 that acts in place of one.
1200 :class:`.SQLColumnExpression` is a base of
1201 :class:`.ColumnElement`, as well as within the bases of ORM elements
1202 such as :class:`.InstrumentedAttribute`, and may be used in :pep:`484`
1203 typing to indicate arguments or return values that should behave
1204 as column expressions.
1206 .. versionadded:: 2.0.0b4
1209 """
1211 __slots__ = ()
1214_SQO = SQLCoreOperations
1217class ColumnElement(
1218 roles.ColumnArgumentOrKeyRole,
1219 roles.StatementOptionRole,
1220 roles.WhereHavingRole,
1221 roles.BinaryElementRole[_T],
1222 roles.OrderByRole,
1223 roles.ColumnsClauseRole,
1224 roles.LimitOffsetRole,
1225 roles.DMLColumnRole,
1226 roles.DDLConstraintColumnRole,
1227 roles.DDLExpressionRole,
1228 SQLColumnExpression[_T],
1229 DQLDMLClauseElement,
1230):
1231 """Represent a column-oriented SQL expression suitable for usage in the
1232 "columns" clause, WHERE clause etc. of a statement.
1234 While the most familiar kind of :class:`_expression.ColumnElement` is the
1235 :class:`_schema.Column` object, :class:`_expression.ColumnElement`
1236 serves as the basis
1237 for any unit that may be present in a SQL expression, including
1238 the expressions themselves, SQL functions, bound parameters,
1239 literal expressions, keywords such as ``NULL``, etc.
1240 :class:`_expression.ColumnElement`
1241 is the ultimate base class for all such elements.
1243 A wide variety of SQLAlchemy Core functions work at the SQL expression
1244 level, and are intended to accept instances of
1245 :class:`_expression.ColumnElement` as
1246 arguments. These functions will typically document that they accept a
1247 "SQL expression" as an argument. What this means in terms of SQLAlchemy
1248 usually refers to an input which is either already in the form of a
1249 :class:`_expression.ColumnElement` object,
1250 or a value which can be **coerced** into
1251 one. The coercion rules followed by most, but not all, SQLAlchemy Core
1252 functions with regards to SQL expressions are as follows:
1254 * a literal Python value, such as a string, integer or floating
1255 point value, boolean, datetime, ``Decimal`` object, or virtually
1256 any other Python object, will be coerced into a "literal bound
1257 value". This generally means that a :func:`.bindparam` will be
1258 produced featuring the given value embedded into the construct; the
1259 resulting :class:`.BindParameter` object is an instance of
1260 :class:`_expression.ColumnElement`.
1261 The Python value will ultimately be sent
1262 to the DBAPI at execution time as a parameterized argument to the
1263 ``execute()`` or ``executemany()`` methods, after SQLAlchemy
1264 type-specific converters (e.g. those provided by any associated
1265 :class:`.TypeEngine` objects) are applied to the value.
1267 * any special object value, typically ORM-level constructs, which
1268 feature an accessor called ``__clause_element__()``. The Core
1269 expression system looks for this method when an object of otherwise
1270 unknown type is passed to a function that is looking to coerce the
1271 argument into a :class:`_expression.ColumnElement` and sometimes a
1272 :class:`_expression.SelectBase` expression.
1273 It is used within the ORM to
1274 convert from ORM-specific objects like mapped classes and
1275 mapped attributes into Core expression objects.
1277 * The Python ``None`` value is typically interpreted as ``NULL``,
1278 which in SQLAlchemy Core produces an instance of :func:`.null`.
1280 A :class:`_expression.ColumnElement` provides the ability to generate new
1281 :class:`_expression.ColumnElement`
1282 objects using Python expressions. This means that Python operators
1283 such as ``==``, ``!=`` and ``<`` are overloaded to mimic SQL operations,
1284 and allow the instantiation of further :class:`_expression.ColumnElement`
1285 instances
1286 which are composed from other, more fundamental
1287 :class:`_expression.ColumnElement`
1288 objects. For example, two :class:`.ColumnClause` objects can be added
1289 together with the addition operator ``+`` to produce
1290 a :class:`.BinaryExpression`.
1291 Both :class:`.ColumnClause` and :class:`.BinaryExpression` are subclasses
1292 of :class:`_expression.ColumnElement`:
1294 .. sourcecode:: pycon+sql
1296 >>> from sqlalchemy.sql import column
1297 >>> column("a") + column("b")
1298 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
1299 >>> print(column("a") + column("b"))
1300 {printsql}a + b
1302 .. seealso::
1304 :class:`_schema.Column`
1306 :func:`_expression.column`
1308 """
1310 __visit_name__ = "column_element"
1312 primary_key: bool = False
1313 _is_clone_of: Optional[ColumnElement[_T]]
1314 _is_column_element = True
1315 _insert_sentinel: bool = False
1316 _omit_from_statements = False
1317 _is_collection_aggregate = False
1319 foreign_keys: AbstractSet[ForeignKey] = frozenset()
1321 @util.memoized_property
1322 def _proxies(self) -> List[ColumnElement[Any]]:
1323 return []
1325 @util.non_memoized_property
1326 def _tq_label(self) -> Optional[str]:
1327 """The named label that can be used to target
1328 this column in a result set in a "table qualified" context.
1330 This label is almost always the label used when
1331 rendering <expr> AS <label> in a SELECT statement when using
1332 the LABEL_STYLE_TABLENAME_PLUS_COL label style, which is what the
1333 legacy ORM ``Query`` object uses as well.
1335 For a regular Column bound to a Table, this is typically the label
1336 <tablename>_<columnname>. For other constructs, different rules
1337 may apply, such as anonymized labels and others.
1339 .. versionchanged:: 1.4.21 renamed from ``._label``
1341 """
1342 return None
1344 key: Optional[str] = None
1345 """The 'key' that in some circumstances refers to this object in a
1346 Python namespace.
1348 This typically refers to the "key" of the column as present in the
1349 ``.c`` collection of a selectable, e.g. ``sometable.c["somekey"]`` would
1350 return a :class:`_schema.Column` with a ``.key`` of "somekey".
1352 """
1354 @HasMemoized.memoized_attribute
1355 def _tq_key_label(self) -> Optional[str]:
1356 """A label-based version of 'key' that in some circumstances refers
1357 to this object in a Python namespace.
1360 _tq_key_label comes into play when a select() statement is constructed
1361 with apply_labels(); in this case, all Column objects in the ``.c``
1362 collection are rendered as <tablename>_<columnname> in SQL; this is
1363 essentially the value of ._label. But to locate those columns in the
1364 ``.c`` collection, the name is along the lines of <tablename>_<key>;
1365 that's the typical value of .key_label.
1367 .. versionchanged:: 1.4.21 renamed from ``._key_label``
1369 """
1370 return self._proxy_key
1372 @property
1373 def _key_label(self) -> Optional[str]:
1374 """legacy; renamed to _tq_key_label"""
1375 return self._tq_key_label
1377 @property
1378 def _label(self) -> Optional[str]:
1379 """legacy; renamed to _tq_label"""
1380 return self._tq_label
1382 @property
1383 def _non_anon_label(self) -> Optional[str]:
1384 """the 'name' that naturally applies this element when rendered in
1385 SQL.
1387 Concretely, this is the "name" of a column or a label in a
1388 SELECT statement; ``<columnname>`` and ``<labelname>`` below:
1390 .. sourcecode:: sql
1392 SELECT <columnmame> FROM table
1394 SELECT column AS <labelname> FROM table
1396 Above, the two names noted will be what's present in the DBAPI
1397 ``cursor.description`` as the names.
1399 If this attribute returns ``None``, it means that the SQL element as
1400 written does not have a 100% fully predictable "name" that would appear
1401 in the ``cursor.description``. Examples include SQL functions, CAST
1402 functions, etc. While such things do return names in
1403 ``cursor.description``, they are only predictable on a
1404 database-specific basis; e.g. an expression like ``MAX(table.col)`` may
1405 appear as the string ``max`` on one database (like PostgreSQL) or may
1406 appear as the whole expression ``max(table.col)`` on SQLite.
1408 The default implementation looks for a ``.name`` attribute on the
1409 object, as has been the precedent established in SQLAlchemy for many
1410 years. An exception is made on the ``FunctionElement`` subclass
1411 so that the return value is always ``None``.
1413 .. versionadded:: 1.4.21
1417 """
1418 return getattr(self, "name", None)
1420 _render_label_in_columns_clause = True
1421 """A flag used by select._columns_plus_names that helps to determine
1422 we are actually going to render in terms of "SELECT <col> AS <label>".
1423 This flag can be returned as False for some Column objects that want
1424 to be rendered as simple "SELECT <col>"; typically columns that don't have
1425 any parent table and are named the same as what the label would be
1426 in any case.
1428 """
1430 _allow_label_resolve = True
1431 """A flag that can be flipped to prevent a column from being resolvable
1432 by string label name.
1434 The joined eager loader strategy in the ORM uses this, for example.
1436 """
1438 _is_implicitly_boolean = False
1440 _alt_names: Sequence[str] = ()
1442 @overload
1443 def self_group(self, against: None = None) -> ColumnElement[_T]: ...
1445 @overload
1446 def self_group(
1447 self, against: Optional[OperatorType] = None
1448 ) -> ColumnElement[Any]: ...
1450 def self_group(
1451 self, against: Optional[OperatorType] = None
1452 ) -> ColumnElement[Any]:
1453 if (
1454 against in (operators.and_, operators.or_, operators._asbool)
1455 and self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity
1456 ):
1457 return AsBoolean(self, operators.is_true, operators.is_false)
1458 elif against in (operators.any_op, operators.all_op):
1459 return Grouping(self)
1460 else:
1461 return self
1463 @overload
1464 def _negate(self: ColumnElement[bool]) -> ColumnElement[bool]: ...
1466 @overload
1467 def _negate(self: ColumnElement[_T]) -> ColumnElement[_T]: ...
1469 def _negate(self) -> ColumnElement[Any]:
1470 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
1471 return AsBoolean(self, operators.is_false, operators.is_true)
1472 else:
1473 grouped = self.self_group(against=operators.inv)
1474 assert isinstance(grouped, ColumnElement)
1475 return UnaryExpression(
1476 grouped, operator=operators.inv, wraps_column_expression=True
1477 )
1479 type: TypeEngine[_T]
1481 if not TYPE_CHECKING:
1483 @util.memoized_property
1484 def type(self) -> TypeEngine[_T]: # noqa: A001
1485 # used for delayed setup of
1486 # type_api
1487 return type_api.NULLTYPE
1489 @HasMemoized.memoized_attribute
1490 def comparator(self) -> TypeEngine.Comparator[_T]:
1491 try:
1492 comparator_factory = self.type.comparator_factory
1493 except AttributeError as err:
1494 raise TypeError(
1495 "Object %r associated with '.type' attribute "
1496 "is not a TypeEngine class or object" % self.type
1497 ) from err
1498 else:
1499 return comparator_factory(self)
1501 def __setstate__(self, state):
1502 self.__dict__.update(state)
1504 def __getattr__(self, key: str) -> Any:
1505 try:
1506 return getattr(self.comparator, key)
1507 except AttributeError as err:
1508 raise AttributeError(
1509 "Neither %r object nor %r object has an attribute %r"
1510 % (
1511 type(self).__name__,
1512 type(self.comparator).__name__,
1513 key,
1514 )
1515 ) from err
1517 def operate(
1518 self,
1519 op: operators.OperatorType,
1520 *other: Any,
1521 **kwargs: Any,
1522 ) -> ColumnElement[Any]:
1523 return op(self.comparator, *other, **kwargs) # type: ignore[no-any-return] # noqa: E501
1525 def reverse_operate(
1526 self, op: operators.OperatorType, other: Any, **kwargs: Any
1527 ) -> ColumnElement[Any]:
1528 return op(other, self.comparator, **kwargs) # type: ignore[no-any-return] # noqa: E501
1530 def _bind_param(
1531 self,
1532 operator: operators.OperatorType,
1533 obj: Any,
1534 type_: Optional[TypeEngine[_T]] = None,
1535 expanding: bool = False,
1536 ) -> BindParameter[_T]:
1537 return BindParameter(
1538 None,
1539 obj,
1540 _compared_to_operator=operator,
1541 type_=type_,
1542 _compared_to_type=self.type,
1543 unique=True,
1544 expanding=expanding,
1545 )
1547 @property
1548 def expression(self) -> ColumnElement[Any]:
1549 """Return a column expression.
1551 Part of the inspection interface; returns self.
1553 """
1554 return self
1556 @property
1557 def _select_iterable(self) -> _SelectIterable:
1558 return (self,)
1560 @util.memoized_property
1561 def base_columns(self) -> FrozenSet[ColumnElement[Any]]:
1562 return frozenset(c for c in self.proxy_set if not c._proxies)
1564 @util.memoized_property
1565 def proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1566 """set of all columns we are proxying
1568 as of 2.0 this is explicitly deannotated columns. previously it was
1569 effectively deannotated columns but wasn't enforced. annotated
1570 columns should basically not go into sets if at all possible because
1571 their hashing behavior is very non-performant.
1573 """
1574 return frozenset([self._deannotate()]).union(
1575 itertools.chain(*[c.proxy_set for c in self._proxies])
1576 )
1578 @util.memoized_property
1579 def _expanded_proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1580 return frozenset(_expand_cloned(self.proxy_set))
1582 def _uncached_proxy_list(self) -> List[ColumnElement[Any]]:
1583 """An 'uncached' version of proxy set.
1585 This list includes annotated columns which perform very poorly in
1586 set operations.
1588 """
1590 return [self] + list(
1591 itertools.chain(*[c._uncached_proxy_list() for c in self._proxies])
1592 )
1594 def shares_lineage(self, othercolumn: ColumnElement[Any]) -> bool:
1595 """Return True if the given :class:`_expression.ColumnElement`
1596 has a common ancestor to this :class:`_expression.ColumnElement`."""
1598 return bool(self.proxy_set.intersection(othercolumn.proxy_set))
1600 def _compare_name_for_result(self, other: ColumnElement[Any]) -> bool:
1601 """Return True if the given column element compares to this one
1602 when targeting within a result row."""
1604 return (
1605 hasattr(other, "name")
1606 and hasattr(self, "name")
1607 and other.name == self.name
1608 )
1610 @HasMemoized.memoized_attribute
1611 def _proxy_key(self) -> Optional[str]:
1612 if self._annotations and "proxy_key" in self._annotations:
1613 return cast(str, self._annotations["proxy_key"])
1615 name = self.key
1616 if not name:
1617 # there's a bit of a seeming contradiction which is that the
1618 # "_non_anon_label" of a column can in fact be an
1619 # "_anonymous_label"; this is when it's on a column that is
1620 # proxying for an anonymous expression in a subquery.
1621 name = self._non_anon_label
1623 if isinstance(name, _anonymous_label):
1624 return None
1625 else:
1626 return name
1628 @HasMemoized.memoized_attribute
1629 def _expression_label(self) -> Optional[str]:
1630 """a suggested label to use in the case that the column has no name,
1631 which should be used if possible as the explicit 'AS <label>'
1632 where this expression would normally have an anon label.
1634 this is essentially mostly what _proxy_key does except it returns
1635 None if the column has a normal name that can be used.
1637 """
1639 if getattr(self, "name", None) is not None:
1640 return None
1641 elif self._annotations and "proxy_key" in self._annotations:
1642 return cast(str, self._annotations["proxy_key"])
1643 else:
1644 return None
1646 def _make_proxy(
1647 self,
1648 selectable: FromClause,
1649 *,
1650 primary_key: ColumnSet,
1651 foreign_keys: Set[KeyedColumnElement[Any]],
1652 name: Optional[str] = None,
1653 key: Optional[str] = None,
1654 name_is_truncatable: bool = False,
1655 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
1656 **kw: Any,
1657 ) -> typing_Tuple[str, ColumnClause[_T]]:
1658 """Create a new :class:`_expression.ColumnElement` representing this
1659 :class:`_expression.ColumnElement` as it appears in the select list of
1660 a descending selectable.
1662 """
1663 if name is None:
1664 name = self._anon_name_label
1665 if key is None:
1666 key = self._proxy_key
1667 else:
1668 key = name
1670 assert key is not None
1672 co: ColumnClause[_T] = ColumnClause(
1673 (
1674 coercions.expect(roles.TruncatedLabelRole, name)
1675 if name_is_truncatable
1676 else name
1677 ),
1678 type_=getattr(self, "type", None),
1679 _selectable=selectable,
1680 )
1682 co._propagate_attrs = selectable._propagate_attrs
1683 if compound_select_cols:
1684 co._proxies = list(compound_select_cols)
1685 else:
1686 co._proxies = [self]
1687 if selectable._is_clone_of is not None:
1688 co._is_clone_of = selectable._is_clone_of.columns.get(key)
1689 return key, co
1691 def cast(self, type_: _TypeEngineArgument[_OPT]) -> Cast[_OPT]:
1692 """Produce a type cast, i.e. ``CAST(<expression> AS <type>)``.
1694 This is a shortcut to the :func:`_expression.cast` function.
1696 .. seealso::
1698 :ref:`tutorial_casts`
1700 :func:`_expression.cast`
1702 :func:`_expression.type_coerce`
1704 """
1705 return Cast(self, type_)
1707 def label(self, name: Optional[str]) -> Label[_T]:
1708 """Produce a column label, i.e. ``<columnname> AS <name>``.
1710 This is a shortcut to the :func:`_expression.label` function.
1712 If 'name' is ``None``, an anonymous label name will be generated.
1714 """
1715 return Label(name, self, self.type)
1717 def _anon_label(
1718 self, seed: Optional[str], add_hash: Optional[int] = None
1719 ) -> _anonymous_label:
1720 while self._is_clone_of is not None:
1721 self = self._is_clone_of
1723 # as of 1.4 anonymous label for ColumnElement uses hash(), not id(),
1724 # as the identifier, because a column and its annotated version are
1725 # the same thing in a SQL statement
1726 hash_value = hash(self)
1728 if add_hash:
1729 # this path is used for disambiguating anon labels that would
1730 # otherwise be the same name for the same element repeated.
1731 # an additional numeric value is factored in for each label.
1733 # shift hash(self) (which is id(self), typically 8 byte integer)
1734 # 16 bits leftward. fill extra add_hash on right
1735 assert add_hash < (2 << 15)
1736 assert seed
1737 hash_value = (hash_value << 16) | add_hash
1739 # extra underscore is added for labels with extra hash
1740 # values, to isolate the "deduped anon" namespace from the
1741 # regular namespace. eliminates chance of these
1742 # manufactured hash values overlapping with regular ones for some
1743 # undefined python interpreter
1744 seed = seed + "_"
1746 if isinstance(seed, _anonymous_label):
1747 return _anonymous_label.safe_construct(
1748 hash_value, "", enclosing_label=seed
1749 )
1751 return _anonymous_label.safe_construct(hash_value, seed or "anon")
1753 @util.memoized_property
1754 def _anon_name_label(self) -> str:
1755 """Provides a constant 'anonymous label' for this ColumnElement.
1757 This is a label() expression which will be named at compile time.
1758 The same label() is returned each time ``anon_label`` is called so
1759 that expressions can reference ``anon_label`` multiple times,
1760 producing the same label name at compile time.
1762 The compiler uses this function automatically at compile time
1763 for expressions that are known to be 'unnamed' like binary
1764 expressions and function calls.
1766 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1767 public and is renamed to _anon_name_label. anon_name exists
1768 for backwards compat
1770 """
1771 name = getattr(self, "name", None)
1772 return self._anon_label(name)
1774 @util.memoized_property
1775 def _anon_key_label(self) -> _anonymous_label:
1776 """Provides a constant 'anonymous key label' for this ColumnElement.
1778 Compare to ``anon_label``, except that the "key" of the column,
1779 if available, is used to generate the label.
1781 This is used when a deduplicating key is placed into the columns
1782 collection of a selectable.
1784 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1785 public and is renamed to _anon_key_label. anon_key_label exists
1786 for backwards compat
1788 """
1789 return self._anon_label(self._proxy_key)
1791 @property
1792 @util.deprecated(
1793 "1.4",
1794 "The :attr:`_expression.ColumnElement.anon_label` attribute is now "
1795 "private, and the public accessor is deprecated.",
1796 )
1797 def anon_label(self) -> str:
1798 return self._anon_name_label
1800 @property
1801 @util.deprecated(
1802 "1.4",
1803 "The :attr:`_expression.ColumnElement.anon_key_label` attribute is "
1804 "now private, and the public accessor is deprecated.",
1805 )
1806 def anon_key_label(self) -> str:
1807 return self._anon_key_label
1809 def _dedupe_anon_label_idx(self, idx: int) -> str:
1810 """label to apply to a column that is anon labeled, but repeated
1811 in the SELECT, so that we have to make an "extra anon" label that
1812 disambiguates it from the previous appearance.
1814 these labels come out like "foo_bar_id__1" and have double underscores
1815 in them.
1817 """
1818 label = getattr(self, "name", None)
1820 # current convention is that if the element doesn't have a
1821 # ".name" (usually because it is not NamedColumn), we try to
1822 # use a "table qualified" form for the "dedupe anon" label,
1823 # based on the notion that a label like
1824 # "CAST(casttest.v1 AS DECIMAL) AS casttest_v1__1" looks better than
1825 # "CAST(casttest.v1 AS DECIMAL) AS anon__1"
1827 if label is None:
1828 return self._dedupe_anon_tq_label_idx(idx)
1829 else:
1830 return self._anon_label(label, add_hash=idx)
1832 @util.memoized_property
1833 def _anon_tq_label(self) -> _anonymous_label:
1834 return self._anon_label(getattr(self, "_tq_label", None))
1836 @util.memoized_property
1837 def _anon_tq_key_label(self) -> _anonymous_label:
1838 return self._anon_label(getattr(self, "_tq_key_label", None))
1840 def _dedupe_anon_tq_label_idx(self, idx: int) -> _anonymous_label:
1841 label = getattr(self, "_tq_label", None) or "anon"
1843 return self._anon_label(label, add_hash=idx)
1846class KeyedColumnElement(ColumnElement[_T]):
1847 """ColumnElement where ``.key`` is non-None."""
1849 _is_keyed_column_element = True
1851 key: str
1854class WrapsColumnExpression(ColumnElement[_T]):
1855 """Mixin that defines a :class:`_expression.ColumnElement`
1856 as a wrapper with special
1857 labeling behavior for an expression that already has a name.
1859 .. versionadded:: 1.4
1861 .. seealso::
1863 :ref:`change_4449`
1866 """
1868 @property
1869 def wrapped_column_expression(self) -> ColumnElement[_T]:
1870 raise NotImplementedError()
1872 @util.non_memoized_property
1873 def _tq_label(self) -> Optional[str]:
1874 wce = self.wrapped_column_expression
1875 if hasattr(wce, "_tq_label"):
1876 return wce._tq_label
1877 else:
1878 return None
1880 @property
1881 def _label(self) -> Optional[str]:
1882 return self._tq_label
1884 @property
1885 def _non_anon_label(self) -> Optional[str]:
1886 return None
1888 @util.non_memoized_property
1889 def _anon_name_label(self) -> str:
1890 wce = self.wrapped_column_expression
1892 # this logic tries to get the WrappedColumnExpression to render
1893 # with "<expr> AS <name>", where "<name>" is the natural name
1894 # within the expression itself. e.g. "CAST(table.foo) AS foo".
1895 if not wce._is_text_clause:
1896 nal = wce._non_anon_label
1897 if nal:
1898 return nal
1899 elif hasattr(wce, "_anon_name_label"):
1900 return wce._anon_name_label
1901 return super()._anon_name_label
1903 def _dedupe_anon_label_idx(self, idx: int) -> str:
1904 wce = self.wrapped_column_expression
1905 nal = wce._non_anon_label
1906 if nal:
1907 return self._anon_label(nal + "_")
1908 else:
1909 return self._dedupe_anon_tq_label_idx(idx)
1911 @property
1912 def _proxy_key(self):
1913 wce = self.wrapped_column_expression
1915 if not wce._is_text_clause:
1916 return wce._proxy_key
1917 return super()._proxy_key
1920class BindParameter(roles.InElementRole, KeyedColumnElement[_T]):
1921 r"""Represent a "bound expression".
1923 :class:`.BindParameter` is invoked explicitly using the
1924 :func:`.bindparam` function, as in::
1926 from sqlalchemy import bindparam
1928 stmt = select(users_table).where(
1929 users_table.c.name == bindparam("username")
1930 )
1932 Detailed discussion of how :class:`.BindParameter` is used is
1933 at :func:`.bindparam`.
1935 .. seealso::
1937 :func:`.bindparam`
1939 """
1941 __visit_name__ = "bindparam"
1943 _traverse_internals: _TraverseInternalsType = [
1944 ("key", InternalTraversal.dp_anon_name),
1945 ("type", InternalTraversal.dp_type),
1946 ("callable", InternalTraversal.dp_plain_dict),
1947 ("value", InternalTraversal.dp_plain_obj),
1948 ("literal_execute", InternalTraversal.dp_boolean),
1949 ]
1951 key: str
1952 type: TypeEngine[_T]
1953 value: Optional[_T]
1955 _is_crud = False
1956 _is_bind_parameter = True
1957 _key_is_anon = False
1959 # bindparam implements its own _gen_cache_key() method however
1960 # we check subclasses for this flag, else no cache key is generated
1961 inherit_cache = True
1963 def __init__(
1964 self,
1965 key: Optional[str],
1966 value: Any = _NoArg.NO_ARG,
1967 type_: Optional[_TypeEngineArgument[_T]] = None,
1968 unique: bool = False,
1969 required: Union[bool, Literal[_NoArg.NO_ARG]] = _NoArg.NO_ARG,
1970 quote: Optional[bool] = None,
1971 callable_: Optional[Callable[[], Any]] = None,
1972 expanding: bool = False,
1973 isoutparam: bool = False,
1974 literal_execute: bool = False,
1975 _compared_to_operator: Optional[OperatorType] = None,
1976 _compared_to_type: Optional[TypeEngine[Any]] = None,
1977 _is_crud: bool = False,
1978 ):
1979 if required is _NoArg.NO_ARG:
1980 required = value is _NoArg.NO_ARG and callable_ is None
1981 if value is _NoArg.NO_ARG:
1982 value = None
1984 if quote is not None:
1985 key = quoted_name.construct(key, quote)
1987 if unique:
1988 self.key = _anonymous_label.safe_construct(
1989 id(self),
1990 (
1991 key
1992 if key is not None
1993 and not isinstance(key, _anonymous_label)
1994 else "param"
1995 ),
1996 sanitize_key=True,
1997 )
1998 self._key_is_anon = True
1999 elif key:
2000 self.key = key
2001 else:
2002 self.key = _anonymous_label.safe_construct(id(self), "param")
2003 self._key_is_anon = True
2005 # identifying key that won't change across
2006 # clones, used to identify the bind's logical
2007 # identity
2008 self._identifying_key = self.key
2010 # key that was passed in the first place, used to
2011 # generate new keys
2012 self._orig_key = key or "param"
2014 self.unique = unique
2015 self.value = value
2016 self.callable = callable_
2017 self.isoutparam = isoutparam
2018 self.required = required
2020 # indicate an "expanding" parameter; the compiler sets this
2021 # automatically in the compiler _render_in_expr_w_bindparam method
2022 # for an IN expression
2023 self.expanding = expanding
2025 # this is another hint to help w/ expanding and is typically
2026 # set in the compiler _render_in_expr_w_bindparam method for an
2027 # IN expression
2028 self.expand_op = None
2030 self.literal_execute = literal_execute
2031 if _is_crud:
2032 self._is_crud = True
2034 if type_ is None:
2035 if expanding:
2036 if value:
2037 check_value = value[0]
2038 else:
2039 check_value = type_api._NO_VALUE_IN_LIST
2040 else:
2041 check_value = value
2042 if _compared_to_type is not None:
2043 self.type = _compared_to_type.coerce_compared_value(
2044 _compared_to_operator, check_value
2045 )
2046 else:
2047 self.type = type_api._resolve_value_to_type(check_value)
2048 elif isinstance(type_, type):
2049 self.type = type_()
2050 elif is_tuple_type(type_):
2051 if value:
2052 if expanding:
2053 check_value = value[0]
2054 else:
2055 check_value = value
2056 cast("BindParameter[typing_Tuple[Any, ...]]", self).type = (
2057 type_._resolve_values_to_types(check_value)
2058 )
2059 else:
2060 cast("BindParameter[typing_Tuple[Any, ...]]", self).type = (
2061 type_
2062 )
2063 else:
2064 self.type = type_
2066 def _with_value(self, value, maintain_key=False, required=NO_ARG):
2067 """Return a copy of this :class:`.BindParameter` with the given value
2068 set.
2069 """
2070 cloned = self._clone(maintain_key=maintain_key)
2071 cloned.value = value
2072 cloned.callable = None
2073 cloned.required = required if required is not NO_ARG else self.required
2074 if cloned.type is type_api.NULLTYPE:
2075 cloned.type = type_api._resolve_value_to_type(value)
2076 return cloned
2078 @property
2079 def effective_value(self) -> Optional[_T]:
2080 """Return the value of this bound parameter,
2081 taking into account if the ``callable`` parameter
2082 was set.
2084 The ``callable`` value will be evaluated
2085 and returned if present, else ``value``.
2087 """
2088 if self.callable:
2089 # TODO: set up protocol for bind parameter callable
2090 return self.callable() # type: ignore
2091 else:
2092 return self.value
2094 def render_literal_execute(self) -> BindParameter[_T]:
2095 """Produce a copy of this bound parameter that will enable the
2096 :paramref:`_sql.BindParameter.literal_execute` flag.
2098 The :paramref:`_sql.BindParameter.literal_execute` flag will
2099 have the effect of the parameter rendered in the compiled SQL
2100 string using ``[POSTCOMPILE]`` form, which is a special form that
2101 is converted to be a rendering of the literal value of the parameter
2102 at SQL execution time. The rationale is to support caching
2103 of SQL statement strings that can embed per-statement literal values,
2104 such as LIMIT and OFFSET parameters, in the final SQL string that
2105 is passed to the DBAPI. Dialects in particular may want to use
2106 this method within custom compilation schemes.
2108 .. versionadded:: 1.4.5
2110 .. seealso::
2112 :ref:`engine_thirdparty_caching`
2114 """
2115 c = ClauseElement._clone(self)
2116 c.literal_execute = True
2117 return c
2119 def _negate_in_binary(self, negated_op, original_op):
2120 if self.expand_op is original_op:
2121 bind = self._clone()
2122 bind.expand_op = negated_op
2123 return bind
2124 else:
2125 return self
2127 def _with_binary_element_type(self, type_):
2128 c = ClauseElement._clone(self)
2129 c.type = type_
2130 return c
2132 def _clone(self, maintain_key: bool = False, **kw: Any) -> Self:
2133 c = ClauseElement._clone(self, **kw)
2134 # ensure all the BindParameter objects stay in cloned set.
2135 # in #7823, we changed "clone" so that a clone only keeps a reference
2136 # to the "original" element, since for column correspondence, that's
2137 # all we need. However, for BindParam, _cloned_set is used by
2138 # the "cache key bind match" lookup, which means if any of those
2139 # interim BindParameter objects became part of a cache key in the
2140 # cache, we need it. So here, make sure all clones keep carrying
2141 # forward.
2142 c._cloned_set.update(self._cloned_set)
2143 if not maintain_key and self.unique:
2144 c.key = _anonymous_label.safe_construct(
2145 id(c), c._orig_key or "param", sanitize_key=True
2146 )
2147 return c
2149 def _gen_cache_key(self, anon_map, bindparams):
2150 _gen_cache_ok = self.__class__.__dict__.get("inherit_cache", False)
2152 if not _gen_cache_ok:
2153 if anon_map is not None:
2154 anon_map[NO_CACHE] = True
2155 return None
2157 id_, found = anon_map.get_anon(self)
2158 if found:
2159 return (id_, self.__class__)
2161 if bindparams is not None:
2162 bindparams.append(self)
2164 return (
2165 id_,
2166 self.__class__,
2167 self.type._static_cache_key,
2168 self.key % anon_map if self._key_is_anon else self.key,
2169 self.literal_execute,
2170 )
2172 def _convert_to_unique(self):
2173 if not self.unique:
2174 self.unique = True
2175 self.key = _anonymous_label.safe_construct(
2176 id(self), self._orig_key or "param", sanitize_key=True
2177 )
2179 def __getstate__(self):
2180 """execute a deferred value for serialization purposes."""
2182 d = self.__dict__.copy()
2183 v = self.value
2184 if self.callable:
2185 v = self.callable()
2186 d["callable"] = None
2187 d["value"] = v
2188 return d
2190 def __setstate__(self, state):
2191 if state.get("unique", False):
2192 state["key"] = _anonymous_label.safe_construct(
2193 id(self), state.get("_orig_key", "param"), sanitize_key=True
2194 )
2195 self.__dict__.update(state)
2197 def __repr__(self):
2198 return "%s(%r, %r, type_=%r)" % (
2199 self.__class__.__name__,
2200 self.key,
2201 self.value,
2202 self.type,
2203 )
2206class TypeClause(DQLDMLClauseElement):
2207 """Handle a type keyword in a SQL statement.
2209 Used by the ``Case`` statement.
2211 """
2213 __visit_name__ = "typeclause"
2215 _traverse_internals: _TraverseInternalsType = [
2216 ("type", InternalTraversal.dp_type)
2217 ]
2218 type: TypeEngine[Any]
2220 def __init__(self, type_: TypeEngine[Any]):
2221 self.type = type_
2224class TextClause(
2225 roles.DDLConstraintColumnRole,
2226 roles.DDLExpressionRole,
2227 roles.StatementOptionRole,
2228 roles.WhereHavingRole,
2229 roles.OrderByRole,
2230 roles.FromClauseRole,
2231 roles.SelectStatementRole,
2232 roles.InElementRole,
2233 Generative,
2234 Executable,
2235 DQLDMLClauseElement,
2236 roles.BinaryElementRole[Any],
2237 inspection.Inspectable["TextClause"],
2238):
2239 """Represent a literal SQL text fragment.
2241 E.g.::
2243 from sqlalchemy import text
2245 t = text("SELECT * FROM users")
2246 result = connection.execute(t)
2248 The :class:`_expression.TextClause` construct is produced using the
2249 :func:`_expression.text`
2250 function; see that function for full documentation.
2252 .. seealso::
2254 :func:`_expression.text`
2256 """
2258 __visit_name__ = "textclause"
2260 _traverse_internals: _TraverseInternalsType = [
2261 ("_bindparams", InternalTraversal.dp_string_clauseelement_dict),
2262 ("text", InternalTraversal.dp_string),
2263 ]
2265 _is_text_clause = True
2267 _is_textual = True
2269 _bind_params_regex = re.compile(r"(?<![:\w\x5c]):(\w+)(?!:)", re.UNICODE)
2270 _is_implicitly_boolean = False
2272 _render_label_in_columns_clause = False
2274 _omit_from_statements = False
2276 _is_collection_aggregate = False
2278 @property
2279 def _hide_froms(self) -> Iterable[FromClause]:
2280 return ()
2282 def __and__(self, other):
2283 # support use in select.where(), query.filter()
2284 return and_(self, other)
2286 @property
2287 def _select_iterable(self) -> _SelectIterable:
2288 return (self,)
2290 # help in those cases where text() is
2291 # interpreted in a column expression situation
2292 key: Optional[str] = None
2293 _label: Optional[str] = None
2295 _allow_label_resolve = False
2297 @property
2298 def _is_star(self):
2299 return self.text == "*"
2301 def __init__(self, text: str):
2302 self._bindparams: Dict[str, BindParameter[Any]] = {}
2304 def repl(m):
2305 self._bindparams[m.group(1)] = BindParameter(m.group(1))
2306 return ":%s" % m.group(1)
2308 # scan the string and search for bind parameter names, add them
2309 # to the list of bindparams
2310 self.text = self._bind_params_regex.sub(repl, text)
2312 @_generative
2313 def bindparams(
2314 self,
2315 *binds: BindParameter[Any],
2316 **names_to_values: Any,
2317 ) -> Self:
2318 """Establish the values and/or types of bound parameters within
2319 this :class:`_expression.TextClause` construct.
2321 Given a text construct such as::
2323 from sqlalchemy import text
2325 stmt = text(
2326 "SELECT id, name FROM user WHERE name=:name AND timestamp=:timestamp"
2327 )
2329 the :meth:`_expression.TextClause.bindparams`
2330 method can be used to establish
2331 the initial value of ``:name`` and ``:timestamp``,
2332 using simple keyword arguments::
2334 stmt = stmt.bindparams(
2335 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2336 )
2338 Where above, new :class:`.BindParameter` objects
2339 will be generated with the names ``name`` and ``timestamp``, and
2340 values of ``jack`` and ``datetime.datetime(2012, 10, 8, 15, 12, 5)``,
2341 respectively. The types will be
2342 inferred from the values given, in this case :class:`.String` and
2343 :class:`.DateTime`.
2345 When specific typing behavior is needed, the positional ``*binds``
2346 argument can be used in which to specify :func:`.bindparam` constructs
2347 directly. These constructs must include at least the ``key``
2348 argument, then an optional value and type::
2350 from sqlalchemy import bindparam
2352 stmt = stmt.bindparams(
2353 bindparam("name", value="jack", type_=String),
2354 bindparam("timestamp", type_=DateTime),
2355 )
2357 Above, we specified the type of :class:`.DateTime` for the
2358 ``timestamp`` bind, and the type of :class:`.String` for the ``name``
2359 bind. In the case of ``name`` we also set the default value of
2360 ``"jack"``.
2362 Additional bound parameters can be supplied at statement execution
2363 time, e.g.::
2365 result = connection.execute(
2366 stmt, timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2367 )
2369 The :meth:`_expression.TextClause.bindparams`
2370 method can be called repeatedly,
2371 where it will re-use existing :class:`.BindParameter` objects to add
2372 new information. For example, we can call
2373 :meth:`_expression.TextClause.bindparams`
2374 first with typing information, and a
2375 second time with value information, and it will be combined::
2377 stmt = text(
2378 "SELECT id, name FROM user WHERE name=:name "
2379 "AND timestamp=:timestamp"
2380 )
2381 stmt = stmt.bindparams(
2382 bindparam("name", type_=String), bindparam("timestamp", type_=DateTime)
2383 )
2384 stmt = stmt.bindparams(
2385 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2386 )
2388 The :meth:`_expression.TextClause.bindparams`
2389 method also supports the concept of
2390 **unique** bound parameters. These are parameters that are
2391 "uniquified" on name at statement compilation time, so that multiple
2392 :func:`_expression.text`
2393 constructs may be combined together without the names
2394 conflicting. To use this feature, specify the
2395 :paramref:`.BindParameter.unique` flag on each :func:`.bindparam`
2396 object::
2398 stmt1 = text("select id from table where name=:name").bindparams(
2399 bindparam("name", value="name1", unique=True)
2400 )
2401 stmt2 = text("select id from table where name=:name").bindparams(
2402 bindparam("name", value="name2", unique=True)
2403 )
2405 union = union_all(stmt1.columns(column("id")), stmt2.columns(column("id")))
2407 The above statement will render as:
2409 .. sourcecode:: sql
2411 select id from table where name=:name_1
2412 UNION ALL select id from table where name=:name_2
2414 .. versionadded:: 1.3.11 Added support for the
2415 :paramref:`.BindParameter.unique` flag to work with
2416 :func:`_expression.text`
2417 constructs.
2419 """ # noqa: E501
2420 self._bindparams = new_params = self._bindparams.copy()
2422 for bind in binds:
2423 try:
2424 # the regex used for text() currently will not match
2425 # a unique/anonymous key in any case, so use the _orig_key
2426 # so that a text() construct can support unique parameters
2427 existing = new_params[bind._orig_key]
2428 except KeyError as err:
2429 raise exc.ArgumentError(
2430 "This text() construct doesn't define a "
2431 "bound parameter named %r" % bind._orig_key
2432 ) from err
2433 else:
2434 new_params[existing._orig_key] = bind
2436 for key, value in names_to_values.items():
2437 try:
2438 existing = new_params[key]
2439 except KeyError as err:
2440 raise exc.ArgumentError(
2441 "This text() construct doesn't define a "
2442 "bound parameter named %r" % key
2443 ) from err
2444 else:
2445 new_params[key] = existing._with_value(value, required=False)
2446 return self
2448 @util.preload_module("sqlalchemy.sql.selectable")
2449 def columns(
2450 self,
2451 *cols: _ColumnExpressionArgument[Any],
2452 **types: _TypeEngineArgument[Any],
2453 ) -> TextualSelect:
2454 r"""Turn this :class:`_expression.TextClause` object into a
2455 :class:`_expression.TextualSelect`
2456 object that serves the same role as a SELECT
2457 statement.
2459 The :class:`_expression.TextualSelect` is part of the
2460 :class:`_expression.SelectBase`
2461 hierarchy and can be embedded into another statement by using the
2462 :meth:`_expression.TextualSelect.subquery` method to produce a
2463 :class:`.Subquery`
2464 object, which can then be SELECTed from.
2466 This function essentially bridges the gap between an entirely
2467 textual SELECT statement and the SQL expression language concept
2468 of a "selectable"::
2470 from sqlalchemy.sql import column, text
2472 stmt = text("SELECT id, name FROM some_table")
2473 stmt = stmt.columns(column("id"), column("name")).subquery("st")
2475 stmt = (
2476 select(mytable)
2477 .select_from(mytable.join(stmt, mytable.c.name == stmt.c.name))
2478 .where(stmt.c.id > 5)
2479 )
2481 Above, we pass a series of :func:`_expression.column` elements to the
2482 :meth:`_expression.TextClause.columns` method positionally. These
2483 :func:`_expression.column`
2484 elements now become first class elements upon the
2485 :attr:`_expression.TextualSelect.selected_columns` column collection,
2486 which then
2487 become part of the :attr:`.Subquery.c` collection after
2488 :meth:`_expression.TextualSelect.subquery` is invoked.
2490 The column expressions we pass to
2491 :meth:`_expression.TextClause.columns` may
2492 also be typed; when we do so, these :class:`.TypeEngine` objects become
2493 the effective return type of the column, so that SQLAlchemy's
2494 result-set-processing systems may be used on the return values.
2495 This is often needed for types such as date or boolean types, as well
2496 as for unicode processing on some dialect configurations::
2498 stmt = text("SELECT id, name, timestamp FROM some_table")
2499 stmt = stmt.columns(
2500 column("id", Integer),
2501 column("name", Unicode),
2502 column("timestamp", DateTime),
2503 )
2505 for id, name, timestamp in connection.execute(stmt):
2506 print(id, name, timestamp)
2508 As a shortcut to the above syntax, keyword arguments referring to
2509 types alone may be used, if only type conversion is needed::
2511 stmt = text("SELECT id, name, timestamp FROM some_table")
2512 stmt = stmt.columns(id=Integer, name=Unicode, timestamp=DateTime)
2514 for id, name, timestamp in connection.execute(stmt):
2515 print(id, name, timestamp)
2517 The positional form of :meth:`_expression.TextClause.columns`
2518 also provides the
2519 unique feature of **positional column targeting**, which is
2520 particularly useful when using the ORM with complex textual queries. If
2521 we specify the columns from our model to
2522 :meth:`_expression.TextClause.columns`,
2523 the result set will match to those columns positionally, meaning the
2524 name or origin of the column in the textual SQL doesn't matter::
2526 stmt = text(
2527 "SELECT users.id, addresses.id, users.id, "
2528 "users.name, addresses.email_address AS email "
2529 "FROM users JOIN addresses ON users.id=addresses.user_id "
2530 "WHERE users.id = 1"
2531 ).columns(
2532 User.id,
2533 Address.id,
2534 Address.user_id,
2535 User.name,
2536 Address.email_address,
2537 )
2539 query = (
2540 session.query(User)
2541 .from_statement(stmt)
2542 .options(contains_eager(User.addresses))
2543 )
2545 The :meth:`_expression.TextClause.columns` method provides a direct
2546 route to calling :meth:`_expression.FromClause.subquery` as well as
2547 :meth:`_expression.SelectBase.cte`
2548 against a textual SELECT statement::
2550 stmt = stmt.columns(id=Integer, name=String).cte("st")
2552 stmt = select(sometable).where(sometable.c.id == stmt.c.id)
2554 :param \*cols: A series of :class:`_expression.ColumnElement` objects,
2555 typically
2556 :class:`_schema.Column` objects from a :class:`_schema.Table`
2557 or ORM level
2558 column-mapped attributes, representing a set of columns that this
2559 textual string will SELECT from.
2561 :param \**types: A mapping of string names to :class:`.TypeEngine`
2562 type objects indicating the datatypes to use for names that are
2563 SELECTed from the textual string. Prefer to use the ``*cols``
2564 argument as it also indicates positional ordering.
2566 """
2567 selectable = util.preloaded.sql_selectable
2569 input_cols: List[NamedColumn[Any]] = [
2570 coercions.expect(roles.LabeledColumnExprRole, col) for col in cols
2571 ]
2573 positional_input_cols = [
2574 (
2575 ColumnClause(col.key, types.pop(col.key))
2576 if col.key in types
2577 else col
2578 )
2579 for col in input_cols
2580 ]
2581 keyed_input_cols: List[NamedColumn[Any]] = [
2582 ColumnClause(key, type_) for key, type_ in types.items()
2583 ]
2585 elem = selectable.TextualSelect.__new__(selectable.TextualSelect)
2586 elem._init(
2587 self,
2588 positional_input_cols + keyed_input_cols,
2589 positional=bool(positional_input_cols) and not keyed_input_cols,
2590 )
2591 return elem
2593 @property
2594 def type(self) -> TypeEngine[Any]:
2595 return type_api.NULLTYPE
2597 @property
2598 def comparator(self):
2599 # TODO: this seems wrong, it seems like we might not
2600 # be using this method.
2601 return self.type.comparator_factory(self) # type: ignore
2603 def self_group(
2604 self, against: Optional[OperatorType] = None
2605 ) -> Union[Self, Grouping[Any]]:
2606 if against is operators.in_op:
2607 return Grouping(self)
2608 else:
2609 return self
2612class Null(SingletonConstant, roles.ConstExprRole[None], ColumnElement[None]):
2613 """Represent the NULL keyword in a SQL statement.
2615 :class:`.Null` is accessed as a constant via the
2616 :func:`.null` function.
2618 """
2620 __visit_name__ = "null"
2622 _traverse_internals: _TraverseInternalsType = []
2623 _singleton: Null
2625 if not TYPE_CHECKING:
2627 @util.memoized_property
2628 def type(self) -> TypeEngine[_T]: # noqa: A001
2629 return type_api.NULLTYPE
2631 @classmethod
2632 def _instance(cls) -> Null:
2633 """Return a constant :class:`.Null` construct."""
2635 return Null._singleton
2638Null._create_singleton()
2641class False_(
2642 SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]
2643):
2644 """Represent the ``false`` keyword, or equivalent, in a SQL statement.
2646 :class:`.False_` is accessed as a constant via the
2647 :func:`.false` function.
2649 """
2651 __visit_name__ = "false"
2652 _traverse_internals: _TraverseInternalsType = []
2653 _singleton: False_
2655 if not TYPE_CHECKING:
2657 @util.memoized_property
2658 def type(self) -> TypeEngine[_T]: # noqa: A001
2659 return type_api.BOOLEANTYPE
2661 def _negate(self) -> True_:
2662 return True_._singleton
2664 @classmethod
2665 def _instance(cls) -> False_:
2666 return False_._singleton
2669False_._create_singleton()
2672class True_(SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]):
2673 """Represent the ``true`` keyword, or equivalent, in a SQL statement.
2675 :class:`.True_` is accessed as a constant via the
2676 :func:`.true` function.
2678 """
2680 __visit_name__ = "true"
2682 _traverse_internals: _TraverseInternalsType = []
2683 _singleton: True_
2685 if not TYPE_CHECKING:
2687 @util.memoized_property
2688 def type(self) -> TypeEngine[_T]: # noqa: A001
2689 return type_api.BOOLEANTYPE
2691 def _negate(self) -> False_:
2692 return False_._singleton
2694 @classmethod
2695 def _ifnone(
2696 cls, other: Optional[ColumnElement[Any]]
2697 ) -> ColumnElement[Any]:
2698 if other is None:
2699 return cls._instance()
2700 else:
2701 return other
2703 @classmethod
2704 def _instance(cls) -> True_:
2705 return True_._singleton
2708True_._create_singleton()
2711class ClauseList(
2712 roles.InElementRole,
2713 roles.OrderByRole,
2714 roles.ColumnsClauseRole,
2715 roles.DMLColumnRole,
2716 DQLDMLClauseElement,
2717):
2718 """Describe a list of clauses, separated by an operator.
2720 By default, is comma-separated, such as a column listing.
2722 """
2724 __visit_name__ = "clauselist"
2726 # this is used only by the ORM in a legacy use case for
2727 # composite attributes
2728 _is_clause_list = True
2730 _traverse_internals: _TraverseInternalsType = [
2731 ("clauses", InternalTraversal.dp_clauseelement_list),
2732 ("operator", InternalTraversal.dp_operator),
2733 ]
2735 clauses: List[ColumnElement[Any]]
2737 def __init__(
2738 self,
2739 *clauses: _ColumnExpressionArgument[Any],
2740 operator: OperatorType = operators.comma_op,
2741 group: bool = True,
2742 group_contents: bool = True,
2743 _literal_as_text_role: Type[roles.SQLRole] = roles.WhereHavingRole,
2744 ):
2745 self.operator = operator
2746 self.group = group
2747 self.group_contents = group_contents
2748 clauses_iterator: Iterable[_ColumnExpressionArgument[Any]] = clauses
2749 text_converter_role: Type[roles.SQLRole] = _literal_as_text_role
2750 self._text_converter_role = text_converter_role
2752 if self.group_contents:
2753 self.clauses = [
2754 coercions.expect(
2755 text_converter_role, clause, apply_propagate_attrs=self
2756 ).self_group(against=self.operator)
2757 for clause in clauses_iterator
2758 ]
2759 else:
2760 self.clauses = [
2761 coercions.expect(
2762 text_converter_role, clause, apply_propagate_attrs=self
2763 )
2764 for clause in clauses_iterator
2765 ]
2766 self._is_implicitly_boolean = operators.is_boolean(self.operator)
2768 @classmethod
2769 def _construct_raw(
2770 cls,
2771 operator: OperatorType,
2772 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
2773 ) -> ClauseList:
2774 self = cls.__new__(cls)
2775 self.clauses = list(clauses) if clauses else []
2776 self.group = True
2777 self.operator = operator
2778 self.group_contents = True
2779 self._is_implicitly_boolean = False
2780 return self
2782 def __iter__(self) -> Iterator[ColumnElement[Any]]:
2783 return iter(self.clauses)
2785 def __len__(self) -> int:
2786 return len(self.clauses)
2788 @property
2789 def _select_iterable(self) -> _SelectIterable:
2790 return itertools.chain.from_iterable(
2791 [elem._select_iterable for elem in self.clauses]
2792 )
2794 def append(self, clause):
2795 if self.group_contents:
2796 self.clauses.append(
2797 coercions.expect(self._text_converter_role, clause).self_group(
2798 against=self.operator
2799 )
2800 )
2801 else:
2802 self.clauses.append(
2803 coercions.expect(self._text_converter_role, clause)
2804 )
2806 @util.ro_non_memoized_property
2807 def _from_objects(self) -> List[FromClause]:
2808 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
2810 def self_group(
2811 self, against: Optional[OperatorType] = None
2812 ) -> Union[Self, Grouping[Any]]:
2813 if self.group and operators.is_precedent(self.operator, against):
2814 return Grouping(self)
2815 else:
2816 return self
2819class OperatorExpression(ColumnElement[_T]):
2820 """base for expressions that contain an operator and operands
2822 .. versionadded:: 2.0
2824 """
2826 operator: OperatorType
2827 type: TypeEngine[_T]
2829 group: bool = True
2831 @property
2832 def is_comparison(self):
2833 return operators.is_comparison(self.operator)
2835 def self_group(
2836 self, against: Optional[OperatorType] = None
2837 ) -> Union[Self, Grouping[_T]]:
2838 if (
2839 self.group
2840 and operators.is_precedent(self.operator, against)
2841 or (
2842 # a negate against a non-boolean operator
2843 # doesn't make too much sense but we should
2844 # group for that
2845 against is operators.inv
2846 and not operators.is_boolean(self.operator)
2847 )
2848 ):
2849 return Grouping(self)
2850 else:
2851 return self
2853 @property
2854 def _flattened_operator_clauses(
2855 self,
2856 ) -> typing_Tuple[ColumnElement[Any], ...]:
2857 raise NotImplementedError()
2859 @classmethod
2860 def _construct_for_op(
2861 cls,
2862 left: ColumnElement[Any],
2863 right: ColumnElement[Any],
2864 op: OperatorType,
2865 *,
2866 type_: TypeEngine[_T],
2867 negate: Optional[OperatorType] = None,
2868 modifiers: Optional[Mapping[str, Any]] = None,
2869 ) -> OperatorExpression[_T]:
2870 if operators.is_associative(op):
2871 assert (
2872 negate is None
2873 ), f"negate not supported for associative operator {op}"
2875 multi = False
2876 if getattr(
2877 left, "operator", None
2878 ) is op and type_._compare_type_affinity(left.type):
2879 multi = True
2880 left_flattened = left._flattened_operator_clauses
2881 else:
2882 left_flattened = (left,)
2884 if getattr(
2885 right, "operator", None
2886 ) is op and type_._compare_type_affinity(right.type):
2887 multi = True
2888 right_flattened = right._flattened_operator_clauses
2889 else:
2890 right_flattened = (right,)
2892 if multi:
2893 return ExpressionClauseList._construct_for_list(
2894 op,
2895 type_,
2896 *(left_flattened + right_flattened),
2897 )
2899 if right._is_collection_aggregate:
2900 negate = None
2902 return BinaryExpression(
2903 left, right, op, type_=type_, negate=negate, modifiers=modifiers
2904 )
2907class ExpressionClauseList(OperatorExpression[_T]):
2908 """Describe a list of clauses, separated by an operator,
2909 in a column expression context.
2911 :class:`.ExpressionClauseList` differs from :class:`.ClauseList` in that
2912 it represents a column-oriented DQL expression only, not an open ended
2913 list of anything comma separated.
2915 .. versionadded:: 2.0
2917 """
2919 __visit_name__ = "expression_clauselist"
2921 _traverse_internals: _TraverseInternalsType = [
2922 ("clauses", InternalTraversal.dp_clauseelement_tuple),
2923 ("operator", InternalTraversal.dp_operator),
2924 ]
2926 clauses: typing_Tuple[ColumnElement[Any], ...]
2928 group: bool
2930 def __init__(
2931 self,
2932 operator: OperatorType,
2933 *clauses: _ColumnExpressionArgument[Any],
2934 type_: Optional[_TypeEngineArgument[_T]] = None,
2935 ):
2936 self.operator = operator
2938 self.clauses = tuple(
2939 coercions.expect(
2940 roles.ExpressionElementRole, clause, apply_propagate_attrs=self
2941 )
2942 for clause in clauses
2943 )
2944 self._is_implicitly_boolean = operators.is_boolean(self.operator)
2945 self.type = type_api.to_instance(type_) # type: ignore
2947 @property
2948 def _flattened_operator_clauses(
2949 self,
2950 ) -> typing_Tuple[ColumnElement[Any], ...]:
2951 return self.clauses
2953 def __iter__(self) -> Iterator[ColumnElement[Any]]:
2954 return iter(self.clauses)
2956 def __len__(self) -> int:
2957 return len(self.clauses)
2959 @property
2960 def _select_iterable(self) -> _SelectIterable:
2961 return (self,)
2963 @util.ro_non_memoized_property
2964 def _from_objects(self) -> List[FromClause]:
2965 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
2967 def _append_inplace(self, clause: ColumnElement[Any]) -> None:
2968 self.clauses += (clause,)
2970 @classmethod
2971 def _construct_for_list(
2972 cls,
2973 operator: OperatorType,
2974 type_: TypeEngine[_T],
2975 *clauses: ColumnElement[Any],
2976 group: bool = True,
2977 ) -> ExpressionClauseList[_T]:
2978 self = cls.__new__(cls)
2979 self.group = group
2980 if group:
2981 self.clauses = tuple(
2982 c.self_group(against=operator) for c in clauses
2983 )
2984 else:
2985 self.clauses = clauses
2986 self.operator = operator
2987 self.type = type_
2988 for c in clauses:
2989 if c._propagate_attrs:
2990 self._propagate_attrs = c._propagate_attrs
2991 break
2992 return self
2994 def _negate(self) -> Any:
2995 grouped = self.self_group(against=operators.inv)
2996 assert isinstance(grouped, ColumnElement)
2997 return UnaryExpression(
2998 grouped, operator=operators.inv, wraps_column_expression=True
2999 )
3002class BooleanClauseList(ExpressionClauseList[bool]):
3003 __visit_name__ = "expression_clauselist"
3004 inherit_cache = True
3006 def __init__(self, *arg, **kw):
3007 raise NotImplementedError(
3008 "BooleanClauseList has a private constructor"
3009 )
3011 @classmethod
3012 def _process_clauses_for_boolean(
3013 cls,
3014 operator: OperatorType,
3015 continue_on: Any,
3016 skip_on: Any,
3017 clauses: Iterable[ColumnElement[Any]],
3018 ) -> typing_Tuple[int, List[ColumnElement[Any]]]:
3019 has_continue_on = None
3021 convert_clauses = []
3023 against = operators._asbool
3024 lcc = 0
3026 for clause in clauses:
3027 if clause is continue_on:
3028 # instance of continue_on, like and_(x, y, True, z), store it
3029 # if we didn't find one already, we will use it if there
3030 # are no other expressions here.
3031 has_continue_on = clause
3032 elif clause is skip_on:
3033 # instance of skip_on, e.g. and_(x, y, False, z), cancels
3034 # the rest out
3035 convert_clauses = [clause]
3036 lcc = 1
3037 break
3038 else:
3039 if not lcc:
3040 lcc = 1
3041 else:
3042 against = operator
3043 # technically this would be len(convert_clauses) + 1
3044 # however this only needs to indicate "greater than one"
3045 lcc = 2
3046 convert_clauses.append(clause)
3048 if not convert_clauses and has_continue_on is not None:
3049 convert_clauses = [has_continue_on]
3050 lcc = 1
3052 return lcc, [c.self_group(against=against) for c in convert_clauses]
3054 @classmethod
3055 def _construct(
3056 cls,
3057 operator: OperatorType,
3058 continue_on: Any,
3059 skip_on: Any,
3060 initial_clause: Any = _NoArg.NO_ARG,
3061 *clauses: Any,
3062 **kw: Any,
3063 ) -> ColumnElement[Any]:
3064 if initial_clause is _NoArg.NO_ARG:
3065 # no elements period. deprecated use case. return an empty
3066 # ClauseList construct that generates nothing unless it has
3067 # elements added to it.
3068 name = operator.__name__
3070 util.warn_deprecated(
3071 f"Invoking {name}() without arguments is deprecated, and "
3072 f"will be disallowed in a future release. For an empty "
3073 f"""{name}() construct, use '{name}({
3074 'true()' if continue_on is True_._singleton else 'false()'
3075 }, *args)' """
3076 f"""or '{name}({
3077 'True' if continue_on is True_._singleton else 'False'
3078 }, *args)'.""",
3079 version="1.4",
3080 )
3081 return cls._construct_raw(operator)
3083 lcc, convert_clauses = cls._process_clauses_for_boolean(
3084 operator,
3085 continue_on,
3086 skip_on,
3087 [
3088 coercions.expect(roles.WhereHavingRole, clause)
3089 for clause in util.coerce_generator_arg(
3090 (initial_clause,) + clauses
3091 )
3092 ],
3093 )
3095 if lcc > 1:
3096 # multiple elements. Return regular BooleanClauseList
3097 # which will link elements against the operator.
3099 flattened_clauses = itertools.chain.from_iterable(
3100 (
3101 (c for c in to_flat._flattened_operator_clauses)
3102 if getattr(to_flat, "operator", None) is operator
3103 else (to_flat,)
3104 )
3105 for to_flat in convert_clauses
3106 )
3108 return cls._construct_raw(operator, flattened_clauses) # type: ignore # noqa: E501
3109 else:
3110 assert lcc
3111 # just one element. return it as a single boolean element,
3112 # not a list and discard the operator.
3113 return convert_clauses[0]
3115 @classmethod
3116 def _construct_for_whereclause(
3117 cls, clauses: Iterable[ColumnElement[Any]]
3118 ) -> Optional[ColumnElement[bool]]:
3119 operator, continue_on, skip_on = (
3120 operators.and_,
3121 True_._singleton,
3122 False_._singleton,
3123 )
3125 lcc, convert_clauses = cls._process_clauses_for_boolean(
3126 operator,
3127 continue_on,
3128 skip_on,
3129 clauses, # these are assumed to be coerced already
3130 )
3132 if lcc > 1:
3133 # multiple elements. Return regular BooleanClauseList
3134 # which will link elements against the operator.
3135 return cls._construct_raw(operator, convert_clauses)
3136 elif lcc == 1:
3137 # just one element. return it as a single boolean element,
3138 # not a list and discard the operator.
3139 return convert_clauses[0]
3140 else:
3141 return None
3143 @classmethod
3144 def _construct_raw(
3145 cls,
3146 operator: OperatorType,
3147 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
3148 ) -> BooleanClauseList:
3149 self = cls.__new__(cls)
3150 self.clauses = tuple(clauses) if clauses else ()
3151 self.group = True
3152 self.operator = operator
3153 self.type = type_api.BOOLEANTYPE
3154 self._is_implicitly_boolean = True
3155 return self
3157 @classmethod
3158 def and_(
3159 cls,
3160 initial_clause: Union[
3161 Literal[True], _ColumnExpressionArgument[bool], _NoArg
3162 ] = _NoArg.NO_ARG,
3163 *clauses: _ColumnExpressionArgument[bool],
3164 ) -> ColumnElement[bool]:
3165 r"""Produce a conjunction of expressions joined by ``AND``.
3167 See :func:`_sql.and_` for full documentation.
3168 """
3169 return cls._construct(
3170 operators.and_,
3171 True_._singleton,
3172 False_._singleton,
3173 initial_clause,
3174 *clauses,
3175 )
3177 @classmethod
3178 def or_(
3179 cls,
3180 initial_clause: Union[
3181 Literal[False], _ColumnExpressionArgument[bool], _NoArg
3182 ] = _NoArg.NO_ARG,
3183 *clauses: _ColumnExpressionArgument[bool],
3184 ) -> ColumnElement[bool]:
3185 """Produce a conjunction of expressions joined by ``OR``.
3187 See :func:`_sql.or_` for full documentation.
3188 """
3189 return cls._construct(
3190 operators.or_,
3191 False_._singleton,
3192 True_._singleton,
3193 initial_clause,
3194 *clauses,
3195 )
3197 @property
3198 def _select_iterable(self) -> _SelectIterable:
3199 return (self,)
3201 def self_group(
3202 self, against: Optional[OperatorType] = None
3203 ) -> Union[Self, Grouping[bool]]:
3204 if not self.clauses:
3205 return self
3206 else:
3207 return super().self_group(against=against)
3210and_ = BooleanClauseList.and_
3211or_ = BooleanClauseList.or_
3214class Tuple(ClauseList, ColumnElement[typing_Tuple[Any, ...]]):
3215 """Represent a SQL tuple."""
3217 __visit_name__ = "tuple"
3219 _traverse_internals: _TraverseInternalsType = (
3220 ClauseList._traverse_internals + []
3221 )
3223 type: TupleType
3225 @util.preload_module("sqlalchemy.sql.sqltypes")
3226 def __init__(
3227 self,
3228 *clauses: _ColumnExpressionArgument[Any],
3229 types: Optional[Sequence[_TypeEngineArgument[Any]]] = None,
3230 ):
3231 sqltypes = util.preloaded.sql_sqltypes
3233 if types is None:
3234 init_clauses: List[ColumnElement[Any]] = [
3235 coercions.expect(roles.ExpressionElementRole, c)
3236 for c in clauses
3237 ]
3238 else:
3239 if len(types) != len(clauses):
3240 raise exc.ArgumentError(
3241 "Wrong number of elements for %d-tuple: %r "
3242 % (len(types), clauses)
3243 )
3244 init_clauses = [
3245 coercions.expect(
3246 roles.ExpressionElementRole,
3247 c,
3248 type_=typ if not typ._isnull else None,
3249 )
3250 for typ, c in zip(types, clauses)
3251 ]
3253 self.type = sqltypes.TupleType(*[arg.type for arg in init_clauses])
3254 super().__init__(*init_clauses)
3256 @property
3257 def _select_iterable(self) -> _SelectIterable:
3258 return (self,)
3260 def _bind_param(self, operator, obj, type_=None, expanding=False):
3261 if expanding:
3262 return BindParameter(
3263 None,
3264 value=obj,
3265 _compared_to_operator=operator,
3266 unique=True,
3267 expanding=True,
3268 type_=type_,
3269 _compared_to_type=self.type,
3270 )
3271 else:
3272 return Tuple(
3273 *[
3274 BindParameter(
3275 None,
3276 o,
3277 _compared_to_operator=operator,
3278 _compared_to_type=compared_to_type,
3279 unique=True,
3280 type_=type_,
3281 )
3282 for o, compared_to_type in zip(obj, self.type.types)
3283 ]
3284 )
3286 def self_group(self, against: Optional[OperatorType] = None) -> Self:
3287 # Tuple is parenthesized by definition.
3288 return self
3291class Case(ColumnElement[_T]):
3292 """Represent a ``CASE`` expression.
3294 :class:`.Case` is produced using the :func:`.case` factory function,
3295 as in::
3297 from sqlalchemy import case
3299 stmt = select(users_table).where(
3300 case(
3301 (users_table.c.name == "wendy", "W"),
3302 (users_table.c.name == "jack", "J"),
3303 else_="E",
3304 )
3305 )
3307 Details on :class:`.Case` usage is at :func:`.case`.
3309 .. seealso::
3311 :func:`.case`
3313 """
3315 __visit_name__ = "case"
3317 _traverse_internals: _TraverseInternalsType = [
3318 ("value", InternalTraversal.dp_clauseelement),
3319 ("whens", InternalTraversal.dp_clauseelement_tuples),
3320 ("else_", InternalTraversal.dp_clauseelement),
3321 ]
3323 # for case(), the type is derived from the whens. so for the moment
3324 # users would have to cast() the case to get a specific type
3326 whens: List[typing_Tuple[ColumnElement[bool], ColumnElement[_T]]]
3327 else_: Optional[ColumnElement[_T]]
3328 value: Optional[ColumnElement[Any]]
3330 def __init__(
3331 self,
3332 *whens: Union[
3333 typing_Tuple[_ColumnExpressionArgument[bool], Any],
3334 Mapping[Any, Any],
3335 ],
3336 value: Optional[Any] = None,
3337 else_: Optional[Any] = None,
3338 ):
3339 new_whens: Iterable[Any] = coercions._expression_collection_was_a_list(
3340 "whens", "case", whens
3341 )
3342 try:
3343 new_whens = util.dictlike_iteritems(new_whens)
3344 except TypeError:
3345 pass
3347 self.whens = [
3348 (
3349 coercions.expect(
3350 roles.ExpressionElementRole,
3351 c,
3352 apply_propagate_attrs=self,
3353 ).self_group(),
3354 coercions.expect(roles.ExpressionElementRole, r),
3355 )
3356 for (c, r) in new_whens
3357 ]
3359 if value is None:
3360 self.value = None
3361 else:
3362 self.value = coercions.expect(roles.ExpressionElementRole, value)
3364 if else_ is not None:
3365 self.else_ = coercions.expect(roles.ExpressionElementRole, else_)
3366 else:
3367 self.else_ = None
3369 type_ = next(
3370 (
3371 then.type
3372 # Iterate `whens` in reverse to match previous behaviour
3373 # where type of final element took priority
3374 for *_, then in reversed(self.whens)
3375 if not then.type._isnull
3376 ),
3377 self.else_.type if self.else_ is not None else type_api.NULLTYPE,
3378 )
3379 self.type = cast(_T, type_)
3381 @util.ro_non_memoized_property
3382 def _from_objects(self) -> List[FromClause]:
3383 return list(
3384 itertools.chain(*[x._from_objects for x in self.get_children()])
3385 )
3388class Cast(WrapsColumnExpression[_T]):
3389 """Represent a ``CAST`` expression.
3391 :class:`.Cast` is produced using the :func:`.cast` factory function,
3392 as in::
3394 from sqlalchemy import cast, Numeric
3396 stmt = select(cast(product_table.c.unit_price, Numeric(10, 4)))
3398 Details on :class:`.Cast` usage is at :func:`.cast`.
3400 .. seealso::
3402 :ref:`tutorial_casts`
3404 :func:`.cast`
3406 :func:`.try_cast`
3408 :func:`.type_coerce` - an alternative to CAST that coerces the type
3409 on the Python side only, which is often sufficient to generate the
3410 correct SQL and data coercion.
3412 """
3414 __visit_name__ = "cast"
3416 _traverse_internals: _TraverseInternalsType = [
3417 ("clause", InternalTraversal.dp_clauseelement),
3418 ("type", InternalTraversal.dp_type),
3419 ]
3421 clause: ColumnElement[Any]
3422 type: TypeEngine[_T]
3423 typeclause: TypeClause
3425 def __init__(
3426 self,
3427 expression: _ColumnExpressionArgument[Any],
3428 type_: _TypeEngineArgument[_T],
3429 ):
3430 self.type = type_api.to_instance(type_)
3431 self.clause = coercions.expect(
3432 roles.ExpressionElementRole,
3433 expression,
3434 type_=self.type,
3435 apply_propagate_attrs=self,
3436 )
3437 self.typeclause = TypeClause(self.type)
3439 @util.ro_non_memoized_property
3440 def _from_objects(self) -> List[FromClause]:
3441 return self.clause._from_objects
3443 @property
3444 def wrapped_column_expression(self):
3445 return self.clause
3448class TryCast(Cast[_T]):
3449 """Represent a TRY_CAST expression.
3451 Details on :class:`.TryCast` usage is at :func:`.try_cast`.
3453 .. seealso::
3455 :func:`.try_cast`
3457 :ref:`tutorial_casts`
3458 """
3460 __visit_name__ = "try_cast"
3461 inherit_cache = True
3464class TypeCoerce(WrapsColumnExpression[_T]):
3465 """Represent a Python-side type-coercion wrapper.
3467 :class:`.TypeCoerce` supplies the :func:`_expression.type_coerce`
3468 function; see that function for usage details.
3470 .. seealso::
3472 :func:`_expression.type_coerce`
3474 :func:`.cast`
3476 """
3478 __visit_name__ = "type_coerce"
3480 _traverse_internals: _TraverseInternalsType = [
3481 ("clause", InternalTraversal.dp_clauseelement),
3482 ("type", InternalTraversal.dp_type),
3483 ]
3485 clause: ColumnElement[Any]
3486 type: TypeEngine[_T]
3488 def __init__(
3489 self,
3490 expression: _ColumnExpressionArgument[Any],
3491 type_: _TypeEngineArgument[_T],
3492 ):
3493 self.type = type_api.to_instance(type_)
3494 self.clause = coercions.expect(
3495 roles.ExpressionElementRole,
3496 expression,
3497 type_=self.type,
3498 apply_propagate_attrs=self,
3499 )
3501 @util.ro_non_memoized_property
3502 def _from_objects(self) -> List[FromClause]:
3503 return self.clause._from_objects
3505 @HasMemoized.memoized_attribute
3506 def typed_expression(self):
3507 if isinstance(self.clause, BindParameter):
3508 bp = self.clause._clone()
3509 bp.type = self.type
3510 return bp
3511 else:
3512 return self.clause
3514 @property
3515 def wrapped_column_expression(self):
3516 return self.clause
3518 def self_group(
3519 self, against: Optional[OperatorType] = None
3520 ) -> TypeCoerce[_T]:
3521 grouped = self.clause.self_group(against=against)
3522 if grouped is not self.clause:
3523 return TypeCoerce(grouped, self.type)
3524 else:
3525 return self
3528class Extract(ColumnElement[int]):
3529 """Represent a SQL EXTRACT clause, ``extract(field FROM expr)``."""
3531 __visit_name__ = "extract"
3533 _traverse_internals: _TraverseInternalsType = [
3534 ("expr", InternalTraversal.dp_clauseelement),
3535 ("field", InternalTraversal.dp_string),
3536 ]
3538 expr: ColumnElement[Any]
3539 field: str
3541 def __init__(self, field: str, expr: _ColumnExpressionArgument[Any]):
3542 self.type = type_api.INTEGERTYPE
3543 self.field = field
3544 self.expr = coercions.expect(roles.ExpressionElementRole, expr)
3546 @util.ro_non_memoized_property
3547 def _from_objects(self) -> List[FromClause]:
3548 return self.expr._from_objects
3551class _label_reference(ColumnElement[_T]):
3552 """Wrap a column expression as it appears in a 'reference' context.
3554 This expression is any that includes an _order_by_label_element,
3555 which is a Label, or a DESC / ASC construct wrapping a Label.
3557 The production of _label_reference() should occur when an expression
3558 is added to this context; this includes the ORDER BY or GROUP BY of a
3559 SELECT statement, as well as a few other places, such as the ORDER BY
3560 within an OVER clause.
3562 """
3564 __visit_name__ = "label_reference"
3566 _traverse_internals: _TraverseInternalsType = [
3567 ("element", InternalTraversal.dp_clauseelement)
3568 ]
3570 element: ColumnElement[_T]
3572 def __init__(self, element: ColumnElement[_T]):
3573 self.element = element
3575 @util.ro_non_memoized_property
3576 def _from_objects(self) -> List[FromClause]:
3577 return []
3580class _textual_label_reference(ColumnElement[Any]):
3581 __visit_name__ = "textual_label_reference"
3583 _traverse_internals: _TraverseInternalsType = [
3584 ("element", InternalTraversal.dp_string)
3585 ]
3587 def __init__(self, element: str):
3588 self.element = element
3590 @util.memoized_property
3591 def _text_clause(self) -> TextClause:
3592 return TextClause(self.element)
3595class UnaryExpression(ColumnElement[_T]):
3596 """Define a 'unary' expression.
3598 A unary expression has a single column expression
3599 and an operator. The operator can be placed on the left
3600 (where it is called the 'operator') or right (where it is called the
3601 'modifier') of the column expression.
3603 :class:`.UnaryExpression` is the basis for several unary operators
3604 including those used by :func:`.desc`, :func:`.asc`, :func:`.distinct`,
3605 :func:`.nulls_first` and :func:`.nulls_last`.
3607 """
3609 __visit_name__ = "unary"
3611 _traverse_internals: _TraverseInternalsType = [
3612 ("element", InternalTraversal.dp_clauseelement),
3613 ("operator", InternalTraversal.dp_operator),
3614 ("modifier", InternalTraversal.dp_operator),
3615 ]
3617 element: ClauseElement
3619 def __init__(
3620 self,
3621 element: ColumnElement[Any],
3622 operator: Optional[OperatorType] = None,
3623 modifier: Optional[OperatorType] = None,
3624 type_: Optional[_TypeEngineArgument[_T]] = None,
3625 wraps_column_expression: bool = False,
3626 ):
3627 self.operator = operator
3628 self.modifier = modifier
3629 self._propagate_attrs = element._propagate_attrs
3630 self.element = element.self_group(
3631 against=self.operator or self.modifier
3632 )
3634 # if type is None, we get NULLTYPE, which is our _T. But I don't
3635 # know how to get the overloads to express that correctly
3636 self.type = type_api.to_instance(type_) # type: ignore
3638 self.wraps_column_expression = wraps_column_expression
3640 @classmethod
3641 def _create_nulls_first(
3642 cls,
3643 column: _ColumnExpressionArgument[_T],
3644 ) -> UnaryExpression[_T]:
3645 return UnaryExpression(
3646 coercions.expect(roles.ByOfRole, column),
3647 modifier=operators.nulls_first_op,
3648 wraps_column_expression=False,
3649 )
3651 @classmethod
3652 def _create_nulls_last(
3653 cls,
3654 column: _ColumnExpressionArgument[_T],
3655 ) -> UnaryExpression[_T]:
3656 return UnaryExpression(
3657 coercions.expect(roles.ByOfRole, column),
3658 modifier=operators.nulls_last_op,
3659 wraps_column_expression=False,
3660 )
3662 @classmethod
3663 def _create_desc(
3664 cls, column: _ColumnExpressionOrStrLabelArgument[_T]
3665 ) -> UnaryExpression[_T]:
3666 return UnaryExpression(
3667 coercions.expect(roles.ByOfRole, column),
3668 modifier=operators.desc_op,
3669 wraps_column_expression=False,
3670 )
3672 @classmethod
3673 def _create_asc(
3674 cls,
3675 column: _ColumnExpressionOrStrLabelArgument[_T],
3676 ) -> UnaryExpression[_T]:
3677 return UnaryExpression(
3678 coercions.expect(roles.ByOfRole, column),
3679 modifier=operators.asc_op,
3680 wraps_column_expression=False,
3681 )
3683 @classmethod
3684 def _create_distinct(
3685 cls,
3686 expr: _ColumnExpressionArgument[_T],
3687 ) -> UnaryExpression[_T]:
3688 col_expr: ColumnElement[_T] = coercions.expect(
3689 roles.ExpressionElementRole, expr
3690 )
3691 return UnaryExpression(
3692 col_expr,
3693 operator=operators.distinct_op,
3694 type_=col_expr.type,
3695 wraps_column_expression=False,
3696 )
3698 @classmethod
3699 def _create_bitwise_not(
3700 cls,
3701 expr: _ColumnExpressionArgument[_T],
3702 ) -> UnaryExpression[_T]:
3703 col_expr: ColumnElement[_T] = coercions.expect(
3704 roles.ExpressionElementRole, expr
3705 )
3706 return UnaryExpression(
3707 col_expr,
3708 operator=operators.bitwise_not_op,
3709 type_=col_expr.type,
3710 wraps_column_expression=False,
3711 )
3713 @property
3714 def _order_by_label_element(self) -> Optional[Label[Any]]:
3715 if operators.is_order_by_modifier(self.modifier):
3716 return self.element._order_by_label_element
3717 else:
3718 return None
3720 @util.ro_non_memoized_property
3721 def _from_objects(self) -> List[FromClause]:
3722 return self.element._from_objects
3724 def _negate(self):
3725 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
3726 return UnaryExpression(
3727 self.self_group(against=operators.inv),
3728 operator=operators.inv,
3729 type_=type_api.BOOLEANTYPE,
3730 wraps_column_expression=self.wraps_column_expression,
3731 )
3732 else:
3733 return ClauseElement._negate(self)
3735 def self_group(
3736 self, against: Optional[OperatorType] = None
3737 ) -> Union[Self, Grouping[_T]]:
3738 if self.operator and operators.is_precedent(self.operator, against):
3739 return Grouping(self)
3740 else:
3741 return self
3744class CollectionAggregate(UnaryExpression[_T]):
3745 """Forms the basis for right-hand collection operator modifiers
3746 ANY and ALL.
3748 The ANY and ALL keywords are available in different ways on different
3749 backends. On PostgreSQL, they only work for an ARRAY type. On
3750 MySQL, they only work for subqueries.
3752 """
3754 inherit_cache = True
3755 _is_collection_aggregate = True
3757 @classmethod
3758 def _create_any(
3759 cls, expr: _ColumnExpressionArgument[_T]
3760 ) -> CollectionAggregate[bool]:
3761 col_expr: ColumnElement[_T] = coercions.expect(
3762 roles.ExpressionElementRole,
3763 expr,
3764 )
3765 col_expr = col_expr.self_group()
3766 return CollectionAggregate(
3767 col_expr,
3768 operator=operators.any_op,
3769 type_=type_api.BOOLEANTYPE,
3770 wraps_column_expression=False,
3771 )
3773 @classmethod
3774 def _create_all(
3775 cls, expr: _ColumnExpressionArgument[_T]
3776 ) -> CollectionAggregate[bool]:
3777 col_expr: ColumnElement[_T] = coercions.expect(
3778 roles.ExpressionElementRole,
3779 expr,
3780 )
3781 col_expr = col_expr.self_group()
3782 return CollectionAggregate(
3783 col_expr,
3784 operator=operators.all_op,
3785 type_=type_api.BOOLEANTYPE,
3786 wraps_column_expression=False,
3787 )
3789 # operate and reverse_operate are hardwired to
3790 # dispatch onto the type comparator directly, so that we can
3791 # ensure "reversed" behavior.
3792 def operate(
3793 self, op: OperatorType, *other: Any, **kwargs: Any
3794 ) -> ColumnElement[_T]:
3795 if not operators.is_comparison(op):
3796 raise exc.ArgumentError(
3797 "Only comparison operators may be used with ANY/ALL"
3798 )
3799 kwargs["reverse"] = True
3800 return self.comparator.operate(operators.mirror(op), *other, **kwargs)
3802 def reverse_operate(
3803 self, op: OperatorType, other: Any, **kwargs: Any
3804 ) -> ColumnElement[_T]:
3805 # comparison operators should never call reverse_operate
3806 assert not operators.is_comparison(op)
3807 raise exc.ArgumentError(
3808 "Only comparison operators may be used with ANY/ALL"
3809 )
3812class AsBoolean(WrapsColumnExpression[bool], UnaryExpression[bool]):
3813 inherit_cache = True
3815 def __init__(self, element, operator, negate):
3816 self.element = element
3817 self.type = type_api.BOOLEANTYPE
3818 self.operator = operator
3819 self.negate = negate
3820 self.modifier = None
3821 self.wraps_column_expression = True
3822 self._is_implicitly_boolean = element._is_implicitly_boolean
3824 @property
3825 def wrapped_column_expression(self):
3826 return self.element
3828 def self_group(self, against: Optional[OperatorType] = None) -> Self:
3829 return self
3831 def _negate(self):
3832 if isinstance(self.element, (True_, False_)):
3833 return self.element._negate()
3834 else:
3835 return AsBoolean(self.element, self.negate, self.operator)
3838class BinaryExpression(OperatorExpression[_T]):
3839 """Represent an expression that is ``LEFT <operator> RIGHT``.
3841 A :class:`.BinaryExpression` is generated automatically
3842 whenever two column expressions are used in a Python binary expression:
3844 .. sourcecode:: pycon+sql
3846 >>> from sqlalchemy.sql import column
3847 >>> column("a") + column("b")
3848 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
3849 >>> print(column("a") + column("b"))
3850 {printsql}a + b
3852 """
3854 __visit_name__ = "binary"
3856 _traverse_internals: _TraverseInternalsType = [
3857 ("left", InternalTraversal.dp_clauseelement),
3858 ("right", InternalTraversal.dp_clauseelement),
3859 ("operator", InternalTraversal.dp_operator),
3860 ("negate", InternalTraversal.dp_operator),
3861 ("modifiers", InternalTraversal.dp_plain_dict),
3862 (
3863 "type",
3864 InternalTraversal.dp_type,
3865 ),
3866 ]
3868 _cache_key_traversal = [
3869 ("left", InternalTraversal.dp_clauseelement),
3870 ("right", InternalTraversal.dp_clauseelement),
3871 ("operator", InternalTraversal.dp_operator),
3872 ("modifiers", InternalTraversal.dp_plain_dict),
3873 # "type" affects JSON CAST operators, so while redundant in most cases,
3874 # is needed for that one
3875 (
3876 "type",
3877 InternalTraversal.dp_type,
3878 ),
3879 ]
3881 _is_implicitly_boolean = True
3882 """Indicates that any database will know this is a boolean expression
3883 even if the database does not have an explicit boolean datatype.
3885 """
3887 left: ColumnElement[Any]
3888 right: ColumnElement[Any]
3889 modifiers: Mapping[str, Any]
3891 def __init__(
3892 self,
3893 left: ColumnElement[Any],
3894 right: ColumnElement[Any],
3895 operator: OperatorType,
3896 type_: Optional[_TypeEngineArgument[_T]] = None,
3897 negate: Optional[OperatorType] = None,
3898 modifiers: Optional[Mapping[str, Any]] = None,
3899 ):
3900 # allow compatibility with libraries that
3901 # refer to BinaryExpression directly and pass strings
3902 if isinstance(operator, str):
3903 operator = operators.custom_op(operator)
3904 self._orig = (left.__hash__(), right.__hash__())
3905 self._propagate_attrs = left._propagate_attrs or right._propagate_attrs
3906 self.left = left.self_group(against=operator)
3907 self.right = right.self_group(against=operator)
3908 self.operator = operator
3910 # if type is None, we get NULLTYPE, which is our _T. But I don't
3911 # know how to get the overloads to express that correctly
3912 self.type = type_api.to_instance(type_) # type: ignore
3914 self.negate = negate
3915 self._is_implicitly_boolean = operators.is_boolean(operator)
3917 if modifiers is None:
3918 self.modifiers = {}
3919 else:
3920 self.modifiers = modifiers
3922 @property
3923 def _flattened_operator_clauses(
3924 self,
3925 ) -> typing_Tuple[ColumnElement[Any], ...]:
3926 return (self.left, self.right)
3928 def __bool__(self):
3929 """Implement Python-side "bool" for BinaryExpression as a
3930 simple "identity" check for the left and right attributes,
3931 if the operator is "eq" or "ne". Otherwise the expression
3932 continues to not support "bool" like all other column expressions.
3934 The rationale here is so that ColumnElement objects can be hashable.
3935 What? Well, suppose you do this::
3937 c1, c2 = column("x"), column("y")
3938 s1 = set([c1, c2])
3940 We do that **a lot**, columns inside of sets is an extremely basic
3941 thing all over the ORM for example.
3943 So what happens if we do this? ::
3945 c1 in s1
3947 Hashing means it will normally use ``__hash__()`` of the object,
3948 but in case of hash collision, it's going to also do ``c1 == c1``
3949 and/or ``c1 == c2`` inside. Those operations need to return a
3950 True/False value. But because we override ``==`` and ``!=``, they're
3951 going to get a BinaryExpression. Hence we implement ``__bool__`` here
3952 so that these comparisons behave in this particular context mostly
3953 like regular object comparisons. Thankfully Python is OK with
3954 that! Otherwise we'd have to use special set classes for columns
3955 (which we used to do, decades ago).
3957 """
3958 if self.operator in (operators.eq, operators.ne):
3959 # this is using the eq/ne operator given int hash values,
3960 # rather than Operator, so that "bool" can be based on
3961 # identity
3962 return self.operator(*self._orig) # type: ignore
3963 else:
3964 raise TypeError("Boolean value of this clause is not defined")
3966 if typing.TYPE_CHECKING:
3968 def __invert__(
3969 self: BinaryExpression[_T],
3970 ) -> BinaryExpression[_T]: ...
3972 @util.ro_non_memoized_property
3973 def _from_objects(self) -> List[FromClause]:
3974 return self.left._from_objects + self.right._from_objects
3976 def _negate(self):
3977 if self.negate is not None:
3978 return BinaryExpression(
3979 self.left,
3980 self.right._negate_in_binary(self.negate, self.operator),
3981 self.negate,
3982 negate=self.operator,
3983 type_=self.type,
3984 modifiers=self.modifiers,
3985 )
3986 else:
3987 return self.self_group()._negate()
3990class Slice(ColumnElement[Any]):
3991 """Represent SQL for a Python array-slice object.
3993 This is not a specific SQL construct at this level, but
3994 may be interpreted by specific dialects, e.g. PostgreSQL.
3996 """
3998 __visit_name__ = "slice"
4000 _traverse_internals: _TraverseInternalsType = [
4001 ("start", InternalTraversal.dp_clauseelement),
4002 ("stop", InternalTraversal.dp_clauseelement),
4003 ("step", InternalTraversal.dp_clauseelement),
4004 ]
4006 def __init__(self, start, stop, step, _name=None):
4007 self.start = coercions.expect(
4008 roles.ExpressionElementRole,
4009 start,
4010 name=_name,
4011 type_=type_api.INTEGERTYPE,
4012 )
4013 self.stop = coercions.expect(
4014 roles.ExpressionElementRole,
4015 stop,
4016 name=_name,
4017 type_=type_api.INTEGERTYPE,
4018 )
4019 self.step = coercions.expect(
4020 roles.ExpressionElementRole,
4021 step,
4022 name=_name,
4023 type_=type_api.INTEGERTYPE,
4024 )
4025 self.type = type_api.NULLTYPE
4027 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4028 assert against is operator.getitem
4029 return self
4032class IndexExpression(BinaryExpression[Any]):
4033 """Represent the class of expressions that are like an "index"
4034 operation."""
4036 inherit_cache = True
4039class GroupedElement(DQLDMLClauseElement):
4040 """Represent any parenthesized expression"""
4042 __visit_name__ = "grouping"
4044 element: ClauseElement
4046 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4047 return self
4049 def _ungroup(self):
4050 return self.element._ungroup()
4053class Grouping(GroupedElement, ColumnElement[_T]):
4054 """Represent a grouping within a column expression"""
4056 _traverse_internals: _TraverseInternalsType = [
4057 ("element", InternalTraversal.dp_clauseelement),
4058 ("type", InternalTraversal.dp_type),
4059 ]
4061 _cache_key_traversal = [
4062 ("element", InternalTraversal.dp_clauseelement),
4063 ]
4065 element: Union[TextClause, ClauseList, ColumnElement[_T]]
4067 def __init__(
4068 self, element: Union[TextClause, ClauseList, ColumnElement[_T]]
4069 ):
4070 self.element = element
4072 # nulltype assignment issue
4073 self.type = getattr(element, "type", type_api.NULLTYPE) # type: ignore
4074 self._propagate_attrs = element._propagate_attrs
4076 def _with_binary_element_type(self, type_):
4077 return self.__class__(self.element._with_binary_element_type(type_))
4079 @util.memoized_property
4080 def _is_implicitly_boolean(self):
4081 return self.element._is_implicitly_boolean
4083 @util.non_memoized_property
4084 def _tq_label(self) -> Optional[str]:
4085 return (
4086 getattr(self.element, "_tq_label", None) or self._anon_name_label
4087 )
4089 @util.non_memoized_property
4090 def _proxies(self) -> List[ColumnElement[Any]]:
4091 if isinstance(self.element, ColumnElement):
4092 return [self.element]
4093 else:
4094 return []
4096 @util.ro_non_memoized_property
4097 def _from_objects(self) -> List[FromClause]:
4098 return self.element._from_objects
4100 def __getattr__(self, attr):
4101 return getattr(self.element, attr)
4103 def __getstate__(self):
4104 return {"element": self.element, "type": self.type}
4106 def __setstate__(self, state):
4107 self.element = state["element"]
4108 self.type = state["type"]
4110 if TYPE_CHECKING:
4112 def self_group(
4113 self, against: Optional[OperatorType] = None
4114 ) -> Self: ...
4117class _OverrideBinds(Grouping[_T]):
4118 """used by cache_key->_apply_params_to_element to allow compilation /
4119 execution of a SQL element that's been cached, using an alternate set of
4120 bound parameter values.
4122 This is used by the ORM to swap new parameter values into expressions
4123 that are embedded into loader options like with_expression(),
4124 selectinload(). Previously, this task was accomplished using the
4125 .params() method which would perform a deep-copy instead. This deep
4126 copy proved to be too expensive for more complex expressions.
4128 See #11085
4130 """
4132 __visit_name__ = "override_binds"
4134 def __init__(
4135 self,
4136 element: ColumnElement[_T],
4137 bindparams: Sequence[BindParameter[Any]],
4138 replaces_params: Sequence[BindParameter[Any]],
4139 ):
4140 self.element = element
4141 self.translate = {
4142 k.key: v.value for k, v in zip(replaces_params, bindparams)
4143 }
4145 def _gen_cache_key(
4146 self, anon_map: anon_map, bindparams: List[BindParameter[Any]]
4147 ) -> Optional[typing_Tuple[Any, ...]]:
4148 """generate a cache key for the given element, substituting its bind
4149 values for the translation values present."""
4151 existing_bps: List[BindParameter[Any]] = []
4152 ck = self.element._gen_cache_key(anon_map, existing_bps)
4154 bindparams.extend(
4155 (
4156 bp._with_value(
4157 self.translate[bp.key], maintain_key=True, required=False
4158 )
4159 if bp.key in self.translate
4160 else bp
4161 )
4162 for bp in existing_bps
4163 )
4165 return ck
4168class _OverRange(Enum):
4169 RANGE_UNBOUNDED = 0
4170 RANGE_CURRENT = 1
4173RANGE_UNBOUNDED = _OverRange.RANGE_UNBOUNDED
4174RANGE_CURRENT = _OverRange.RANGE_CURRENT
4176_IntOrRange = Union[int, _OverRange]
4179class Over(ColumnElement[_T]):
4180 """Represent an OVER clause.
4182 This is a special operator against a so-called
4183 "window" function, as well as any aggregate function,
4184 which produces results relative to the result set
4185 itself. Most modern SQL backends now support window functions.
4187 """
4189 __visit_name__ = "over"
4191 _traverse_internals: _TraverseInternalsType = [
4192 ("element", InternalTraversal.dp_clauseelement),
4193 ("order_by", InternalTraversal.dp_clauseelement),
4194 ("partition_by", InternalTraversal.dp_clauseelement),
4195 ("range_", InternalTraversal.dp_plain_obj),
4196 ("rows", InternalTraversal.dp_plain_obj),
4197 ("groups", InternalTraversal.dp_plain_obj),
4198 ]
4200 order_by: Optional[ClauseList] = None
4201 partition_by: Optional[ClauseList] = None
4203 element: ColumnElement[_T]
4204 """The underlying expression object to which this :class:`.Over`
4205 object refers."""
4207 range_: Optional[typing_Tuple[_IntOrRange, _IntOrRange]]
4208 rows: Optional[typing_Tuple[_IntOrRange, _IntOrRange]]
4209 groups: Optional[typing_Tuple[_IntOrRange, _IntOrRange]]
4211 def __init__(
4212 self,
4213 element: ColumnElement[_T],
4214 partition_by: Optional[_ByArgument] = None,
4215 order_by: Optional[_ByArgument] = None,
4216 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4217 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4218 groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4219 ):
4220 self.element = element
4221 if order_by is not None:
4222 self.order_by = ClauseList(
4223 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4224 )
4225 if partition_by is not None:
4226 self.partition_by = ClauseList(
4227 *util.to_list(partition_by),
4228 _literal_as_text_role=roles.ByOfRole,
4229 )
4231 if sum(bool(item) for item in (range_, rows, groups)) > 1:
4232 raise exc.ArgumentError(
4233 "only one of 'rows', 'range_', or 'groups' may be provided"
4234 )
4235 else:
4236 self.range_ = self._interpret_range(range_) if range_ else None
4237 self.rows = self._interpret_range(rows) if rows else None
4238 self.groups = self._interpret_range(groups) if groups else None
4240 def __reduce__(self):
4241 return self.__class__, (
4242 self.element,
4243 self.partition_by,
4244 self.order_by,
4245 self.range_,
4246 self.rows,
4247 self.groups,
4248 )
4250 def _interpret_range(
4251 self,
4252 range_: typing_Tuple[Optional[_IntOrRange], Optional[_IntOrRange]],
4253 ) -> typing_Tuple[_IntOrRange, _IntOrRange]:
4254 if not isinstance(range_, tuple) or len(range_) != 2:
4255 raise exc.ArgumentError("2-tuple expected for range/rows")
4257 r0, r1 = range_
4259 lower: _IntOrRange
4260 upper: _IntOrRange
4262 if r0 is None:
4263 lower = RANGE_UNBOUNDED
4264 elif isinstance(r0, _OverRange):
4265 lower = r0
4266 else:
4267 try:
4268 lower = int(r0)
4269 except ValueError as err:
4270 raise exc.ArgumentError(
4271 "Integer or None expected for range value"
4272 ) from err
4273 else:
4274 if lower == 0:
4275 lower = RANGE_CURRENT
4277 if r1 is None:
4278 upper = RANGE_UNBOUNDED
4279 elif isinstance(r1, _OverRange):
4280 upper = r1
4281 else:
4282 try:
4283 upper = int(r1)
4284 except ValueError as err:
4285 raise exc.ArgumentError(
4286 "Integer or None expected for range value"
4287 ) from err
4288 else:
4289 if upper == 0:
4290 upper = RANGE_CURRENT
4292 return lower, upper
4294 if not TYPE_CHECKING:
4296 @util.memoized_property
4297 def type(self) -> TypeEngine[_T]: # noqa: A001
4298 return self.element.type
4300 @util.ro_non_memoized_property
4301 def _from_objects(self) -> List[FromClause]:
4302 return list(
4303 itertools.chain(
4304 *[
4305 c._from_objects
4306 for c in (self.element, self.partition_by, self.order_by)
4307 if c is not None
4308 ]
4309 )
4310 )
4313class WithinGroup(ColumnElement[_T]):
4314 """Represent a WITHIN GROUP (ORDER BY) clause.
4316 This is a special operator against so-called
4317 "ordered set aggregate" and "hypothetical
4318 set aggregate" functions, including ``percentile_cont()``,
4319 ``rank()``, ``dense_rank()``, etc.
4321 It's supported only by certain database backends, such as PostgreSQL,
4322 Oracle Database and MS SQL Server.
4324 The :class:`.WithinGroup` construct extracts its type from the
4325 method :meth:`.FunctionElement.within_group_type`. If this returns
4326 ``None``, the function's ``.type`` is used.
4328 """
4330 __visit_name__ = "withingroup"
4332 _traverse_internals: _TraverseInternalsType = [
4333 ("element", InternalTraversal.dp_clauseelement),
4334 ("order_by", InternalTraversal.dp_clauseelement),
4335 ]
4337 order_by: Optional[ClauseList] = None
4339 def __init__(
4340 self,
4341 element: Union[FunctionElement[_T], FunctionFilter[_T]],
4342 *order_by: _ColumnExpressionArgument[Any],
4343 ):
4344 self.element = element
4345 if order_by is not None:
4346 self.order_by = ClauseList(
4347 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4348 )
4350 def __reduce__(self):
4351 return self.__class__, (self.element,) + (
4352 tuple(self.order_by) if self.order_by is not None else ()
4353 )
4355 def over(
4356 self,
4357 *,
4358 partition_by: Optional[_ByArgument] = None,
4359 order_by: Optional[_ByArgument] = None,
4360 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4361 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4362 groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4363 ) -> Over[_T]:
4364 """Produce an OVER clause against this :class:`.WithinGroup`
4365 construct.
4367 This function has the same signature as that of
4368 :meth:`.FunctionElement.over`.
4370 """
4371 return Over(
4372 self,
4373 partition_by=partition_by,
4374 order_by=order_by,
4375 range_=range_,
4376 rows=rows,
4377 groups=groups,
4378 )
4380 @overload
4381 def filter(self) -> Self: ...
4383 @overload
4384 def filter(
4385 self,
4386 __criterion0: _ColumnExpressionArgument[bool],
4387 *criterion: _ColumnExpressionArgument[bool],
4388 ) -> FunctionFilter[_T]: ...
4390 def filter(
4391 self, *criterion: _ColumnExpressionArgument[bool]
4392 ) -> Union[Self, FunctionFilter[_T]]:
4393 """Produce a FILTER clause against this function."""
4394 if not criterion:
4395 return self
4396 return FunctionFilter(self, *criterion)
4398 if not TYPE_CHECKING:
4400 @util.memoized_property
4401 def type(self) -> TypeEngine[_T]: # noqa: A001
4402 wgt = self.element.within_group_type(self)
4403 if wgt is not None:
4404 return wgt
4405 else:
4406 return self.element.type
4408 @util.ro_non_memoized_property
4409 def _from_objects(self) -> List[FromClause]:
4410 return list(
4411 itertools.chain(
4412 *[
4413 c._from_objects
4414 for c in (self.element, self.order_by)
4415 if c is not None
4416 ]
4417 )
4418 )
4421class FunctionFilter(Generative, ColumnElement[_T]):
4422 """Represent a function FILTER clause.
4424 This is a special operator against aggregate and window functions,
4425 which controls which rows are passed to it.
4426 It's supported only by certain database backends.
4428 Invocation of :class:`.FunctionFilter` is via
4429 :meth:`.FunctionElement.filter`::
4431 func.count(1).filter(True)
4433 .. seealso::
4435 :meth:`.FunctionElement.filter`
4437 """
4439 __visit_name__ = "funcfilter"
4441 _traverse_internals: _TraverseInternalsType = [
4442 ("func", InternalTraversal.dp_clauseelement),
4443 ("criterion", InternalTraversal.dp_clauseelement),
4444 ]
4446 criterion: Optional[ColumnElement[bool]] = None
4448 def __init__(
4449 self,
4450 func: Union[FunctionElement[_T], WithinGroup[_T]],
4451 *criterion: _ColumnExpressionArgument[bool],
4452 ):
4453 self.func = func
4454 self.filter.non_generative(self, *criterion) # type: ignore
4456 @_generative
4457 def filter(self, *criterion: _ColumnExpressionArgument[bool]) -> Self:
4458 """Produce an additional FILTER against the function.
4460 This method adds additional criteria to the initial criteria
4461 set up by :meth:`.FunctionElement.filter`.
4463 Multiple criteria are joined together at SQL render time
4464 via ``AND``.
4467 """
4469 for crit in list(criterion):
4470 crit = coercions.expect(roles.WhereHavingRole, crit)
4472 if self.criterion is not None:
4473 self.criterion = self.criterion & crit
4474 else:
4475 self.criterion = crit
4477 return self
4479 def over(
4480 self,
4481 partition_by: Optional[
4482 Union[
4483 Iterable[_ColumnExpressionArgument[Any]],
4484 _ColumnExpressionArgument[Any],
4485 ]
4486 ] = None,
4487 order_by: Optional[
4488 Union[
4489 Iterable[_ColumnExpressionArgument[Any]],
4490 _ColumnExpressionArgument[Any],
4491 ]
4492 ] = None,
4493 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4494 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4495 groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4496 ) -> Over[_T]:
4497 """Produce an OVER clause against this filtered function.
4499 Used against aggregate or so-called "window" functions,
4500 for database backends that support window functions.
4502 The expression::
4504 func.rank().filter(MyClass.y > 5).over(order_by="x")
4506 is shorthand for::
4508 from sqlalchemy import over, funcfilter
4510 over(funcfilter(func.rank(), MyClass.y > 5), order_by="x")
4512 See :func:`_expression.over` for a full description.
4514 """
4515 return Over(
4516 self,
4517 partition_by=partition_by,
4518 order_by=order_by,
4519 range_=range_,
4520 rows=rows,
4521 groups=groups,
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) -> Optional[str]:
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 primary_key: ColumnSet,
4639 foreign_keys: Set[KeyedColumnElement[Any]],
4640 name: Optional[str] = None,
4641 key: Optional[str] = None,
4642 name_is_truncatable: bool = False,
4643 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
4644 disallow_is_literal: bool = False,
4645 **kw: Any,
4646 ) -> typing_Tuple[str, ColumnClause[_T]]:
4647 c = ColumnClause(
4648 (
4649 coercions.expect(roles.TruncatedLabelRole, name or self.name)
4650 if name_is_truncatable
4651 else (name or self.name)
4652 ),
4653 type_=self.type,
4654 _selectable=selectable,
4655 is_literal=False,
4656 )
4658 c._propagate_attrs = selectable._propagate_attrs
4659 if name is None:
4660 c.key = self.key
4661 if compound_select_cols:
4662 c._proxies = list(compound_select_cols)
4663 else:
4664 c._proxies = [self]
4666 if selectable._is_clone_of is not None:
4667 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
4668 return c.key, c
4671_PS = ParamSpec("_PS")
4674class Label(roles.LabeledColumnExprRole[_T], NamedColumn[_T]):
4675 """Represents a column label (AS).
4677 Represent a label, as typically applied to any column-level
4678 element using the ``AS`` sql keyword.
4680 """
4682 __visit_name__ = "label"
4684 _traverse_internals: _TraverseInternalsType = [
4685 ("name", InternalTraversal.dp_anon_name),
4686 ("type", InternalTraversal.dp_type),
4687 ("_element", InternalTraversal.dp_clauseelement),
4688 ]
4690 _cache_key_traversal = [
4691 ("name", InternalTraversal.dp_anon_name),
4692 ("_element", InternalTraversal.dp_clauseelement),
4693 ]
4695 _element: ColumnElement[_T]
4696 name: str
4698 def __init__(
4699 self,
4700 name: Optional[str],
4701 element: _ColumnExpressionArgument[_T],
4702 type_: Optional[_TypeEngineArgument[_T]] = None,
4703 ):
4704 orig_element = element
4705 element = coercions.expect(
4706 roles.ExpressionElementRole,
4707 element,
4708 apply_propagate_attrs=self,
4709 )
4710 while isinstance(element, Label):
4711 # TODO: this is only covered in test_text.py, but nothing
4712 # fails if it's removed. determine rationale
4713 element = element.element
4715 if name:
4716 self.name = name
4717 else:
4718 self.name = _anonymous_label.safe_construct(
4719 id(self), getattr(element, "name", "anon")
4720 )
4721 if isinstance(orig_element, Label):
4722 # TODO: no coverage for this block, again would be in
4723 # test_text.py where the resolve_label concept is important
4724 self._resolve_label = orig_element._label
4726 self.key = self._tq_label = self._tq_key_label = self.name
4727 self._element = element
4729 self.type = (
4730 type_api.to_instance(type_)
4731 if type_ is not None
4732 else self._element.type
4733 )
4735 self._proxies = [element]
4737 def __reduce__(self):
4738 return self.__class__, (self.name, self._element, self.type)
4740 @HasMemoized.memoized_attribute
4741 def _render_label_in_columns_clause(self):
4742 return True
4744 def _bind_param(self, operator, obj, type_=None, expanding=False):
4745 return BindParameter(
4746 None,
4747 obj,
4748 _compared_to_operator=operator,
4749 type_=type_,
4750 _compared_to_type=self.type,
4751 unique=True,
4752 expanding=expanding,
4753 )
4755 @util.memoized_property
4756 def _is_implicitly_boolean(self):
4757 return self.element._is_implicitly_boolean
4759 @HasMemoized.memoized_attribute
4760 def _allow_label_resolve(self):
4761 return self.element._allow_label_resolve
4763 @property
4764 def _order_by_label_element(self):
4765 return self
4767 @HasMemoized.memoized_attribute
4768 def element(self) -> ColumnElement[_T]:
4769 return self._element.self_group(against=operators.as_)
4771 def self_group(self, against: Optional[OperatorType] = None) -> Label[_T]:
4772 return self._apply_to_inner(self._element.self_group, against=against)
4774 def _negate(self):
4775 return self._apply_to_inner(self._element._negate)
4777 def _apply_to_inner(
4778 self,
4779 fn: Callable[_PS, ColumnElement[_T]],
4780 *arg: _PS.args,
4781 **kw: _PS.kwargs,
4782 ) -> Label[_T]:
4783 sub_element = fn(*arg, **kw)
4784 if sub_element is not self._element:
4785 return Label(self.name, sub_element, type_=self.type)
4786 else:
4787 return self
4789 @property
4790 def primary_key(self):
4791 return self.element.primary_key
4793 @property
4794 def foreign_keys(self):
4795 return self.element.foreign_keys
4797 def _copy_internals(
4798 self,
4799 *,
4800 clone: _CloneCallableType = _clone,
4801 anonymize_labels: bool = False,
4802 **kw: Any,
4803 ) -> None:
4804 self._reset_memoizations()
4805 self._element = clone(self._element, **kw)
4806 if anonymize_labels:
4807 self.name = _anonymous_label.safe_construct(
4808 id(self), getattr(self.element, "name", "anon")
4809 )
4810 self.key = self._tq_label = self._tq_key_label = self.name
4812 @util.ro_non_memoized_property
4813 def _from_objects(self) -> List[FromClause]:
4814 return self.element._from_objects
4816 def _make_proxy(
4817 self,
4818 selectable: FromClause,
4819 *,
4820 primary_key: ColumnSet,
4821 foreign_keys: Set[KeyedColumnElement[Any]],
4822 name: Optional[str] = None,
4823 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
4824 **kw: Any,
4825 ) -> typing_Tuple[str, ColumnClause[_T]]:
4826 name = self.name if not name else name
4828 key, e = self.element._make_proxy(
4829 selectable,
4830 name=name,
4831 disallow_is_literal=True,
4832 name_is_truncatable=isinstance(name, _truncated_label),
4833 compound_select_cols=compound_select_cols,
4834 primary_key=primary_key,
4835 foreign_keys=foreign_keys,
4836 )
4838 # there was a note here to remove this assertion, which was here
4839 # to determine if we later could support a use case where
4840 # the key and name of a label are separate. But I don't know what
4841 # that case was. For now, this is an unexpected case that occurs
4842 # when a label name conflicts with other columns and select()
4843 # is attempting to disambiguate an explicit label, which is not what
4844 # the user would want. See issue #6090.
4845 if key != self.name and not isinstance(self.name, _anonymous_label):
4846 raise exc.InvalidRequestError(
4847 "Label name %s is being renamed to an anonymous label due "
4848 "to disambiguation "
4849 "which is not supported right now. Please use unique names "
4850 "for explicit labels." % (self.name)
4851 )
4853 e._propagate_attrs = selectable._propagate_attrs
4854 e._proxies.append(self)
4855 if self.type is not None:
4856 e.type = self.type
4858 return self.key, e
4861class ColumnClause(
4862 roles.DDLReferredColumnRole,
4863 roles.LabeledColumnExprRole[_T],
4864 roles.StrAsPlainColumnRole,
4865 Immutable,
4866 NamedColumn[_T],
4867):
4868 """Represents a column expression from any textual string.
4870 The :class:`.ColumnClause`, a lightweight analogue to the
4871 :class:`_schema.Column` class, is typically invoked using the
4872 :func:`_expression.column` function, as in::
4874 from sqlalchemy import column
4876 id, name = column("id"), column("name")
4877 stmt = select(id, name).select_from("user")
4879 The above statement would produce SQL like:
4881 .. sourcecode:: sql
4883 SELECT id, name FROM user
4885 :class:`.ColumnClause` is the immediate superclass of the schema-specific
4886 :class:`_schema.Column` object. While the :class:`_schema.Column`
4887 class has all the
4888 same capabilities as :class:`.ColumnClause`, the :class:`.ColumnClause`
4889 class is usable by itself in those cases where behavioral requirements
4890 are limited to simple SQL expression generation. The object has none of
4891 the associations with schema-level metadata or with execution-time
4892 behavior that :class:`_schema.Column` does,
4893 so in that sense is a "lightweight"
4894 version of :class:`_schema.Column`.
4896 Full details on :class:`.ColumnClause` usage is at
4897 :func:`_expression.column`.
4899 .. seealso::
4901 :func:`_expression.column`
4903 :class:`_schema.Column`
4905 """
4907 table: Optional[FromClause]
4908 is_literal: bool
4910 __visit_name__ = "column"
4912 _traverse_internals: _TraverseInternalsType = [
4913 ("name", InternalTraversal.dp_anon_name),
4914 ("type", InternalTraversal.dp_type),
4915 ("table", InternalTraversal.dp_clauseelement),
4916 ("is_literal", InternalTraversal.dp_boolean),
4917 ]
4919 onupdate: Optional[DefaultGenerator] = None
4920 default: Optional[DefaultGenerator] = None
4921 server_default: Optional[FetchedValue] = None
4922 server_onupdate: Optional[FetchedValue] = None
4924 _is_multiparam_column = False
4926 @property
4927 def _is_star(self):
4928 return self.is_literal and self.name == "*"
4930 def __init__(
4931 self,
4932 text: str,
4933 type_: Optional[_TypeEngineArgument[_T]] = None,
4934 is_literal: bool = False,
4935 _selectable: Optional[FromClause] = None,
4936 ):
4937 self.key = self.name = text
4938 self.table = _selectable
4940 # if type is None, we get NULLTYPE, which is our _T. But I don't
4941 # know how to get the overloads to express that correctly
4942 self.type = type_api.to_instance(type_) # type: ignore
4944 self.is_literal = is_literal
4946 def get_children(self, *, column_tables=False, **kw):
4947 # override base get_children() to not return the Table
4948 # or selectable that is parent to this column. Traversals
4949 # expect the columns of tables and subqueries to be leaf nodes.
4950 return []
4952 @property
4953 def entity_namespace(self):
4954 if self.table is not None:
4955 return self.table.entity_namespace
4956 else:
4957 return super().entity_namespace
4959 def _clone(self, detect_subquery_cols=False, **kw):
4960 if (
4961 detect_subquery_cols
4962 and self.table is not None
4963 and self.table._is_subquery
4964 ):
4965 clone = kw.pop("clone")
4966 table = clone(self.table, **kw)
4967 new = table.c.corresponding_column(self)
4968 return new
4970 return super()._clone(**kw)
4972 @HasMemoized_ro_memoized_attribute
4973 def _from_objects(self) -> List[FromClause]:
4974 t = self.table
4975 if t is not None:
4976 return [t]
4977 else:
4978 return []
4980 @HasMemoized.memoized_attribute
4981 def _render_label_in_columns_clause(self):
4982 return self.table is not None
4984 @property
4985 def _ddl_label(self):
4986 return self._gen_tq_label(self.name, dedupe_on_key=False)
4988 def _compare_name_for_result(self, other):
4989 if (
4990 self.is_literal
4991 or self.table is None
4992 or self.table._is_textual
4993 or not hasattr(other, "proxy_set")
4994 or (
4995 isinstance(other, ColumnClause)
4996 and (
4997 other.is_literal
4998 or other.table is None
4999 or other.table._is_textual
5000 )
5001 )
5002 ):
5003 return (hasattr(other, "name") and self.name == other.name) or (
5004 hasattr(other, "_tq_label")
5005 and self._tq_label == other._tq_label
5006 )
5007 else:
5008 return other.proxy_set.intersection(self.proxy_set)
5010 def _gen_tq_label(
5011 self, name: str, dedupe_on_key: bool = True
5012 ) -> Optional[str]:
5013 """generate table-qualified label
5015 for a table-bound column this is <tablename>_<columnname>.
5017 used primarily for LABEL_STYLE_TABLENAME_PLUS_COL
5018 as well as the .columns collection on a Join object.
5020 """
5021 label: str
5022 t = self.table
5023 if self.is_literal:
5024 return None
5025 elif t is not None and is_named_from_clause(t):
5026 if has_schema_attr(t) and t.schema:
5027 label = t.schema.replace(".", "_") + "_" + t.name + "_" + name
5028 else:
5029 assert not TYPE_CHECKING or isinstance(t, NamedFromClause)
5030 label = t.name + "_" + name
5032 # propagate name quoting rules for labels.
5033 if is_quoted_name(name) and name.quote is not None:
5034 if is_quoted_name(label):
5035 label.quote = name.quote
5036 else:
5037 label = quoted_name(label, name.quote)
5038 elif is_quoted_name(t.name) and t.name.quote is not None:
5039 # can't get this situation to occur, so let's
5040 # assert false on it for now
5041 assert not isinstance(label, quoted_name)
5042 label = quoted_name(label, t.name.quote)
5044 if dedupe_on_key:
5045 # ensure the label name doesn't conflict with that of an
5046 # existing column. note that this implies that any Column
5047 # must **not** set up its _label before its parent table has
5048 # all of its other Column objects set up. There are several
5049 # tables in the test suite which will fail otherwise; example:
5050 # table "owner" has columns "name" and "owner_name". Therefore
5051 # column owner.name cannot use the label "owner_name", it has
5052 # to be "owner_name_1".
5053 if label in t.c:
5054 _label = label
5055 counter = 1
5056 while _label in t.c:
5057 _label = label + "_" + str(counter)
5058 counter += 1
5059 label = _label
5061 return coercions.expect(roles.TruncatedLabelRole, label)
5063 else:
5064 return name
5066 def _make_proxy(
5067 self,
5068 selectable: FromClause,
5069 *,
5070 primary_key: ColumnSet,
5071 foreign_keys: Set[KeyedColumnElement[Any]],
5072 name: Optional[str] = None,
5073 key: Optional[str] = None,
5074 name_is_truncatable: bool = False,
5075 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5076 disallow_is_literal: bool = False,
5077 **kw: Any,
5078 ) -> typing_Tuple[str, ColumnClause[_T]]:
5079 # the "is_literal" flag normally should never be propagated; a proxied
5080 # column is always a SQL identifier and never the actual expression
5081 # being evaluated. however, there is a case where the "is_literal" flag
5082 # might be used to allow the given identifier to have a fixed quoting
5083 # pattern already, so maintain the flag for the proxy unless a
5084 # :class:`.Label` object is creating the proxy. See [ticket:4730].
5085 is_literal = (
5086 not disallow_is_literal
5087 and self.is_literal
5088 and (
5089 # note this does not accommodate for quoted_name differences
5090 # right now
5091 name is None
5092 or name == self.name
5093 )
5094 )
5095 c = self._constructor(
5096 (
5097 coercions.expect(roles.TruncatedLabelRole, name or self.name)
5098 if name_is_truncatable
5099 else (name or self.name)
5100 ),
5101 type_=self.type,
5102 _selectable=selectable,
5103 is_literal=is_literal,
5104 )
5105 c._propagate_attrs = selectable._propagate_attrs
5106 if name is None:
5107 c.key = self.key
5108 if compound_select_cols:
5109 c._proxies = list(compound_select_cols)
5110 else:
5111 c._proxies = [self]
5113 if selectable._is_clone_of is not None:
5114 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
5115 return c.key, c
5118class TableValuedColumn(NamedColumn[_T]):
5119 __visit_name__ = "table_valued_column"
5121 _traverse_internals: _TraverseInternalsType = [
5122 ("name", InternalTraversal.dp_anon_name),
5123 ("type", InternalTraversal.dp_type),
5124 ("scalar_alias", InternalTraversal.dp_clauseelement),
5125 ]
5127 def __init__(self, scalar_alias: NamedFromClause, type_: TypeEngine[_T]):
5128 self.scalar_alias = scalar_alias
5129 self.key = self.name = scalar_alias.name
5130 self.type = type_
5132 def _copy_internals(
5133 self, clone: _CloneCallableType = _clone, **kw: Any
5134 ) -> None:
5135 self.scalar_alias = clone(self.scalar_alias, **kw)
5136 self.key = self.name = self.scalar_alias.name
5138 @util.ro_non_memoized_property
5139 def _from_objects(self) -> List[FromClause]:
5140 return [self.scalar_alias]
5143class CollationClause(ColumnElement[str]):
5144 __visit_name__ = "collation"
5146 _traverse_internals: _TraverseInternalsType = [
5147 ("collation", InternalTraversal.dp_string)
5148 ]
5150 @classmethod
5151 @util.preload_module("sqlalchemy.sql.sqltypes")
5152 def _create_collation_expression(
5153 cls, expression: _ColumnExpressionArgument[str], collation: str
5154 ) -> BinaryExpression[str]:
5156 sqltypes = util.preloaded.sql_sqltypes
5158 expr = coercions.expect(roles.ExpressionElementRole[str], expression)
5160 if expr.type._type_affinity is sqltypes.String:
5161 collate_type = expr.type._with_collation(collation)
5162 else:
5163 collate_type = expr.type
5165 return BinaryExpression(
5166 expr,
5167 CollationClause(collation),
5168 operators.collate,
5169 type_=collate_type,
5170 )
5172 def __init__(self, collation):
5173 self.collation = collation
5176class _IdentifiedClause(Executable, ClauseElement):
5177 __visit_name__ = "identified"
5179 def __init__(self, ident):
5180 self.ident = ident
5183class SavepointClause(_IdentifiedClause):
5184 __visit_name__ = "savepoint"
5185 inherit_cache = False
5188class RollbackToSavepointClause(_IdentifiedClause):
5189 __visit_name__ = "rollback_to_savepoint"
5190 inherit_cache = False
5193class ReleaseSavepointClause(_IdentifiedClause):
5194 __visit_name__ = "release_savepoint"
5195 inherit_cache = False
5198class quoted_name(util.MemoizedSlots, str):
5199 """Represent a SQL identifier combined with quoting preferences.
5201 :class:`.quoted_name` is a Python unicode/str subclass which
5202 represents a particular identifier name along with a
5203 ``quote`` flag. This ``quote`` flag, when set to
5204 ``True`` or ``False``, overrides automatic quoting behavior
5205 for this identifier in order to either unconditionally quote
5206 or to not quote the name. If left at its default of ``None``,
5207 quoting behavior is applied to the identifier on a per-backend basis
5208 based on an examination of the token itself.
5210 A :class:`.quoted_name` object with ``quote=True`` is also
5211 prevented from being modified in the case of a so-called
5212 "name normalize" option. Certain database backends, such as
5213 Oracle Database, Firebird, and DB2 "normalize" case-insensitive names
5214 as uppercase. The SQLAlchemy dialects for these backends
5215 convert from SQLAlchemy's lower-case-means-insensitive convention
5216 to the upper-case-means-insensitive conventions of those backends.
5217 The ``quote=True`` flag here will prevent this conversion from occurring
5218 to support an identifier that's quoted as all lower case against
5219 such a backend.
5221 The :class:`.quoted_name` object is normally created automatically
5222 when specifying the name for key schema constructs such as
5223 :class:`_schema.Table`, :class:`_schema.Column`, and others.
5224 The class can also be
5225 passed explicitly as the name to any function that receives a name which
5226 can be quoted. Such as to use the :meth:`_engine.Engine.has_table`
5227 method with
5228 an unconditionally quoted name::
5230 from sqlalchemy import create_engine
5231 from sqlalchemy import inspect
5232 from sqlalchemy.sql import quoted_name
5234 engine = create_engine("oracle+oracledb://some_dsn")
5235 print(inspect(engine).has_table(quoted_name("some_table", True)))
5237 The above logic will run the "has table" logic against the Oracle Database
5238 backend, passing the name exactly as ``"some_table"`` without converting to
5239 upper case.
5241 .. versionchanged:: 1.2 The :class:`.quoted_name` construct is now
5242 importable from ``sqlalchemy.sql``, in addition to the previous
5243 location of ``sqlalchemy.sql.elements``.
5245 """
5247 __slots__ = "quote", "lower", "upper"
5249 quote: Optional[bool]
5251 @overload
5252 @classmethod
5253 def construct(cls, value: str, quote: Optional[bool]) -> quoted_name: ...
5255 @overload
5256 @classmethod
5257 def construct(cls, value: None, quote: Optional[bool]) -> None: ...
5259 @classmethod
5260 def construct(
5261 cls, value: Optional[str], quote: Optional[bool]
5262 ) -> Optional[quoted_name]:
5263 if value is None:
5264 return None
5265 else:
5266 return quoted_name(value, quote)
5268 def __new__(cls, value: str, quote: Optional[bool]) -> quoted_name:
5269 assert (
5270 value is not None
5271 ), "use quoted_name.construct() for None passthrough"
5272 if isinstance(value, cls) and (quote is None or value.quote == quote):
5273 return value
5274 self = super().__new__(cls, value)
5276 self.quote = quote
5277 return self
5279 def __reduce__(self):
5280 return quoted_name, (str(self), self.quote)
5282 def _memoized_method_lower(self):
5283 if self.quote:
5284 return self
5285 else:
5286 return str(self).lower()
5288 def _memoized_method_upper(self):
5289 if self.quote:
5290 return self
5291 else:
5292 return str(self).upper()
5295def _find_columns(clause: ClauseElement) -> Set[ColumnClause[Any]]:
5296 """locate Column objects within the given expression."""
5298 cols: Set[ColumnClause[Any]] = set()
5299 traverse(clause, {}, {"column": cols.add})
5300 return cols
5303def _type_from_args(args: Sequence[ColumnElement[_T]]) -> TypeEngine[_T]:
5304 for a in args:
5305 if not a.type._isnull:
5306 return a.type
5307 else:
5308 return type_api.NULLTYPE # type: ignore
5311def _corresponding_column_or_error(fromclause, column, require_embedded=False):
5312 c = fromclause.corresponding_column(
5313 column, require_embedded=require_embedded
5314 )
5315 if c is None:
5316 raise exc.InvalidRequestError(
5317 "Given column '%s', attached to table '%s', "
5318 "failed to locate a corresponding column from table '%s'"
5319 % (column, getattr(column, "table", None), fromclause.description)
5320 )
5321 return c
5324class _memoized_property_but_not_nulltype(
5325 util.memoized_property["TypeEngine[_T]"]
5326):
5327 """memoized property, but dont memoize NullType"""
5329 def __get__(self, obj, cls):
5330 if obj is None:
5331 return self
5332 result = self.fget(obj)
5333 if not result._isnull:
5334 obj.__dict__[self.__name__] = result
5335 return result
5338class AnnotatedColumnElement(Annotated):
5339 _Annotated__element: ColumnElement[Any]
5341 def __init__(self, element, values):
5342 Annotated.__init__(self, element, values)
5343 for attr in (
5344 "comparator",
5345 "_proxy_key",
5346 "_tq_key_label",
5347 "_tq_label",
5348 "_non_anon_label",
5349 "type",
5350 ):
5351 self.__dict__.pop(attr, None)
5352 for attr in ("name", "key", "table"):
5353 if self.__dict__.get(attr, False) is None:
5354 self.__dict__.pop(attr)
5356 def _with_annotations(self, values):
5357 clone = super()._with_annotations(values)
5358 for attr in (
5359 "comparator",
5360 "_proxy_key",
5361 "_tq_key_label",
5362 "_tq_label",
5363 "_non_anon_label",
5364 ):
5365 clone.__dict__.pop(attr, None)
5366 return clone
5368 @util.memoized_property
5369 def name(self):
5370 """pull 'name' from parent, if not present"""
5371 return self._Annotated__element.name
5373 @_memoized_property_but_not_nulltype
5374 def type(self):
5375 """pull 'type' from parent and don't cache if null.
5377 type is routinely changed on existing columns within the
5378 mapped_column() initialization process, and "type" is also consulted
5379 during the creation of SQL expressions. Therefore it can change after
5380 it was already retrieved. At the same time we don't want annotated
5381 objects having overhead when expressions are produced, so continue
5382 to memoize, but only when we have a non-null type.
5384 """
5385 return self._Annotated__element.type
5387 @util.memoized_property
5388 def table(self):
5389 """pull 'table' from parent, if not present"""
5390 return self._Annotated__element.table
5392 @util.memoized_property
5393 def key(self):
5394 """pull 'key' from parent, if not present"""
5395 return self._Annotated__element.key
5397 @util.memoized_property
5398 def info(self) -> _InfoType:
5399 if TYPE_CHECKING:
5400 assert isinstance(self._Annotated__element, Column)
5401 return self._Annotated__element.info
5403 @util.memoized_property
5404 def _anon_name_label(self) -> str:
5405 return self._Annotated__element._anon_name_label
5408class _truncated_label(quoted_name):
5409 """A unicode subclass used to identify symbolic "
5410 "names that may require truncation."""
5412 __slots__ = ()
5414 def __new__(cls, value: str, quote: Optional[bool] = None) -> Any:
5415 quote = getattr(value, "quote", quote)
5416 # return super(_truncated_label, cls).__new__(cls, value, quote, True)
5417 return super().__new__(cls, value, quote)
5419 def __reduce__(self) -> Any:
5420 return self.__class__, (str(self), self.quote)
5422 def apply_map(self, map_: Mapping[str, Any]) -> str:
5423 return self
5426class conv(_truncated_label):
5427 """Mark a string indicating that a name has already been converted
5428 by a naming convention.
5430 This is a string subclass that indicates a name that should not be
5431 subject to any further naming conventions.
5433 E.g. when we create a :class:`.Constraint` using a naming convention
5434 as follows::
5436 m = MetaData(
5437 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5438 )
5439 t = Table(
5440 "t", m, Column("x", Integer), CheckConstraint("x > 5", name="x5")
5441 )
5443 The name of the above constraint will be rendered as ``"ck_t_x5"``.
5444 That is, the existing name ``x5`` is used in the naming convention as the
5445 ``constraint_name`` token.
5447 In some situations, such as in migration scripts, we may be rendering
5448 the above :class:`.CheckConstraint` with a name that's already been
5449 converted. In order to make sure the name isn't double-modified, the
5450 new name is applied using the :func:`_schema.conv` marker. We can
5451 use this explicitly as follows::
5454 m = MetaData(
5455 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5456 )
5457 t = Table(
5458 "t",
5459 m,
5460 Column("x", Integer),
5461 CheckConstraint("x > 5", name=conv("ck_t_x5")),
5462 )
5464 Where above, the :func:`_schema.conv` marker indicates that the constraint
5465 name here is final, and the name will render as ``"ck_t_x5"`` and not
5466 ``"ck_t_ck_t_x5"``
5468 .. seealso::
5470 :ref:`constraint_naming_conventions`
5472 """
5474 __slots__ = ()
5477# for backwards compatibility in case
5478# someone is re-implementing the
5479# _truncated_identifier() sequence in a custom
5480# compiler
5481_generated_label = _truncated_label
5484class _anonymous_label(_truncated_label):
5485 """A unicode subclass used to identify anonymously
5486 generated names."""
5488 __slots__ = ()
5490 @classmethod
5491 def safe_construct(
5492 cls,
5493 seed: int,
5494 body: str,
5495 enclosing_label: Optional[str] = None,
5496 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 = re.sub(r"[%\(\) \$]+", "_", body)
5502 if sanitize_key:
5503 # sanitize_key is then an extra step used by BindParameter
5504 body = body.strip("_")
5506 label = "%%(%d %s)s" % (seed, body.replace("%", "%%"))
5507 if enclosing_label:
5508 label = "%s%s" % (enclosing_label, label)
5510 return _anonymous_label(label)
5512 def __add__(self, other):
5513 if "%" in other and not isinstance(other, _anonymous_label):
5514 other = str(other).replace("%", "%%")
5515 else:
5516 other = str(other)
5518 return _anonymous_label(
5519 quoted_name(
5520 str.__add__(self, other),
5521 self.quote,
5522 )
5523 )
5525 def __radd__(self, other):
5526 if "%" in other and not isinstance(other, _anonymous_label):
5527 other = str(other).replace("%", "%%")
5528 else:
5529 other = str(other)
5531 return _anonymous_label(
5532 quoted_name(
5533 str.__add__(other, self),
5534 self.quote,
5535 )
5536 )
5538 def apply_map(self, map_):
5539 if self.quote is not None:
5540 # preserve quoting only if necessary
5541 return quoted_name(self % map_, self.quote)
5542 else:
5543 # else skip the constructor call
5544 return self % map_