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