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 _OnlyColumnArgument
91 from ._typing import _PropagateAttrsType
92 from ._typing import _TypeEngineArgument
93 from .base import _EntityNamespace
94 from .base import ColumnSet
95 from .cache_key import _CacheKeyTraversalType
96 from .cache_key import CacheKey
97 from .compiler import Compiled
98 from .compiler import SQLCompiler
99 from .functions import FunctionElement
100 from .operators import OperatorType
101 from .schema import Column
102 from .schema import DefaultGenerator
103 from .schema import FetchedValue
104 from .schema import ForeignKey
105 from .selectable import _SelectIterable
106 from .selectable import FromClause
107 from .selectable import NamedFromClause
108 from .selectable import TextualSelect
109 from .sqltypes import TupleType
110 from .type_api import TypeEngine
111 from .visitors import _CloneCallableType
112 from .visitors import _TraverseInternalsType
113 from .visitors import anon_map
114 from ..engine import Connection
115 from ..engine import Dialect
116 from ..engine.interfaces import _CoreMultiExecuteParams
117 from ..engine.interfaces import CacheStats
118 from ..engine.interfaces import CompiledCacheType
119 from ..engine.interfaces import CoreExecuteOptionsParameter
120 from ..engine.interfaces import SchemaTranslateMapType
121 from ..engine.result import Result
124_NUMERIC = Union[float, Decimal]
125_NUMBER = Union[float, int, Decimal]
127_T = TypeVar("_T", bound="Any")
128_T_co = TypeVar("_T_co", bound=Any, covariant=True)
129_OPT = TypeVar("_OPT", bound="Any")
130_NT = TypeVar("_NT", bound="_NUMERIC")
132_NMT = TypeVar("_NMT", bound="_NUMBER")
135@overload
136def literal(
137 value: Any,
138 type_: _TypeEngineArgument[_T],
139 literal_execute: bool = False,
140) -> BindParameter[_T]: ...
143@overload
144def literal(
145 value: _T,
146 type_: None = None,
147 literal_execute: bool = False,
148) -> BindParameter[_T]: ...
151@overload
152def literal(
153 value: Any,
154 type_: Optional[_TypeEngineArgument[Any]] = None,
155 literal_execute: bool = False,
156) -> BindParameter[Any]: ...
159def literal(
160 value: Any,
161 type_: Optional[_TypeEngineArgument[Any]] = None,
162 literal_execute: bool = False,
163) -> BindParameter[Any]:
164 r"""Return a literal clause, bound to a bind parameter.
166 Literal clauses are created automatically when non-
167 :class:`_expression.ClauseElement` objects (such as strings, ints, dates,
168 etc.) are
169 used in a comparison operation with a :class:`_expression.ColumnElement`
170 subclass,
171 such as a :class:`~sqlalchemy.schema.Column` object. Use this function
172 to force the generation of a literal clause, which will be created as a
173 :class:`BindParameter` with a bound value.
175 :param value: the value to be bound. Can be any Python object supported by
176 the underlying DB-API, or is translatable via the given type argument.
178 :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine` which will
179 provide bind-parameter translation for this literal.
181 :param literal_execute: optional bool, when True, the SQL engine will
182 attempt to render the bound value directly in the SQL statement at
183 execution time rather than providing as a parameter value.
185 .. versionadded:: 2.0
187 """
188 return coercions.expect(
189 roles.LiteralValueRole,
190 value,
191 type_=type_,
192 literal_execute=literal_execute,
193 )
196def literal_column(
197 text: str, type_: Optional[_TypeEngineArgument[_T]] = None
198) -> ColumnClause[_T]:
199 r"""Produce a :class:`.ColumnClause` object that has the
200 :paramref:`_expression.column.is_literal` flag set to True.
202 :func:`_expression.literal_column` is similar to
203 :func:`_expression.column`, except that
204 it is more often used as a "standalone" column expression that renders
205 exactly as stated; while :func:`_expression.column`
206 stores a string name that
207 will be assumed to be part of a table and may be quoted as such,
208 :func:`_expression.literal_column` can be that,
209 or any other arbitrary column-oriented
210 expression.
212 :param text: the text of the expression; can be any SQL expression.
213 Quoting rules will not be applied. To specify a column-name expression
214 which should be subject to quoting rules, use the :func:`column`
215 function.
217 :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine`
218 object which will
219 provide result-set translation and additional expression semantics for
220 this column. If left as ``None`` the type will be :class:`.NullType`.
222 .. seealso::
224 :func:`_expression.column`
226 :func:`_expression.text`
228 :ref:`tutorial_select_arbitrary_text`
230 """
231 return ColumnClause(text, type_=type_, is_literal=True)
234class CompilerElement(Visitable):
235 """base class for SQL elements that can be compiled to produce a
236 SQL string.
238 .. versionadded:: 2.0
240 """
242 __slots__ = ()
243 __visit_name__ = "compiler_element"
245 supports_execution = False
247 stringify_dialect = "default"
249 @util.preload_module("sqlalchemy.engine.default")
250 @util.preload_module("sqlalchemy.engine.url")
251 def compile(
252 self,
253 bind: Optional[_HasDialect] = None,
254 dialect: Optional[Dialect] = None,
255 **kw: Any,
256 ) -> Compiled:
257 """Compile this SQL expression.
259 The return value is a :class:`~.Compiled` object.
260 Calling ``str()`` or ``unicode()`` on the returned value will yield a
261 string representation of the result. The
262 :class:`~.Compiled` object also can return a
263 dictionary of bind parameter names and values
264 using the ``params`` accessor.
266 :param bind: An :class:`.Connection` or :class:`.Engine` which
267 can provide a :class:`.Dialect` in order to generate a
268 :class:`.Compiled` object. If the ``bind`` and
269 ``dialect`` parameters are both omitted, a default SQL compiler
270 is used.
272 :param column_keys: Used for INSERT and UPDATE statements, a list of
273 column names which should be present in the VALUES clause of the
274 compiled statement. If ``None``, all columns from the target table
275 object are rendered.
277 :param dialect: A :class:`.Dialect` instance which can generate
278 a :class:`.Compiled` object. This argument takes precedence over
279 the ``bind`` argument.
281 :param compile_kwargs: optional dictionary of additional parameters
282 that will be passed through to the compiler within all "visit"
283 methods. This allows any custom flag to be passed through to
284 a custom compilation construct, for example. It is also used
285 for the case of passing the ``literal_binds`` flag through::
287 from sqlalchemy.sql import table, column, select
289 t = table("t", column("x"))
291 s = select(t).where(t.c.x == 5)
293 print(s.compile(compile_kwargs={"literal_binds": True}))
295 .. seealso::
297 :ref:`faq_sql_expression_string`
299 """
301 if dialect is None:
302 if bind:
303 dialect = bind.dialect
304 elif self.stringify_dialect == "default":
305 dialect = self._default_dialect()
306 else:
307 url = util.preloaded.engine_url
308 dialect = url.URL.create(
309 self.stringify_dialect
310 ).get_dialect()()
312 return self._compiler(dialect, **kw)
314 def _default_dialect(self):
315 default = util.preloaded.engine_default
316 return default.StrCompileDialect()
318 def _compiler(self, dialect: Dialect, **kw: Any) -> Compiled:
319 """Return a compiler appropriate for this ClauseElement, given a
320 Dialect."""
322 if TYPE_CHECKING:
323 assert isinstance(self, ClauseElement)
324 return dialect.statement_compiler(dialect, self, **kw)
326 def __str__(self) -> str:
327 return str(self.compile())
330@inspection._self_inspects
331class ClauseElement(
332 SupportsWrappingAnnotations,
333 MemoizedHasCacheKey,
334 HasCopyInternals,
335 ExternallyTraversible,
336 CompilerElement,
337):
338 """Base class for elements of a programmatically constructed SQL
339 expression.
341 """
343 __visit_name__ = "clause"
345 if TYPE_CHECKING:
347 @util.memoized_property
348 def _propagate_attrs(self) -> _PropagateAttrsType:
349 """like annotations, however these propagate outwards liberally
350 as SQL constructs are built, and are set up at construction time.
352 """
353 ...
355 else:
356 _propagate_attrs = util.EMPTY_DICT
358 @util.ro_memoized_property
359 def description(self) -> Optional[str]:
360 return None
362 _is_clone_of: Optional[Self] = None
364 is_clause_element = True
365 is_selectable = False
366 is_dml = False
367 _is_column_element = False
368 _is_keyed_column_element = False
369 _is_table = False
370 _gen_static_annotations_cache_key = False
371 _is_textual = False
372 _is_from_clause = False
373 _is_returns_rows = False
374 _is_text_clause = False
375 _is_from_container = False
376 _is_select_container = False
377 _is_select_base = False
378 _is_select_statement = False
379 _is_bind_parameter = False
380 _is_clause_list = False
381 _is_lambda_element = False
382 _is_singleton_constant = False
383 _is_immutable = False
384 _is_star = False
386 @property
387 def _order_by_label_element(self) -> Optional[Label[Any]]:
388 return None
390 _cache_key_traversal: _CacheKeyTraversalType = None
392 negation_clause: ColumnElement[bool]
394 if typing.TYPE_CHECKING:
396 def get_children(
397 self, *, omit_attrs: typing_Tuple[str, ...] = ..., **kw: Any
398 ) -> Iterable[ClauseElement]: ...
400 @util.ro_non_memoized_property
401 def _from_objects(self) -> List[FromClause]:
402 return []
404 def _set_propagate_attrs(self, values: Mapping[str, Any]) -> Self:
405 # usually, self._propagate_attrs is empty here. one case where it's
406 # not is a subquery against ORM select, that is then pulled as a
407 # property of an aliased class. should all be good
409 # assert not self._propagate_attrs
411 self._propagate_attrs = util.immutabledict(values)
412 return self
414 def _default_compiler(self) -> SQLCompiler:
415 dialect = self._default_dialect()
416 return dialect.statement_compiler(dialect, self) # type: ignore
418 def _clone(self, **kw: Any) -> Self:
419 """Create a shallow copy of this ClauseElement.
421 This method may be used by a generative API. Its also used as
422 part of the "deep" copy afforded by a traversal that combines
423 the _copy_internals() method.
425 """
427 skip = self._memoized_keys
428 c = self.__class__.__new__(self.__class__)
430 if skip:
431 # ensure this iteration remains atomic
432 c.__dict__ = {
433 k: v for k, v in self.__dict__.copy().items() if k not in skip
434 }
435 else:
436 c.__dict__ = self.__dict__.copy()
438 # this is a marker that helps to "equate" clauses to each other
439 # when a Select returns its list of FROM clauses. the cloning
440 # process leaves around a lot of remnants of the previous clause
441 # typically in the form of column expressions still attached to the
442 # old table.
443 cc = self._is_clone_of
444 c._is_clone_of = cc if cc is not None else self
445 return c
447 def _negate_in_binary(self, negated_op, original_op):
448 """a hook to allow the right side of a binary expression to respond
449 to a negation of the binary expression.
451 Used for the special case of expanding bind parameter with IN.
453 """
454 return self
456 def _with_binary_element_type(self, type_):
457 """in the context of binary expression, convert the type of this
458 object to the one given.
460 applies only to :class:`_expression.ColumnElement` classes.
462 """
463 return self
465 @property
466 def _constructor(self): # type: ignore[override]
467 """return the 'constructor' for this ClauseElement.
469 This is for the purposes for creating a new object of
470 this type. Usually, its just the element's __class__.
471 However, the "Annotated" version of the object overrides
472 to return the class of its proxied element.
474 """
475 return self.__class__
477 @HasMemoized.memoized_attribute
478 def _cloned_set(self):
479 """Return the set consisting all cloned ancestors of this
480 ClauseElement.
482 Includes this ClauseElement. This accessor tends to be used for
483 FromClause objects to identify 'equivalent' FROM clauses, regardless
484 of transformative operations.
486 """
487 s = util.column_set()
488 f: Optional[ClauseElement] = self
490 # note this creates a cycle, asserted in test_memusage. however,
491 # turning this into a plain @property adds tends of thousands of method
492 # calls to Core / ORM performance tests, so the small overhead
493 # introduced by the relatively small amount of short term cycles
494 # produced here is preferable
495 while f is not None:
496 s.add(f)
497 f = f._is_clone_of
498 return s
500 def _de_clone(self):
501 while self._is_clone_of is not None:
502 self = self._is_clone_of
503 return self
505 @util.ro_non_memoized_property
506 def entity_namespace(self) -> _EntityNamespace:
507 raise AttributeError(
508 "This SQL expression has no entity namespace "
509 "with which to filter from."
510 )
512 def __getstate__(self):
513 d = self.__dict__.copy()
514 d.pop("_is_clone_of", None)
515 d.pop("_generate_cache_key", None)
516 return d
518 def _execute_on_connection(
519 self,
520 connection: Connection,
521 distilled_params: _CoreMultiExecuteParams,
522 execution_options: CoreExecuteOptionsParameter,
523 ) -> Result[Any]:
524 if self.supports_execution:
525 if TYPE_CHECKING:
526 assert isinstance(self, Executable)
527 return connection._execute_clauseelement(
528 self, distilled_params, execution_options
529 )
530 else:
531 raise exc.ObjectNotExecutableError(self)
533 def _execute_on_scalar(
534 self,
535 connection: Connection,
536 distilled_params: _CoreMultiExecuteParams,
537 execution_options: CoreExecuteOptionsParameter,
538 ) -> Any:
539 """an additional hook for subclasses to provide a different
540 implementation for connection.scalar() vs. connection.execute().
542 .. versionadded:: 2.0
544 """
545 return self._execute_on_connection(
546 connection, distilled_params, execution_options
547 ).scalar()
549 def _get_embedded_bindparams(self) -> Sequence[BindParameter[Any]]:
550 """Return the list of :class:`.BindParameter` objects embedded in the
551 object.
553 This accomplishes the same purpose as ``visitors.traverse()`` or
554 similar would provide, however by making use of the cache key
555 it takes advantage of memoization of the key to result in fewer
556 net method calls, assuming the statement is also going to be
557 executed.
559 """
561 key = self._generate_cache_key()
562 if key is None:
563 bindparams: List[BindParameter[Any]] = []
565 traverse(self, {}, {"bindparam": bindparams.append})
566 return bindparams
568 else:
569 return key.bindparams
571 def unique_params(
572 self,
573 __optionaldict: Optional[Dict[str, Any]] = None,
574 **kwargs: Any,
575 ) -> Self:
576 """Return a copy with :func:`_expression.bindparam` elements
577 replaced.
579 Same functionality as :meth:`_expression.ClauseElement.params`,
580 except adds `unique=True`
581 to affected bind parameters so that multiple statements can be
582 used.
584 """
585 return self._replace_params(True, __optionaldict, kwargs)
587 def params(
588 self,
589 __optionaldict: Optional[Mapping[str, Any]] = None,
590 **kwargs: Any,
591 ) -> Self:
592 """Return a copy with :func:`_expression.bindparam` elements
593 replaced.
595 Returns a copy of this ClauseElement with
596 :func:`_expression.bindparam`
597 elements replaced with values taken from the given dictionary::
599 >>> clause = column("x") + bindparam("foo")
600 >>> print(clause.compile().params)
601 {'foo':None}
602 >>> print(clause.params({"foo": 7}).compile().params)
603 {'foo':7}
605 """
606 return self._replace_params(False, __optionaldict, kwargs)
608 def _replace_params(
609 self,
610 unique: bool,
611 optionaldict: Optional[Mapping[str, Any]],
612 kwargs: Dict[str, Any],
613 ) -> Self:
614 if optionaldict:
615 kwargs.update(optionaldict)
617 def visit_bindparam(bind: BindParameter[Any]) -> None:
618 if bind.key in kwargs:
619 bind.value = kwargs[bind.key]
620 bind.required = False
621 if unique:
622 bind._convert_to_unique()
624 return cloned_traverse(
625 self,
626 {"maintain_key": True, "detect_subquery_cols": True},
627 {"bindparam": visit_bindparam},
628 )
630 def compare(self, other: ClauseElement, **kw: Any) -> bool:
631 r"""Compare this :class:`_expression.ClauseElement` to
632 the given :class:`_expression.ClauseElement`.
634 Subclasses should override the default behavior, which is a
635 straight identity comparison.
637 \**kw are arguments consumed by subclass ``compare()`` methods and
638 may be used to modify the criteria for comparison
639 (see :class:`_expression.ColumnElement`).
641 """
642 return traversals.compare(self, other, **kw)
644 def self_group(
645 self, against: Optional[OperatorType] = None
646 ) -> ClauseElement:
647 """Apply a 'grouping' to this :class:`_expression.ClauseElement`.
649 This method is overridden by subclasses to return a "grouping"
650 construct, i.e. parenthesis. In particular it's used by "binary"
651 expressions to provide a grouping around themselves when placed into a
652 larger expression, as well as by :func:`_expression.select`
653 constructs when placed into the FROM clause of another
654 :func:`_expression.select`. (Note that subqueries should be
655 normally created using the :meth:`_expression.Select.alias` method,
656 as many
657 platforms require nested SELECT statements to be named).
659 As expressions are composed together, the application of
660 :meth:`self_group` is automatic - end-user code should never
661 need to use this method directly. Note that SQLAlchemy's
662 clause constructs take operator precedence into account -
663 so parenthesis might not be needed, for example, in
664 an expression like ``x OR (y AND z)`` - AND takes precedence
665 over OR.
667 The base :meth:`self_group` method of
668 :class:`_expression.ClauseElement`
669 just returns self.
670 """
671 return self
673 def _ungroup(self) -> ClauseElement:
674 """Return this :class:`_expression.ClauseElement`
675 without any groupings.
676 """
678 return self
680 def _compile_w_cache(
681 self,
682 dialect: Dialect,
683 *,
684 compiled_cache: Optional[CompiledCacheType],
685 column_keys: List[str],
686 for_executemany: bool = False,
687 schema_translate_map: Optional[SchemaTranslateMapType] = None,
688 **kw: Any,
689 ) -> typing_Tuple[
690 Compiled, Optional[Sequence[BindParameter[Any]]], CacheStats
691 ]:
692 elem_cache_key: Optional[CacheKey]
694 if compiled_cache is not None and dialect._supports_statement_cache:
695 elem_cache_key = self._generate_cache_key()
696 else:
697 elem_cache_key = None
699 extracted_params: Optional[Sequence[BindParameter[Any]]]
700 if elem_cache_key is not None:
701 if TYPE_CHECKING:
702 assert compiled_cache is not None
704 cache_key, extracted_params = elem_cache_key
705 key = (
706 dialect,
707 cache_key,
708 tuple(column_keys),
709 bool(schema_translate_map),
710 for_executemany,
711 )
712 compiled_sql = compiled_cache.get(key)
714 if compiled_sql is None:
715 cache_hit = dialect.CACHE_MISS
716 compiled_sql = self._compiler(
717 dialect,
718 cache_key=elem_cache_key,
719 column_keys=column_keys,
720 for_executemany=for_executemany,
721 schema_translate_map=schema_translate_map,
722 **kw,
723 )
724 compiled_cache[key] = compiled_sql
725 else:
726 cache_hit = dialect.CACHE_HIT
727 else:
728 extracted_params = None
729 compiled_sql = self._compiler(
730 dialect,
731 cache_key=elem_cache_key,
732 column_keys=column_keys,
733 for_executemany=for_executemany,
734 schema_translate_map=schema_translate_map,
735 **kw,
736 )
738 if not dialect._supports_statement_cache:
739 cache_hit = dialect.NO_DIALECT_SUPPORT
740 elif compiled_cache is None:
741 cache_hit = dialect.CACHING_DISABLED
742 else:
743 cache_hit = dialect.NO_CACHE_KEY
745 return compiled_sql, extracted_params, cache_hit
747 def __invert__(self):
748 # undocumented element currently used by the ORM for
749 # relationship.contains()
750 if hasattr(self, "negation_clause"):
751 return self.negation_clause
752 else:
753 return self._negate()
755 def _negate(self) -> ClauseElement:
756 # TODO: this code is uncovered and in all likelihood is not included
757 # in any codepath. So this should raise NotImplementedError in 2.1
758 grouped = self.self_group(against=operators.inv)
759 assert isinstance(grouped, ColumnElement)
760 return UnaryExpression(grouped, operator=operators.inv)
762 def __bool__(self):
763 raise TypeError("Boolean value of this clause is not defined")
765 def __repr__(self):
766 friendly = self.description
767 if friendly is None:
768 return object.__repr__(self)
769 else:
770 return "<%s.%s at 0x%x; %s>" % (
771 self.__module__,
772 self.__class__.__name__,
773 id(self),
774 friendly,
775 )
778class DQLDMLClauseElement(ClauseElement):
779 """represents a :class:`.ClauseElement` that compiles to a DQL or DML
780 expression, not DDL.
782 .. versionadded:: 2.0
784 """
786 if typing.TYPE_CHECKING:
788 def _compiler(self, dialect: Dialect, **kw: Any) -> SQLCompiler:
789 """Return a compiler appropriate for this ClauseElement, given a
790 Dialect."""
791 ...
793 def compile( # noqa: A001
794 self,
795 bind: Optional[_HasDialect] = None,
796 dialect: Optional[Dialect] = None,
797 **kw: Any,
798 ) -> SQLCompiler: ...
801class CompilerColumnElement(
802 roles.DMLColumnRole,
803 roles.DDLConstraintColumnRole,
804 roles.ColumnsClauseRole,
805 CompilerElement,
806):
807 """A compiler-only column element used for ad-hoc string compilations.
809 .. versionadded:: 2.0
811 """
813 __slots__ = ()
815 _propagate_attrs = util.EMPTY_DICT
816 _is_collection_aggregate = False
819# SQLCoreOperations should be suiting the ExpressionElementRole
820# and ColumnsClauseRole. however the MRO issues become too elaborate
821# at the moment.
822class SQLCoreOperations(Generic[_T_co], ColumnOperators, TypingOnly):
823 __slots__ = ()
825 # annotations for comparison methods
826 # these are from operators->Operators / ColumnOperators,
827 # redefined with the specific types returned by ColumnElement hierarchies
828 if typing.TYPE_CHECKING:
830 @util.non_memoized_property
831 def _propagate_attrs(self) -> _PropagateAttrsType: ...
833 def operate(
834 self, op: OperatorType, *other: Any, **kwargs: Any
835 ) -> ColumnElement[Any]: ...
837 def reverse_operate(
838 self, op: OperatorType, other: Any, **kwargs: Any
839 ) -> ColumnElement[Any]: ...
841 @overload
842 def op(
843 self,
844 opstring: str,
845 precedence: int = ...,
846 is_comparison: bool = ...,
847 *,
848 return_type: _TypeEngineArgument[_OPT],
849 python_impl: Optional[Callable[..., Any]] = None,
850 ) -> Callable[[Any], BinaryExpression[_OPT]]: ...
852 @overload
853 def op(
854 self,
855 opstring: str,
856 precedence: int = ...,
857 is_comparison: bool = ...,
858 return_type: Optional[_TypeEngineArgument[Any]] = ...,
859 python_impl: Optional[Callable[..., Any]] = ...,
860 ) -> Callable[[Any], BinaryExpression[Any]]: ...
862 def op(
863 self,
864 opstring: str,
865 precedence: int = 0,
866 is_comparison: bool = False,
867 return_type: Optional[_TypeEngineArgument[Any]] = None,
868 python_impl: Optional[Callable[..., Any]] = None,
869 ) -> Callable[[Any], BinaryExpression[Any]]: ...
871 def bool_op(
872 self,
873 opstring: str,
874 precedence: int = 0,
875 python_impl: Optional[Callable[..., Any]] = None,
876 ) -> Callable[[Any], BinaryExpression[bool]]: ...
878 def __and__(self, other: Any) -> BooleanClauseList: ...
880 def __or__(self, other: Any) -> BooleanClauseList: ...
882 def __invert__(self) -> ColumnElement[_T_co]: ...
884 def __lt__(self, other: Any) -> ColumnElement[bool]: ...
886 def __le__(self, other: Any) -> ColumnElement[bool]: ...
888 # declare also that this class has an hash method otherwise
889 # it may be assumed to be None by type checkers since the
890 # object defines __eq__ and python sets it to None in that case:
891 # https://docs.python.org/3/reference/datamodel.html#object.__hash__
892 def __hash__(self) -> int: ...
894 def __eq__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
895 ...
897 def __ne__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
898 ...
900 def is_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
902 def is_not_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
904 def __gt__(self, other: Any) -> ColumnElement[bool]: ...
906 def __ge__(self, other: Any) -> ColumnElement[bool]: ...
908 def __neg__(self) -> UnaryExpression[_T_co]: ...
910 def __contains__(self, other: Any) -> ColumnElement[bool]: ...
912 def __getitem__(self, index: Any) -> ColumnElement[Any]: ...
914 @overload
915 def __lshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
917 @overload
918 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
920 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
922 @overload
923 def __rshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
925 @overload
926 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
928 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
930 @overload
931 def concat(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
933 @overload
934 def concat(self, other: Any) -> ColumnElement[Any]: ...
936 def concat(self, other: Any) -> ColumnElement[Any]: ...
938 def like(
939 self, other: Any, escape: Optional[str] = None
940 ) -> BinaryExpression[bool]: ...
942 def ilike(
943 self, other: Any, escape: Optional[str] = None
944 ) -> BinaryExpression[bool]: ...
946 def bitwise_xor(self, other: Any) -> BinaryExpression[Any]: ...
948 def bitwise_or(self, other: Any) -> BinaryExpression[Any]: ...
950 def bitwise_and(self, other: Any) -> BinaryExpression[Any]: ...
952 def bitwise_not(self) -> UnaryExpression[_T_co]: ...
954 def bitwise_lshift(self, other: Any) -> BinaryExpression[Any]: ...
956 def bitwise_rshift(self, other: Any) -> BinaryExpression[Any]: ...
958 def in_(
959 self,
960 other: Union[
961 Iterable[Any], BindParameter[Any], roles.InElementRole
962 ],
963 ) -> BinaryExpression[bool]: ...
965 def not_in(
966 self,
967 other: Union[
968 Iterable[Any], BindParameter[Any], roles.InElementRole
969 ],
970 ) -> BinaryExpression[bool]: ...
972 def notin_(
973 self,
974 other: Union[
975 Iterable[Any], BindParameter[Any], roles.InElementRole
976 ],
977 ) -> BinaryExpression[bool]: ...
979 def not_like(
980 self, other: Any, escape: Optional[str] = None
981 ) -> BinaryExpression[bool]: ...
983 def notlike(
984 self, other: Any, escape: Optional[str] = None
985 ) -> BinaryExpression[bool]: ...
987 def not_ilike(
988 self, other: Any, escape: Optional[str] = None
989 ) -> BinaryExpression[bool]: ...
991 def notilike(
992 self, other: Any, escape: Optional[str] = None
993 ) -> BinaryExpression[bool]: ...
995 def is_(self, other: Any) -> BinaryExpression[bool]: ...
997 def is_not(self, other: Any) -> BinaryExpression[bool]: ...
999 def isnot(self, other: Any) -> BinaryExpression[bool]: ...
1001 def startswith(
1002 self,
1003 other: Any,
1004 escape: Optional[str] = None,
1005 autoescape: bool = False,
1006 ) -> ColumnElement[bool]: ...
1008 def istartswith(
1009 self,
1010 other: Any,
1011 escape: Optional[str] = None,
1012 autoescape: bool = False,
1013 ) -> ColumnElement[bool]: ...
1015 def endswith(
1016 self,
1017 other: Any,
1018 escape: Optional[str] = None,
1019 autoescape: bool = False,
1020 ) -> ColumnElement[bool]: ...
1022 def iendswith(
1023 self,
1024 other: Any,
1025 escape: Optional[str] = None,
1026 autoescape: bool = False,
1027 ) -> ColumnElement[bool]: ...
1029 def contains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1031 def icontains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1033 def match(self, other: Any, **kwargs: Any) -> ColumnElement[bool]: ...
1035 def regexp_match(
1036 self, pattern: Any, flags: Optional[str] = None
1037 ) -> ColumnElement[bool]: ...
1039 def regexp_replace(
1040 self, pattern: Any, replacement: Any, flags: Optional[str] = None
1041 ) -> ColumnElement[str]: ...
1043 def desc(self) -> UnaryExpression[_T_co]: ...
1045 def asc(self) -> UnaryExpression[_T_co]: ...
1047 def nulls_first(self) -> UnaryExpression[_T_co]: ...
1049 def nullsfirst(self) -> UnaryExpression[_T_co]: ...
1051 def nulls_last(self) -> UnaryExpression[_T_co]: ...
1053 def nullslast(self) -> UnaryExpression[_T_co]: ...
1055 def collate(self, collation: str) -> CollationClause: ...
1057 def between(
1058 self, cleft: Any, cright: Any, symmetric: bool = False
1059 ) -> BinaryExpression[bool]: ...
1061 def distinct(self: _SQO[_T_co]) -> UnaryExpression[_T_co]: ...
1063 def any_(self) -> CollectionAggregate[Any]: ...
1065 def all_(self) -> CollectionAggregate[Any]: ...
1067 # numeric overloads. These need more tweaking
1068 # in particular they all need to have a variant for Optiona[_T]
1069 # because Optional only applies to the data side, not the expression
1070 # side
1072 @overload
1073 def __add__(
1074 self: _SQO[_NMT],
1075 other: Any,
1076 ) -> ColumnElement[_NMT]: ...
1078 @overload
1079 def __add__(
1080 self: _SQO[str],
1081 other: Any,
1082 ) -> ColumnElement[str]: ...
1084 @overload
1085 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1087 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1089 @overload
1090 def __radd__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1092 @overload
1093 def __radd__(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
1095 def __radd__(self, other: Any) -> ColumnElement[Any]: ...
1097 @overload
1098 def __sub__(
1099 self: _SQO[_NMT],
1100 other: Any,
1101 ) -> ColumnElement[_NMT]: ...
1103 @overload
1104 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1106 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1108 @overload
1109 def __rsub__(
1110 self: _SQO[_NMT],
1111 other: Any,
1112 ) -> ColumnElement[_NMT]: ...
1114 @overload
1115 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1117 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1119 @overload
1120 def __mul__(
1121 self: _SQO[_NMT],
1122 other: Any,
1123 ) -> ColumnElement[_NMT]: ...
1125 @overload
1126 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1128 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1130 @overload
1131 def __rmul__(
1132 self: _SQO[_NMT],
1133 other: Any,
1134 ) -> ColumnElement[_NMT]: ...
1136 @overload
1137 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1139 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1141 @overload
1142 def __mod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1144 @overload
1145 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1147 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1149 @overload
1150 def __rmod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1152 @overload
1153 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1155 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1157 @overload
1158 def __truediv__(
1159 self: _SQO[int], other: Any
1160 ) -> ColumnElement[_NUMERIC]: ...
1162 @overload
1163 def __truediv__(self: _SQO[_NT], other: Any) -> ColumnElement[_NT]: ...
1165 @overload
1166 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1168 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1170 @overload
1171 def __rtruediv__(
1172 self: _SQO[_NMT], other: Any
1173 ) -> ColumnElement[_NUMERIC]: ...
1175 @overload
1176 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1178 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1180 @overload
1181 def __floordiv__(
1182 self: _SQO[_NMT], other: Any
1183 ) -> ColumnElement[_NMT]: ...
1185 @overload
1186 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1188 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1190 @overload
1191 def __rfloordiv__(
1192 self: _SQO[_NMT], other: Any
1193 ) -> ColumnElement[_NMT]: ...
1195 @overload
1196 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1198 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1201class SQLColumnExpression(
1202 SQLCoreOperations[_T_co], roles.ExpressionElementRole[_T_co], TypingOnly
1203):
1204 """A type that may be used to indicate any SQL column element or object
1205 that acts in place of one.
1207 :class:`.SQLColumnExpression` is a base of
1208 :class:`.ColumnElement`, as well as within the bases of ORM elements
1209 such as :class:`.InstrumentedAttribute`, and may be used in :pep:`484`
1210 typing to indicate arguments or return values that should behave
1211 as column expressions.
1213 .. versionadded:: 2.0.0b4
1216 """
1218 __slots__ = ()
1221_SQO = SQLCoreOperations
1224class ColumnElement(
1225 roles.ColumnArgumentOrKeyRole,
1226 roles.StatementOptionRole,
1227 roles.WhereHavingRole,
1228 roles.BinaryElementRole[_T],
1229 roles.OrderByRole,
1230 roles.ColumnsClauseRole,
1231 roles.LimitOffsetRole,
1232 roles.DMLColumnRole,
1233 roles.DDLConstraintColumnRole,
1234 roles.DDLExpressionRole,
1235 SQLColumnExpression[_T],
1236 DQLDMLClauseElement,
1237):
1238 """Represent a column-oriented SQL expression suitable for usage in the
1239 "columns" clause, WHERE clause etc. of a statement.
1241 While the most familiar kind of :class:`_expression.ColumnElement` is the
1242 :class:`_schema.Column` object, :class:`_expression.ColumnElement`
1243 serves as the basis
1244 for any unit that may be present in a SQL expression, including
1245 the expressions themselves, SQL functions, bound parameters,
1246 literal expressions, keywords such as ``NULL``, etc.
1247 :class:`_expression.ColumnElement`
1248 is the ultimate base class for all such elements.
1250 A wide variety of SQLAlchemy Core functions work at the SQL expression
1251 level, and are intended to accept instances of
1252 :class:`_expression.ColumnElement` as
1253 arguments. These functions will typically document that they accept a
1254 "SQL expression" as an argument. What this means in terms of SQLAlchemy
1255 usually refers to an input which is either already in the form of a
1256 :class:`_expression.ColumnElement` object,
1257 or a value which can be **coerced** into
1258 one. The coercion rules followed by most, but not all, SQLAlchemy Core
1259 functions with regards to SQL expressions are as follows:
1261 * a literal Python value, such as a string, integer or floating
1262 point value, boolean, datetime, ``Decimal`` object, or virtually
1263 any other Python object, will be coerced into a "literal bound
1264 value". This generally means that a :func:`.bindparam` will be
1265 produced featuring the given value embedded into the construct; the
1266 resulting :class:`.BindParameter` object is an instance of
1267 :class:`_expression.ColumnElement`.
1268 The Python value will ultimately be sent
1269 to the DBAPI at execution time as a parameterized argument to the
1270 ``execute()`` or ``executemany()`` methods, after SQLAlchemy
1271 type-specific converters (e.g. those provided by any associated
1272 :class:`.TypeEngine` objects) are applied to the value.
1274 * any special object value, typically ORM-level constructs, which
1275 feature an accessor called ``__clause_element__()``. The Core
1276 expression system looks for this method when an object of otherwise
1277 unknown type is passed to a function that is looking to coerce the
1278 argument into a :class:`_expression.ColumnElement` and sometimes a
1279 :class:`_expression.SelectBase` expression.
1280 It is used within the ORM to
1281 convert from ORM-specific objects like mapped classes and
1282 mapped attributes into Core expression objects.
1284 * The Python ``None`` value is typically interpreted as ``NULL``,
1285 which in SQLAlchemy Core produces an instance of :func:`.null`.
1287 A :class:`_expression.ColumnElement` provides the ability to generate new
1288 :class:`_expression.ColumnElement`
1289 objects using Python expressions. This means that Python operators
1290 such as ``==``, ``!=`` and ``<`` are overloaded to mimic SQL operations,
1291 and allow the instantiation of further :class:`_expression.ColumnElement`
1292 instances
1293 which are composed from other, more fundamental
1294 :class:`_expression.ColumnElement`
1295 objects. For example, two :class:`.ColumnClause` objects can be added
1296 together with the addition operator ``+`` to produce
1297 a :class:`.BinaryExpression`.
1298 Both :class:`.ColumnClause` and :class:`.BinaryExpression` are subclasses
1299 of :class:`_expression.ColumnElement`:
1301 .. sourcecode:: pycon+sql
1303 >>> from sqlalchemy.sql import column
1304 >>> column("a") + column("b")
1305 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
1306 >>> print(column("a") + column("b"))
1307 {printsql}a + b
1309 .. seealso::
1311 :class:`_schema.Column`
1313 :func:`_expression.column`
1315 """
1317 __visit_name__ = "column_element"
1319 primary_key: bool = False
1320 _is_clone_of: Optional[ColumnElement[_T]]
1321 _is_column_element = True
1322 _insert_sentinel: bool = False
1323 _omit_from_statements = False
1324 _is_collection_aggregate = False
1326 foreign_keys: AbstractSet[ForeignKey] = frozenset()
1328 @util.memoized_property
1329 def _proxies(self) -> List[ColumnElement[Any]]:
1330 return []
1332 @util.non_memoized_property
1333 def _tq_label(self) -> Optional[str]:
1334 """The named label that can be used to target
1335 this column in a result set in a "table qualified" context.
1337 This label is almost always the label used when
1338 rendering <expr> AS <label> in a SELECT statement when using
1339 the LABEL_STYLE_TABLENAME_PLUS_COL label style, which is what the
1340 legacy ORM ``Query`` object uses as well.
1342 For a regular Column bound to a Table, this is typically the label
1343 <tablename>_<columnname>. For other constructs, different rules
1344 may apply, such as anonymized labels and others.
1346 .. versionchanged:: 1.4.21 renamed from ``._label``
1348 """
1349 return None
1351 key: Optional[str] = None
1352 """The 'key' that in some circumstances refers to this object in a
1353 Python namespace.
1355 This typically refers to the "key" of the column as present in the
1356 ``.c`` collection of a selectable, e.g. ``sometable.c["somekey"]`` would
1357 return a :class:`_schema.Column` with a ``.key`` of "somekey".
1359 """
1361 @HasMemoized.memoized_attribute
1362 def _tq_key_label(self) -> Optional[str]:
1363 """A label-based version of 'key' that in some circumstances refers
1364 to this object in a Python namespace.
1367 _tq_key_label comes into play when a select() statement is constructed
1368 with apply_labels(); in this case, all Column objects in the ``.c``
1369 collection are rendered as <tablename>_<columnname> in SQL; this is
1370 essentially the value of ._label. But to locate those columns in the
1371 ``.c`` collection, the name is along the lines of <tablename>_<key>;
1372 that's the typical value of .key_label.
1374 .. versionchanged:: 1.4.21 renamed from ``._key_label``
1376 """
1377 return self._proxy_key
1379 @property
1380 def _key_label(self) -> Optional[str]:
1381 """legacy; renamed to _tq_key_label"""
1382 return self._tq_key_label
1384 @property
1385 def _label(self) -> Optional[str]:
1386 """legacy; renamed to _tq_label"""
1387 return self._tq_label
1389 @property
1390 def _non_anon_label(self) -> Optional[str]:
1391 """the 'name' that naturally applies this element when rendered in
1392 SQL.
1394 Concretely, this is the "name" of a column or a label in a
1395 SELECT statement; ``<columnname>`` and ``<labelname>`` below:
1397 .. sourcecode:: sql
1399 SELECT <columnmame> FROM table
1401 SELECT column AS <labelname> FROM table
1403 Above, the two names noted will be what's present in the DBAPI
1404 ``cursor.description`` as the names.
1406 If this attribute returns ``None``, it means that the SQL element as
1407 written does not have a 100% fully predictable "name" that would appear
1408 in the ``cursor.description``. Examples include SQL functions, CAST
1409 functions, etc. While such things do return names in
1410 ``cursor.description``, they are only predictable on a
1411 database-specific basis; e.g. an expression like ``MAX(table.col)`` may
1412 appear as the string ``max`` on one database (like PostgreSQL) or may
1413 appear as the whole expression ``max(table.col)`` on SQLite.
1415 The default implementation looks for a ``.name`` attribute on the
1416 object, as has been the precedent established in SQLAlchemy for many
1417 years. An exception is made on the ``FunctionElement`` subclass
1418 so that the return value is always ``None``.
1420 .. versionadded:: 1.4.21
1424 """
1425 return getattr(self, "name", None)
1427 _render_label_in_columns_clause = True
1428 """A flag used by select._columns_plus_names that helps to determine
1429 we are actually going to render in terms of "SELECT <col> AS <label>".
1430 This flag can be returned as False for some Column objects that want
1431 to be rendered as simple "SELECT <col>"; typically columns that don't have
1432 any parent table and are named the same as what the label would be
1433 in any case.
1435 """
1437 _allow_label_resolve = True
1438 """A flag that can be flipped to prevent a column from being resolvable
1439 by string label name.
1441 The joined eager loader strategy in the ORM uses this, for example.
1443 """
1445 _is_implicitly_boolean = False
1447 _alt_names: Sequence[str] = ()
1449 if TYPE_CHECKING:
1451 def _ungroup(self) -> ColumnElement[_T]: ...
1453 @overload
1454 def self_group(self, against: None = None) -> ColumnElement[_T]: ...
1456 @overload
1457 def self_group(
1458 self, against: Optional[OperatorType] = None
1459 ) -> ColumnElement[Any]: ...
1461 def self_group(
1462 self, against: Optional[OperatorType] = None
1463 ) -> ColumnElement[Any]:
1464 if (
1465 against in (operators.and_, operators.or_, operators._asbool)
1466 and self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity
1467 ):
1468 return AsBoolean(self, operators.is_true, operators.is_false)
1469 elif against in (operators.any_op, operators.all_op):
1470 return Grouping(self)
1471 else:
1472 return self
1474 @overload
1475 def _negate(self: ColumnElement[bool]) -> ColumnElement[bool]: ...
1477 @overload
1478 def _negate(self: ColumnElement[_T]) -> ColumnElement[_T]: ...
1480 def _negate(self) -> ColumnElement[Any]:
1481 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
1482 return AsBoolean(self, operators.is_false, operators.is_true)
1483 else:
1484 grouped = self.self_group(against=operators.inv)
1485 assert isinstance(grouped, ColumnElement)
1486 return UnaryExpression(
1487 grouped,
1488 operator=operators.inv,
1489 )
1491 type: TypeEngine[_T]
1493 if not TYPE_CHECKING:
1495 @util.memoized_property
1496 def type(self) -> TypeEngine[_T]: # noqa: A001
1497 # used for delayed setup of
1498 # type_api
1499 return type_api.NULLTYPE
1501 @HasMemoized.memoized_attribute
1502 def comparator(self) -> TypeEngine.Comparator[_T]:
1503 try:
1504 comparator_factory = self.type.comparator_factory
1505 except AttributeError as err:
1506 raise TypeError(
1507 "Object %r associated with '.type' attribute "
1508 "is not a TypeEngine class or object" % self.type
1509 ) from err
1510 else:
1511 return comparator_factory(self)
1513 def __setstate__(self, state):
1514 self.__dict__.update(state)
1516 def __getattr__(self, key: str) -> Any:
1517 try:
1518 return getattr(self.comparator, key)
1519 except AttributeError as err:
1520 raise AttributeError(
1521 "Neither %r object nor %r object has an attribute %r"
1522 % (
1523 type(self).__name__,
1524 type(self.comparator).__name__,
1525 key,
1526 )
1527 ) from err
1529 def operate(
1530 self,
1531 op: operators.OperatorType,
1532 *other: Any,
1533 **kwargs: Any,
1534 ) -> ColumnElement[Any]:
1535 return op(self.comparator, *other, **kwargs) # type: ignore[no-any-return] # noqa: E501
1537 def reverse_operate(
1538 self, op: operators.OperatorType, other: Any, **kwargs: Any
1539 ) -> ColumnElement[Any]:
1540 return op(other, self.comparator, **kwargs) # type: ignore[no-any-return] # noqa: E501
1542 def _bind_param(
1543 self,
1544 operator: operators.OperatorType,
1545 obj: Any,
1546 type_: Optional[TypeEngine[_T]] = None,
1547 expanding: bool = False,
1548 ) -> BindParameter[_T]:
1549 return BindParameter(
1550 None,
1551 obj,
1552 _compared_to_operator=operator,
1553 type_=type_,
1554 _compared_to_type=self.type,
1555 unique=True,
1556 expanding=expanding,
1557 )
1559 @property
1560 def expression(self) -> ColumnElement[Any]:
1561 """Return a column expression.
1563 Part of the inspection interface; returns self.
1565 """
1566 return self
1568 @property
1569 def _select_iterable(self) -> _SelectIterable:
1570 return (self,)
1572 @util.memoized_property
1573 def base_columns(self) -> FrozenSet[ColumnElement[Any]]:
1574 return frozenset(c for c in self.proxy_set if not c._proxies)
1576 @util.memoized_property
1577 def proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1578 """set of all columns we are proxying
1580 as of 2.0 this is explicitly deannotated columns. previously it was
1581 effectively deannotated columns but wasn't enforced. annotated
1582 columns should basically not go into sets if at all possible because
1583 their hashing behavior is very non-performant.
1585 """
1586 return frozenset([self._deannotate()]).union(
1587 itertools.chain(*[c.proxy_set for c in self._proxies])
1588 )
1590 @util.memoized_property
1591 def _expanded_proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1592 return frozenset(_expand_cloned(self.proxy_set))
1594 def _uncached_proxy_list(self) -> List[ColumnElement[Any]]:
1595 """An 'uncached' version of proxy set.
1597 This list includes annotated columns which perform very poorly in
1598 set operations.
1600 """
1602 return [self] + list(
1603 itertools.chain(*[c._uncached_proxy_list() for c in self._proxies])
1604 )
1606 def shares_lineage(self, othercolumn: ColumnElement[Any]) -> bool:
1607 """Return True if the given :class:`_expression.ColumnElement`
1608 has a common ancestor to this :class:`_expression.ColumnElement`."""
1610 return bool(self.proxy_set.intersection(othercolumn.proxy_set))
1612 def _compare_name_for_result(self, other: ColumnElement[Any]) -> bool:
1613 """Return True if the given column element compares to this one
1614 when targeting within a result row."""
1616 return (
1617 hasattr(other, "name")
1618 and hasattr(self, "name")
1619 and other.name == self.name
1620 )
1622 @HasMemoized.memoized_attribute
1623 def _proxy_key(self) -> Optional[str]:
1624 if self._annotations and "proxy_key" in self._annotations:
1625 return cast(str, self._annotations["proxy_key"])
1627 name = self.key
1628 if not name:
1629 # there's a bit of a seeming contradiction which is that the
1630 # "_non_anon_label" of a column can in fact be an
1631 # "_anonymous_label"; this is when it's on a column that is
1632 # proxying for an anonymous expression in a subquery.
1633 name = self._non_anon_label
1635 if isinstance(name, _anonymous_label):
1636 return None
1637 else:
1638 return name
1640 @HasMemoized.memoized_attribute
1641 def _expression_label(self) -> Optional[str]:
1642 """a suggested label to use in the case that the column has no name,
1643 which should be used if possible as the explicit 'AS <label>'
1644 where this expression would normally have an anon label.
1646 this is essentially mostly what _proxy_key does except it returns
1647 None if the column has a normal name that can be used.
1649 """
1651 if getattr(self, "name", None) is not None:
1652 return None
1653 elif self._annotations and "proxy_key" in self._annotations:
1654 return cast(str, self._annotations["proxy_key"])
1655 else:
1656 return None
1658 def _make_proxy(
1659 self,
1660 selectable: FromClause,
1661 *,
1662 primary_key: ColumnSet,
1663 foreign_keys: Set[KeyedColumnElement[Any]],
1664 name: Optional[str] = None,
1665 key: Optional[str] = None,
1666 name_is_truncatable: bool = False,
1667 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
1668 **kw: Any,
1669 ) -> typing_Tuple[str, ColumnClause[_T]]:
1670 """Create a new :class:`_expression.ColumnElement` representing this
1671 :class:`_expression.ColumnElement` as it appears in the select list of
1672 a descending selectable.
1674 """
1675 if name is None:
1676 name = self._anon_name_label
1677 if key is None:
1678 key = self._proxy_key
1679 else:
1680 key = name
1682 assert key is not None
1684 co: ColumnClause[_T] = ColumnClause(
1685 (
1686 coercions.expect(roles.TruncatedLabelRole, name)
1687 if name_is_truncatable
1688 else name
1689 ),
1690 type_=getattr(self, "type", None),
1691 _selectable=selectable,
1692 )
1694 co._propagate_attrs = selectable._propagate_attrs
1695 if compound_select_cols:
1696 co._proxies = list(compound_select_cols)
1697 else:
1698 co._proxies = [self]
1699 if selectable._is_clone_of is not None:
1700 co._is_clone_of = selectable._is_clone_of.columns.get(key)
1701 return key, co
1703 def cast(self, type_: _TypeEngineArgument[_OPT]) -> Cast[_OPT]:
1704 """Produce a type cast, i.e. ``CAST(<expression> AS <type>)``.
1706 This is a shortcut to the :func:`_expression.cast` function.
1708 .. seealso::
1710 :ref:`tutorial_casts`
1712 :func:`_expression.cast`
1714 :func:`_expression.type_coerce`
1716 """
1717 return Cast(self, type_)
1719 def label(self, name: Optional[str]) -> Label[_T]:
1720 """Produce a column label, i.e. ``<columnname> AS <name>``.
1722 This is a shortcut to the :func:`_expression.label` function.
1724 If 'name' is ``None``, an anonymous label name will be generated.
1726 """
1727 return Label(name, self, self.type)
1729 def _anon_label(
1730 self, seed: Optional[str], add_hash: Optional[int] = None
1731 ) -> _anonymous_label:
1732 while self._is_clone_of is not None:
1733 self = self._is_clone_of
1735 # as of 1.4 anonymous label for ColumnElement uses hash(), not id(),
1736 # as the identifier, because a column and its annotated version are
1737 # the same thing in a SQL statement
1738 hash_value = hash(self)
1740 if add_hash:
1741 # this path is used for disambiguating anon labels that would
1742 # otherwise be the same name for the same element repeated.
1743 # an additional numeric value is factored in for each label.
1745 # shift hash(self) (which is id(self), typically 8 byte integer)
1746 # 16 bits leftward. fill extra add_hash on right
1747 assert add_hash < (2 << 15)
1748 assert seed
1749 hash_value = (hash_value << 16) | add_hash
1751 # extra underscore is added for labels with extra hash
1752 # values, to isolate the "deduped anon" namespace from the
1753 # regular namespace. eliminates chance of these
1754 # manufactured hash values overlapping with regular ones for some
1755 # undefined python interpreter
1756 seed = seed + "_"
1758 if isinstance(seed, _anonymous_label):
1759 return _anonymous_label.safe_construct(
1760 hash_value, "", enclosing_label=seed
1761 )
1763 return _anonymous_label.safe_construct(hash_value, seed or "anon")
1765 @util.memoized_property
1766 def _anon_name_label(self) -> str:
1767 """Provides a constant 'anonymous label' for this ColumnElement.
1769 This is a label() expression which will be named at compile time.
1770 The same label() is returned each time ``anon_label`` is called so
1771 that expressions can reference ``anon_label`` multiple times,
1772 producing the same label name at compile time.
1774 The compiler uses this function automatically at compile time
1775 for expressions that are known to be 'unnamed' like binary
1776 expressions and function calls.
1778 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1779 public and is renamed to _anon_name_label. anon_name exists
1780 for backwards compat
1782 """
1783 name = getattr(self, "name", None)
1784 return self._anon_label(name)
1786 @util.memoized_property
1787 def _anon_key_label(self) -> _anonymous_label:
1788 """Provides a constant 'anonymous key label' for this ColumnElement.
1790 Compare to ``anon_label``, except that the "key" of the column,
1791 if available, is used to generate the label.
1793 This is used when a deduplicating key is placed into the columns
1794 collection of a selectable.
1796 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1797 public and is renamed to _anon_key_label. anon_key_label exists
1798 for backwards compat
1800 """
1801 return self._anon_label(self._proxy_key)
1803 @property
1804 @util.deprecated(
1805 "1.4",
1806 "The :attr:`_expression.ColumnElement.anon_label` attribute is now "
1807 "private, and the public accessor is deprecated.",
1808 )
1809 def anon_label(self) -> str:
1810 return self._anon_name_label
1812 @property
1813 @util.deprecated(
1814 "1.4",
1815 "The :attr:`_expression.ColumnElement.anon_key_label` attribute is "
1816 "now private, and the public accessor is deprecated.",
1817 )
1818 def anon_key_label(self) -> str:
1819 return self._anon_key_label
1821 def _dedupe_anon_label_idx(self, idx: int) -> str:
1822 """label to apply to a column that is anon labeled, but repeated
1823 in the SELECT, so that we have to make an "extra anon" label that
1824 disambiguates it from the previous appearance.
1826 these labels come out like "foo_bar_id__1" and have double underscores
1827 in them.
1829 """
1830 label = getattr(self, "name", None)
1832 # current convention is that if the element doesn't have a
1833 # ".name" (usually because it is not NamedColumn), we try to
1834 # use a "table qualified" form for the "dedupe anon" label,
1835 # based on the notion that a label like
1836 # "CAST(casttest.v1 AS DECIMAL) AS casttest_v1__1" looks better than
1837 # "CAST(casttest.v1 AS DECIMAL) AS anon__1"
1839 if label is None:
1840 return self._dedupe_anon_tq_label_idx(idx)
1841 else:
1842 return self._anon_label(label, add_hash=idx)
1844 @util.memoized_property
1845 def _anon_tq_label(self) -> _anonymous_label:
1846 return self._anon_label(getattr(self, "_tq_label", None))
1848 @util.memoized_property
1849 def _anon_tq_key_label(self) -> _anonymous_label:
1850 return self._anon_label(getattr(self, "_tq_key_label", None))
1852 def _dedupe_anon_tq_label_idx(self, idx: int) -> _anonymous_label:
1853 label = getattr(self, "_tq_label", None) or "anon"
1855 return self._anon_label(label, add_hash=idx)
1858class KeyedColumnElement(ColumnElement[_T]):
1859 """ColumnElement where ``.key`` is non-None."""
1861 _is_keyed_column_element = True
1863 key: str
1866class WrapsColumnExpression(ColumnElement[_T]):
1867 """Mixin that defines a :class:`_expression.ColumnElement`
1868 as a wrapper with special
1869 labeling behavior for an expression that already has a name.
1871 .. versionadded:: 1.4
1873 .. seealso::
1875 :ref:`change_4449`
1878 """
1880 @property
1881 def wrapped_column_expression(self) -> ColumnElement[_T]:
1882 raise NotImplementedError()
1884 @util.non_memoized_property
1885 def _tq_label(self) -> Optional[str]:
1886 wce = self.wrapped_column_expression
1887 if hasattr(wce, "_tq_label"):
1888 return wce._tq_label
1889 else:
1890 return None
1892 @property
1893 def _label(self) -> Optional[str]:
1894 return self._tq_label
1896 @property
1897 def _non_anon_label(self) -> Optional[str]:
1898 return None
1900 @util.non_memoized_property
1901 def _anon_name_label(self) -> str:
1902 wce = self.wrapped_column_expression
1904 # this logic tries to get the WrappedColumnExpression to render
1905 # with "<expr> AS <name>", where "<name>" is the natural name
1906 # within the expression itself. e.g. "CAST(table.foo) AS foo".
1907 if not wce._is_text_clause:
1908 nal = wce._non_anon_label
1909 if nal:
1910 return nal
1911 elif hasattr(wce, "_anon_name_label"):
1912 return wce._anon_name_label
1913 return super()._anon_name_label
1915 def _dedupe_anon_label_idx(self, idx: int) -> str:
1916 wce = self.wrapped_column_expression
1917 nal = wce._non_anon_label
1918 if nal:
1919 return self._anon_label(nal + "_")
1920 else:
1921 return self._dedupe_anon_tq_label_idx(idx)
1923 @property
1924 def _proxy_key(self):
1925 wce = self.wrapped_column_expression
1927 if not wce._is_text_clause:
1928 return wce._proxy_key
1929 return super()._proxy_key
1932class BindParameter(roles.InElementRole, KeyedColumnElement[_T]):
1933 r"""Represent a "bound expression".
1935 :class:`.BindParameter` is invoked explicitly using the
1936 :func:`.bindparam` function, as in::
1938 from sqlalchemy import bindparam
1940 stmt = select(users_table).where(
1941 users_table.c.name == bindparam("username")
1942 )
1944 Detailed discussion of how :class:`.BindParameter` is used is
1945 at :func:`.bindparam`.
1947 .. seealso::
1949 :func:`.bindparam`
1951 """
1953 __visit_name__ = "bindparam"
1955 _traverse_internals: _TraverseInternalsType = [
1956 ("key", InternalTraversal.dp_anon_name),
1957 ("type", InternalTraversal.dp_type),
1958 ("callable", InternalTraversal.dp_plain_dict),
1959 ("value", InternalTraversal.dp_plain_obj),
1960 ("literal_execute", InternalTraversal.dp_boolean),
1961 ]
1963 key: str
1964 type: TypeEngine[_T]
1965 value: Optional[_T]
1967 _is_crud = False
1968 _is_bind_parameter = True
1969 _key_is_anon = False
1971 # bindparam implements its own _gen_cache_key() method however
1972 # we check subclasses for this flag, else no cache key is generated
1973 inherit_cache = True
1975 def __init__(
1976 self,
1977 key: Optional[str],
1978 value: Any = _NoArg.NO_ARG,
1979 type_: Optional[_TypeEngineArgument[_T]] = None,
1980 unique: bool = False,
1981 required: Union[bool, Literal[_NoArg.NO_ARG]] = _NoArg.NO_ARG,
1982 quote: Optional[bool] = None,
1983 callable_: Optional[Callable[[], Any]] = None,
1984 expanding: bool = False,
1985 isoutparam: bool = False,
1986 literal_execute: bool = False,
1987 _compared_to_operator: Optional[OperatorType] = None,
1988 _compared_to_type: Optional[TypeEngine[Any]] = None,
1989 _is_crud: bool = False,
1990 ):
1991 if required is _NoArg.NO_ARG:
1992 required = value is _NoArg.NO_ARG and callable_ is None
1993 if value is _NoArg.NO_ARG:
1994 value = None
1996 if quote is not None:
1997 key = quoted_name.construct(key, quote)
1999 if unique:
2000 self.key = _anonymous_label.safe_construct(
2001 id(self),
2002 (
2003 key
2004 if key is not None
2005 and not isinstance(key, _anonymous_label)
2006 else "param"
2007 ),
2008 sanitize_key=True,
2009 )
2010 self._key_is_anon = True
2011 elif key:
2012 self.key = key
2013 else:
2014 self.key = _anonymous_label.safe_construct(id(self), "param")
2015 self._key_is_anon = True
2017 # identifying key that won't change across
2018 # clones, used to identify the bind's logical
2019 # identity
2020 self._identifying_key = self.key
2022 # key that was passed in the first place, used to
2023 # generate new keys
2024 self._orig_key = key or "param"
2026 self.unique = unique
2027 self.value = value
2028 self.callable = callable_
2029 self.isoutparam = isoutparam
2030 self.required = required
2032 # indicate an "expanding" parameter; the compiler sets this
2033 # automatically in the compiler _render_in_expr_w_bindparam method
2034 # for an IN expression
2035 self.expanding = expanding
2037 # this is another hint to help w/ expanding and is typically
2038 # set in the compiler _render_in_expr_w_bindparam method for an
2039 # IN expression
2040 self.expand_op = None
2042 self.literal_execute = literal_execute
2043 if _is_crud:
2044 self._is_crud = True
2046 if type_ is None:
2047 if expanding:
2048 if value:
2049 check_value = value[0]
2050 else:
2051 check_value = type_api._NO_VALUE_IN_LIST
2052 else:
2053 check_value = value
2054 if _compared_to_type is not None:
2055 self.type = _compared_to_type.coerce_compared_value(
2056 _compared_to_operator, check_value
2057 )
2058 else:
2059 self.type = type_api._resolve_value_to_type(check_value)
2060 elif isinstance(type_, type):
2061 self.type = type_()
2062 elif is_tuple_type(type_):
2063 if value:
2064 if expanding:
2065 check_value = value[0]
2066 else:
2067 check_value = value
2068 cast("BindParameter[typing_Tuple[Any, ...]]", self).type = (
2069 type_._resolve_values_to_types(check_value)
2070 )
2071 else:
2072 cast("BindParameter[typing_Tuple[Any, ...]]", self).type = (
2073 type_
2074 )
2075 else:
2076 self.type = type_
2078 def _with_value(self, value, maintain_key=False, required=NO_ARG):
2079 """Return a copy of this :class:`.BindParameter` with the given value
2080 set.
2081 """
2082 cloned = self._clone(maintain_key=maintain_key)
2083 cloned.value = value
2084 cloned.callable = None
2085 cloned.required = required if required is not NO_ARG else self.required
2086 if cloned.type is type_api.NULLTYPE:
2087 cloned.type = type_api._resolve_value_to_type(value)
2088 return cloned
2090 @property
2091 def effective_value(self) -> Optional[_T]:
2092 """Return the value of this bound parameter,
2093 taking into account if the ``callable`` parameter
2094 was set.
2096 The ``callable`` value will be evaluated
2097 and returned if present, else ``value``.
2099 """
2100 if self.callable:
2101 # TODO: set up protocol for bind parameter callable
2102 return self.callable() # type: ignore
2103 else:
2104 return self.value
2106 def render_literal_execute(self) -> BindParameter[_T]:
2107 """Produce a copy of this bound parameter that will enable the
2108 :paramref:`_sql.BindParameter.literal_execute` flag.
2110 The :paramref:`_sql.BindParameter.literal_execute` flag will
2111 have the effect of the parameter rendered in the compiled SQL
2112 string using ``[POSTCOMPILE]`` form, which is a special form that
2113 is converted to be a rendering of the literal value of the parameter
2114 at SQL execution time. The rationale is to support caching
2115 of SQL statement strings that can embed per-statement literal values,
2116 such as LIMIT and OFFSET parameters, in the final SQL string that
2117 is passed to the DBAPI. Dialects in particular may want to use
2118 this method within custom compilation schemes.
2120 .. versionadded:: 1.4.5
2122 .. seealso::
2124 :ref:`engine_thirdparty_caching`
2126 """
2127 c = ClauseElement._clone(self)
2128 c.literal_execute = True
2129 return c
2131 def _negate_in_binary(self, negated_op, original_op):
2132 if self.expand_op is original_op:
2133 bind = self._clone()
2134 bind.expand_op = negated_op
2135 return bind
2136 else:
2137 return self
2139 def _with_binary_element_type(self, type_: TypeEngine[Any]) -> Self:
2140 c: Self = ClauseElement._clone(self)
2141 c.type = type_
2142 return c
2144 def _clone(self, maintain_key: bool = False, **kw: Any) -> Self:
2145 c = ClauseElement._clone(self, **kw)
2146 # ensure all the BindParameter objects stay in cloned set.
2147 # in #7823, we changed "clone" so that a clone only keeps a reference
2148 # to the "original" element, since for column correspondence, that's
2149 # all we need. However, for BindParam, _cloned_set is used by
2150 # the "cache key bind match" lookup, which means if any of those
2151 # interim BindParameter objects became part of a cache key in the
2152 # cache, we need it. So here, make sure all clones keep carrying
2153 # forward.
2154 c._cloned_set.update(self._cloned_set)
2155 if not maintain_key and self.unique:
2156 c.key = _anonymous_label.safe_construct(
2157 id(c), c._orig_key or "param", sanitize_key=True
2158 )
2159 return c
2161 def _gen_cache_key(self, anon_map, bindparams):
2162 _gen_cache_ok = self.__class__.__dict__.get("inherit_cache", False)
2164 if not _gen_cache_ok:
2165 if anon_map is not None:
2166 anon_map[NO_CACHE] = True
2167 return None
2169 id_, found = anon_map.get_anon(self)
2170 if found:
2171 return (id_, self.__class__)
2173 if bindparams is not None:
2174 bindparams.append(self)
2176 return (
2177 id_,
2178 self.__class__,
2179 self.type._static_cache_key,
2180 self.key % anon_map if self._key_is_anon else self.key,
2181 self.literal_execute,
2182 )
2184 def _convert_to_unique(self):
2185 if not self.unique:
2186 self.unique = True
2187 self.key = _anonymous_label.safe_construct(
2188 id(self), self._orig_key or "param", sanitize_key=True
2189 )
2191 def __getstate__(self):
2192 """execute a deferred value for serialization purposes."""
2194 d = self.__dict__.copy()
2195 v = self.value
2196 if self.callable:
2197 v = self.callable()
2198 d["callable"] = None
2199 d["value"] = v
2200 return d
2202 def __setstate__(self, state):
2203 if state.get("unique", False):
2204 state["key"] = _anonymous_label.safe_construct(
2205 id(self), state.get("_orig_key", "param"), sanitize_key=True
2206 )
2207 self.__dict__.update(state)
2209 def __repr__(self):
2210 return "%s(%r, %r, type_=%r)" % (
2211 self.__class__.__name__,
2212 self.key,
2213 self.value,
2214 self.type,
2215 )
2218class TypeClause(DQLDMLClauseElement):
2219 """Handle a type keyword in a SQL statement.
2221 Used by the ``Case`` statement.
2223 """
2225 __visit_name__ = "typeclause"
2227 _traverse_internals: _TraverseInternalsType = [
2228 ("type", InternalTraversal.dp_type)
2229 ]
2230 type: TypeEngine[Any]
2232 def __init__(self, type_: TypeEngine[Any]):
2233 self.type = type_
2236class TextClause(
2237 roles.DDLConstraintColumnRole,
2238 roles.DDLExpressionRole,
2239 roles.StatementOptionRole,
2240 roles.WhereHavingRole,
2241 roles.OrderByRole,
2242 roles.FromClauseRole,
2243 roles.SelectStatementRole,
2244 roles.InElementRole,
2245 Generative,
2246 Executable,
2247 DQLDMLClauseElement,
2248 roles.BinaryElementRole[Any],
2249 inspection.Inspectable["TextClause"],
2250):
2251 """Represent a literal SQL text fragment.
2253 E.g.::
2255 from sqlalchemy import text
2257 t = text("SELECT * FROM users")
2258 result = connection.execute(t)
2260 The :class:`_expression.TextClause` construct is produced using the
2261 :func:`_expression.text`
2262 function; see that function for full documentation.
2264 .. seealso::
2266 :func:`_expression.text`
2268 """
2270 __visit_name__ = "textclause"
2272 _traverse_internals: _TraverseInternalsType = [
2273 ("_bindparams", InternalTraversal.dp_string_clauseelement_dict),
2274 ("text", InternalTraversal.dp_string),
2275 ] + Executable._executable_traverse_internals
2277 _is_text_clause = True
2279 _is_textual = True
2281 _bind_params_regex = re.compile(r"(?<![:\w\x5c]):(\w+)(?!:)", re.UNICODE)
2282 _is_implicitly_boolean = False
2284 _render_label_in_columns_clause = False
2286 _omit_from_statements = False
2288 _is_collection_aggregate = False
2290 @property
2291 def _hide_froms(self) -> Iterable[FromClause]:
2292 return ()
2294 def __and__(self, other):
2295 # support use in select.where(), query.filter()
2296 return and_(self, other)
2298 @property
2299 def _select_iterable(self) -> _SelectIterable:
2300 return (self,)
2302 # help in those cases where text() is
2303 # interpreted in a column expression situation
2304 key: Optional[str] = None
2305 _label: Optional[str] = None
2307 _allow_label_resolve = False
2309 @property
2310 def _is_star(self): # type: ignore[override]
2311 return self.text == "*"
2313 def __init__(self, text: str):
2314 self._bindparams: Dict[str, BindParameter[Any]] = {}
2316 def repl(m):
2317 self._bindparams[m.group(1)] = BindParameter(m.group(1))
2318 return ":%s" % m.group(1)
2320 # scan the string and search for bind parameter names, add them
2321 # to the list of bindparams
2322 self.text = self._bind_params_regex.sub(repl, text)
2324 @_generative
2325 def bindparams(
2326 self,
2327 *binds: BindParameter[Any],
2328 **names_to_values: Any,
2329 ) -> Self:
2330 """Establish the values and/or types of bound parameters within
2331 this :class:`_expression.TextClause` construct.
2333 Given a text construct such as::
2335 from sqlalchemy import text
2337 stmt = text(
2338 "SELECT id, name FROM user WHERE name=:name AND timestamp=:timestamp"
2339 )
2341 the :meth:`_expression.TextClause.bindparams`
2342 method can be used to establish
2343 the initial value of ``:name`` and ``:timestamp``,
2344 using simple keyword arguments::
2346 stmt = stmt.bindparams(
2347 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2348 )
2350 Where above, new :class:`.BindParameter` objects
2351 will be generated with the names ``name`` and ``timestamp``, and
2352 values of ``jack`` and ``datetime.datetime(2012, 10, 8, 15, 12, 5)``,
2353 respectively. The types will be
2354 inferred from the values given, in this case :class:`.String` and
2355 :class:`.DateTime`.
2357 When specific typing behavior is needed, the positional ``*binds``
2358 argument can be used in which to specify :func:`.bindparam` constructs
2359 directly. These constructs must include at least the ``key``
2360 argument, then an optional value and type::
2362 from sqlalchemy import bindparam
2364 stmt = stmt.bindparams(
2365 bindparam("name", value="jack", type_=String),
2366 bindparam("timestamp", type_=DateTime),
2367 )
2369 Above, we specified the type of :class:`.DateTime` for the
2370 ``timestamp`` bind, and the type of :class:`.String` for the ``name``
2371 bind. In the case of ``name`` we also set the default value of
2372 ``"jack"``.
2374 Additional bound parameters can be supplied at statement execution
2375 time, e.g.::
2377 result = connection.execute(
2378 stmt, timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2379 )
2381 The :meth:`_expression.TextClause.bindparams`
2382 method can be called repeatedly,
2383 where it will re-use existing :class:`.BindParameter` objects to add
2384 new information. For example, we can call
2385 :meth:`_expression.TextClause.bindparams`
2386 first with typing information, and a
2387 second time with value information, and it will be combined::
2389 stmt = text(
2390 "SELECT id, name FROM user WHERE name=:name "
2391 "AND timestamp=:timestamp"
2392 )
2393 stmt = stmt.bindparams(
2394 bindparam("name", type_=String), bindparam("timestamp", type_=DateTime)
2395 )
2396 stmt = stmt.bindparams(
2397 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2398 )
2400 The :meth:`_expression.TextClause.bindparams`
2401 method also supports the concept of
2402 **unique** bound parameters. These are parameters that are
2403 "uniquified" on name at statement compilation time, so that multiple
2404 :func:`_expression.text`
2405 constructs may be combined together without the names
2406 conflicting. To use this feature, specify the
2407 :paramref:`.BindParameter.unique` flag on each :func:`.bindparam`
2408 object::
2410 stmt1 = text("select id from table where name=:name").bindparams(
2411 bindparam("name", value="name1", unique=True)
2412 )
2413 stmt2 = text("select id from table where name=:name").bindparams(
2414 bindparam("name", value="name2", unique=True)
2415 )
2417 union = union_all(stmt1.columns(column("id")), stmt2.columns(column("id")))
2419 The above statement will render as:
2421 .. sourcecode:: sql
2423 select id from table where name=:name_1
2424 UNION ALL select id from table where name=:name_2
2426 .. versionadded:: 1.3.11 Added support for the
2427 :paramref:`.BindParameter.unique` flag to work with
2428 :func:`_expression.text`
2429 constructs.
2431 """ # noqa: E501
2432 self._bindparams = new_params = self._bindparams.copy()
2434 for bind in binds:
2435 try:
2436 # the regex used for text() currently will not match
2437 # a unique/anonymous key in any case, so use the _orig_key
2438 # so that a text() construct can support unique parameters
2439 existing = new_params[bind._orig_key]
2440 except KeyError as err:
2441 raise exc.ArgumentError(
2442 "This text() construct doesn't define a "
2443 "bound parameter named %r" % bind._orig_key
2444 ) from err
2445 else:
2446 new_params[existing._orig_key] = bind
2448 for key, value in names_to_values.items():
2449 try:
2450 existing = new_params[key]
2451 except KeyError as err:
2452 raise exc.ArgumentError(
2453 "This text() construct doesn't define a "
2454 "bound parameter named %r" % key
2455 ) from err
2456 else:
2457 new_params[key] = existing._with_value(value, required=False)
2458 return self
2460 @util.preload_module("sqlalchemy.sql.selectable")
2461 def columns(
2462 self,
2463 *cols: _OnlyColumnArgument[Any],
2464 **types: _TypeEngineArgument[Any],
2465 ) -> TextualSelect:
2466 r"""Turn this :class:`_expression.TextClause` object into a
2467 :class:`_expression.TextualSelect`
2468 object that serves the same role as a SELECT
2469 statement.
2471 The :class:`_expression.TextualSelect` is part of the
2472 :class:`_expression.SelectBase`
2473 hierarchy and can be embedded into another statement by using the
2474 :meth:`_expression.TextualSelect.subquery` method to produce a
2475 :class:`.Subquery`
2476 object, which can then be SELECTed from.
2478 This function essentially bridges the gap between an entirely
2479 textual SELECT statement and the SQL expression language concept
2480 of a "selectable"::
2482 from sqlalchemy.sql import column, text
2484 stmt = text("SELECT id, name FROM some_table")
2485 stmt = stmt.columns(column("id"), column("name")).subquery("st")
2487 stmt = (
2488 select(mytable)
2489 .select_from(mytable.join(stmt, mytable.c.name == stmt.c.name))
2490 .where(stmt.c.id > 5)
2491 )
2493 Above, we pass a series of :func:`_expression.column` elements to the
2494 :meth:`_expression.TextClause.columns` method positionally. These
2495 :func:`_expression.column`
2496 elements now become first class elements upon the
2497 :attr:`_expression.TextualSelect.selected_columns` column collection,
2498 which then
2499 become part of the :attr:`.Subquery.c` collection after
2500 :meth:`_expression.TextualSelect.subquery` is invoked.
2502 The column expressions we pass to
2503 :meth:`_expression.TextClause.columns` may
2504 also be typed; when we do so, these :class:`.TypeEngine` objects become
2505 the effective return type of the column, so that SQLAlchemy's
2506 result-set-processing systems may be used on the return values.
2507 This is often needed for types such as date or boolean types, as well
2508 as for unicode processing on some dialect configurations::
2510 stmt = text("SELECT id, name, timestamp FROM some_table")
2511 stmt = stmt.columns(
2512 column("id", Integer),
2513 column("name", Unicode),
2514 column("timestamp", DateTime),
2515 )
2517 for id, name, timestamp in connection.execute(stmt):
2518 print(id, name, timestamp)
2520 As a shortcut to the above syntax, keyword arguments referring to
2521 types alone may be used, if only type conversion is needed::
2523 stmt = text("SELECT id, name, timestamp FROM some_table")
2524 stmt = stmt.columns(id=Integer, name=Unicode, timestamp=DateTime)
2526 for id, name, timestamp in connection.execute(stmt):
2527 print(id, name, timestamp)
2529 The positional form of :meth:`_expression.TextClause.columns`
2530 also provides the
2531 unique feature of **positional column targeting**, which is
2532 particularly useful when using the ORM with complex textual queries. If
2533 we specify the columns from our model to
2534 :meth:`_expression.TextClause.columns`,
2535 the result set will match to those columns positionally, meaning the
2536 name or origin of the column in the textual SQL doesn't matter::
2538 stmt = text(
2539 "SELECT users.id, addresses.id, users.id, "
2540 "users.name, addresses.email_address AS email "
2541 "FROM users JOIN addresses ON users.id=addresses.user_id "
2542 "WHERE users.id = 1"
2543 ).columns(
2544 User.id,
2545 Address.id,
2546 Address.user_id,
2547 User.name,
2548 Address.email_address,
2549 )
2551 query = (
2552 session.query(User)
2553 .from_statement(stmt)
2554 .options(contains_eager(User.addresses))
2555 )
2557 The :meth:`_expression.TextClause.columns` method provides a direct
2558 route to calling :meth:`_expression.FromClause.subquery` as well as
2559 :meth:`_expression.SelectBase.cte`
2560 against a textual SELECT statement::
2562 stmt = stmt.columns(id=Integer, name=String).cte("st")
2564 stmt = select(sometable).where(sometable.c.id == stmt.c.id)
2566 :param \*cols: A series of :class:`_expression.ColumnElement` objects,
2567 typically
2568 :class:`_schema.Column` objects from a :class:`_schema.Table`
2569 or ORM level
2570 column-mapped attributes, representing a set of columns that this
2571 textual string will SELECT from.
2573 :param \**types: A mapping of string names to :class:`.TypeEngine`
2574 type objects indicating the datatypes to use for names that are
2575 SELECTed from the textual string. Prefer to use the ``*cols``
2576 argument as it also indicates positional ordering.
2578 """
2579 selectable = util.preloaded.sql_selectable
2581 input_cols: List[NamedColumn[Any]] = [
2582 coercions.expect(roles.LabeledColumnExprRole, col) for col in cols
2583 ]
2585 positional_input_cols = [
2586 (
2587 ColumnClause(col.key, types.pop(col.key))
2588 if col.key in types
2589 else col
2590 )
2591 for col in input_cols
2592 ]
2593 keyed_input_cols: List[NamedColumn[Any]] = [
2594 ColumnClause(key, type_) for key, type_ in types.items()
2595 ]
2597 elem = selectable.TextualSelect.__new__(selectable.TextualSelect)
2598 elem._init(
2599 self,
2600 positional_input_cols + keyed_input_cols,
2601 positional=bool(positional_input_cols) and not keyed_input_cols,
2602 )
2603 return elem
2605 @property
2606 def type(self) -> TypeEngine[Any]:
2607 return type_api.NULLTYPE
2609 @property
2610 def comparator(self):
2611 # TODO: this seems wrong, it seems like we might not
2612 # be using this method.
2613 return self.type.comparator_factory(self) # type: ignore
2615 def self_group(
2616 self, against: Optional[OperatorType] = None
2617 ) -> Union[Self, Grouping[Any]]:
2618 if against is operators.in_op:
2619 return Grouping(self)
2620 else:
2621 return self
2624class Null(SingletonConstant, roles.ConstExprRole[None], ColumnElement[None]):
2625 """Represent the NULL keyword in a SQL statement.
2627 :class:`.Null` is accessed as a constant via the
2628 :func:`.null` function.
2630 """
2632 __visit_name__ = "null"
2634 _traverse_internals: _TraverseInternalsType = []
2635 _singleton: Null
2637 if not TYPE_CHECKING:
2639 @util.memoized_property
2640 def type(self) -> TypeEngine[_T]: # noqa: A001
2641 return type_api.NULLTYPE
2643 @classmethod
2644 def _instance(cls) -> Null:
2645 """Return a constant :class:`.Null` construct."""
2647 return Null._singleton
2650Null._create_singleton()
2653class False_(
2654 SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]
2655):
2656 """Represent the ``false`` keyword, or equivalent, in a SQL statement.
2658 :class:`.False_` is accessed as a constant via the
2659 :func:`.false` function.
2661 """
2663 __visit_name__ = "false"
2664 _traverse_internals: _TraverseInternalsType = []
2665 _singleton: False_
2667 if not TYPE_CHECKING:
2669 @util.memoized_property
2670 def type(self) -> TypeEngine[_T]: # noqa: A001
2671 return type_api.BOOLEANTYPE
2673 def _negate(self) -> True_:
2674 return True_._singleton
2676 @classmethod
2677 def _instance(cls) -> False_:
2678 return False_._singleton
2681False_._create_singleton()
2684class True_(SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]):
2685 """Represent the ``true`` keyword, or equivalent, in a SQL statement.
2687 :class:`.True_` is accessed as a constant via the
2688 :func:`.true` function.
2690 """
2692 __visit_name__ = "true"
2694 _traverse_internals: _TraverseInternalsType = []
2695 _singleton: True_
2697 if not TYPE_CHECKING:
2699 @util.memoized_property
2700 def type(self) -> TypeEngine[_T]: # noqa: A001
2701 return type_api.BOOLEANTYPE
2703 def _negate(self) -> False_:
2704 return False_._singleton
2706 @classmethod
2707 def _ifnone(
2708 cls, other: Optional[ColumnElement[Any]]
2709 ) -> ColumnElement[Any]:
2710 if other is None:
2711 return cls._instance()
2712 else:
2713 return other
2715 @classmethod
2716 def _instance(cls) -> True_:
2717 return True_._singleton
2720True_._create_singleton()
2723class ClauseList(
2724 roles.InElementRole,
2725 roles.OrderByRole,
2726 roles.ColumnsClauseRole,
2727 roles.DMLColumnRole,
2728 DQLDMLClauseElement,
2729):
2730 """Describe a list of clauses, separated by an operator.
2732 By default, is comma-separated, such as a column listing.
2734 """
2736 __visit_name__ = "clauselist"
2738 # this is used only by the ORM in a legacy use case for
2739 # composite attributes
2740 _is_clause_list = True
2742 _traverse_internals: _TraverseInternalsType = [
2743 ("clauses", InternalTraversal.dp_clauseelement_list),
2744 ("operator", InternalTraversal.dp_operator),
2745 ]
2747 clauses: List[ColumnElement[Any]]
2749 def __init__(
2750 self,
2751 *clauses: _ColumnExpressionArgument[Any],
2752 operator: OperatorType = operators.comma_op,
2753 group: bool = True,
2754 group_contents: bool = True,
2755 _literal_as_text_role: Type[roles.SQLRole] = roles.WhereHavingRole,
2756 ):
2757 self.operator = operator
2758 self.group = group
2759 self.group_contents = group_contents
2760 clauses_iterator: Iterable[_ColumnExpressionArgument[Any]] = clauses
2761 text_converter_role: Type[roles.SQLRole] = _literal_as_text_role
2762 self._text_converter_role = text_converter_role
2764 if self.group_contents:
2765 self.clauses = [
2766 coercions.expect(
2767 text_converter_role, clause, apply_propagate_attrs=self
2768 ).self_group(against=self.operator)
2769 for clause in clauses_iterator
2770 ]
2771 else:
2772 self.clauses = [
2773 coercions.expect(
2774 text_converter_role, clause, apply_propagate_attrs=self
2775 )
2776 for clause in clauses_iterator
2777 ]
2778 self._is_implicitly_boolean = operators.is_boolean(self.operator)
2780 @classmethod
2781 def _construct_raw(
2782 cls,
2783 operator: OperatorType,
2784 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
2785 ) -> ClauseList:
2786 self = cls.__new__(cls)
2787 self.clauses = list(clauses) if clauses else []
2788 self.group = True
2789 self.operator = operator
2790 self.group_contents = True
2791 self._is_implicitly_boolean = False
2792 return self
2794 def __iter__(self) -> Iterator[ColumnElement[Any]]:
2795 return iter(self.clauses)
2797 def __len__(self) -> int:
2798 return len(self.clauses)
2800 @property
2801 def _select_iterable(self) -> _SelectIterable:
2802 return itertools.chain.from_iterable(
2803 [elem._select_iterable for elem in self.clauses]
2804 )
2806 def append(self, clause):
2807 if self.group_contents:
2808 self.clauses.append(
2809 coercions.expect(self._text_converter_role, clause).self_group(
2810 against=self.operator
2811 )
2812 )
2813 else:
2814 self.clauses.append(
2815 coercions.expect(self._text_converter_role, clause)
2816 )
2818 @util.ro_non_memoized_property
2819 def _from_objects(self) -> List[FromClause]:
2820 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
2822 def self_group(
2823 self, against: Optional[OperatorType] = None
2824 ) -> Union[Self, Grouping[Any]]:
2825 if self.group and operators.is_precedent(self.operator, against):
2826 return Grouping(self)
2827 else:
2828 return self
2831class OperatorExpression(ColumnElement[_T]):
2832 """base for expressions that contain an operator and operands
2834 .. versionadded:: 2.0
2836 """
2838 operator: OperatorType
2839 type: TypeEngine[_T]
2841 group: bool = True
2843 @property
2844 def is_comparison(self):
2845 return operators.is_comparison(self.operator)
2847 def self_group(
2848 self, against: Optional[OperatorType] = None
2849 ) -> Union[Self, Grouping[_T]]:
2850 if (
2851 self.group
2852 and operators.is_precedent(self.operator, against)
2853 or (
2854 # a negate against a non-boolean operator
2855 # doesn't make too much sense but we should
2856 # group for that
2857 against is operators.inv
2858 and not operators.is_boolean(self.operator)
2859 )
2860 ):
2861 return Grouping(self)
2862 else:
2863 return self
2865 @property
2866 def _flattened_operator_clauses(
2867 self,
2868 ) -> typing_Tuple[ColumnElement[Any], ...]:
2869 raise NotImplementedError()
2871 @classmethod
2872 def _construct_for_op(
2873 cls,
2874 left: ColumnElement[Any],
2875 right: ColumnElement[Any],
2876 op: OperatorType,
2877 *,
2878 type_: TypeEngine[_T],
2879 negate: Optional[OperatorType] = None,
2880 modifiers: Optional[Mapping[str, Any]] = None,
2881 ) -> OperatorExpression[_T]:
2882 if operators.is_associative(op):
2883 assert (
2884 negate is None
2885 ), f"negate not supported for associative operator {op}"
2887 multi = False
2888 if getattr(
2889 left, "operator", None
2890 ) is op and type_._compare_type_affinity(left.type):
2891 multi = True
2892 left_flattened = left._flattened_operator_clauses
2893 else:
2894 left_flattened = (left,)
2896 if getattr(
2897 right, "operator", None
2898 ) is op and type_._compare_type_affinity(right.type):
2899 multi = True
2900 right_flattened = right._flattened_operator_clauses
2901 else:
2902 right_flattened = (right,)
2904 if multi:
2905 return ExpressionClauseList._construct_for_list(
2906 op,
2907 type_,
2908 *(left_flattened + right_flattened),
2909 )
2911 if right._is_collection_aggregate:
2912 negate = None
2914 return BinaryExpression(
2915 left, right, op, type_=type_, negate=negate, modifiers=modifiers
2916 )
2919class ExpressionClauseList(OperatorExpression[_T]):
2920 """Describe a list of clauses, separated by an operator,
2921 in a column expression context.
2923 :class:`.ExpressionClauseList` differs from :class:`.ClauseList` in that
2924 it represents a column-oriented DQL expression only, not an open ended
2925 list of anything comma separated.
2927 .. versionadded:: 2.0
2929 """
2931 __visit_name__ = "expression_clauselist"
2933 _traverse_internals: _TraverseInternalsType = [
2934 ("clauses", InternalTraversal.dp_clauseelement_tuple),
2935 ("operator", InternalTraversal.dp_operator),
2936 ]
2938 clauses: typing_Tuple[ColumnElement[Any], ...]
2940 group: bool
2942 def __init__(
2943 self,
2944 operator: OperatorType,
2945 *clauses: _ColumnExpressionArgument[Any],
2946 type_: Optional[_TypeEngineArgument[_T]] = None,
2947 ):
2948 self.operator = operator
2950 self.clauses = tuple(
2951 coercions.expect(
2952 roles.ExpressionElementRole, clause, apply_propagate_attrs=self
2953 )
2954 for clause in clauses
2955 )
2956 self._is_implicitly_boolean = operators.is_boolean(self.operator)
2957 self.type = type_api.to_instance(type_) # type: ignore
2959 @property
2960 def _flattened_operator_clauses(
2961 self,
2962 ) -> typing_Tuple[ColumnElement[Any], ...]:
2963 return self.clauses
2965 def __iter__(self) -> Iterator[ColumnElement[Any]]:
2966 return iter(self.clauses)
2968 def __len__(self) -> int:
2969 return len(self.clauses)
2971 @property
2972 def _select_iterable(self) -> _SelectIterable:
2973 return (self,)
2975 @util.ro_non_memoized_property
2976 def _from_objects(self) -> List[FromClause]:
2977 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
2979 def _append_inplace(self, clause: ColumnElement[Any]) -> None:
2980 self.clauses += (clause,)
2982 @classmethod
2983 def _construct_for_list(
2984 cls,
2985 operator: OperatorType,
2986 type_: TypeEngine[_T],
2987 *clauses: ColumnElement[Any],
2988 group: bool = True,
2989 ) -> ExpressionClauseList[_T]:
2990 self = cls.__new__(cls)
2991 self.group = group
2992 if group:
2993 self.clauses = tuple(
2994 c.self_group(against=operator) for c in clauses
2995 )
2996 else:
2997 self.clauses = clauses
2998 self.operator = operator
2999 self.type = type_
3000 for c in clauses:
3001 if c._propagate_attrs:
3002 self._propagate_attrs = c._propagate_attrs
3003 break
3004 return self
3006 def _negate(self) -> Any:
3007 grouped = self.self_group(against=operators.inv)
3008 assert isinstance(grouped, ColumnElement)
3009 return UnaryExpression(grouped, operator=operators.inv)
3012class BooleanClauseList(ExpressionClauseList[bool]):
3013 __visit_name__ = "expression_clauselist"
3014 inherit_cache = True
3016 def __init__(self, *arg, **kw):
3017 raise NotImplementedError(
3018 "BooleanClauseList has a private constructor"
3019 )
3021 @classmethod
3022 def _process_clauses_for_boolean(
3023 cls,
3024 operator: OperatorType,
3025 continue_on: Any,
3026 skip_on: Any,
3027 clauses: Iterable[ColumnElement[Any]],
3028 ) -> typing_Tuple[int, List[ColumnElement[Any]]]:
3029 has_continue_on = None
3031 convert_clauses = []
3033 against = operators._asbool
3034 lcc = 0
3036 for clause in clauses:
3037 if clause is continue_on:
3038 # instance of continue_on, like and_(x, y, True, z), store it
3039 # if we didn't find one already, we will use it if there
3040 # are no other expressions here.
3041 has_continue_on = clause
3042 elif clause is skip_on:
3043 # instance of skip_on, e.g. and_(x, y, False, z), cancels
3044 # the rest out
3045 convert_clauses = [clause]
3046 lcc = 1
3047 break
3048 else:
3049 if not lcc:
3050 lcc = 1
3051 else:
3052 against = operator
3053 # technically this would be len(convert_clauses) + 1
3054 # however this only needs to indicate "greater than one"
3055 lcc = 2
3056 convert_clauses.append(clause)
3058 if not convert_clauses and has_continue_on is not None:
3059 convert_clauses = [has_continue_on]
3060 lcc = 1
3062 return lcc, [c.self_group(against=against) for c in convert_clauses]
3064 @classmethod
3065 def _construct(
3066 cls,
3067 operator: OperatorType,
3068 continue_on: Any,
3069 skip_on: Any,
3070 initial_clause: Any = _NoArg.NO_ARG,
3071 *clauses: Any,
3072 **kw: Any,
3073 ) -> ColumnElement[Any]:
3074 if initial_clause is _NoArg.NO_ARG:
3075 # no elements period. deprecated use case. return an empty
3076 # ClauseList construct that generates nothing unless it has
3077 # elements added to it.
3078 name = operator.__name__
3080 util.warn_deprecated(
3081 f"Invoking {name}() without arguments is deprecated, and "
3082 f"will be disallowed in a future release. For an empty "
3083 f"""{name}() construct, use '{name}({
3084 'true()' if continue_on is True_._singleton else 'false()'
3085 }, *args)' """
3086 f"""or '{name}({
3087 'True' if continue_on is True_._singleton else 'False'
3088 }, *args)'.""",
3089 version="1.4",
3090 )
3091 return cls._construct_raw(operator)
3093 lcc, convert_clauses = cls._process_clauses_for_boolean(
3094 operator,
3095 continue_on,
3096 skip_on,
3097 [
3098 coercions.expect(roles.WhereHavingRole, clause)
3099 for clause in util.coerce_generator_arg(
3100 (initial_clause,) + clauses
3101 )
3102 ],
3103 )
3105 if lcc > 1:
3106 # multiple elements. Return regular BooleanClauseList
3107 # which will link elements against the operator.
3109 flattened_clauses = itertools.chain.from_iterable(
3110 (
3111 (c for c in to_flat._flattened_operator_clauses)
3112 if getattr(to_flat, "operator", None) is operator
3113 else (to_flat,)
3114 )
3115 for to_flat in convert_clauses
3116 )
3118 return cls._construct_raw(operator, flattened_clauses) # type: ignore # noqa: E501
3119 else:
3120 assert lcc
3121 # just one element. return it as a single boolean element,
3122 # not a list and discard the operator.
3123 return convert_clauses[0]
3125 @classmethod
3126 def _construct_for_whereclause(
3127 cls, clauses: Iterable[ColumnElement[Any]]
3128 ) -> Optional[ColumnElement[bool]]:
3129 operator, continue_on, skip_on = (
3130 operators.and_,
3131 True_._singleton,
3132 False_._singleton,
3133 )
3135 lcc, convert_clauses = cls._process_clauses_for_boolean(
3136 operator,
3137 continue_on,
3138 skip_on,
3139 clauses, # these are assumed to be coerced already
3140 )
3142 if lcc > 1:
3143 # multiple elements. Return regular BooleanClauseList
3144 # which will link elements against the operator.
3145 return cls._construct_raw(operator, convert_clauses)
3146 elif lcc == 1:
3147 # just one element. return it as a single boolean element,
3148 # not a list and discard the operator.
3149 return convert_clauses[0]
3150 else:
3151 return None
3153 @classmethod
3154 def _construct_raw(
3155 cls,
3156 operator: OperatorType,
3157 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
3158 ) -> BooleanClauseList:
3159 self = cls.__new__(cls)
3160 self.clauses = tuple(clauses) if clauses else ()
3161 self.group = True
3162 self.operator = operator
3163 self.type = type_api.BOOLEANTYPE
3164 self._is_implicitly_boolean = True
3165 return self
3167 @classmethod
3168 def and_(
3169 cls,
3170 initial_clause: Union[
3171 Literal[True], _ColumnExpressionArgument[bool], _NoArg
3172 ] = _NoArg.NO_ARG,
3173 *clauses: _ColumnExpressionArgument[bool],
3174 ) -> ColumnElement[bool]:
3175 r"""Produce a conjunction of expressions joined by ``AND``.
3177 See :func:`_sql.and_` for full documentation.
3178 """
3179 return cls._construct(
3180 operators.and_,
3181 True_._singleton,
3182 False_._singleton,
3183 initial_clause,
3184 *clauses,
3185 )
3187 @classmethod
3188 def or_(
3189 cls,
3190 initial_clause: Union[
3191 Literal[False], _ColumnExpressionArgument[bool], _NoArg
3192 ] = _NoArg.NO_ARG,
3193 *clauses: _ColumnExpressionArgument[bool],
3194 ) -> ColumnElement[bool]:
3195 """Produce a conjunction of expressions joined by ``OR``.
3197 See :func:`_sql.or_` for full documentation.
3198 """
3199 return cls._construct(
3200 operators.or_,
3201 False_._singleton,
3202 True_._singleton,
3203 initial_clause,
3204 *clauses,
3205 )
3207 @property
3208 def _select_iterable(self) -> _SelectIterable:
3209 return (self,)
3211 def self_group(
3212 self, against: Optional[OperatorType] = None
3213 ) -> Union[Self, Grouping[bool]]:
3214 if not self.clauses:
3215 return self
3216 else:
3217 return super().self_group(against=against)
3220and_ = BooleanClauseList.and_
3221or_ = BooleanClauseList.or_
3224class Tuple(ClauseList, ColumnElement[typing_Tuple[Any, ...]]):
3225 """Represent a SQL tuple."""
3227 __visit_name__ = "tuple"
3229 _traverse_internals: _TraverseInternalsType = (
3230 ClauseList._traverse_internals + []
3231 )
3233 type: TupleType
3235 @util.preload_module("sqlalchemy.sql.sqltypes")
3236 def __init__(
3237 self,
3238 *clauses: _ColumnExpressionArgument[Any],
3239 types: Optional[Sequence[_TypeEngineArgument[Any]]] = None,
3240 ):
3241 sqltypes = util.preloaded.sql_sqltypes
3243 if types is None:
3244 init_clauses: List[ColumnElement[Any]] = [
3245 coercions.expect(roles.ExpressionElementRole, c)
3246 for c in clauses
3247 ]
3248 else:
3249 if len(types) != len(clauses):
3250 raise exc.ArgumentError(
3251 "Wrong number of elements for %d-tuple: %r "
3252 % (len(types), clauses)
3253 )
3254 init_clauses = [
3255 coercions.expect(
3256 roles.ExpressionElementRole,
3257 c,
3258 type_=typ if not typ._isnull else None,
3259 )
3260 for typ, c in zip(types, clauses)
3261 ]
3263 self.type = sqltypes.TupleType(*[arg.type for arg in init_clauses])
3264 super().__init__(*init_clauses)
3266 @property
3267 def _select_iterable(self) -> _SelectIterable:
3268 return (self,)
3270 def _bind_param(self, operator, obj, type_=None, expanding=False):
3271 if expanding:
3272 return BindParameter(
3273 None,
3274 value=obj,
3275 _compared_to_operator=operator,
3276 unique=True,
3277 expanding=True,
3278 type_=type_,
3279 _compared_to_type=self.type,
3280 )
3281 else:
3282 return Tuple(
3283 *[
3284 BindParameter(
3285 None,
3286 o,
3287 _compared_to_operator=operator,
3288 _compared_to_type=compared_to_type,
3289 unique=True,
3290 type_=type_,
3291 )
3292 for o, compared_to_type in zip(obj, self.type.types)
3293 ]
3294 )
3296 def self_group(self, against: Optional[OperatorType] = None) -> Self:
3297 # Tuple is parenthesized by definition.
3298 return self
3301class Case(ColumnElement[_T]):
3302 """Represent a ``CASE`` expression.
3304 :class:`.Case` is produced using the :func:`.case` factory function,
3305 as in::
3307 from sqlalchemy import case
3309 stmt = select(users_table).where(
3310 case(
3311 (users_table.c.name == "wendy", "W"),
3312 (users_table.c.name == "jack", "J"),
3313 else_="E",
3314 )
3315 )
3317 Details on :class:`.Case` usage is at :func:`.case`.
3319 .. seealso::
3321 :func:`.case`
3323 """
3325 __visit_name__ = "case"
3327 _traverse_internals: _TraverseInternalsType = [
3328 ("value", InternalTraversal.dp_clauseelement),
3329 ("whens", InternalTraversal.dp_clauseelement_tuples),
3330 ("else_", InternalTraversal.dp_clauseelement),
3331 ]
3333 # for case(), the type is derived from the whens. so for the moment
3334 # users would have to cast() the case to get a specific type
3336 whens: List[typing_Tuple[ColumnElement[bool], ColumnElement[_T]]]
3337 else_: Optional[ColumnElement[_T]]
3338 value: Optional[ColumnElement[Any]]
3340 def __init__(
3341 self,
3342 *whens: Union[
3343 typing_Tuple[_ColumnExpressionArgument[bool], Any],
3344 Mapping[Any, Any],
3345 ],
3346 value: Optional[Any] = None,
3347 else_: Optional[Any] = None,
3348 ):
3349 new_whens: Iterable[Any] = coercions._expression_collection_was_a_list(
3350 "whens", "case", whens
3351 )
3352 try:
3353 new_whens = util.dictlike_iteritems(new_whens)
3354 except TypeError:
3355 pass
3357 self.whens = [
3358 (
3359 coercions.expect(
3360 roles.ExpressionElementRole,
3361 c,
3362 apply_propagate_attrs=self,
3363 ).self_group(),
3364 coercions.expect(roles.ExpressionElementRole, r),
3365 )
3366 for (c, r) in new_whens
3367 ]
3369 if value is None:
3370 self.value = None
3371 else:
3372 self.value = coercions.expect(roles.ExpressionElementRole, value)
3374 if else_ is not None:
3375 self.else_ = coercions.expect(roles.ExpressionElementRole, else_)
3376 else:
3377 self.else_ = None
3379 type_ = next(
3380 (
3381 then.type
3382 # Iterate `whens` in reverse to match previous behaviour
3383 # where type of final element took priority
3384 for *_, then in reversed(self.whens)
3385 if not then.type._isnull
3386 ),
3387 self.else_.type if self.else_ is not None else type_api.NULLTYPE,
3388 )
3389 self.type = cast(_T, type_)
3391 @util.ro_non_memoized_property
3392 def _from_objects(self) -> List[FromClause]:
3393 return list(
3394 itertools.chain(*[x._from_objects for x in self.get_children()])
3395 )
3398class Cast(WrapsColumnExpression[_T]):
3399 """Represent a ``CAST`` expression.
3401 :class:`.Cast` is produced using the :func:`.cast` factory function,
3402 as in::
3404 from sqlalchemy import cast, Numeric
3406 stmt = select(cast(product_table.c.unit_price, Numeric(10, 4)))
3408 Details on :class:`.Cast` usage is at :func:`.cast`.
3410 .. seealso::
3412 :ref:`tutorial_casts`
3414 :func:`.cast`
3416 :func:`.try_cast`
3418 :func:`.type_coerce` - an alternative to CAST that coerces the type
3419 on the Python side only, which is often sufficient to generate the
3420 correct SQL and data coercion.
3422 """
3424 __visit_name__ = "cast"
3426 _traverse_internals: _TraverseInternalsType = [
3427 ("clause", InternalTraversal.dp_clauseelement),
3428 ("type", InternalTraversal.dp_type),
3429 ]
3431 clause: ColumnElement[Any]
3432 type: TypeEngine[_T]
3433 typeclause: TypeClause
3435 def __init__(
3436 self,
3437 expression: _ColumnExpressionArgument[Any],
3438 type_: _TypeEngineArgument[_T],
3439 ):
3440 self.type = type_api.to_instance(type_)
3441 self.clause = coercions.expect(
3442 roles.ExpressionElementRole,
3443 expression,
3444 type_=self.type,
3445 apply_propagate_attrs=self,
3446 )
3447 self.typeclause = TypeClause(self.type)
3449 @util.ro_non_memoized_property
3450 def _from_objects(self) -> List[FromClause]:
3451 return self.clause._from_objects
3453 @property
3454 def wrapped_column_expression(self):
3455 return self.clause
3458class TryCast(Cast[_T]):
3459 """Represent a TRY_CAST expression.
3461 Details on :class:`.TryCast` usage is at :func:`.try_cast`.
3463 .. seealso::
3465 :func:`.try_cast`
3467 :ref:`tutorial_casts`
3468 """
3470 __visit_name__ = "try_cast"
3471 inherit_cache = True
3474class TypeCoerce(WrapsColumnExpression[_T]):
3475 """Represent a Python-side type-coercion wrapper.
3477 :class:`.TypeCoerce` supplies the :func:`_expression.type_coerce`
3478 function; see that function for usage details.
3480 .. seealso::
3482 :func:`_expression.type_coerce`
3484 :func:`.cast`
3486 """
3488 __visit_name__ = "type_coerce"
3490 _traverse_internals: _TraverseInternalsType = [
3491 ("clause", InternalTraversal.dp_clauseelement),
3492 ("type", InternalTraversal.dp_type),
3493 ]
3495 clause: ColumnElement[Any]
3496 type: TypeEngine[_T]
3498 def __init__(
3499 self,
3500 expression: _ColumnExpressionArgument[Any],
3501 type_: _TypeEngineArgument[_T],
3502 ):
3503 self.type = type_api.to_instance(type_)
3504 self.clause = coercions.expect(
3505 roles.ExpressionElementRole,
3506 expression,
3507 type_=self.type,
3508 apply_propagate_attrs=self,
3509 )
3511 @util.ro_non_memoized_property
3512 def _from_objects(self) -> List[FromClause]:
3513 return self.clause._from_objects
3515 @HasMemoized.memoized_attribute
3516 def typed_expression(self):
3517 if isinstance(self.clause, BindParameter):
3518 bp = self.clause._clone()
3519 bp.type = self.type
3520 return bp
3521 else:
3522 return self.clause
3524 @property
3525 def wrapped_column_expression(self):
3526 return self.clause
3528 def self_group(
3529 self, against: Optional[OperatorType] = None
3530 ) -> TypeCoerce[_T]:
3531 grouped = self.clause.self_group(against=against)
3532 if grouped is not self.clause:
3533 return TypeCoerce(grouped, self.type)
3534 else:
3535 return self
3538class Extract(ColumnElement[int]):
3539 """Represent a SQL EXTRACT clause, ``extract(field FROM expr)``."""
3541 __visit_name__ = "extract"
3543 _traverse_internals: _TraverseInternalsType = [
3544 ("expr", InternalTraversal.dp_clauseelement),
3545 ("field", InternalTraversal.dp_string),
3546 ]
3548 expr: ColumnElement[Any]
3549 field: str
3551 def __init__(self, field: str, expr: _ColumnExpressionArgument[Any]):
3552 self.type = type_api.INTEGERTYPE
3553 self.field = field
3554 self.expr = coercions.expect(roles.ExpressionElementRole, expr)
3556 @util.ro_non_memoized_property
3557 def _from_objects(self) -> List[FromClause]:
3558 return self.expr._from_objects
3561class _label_reference(ColumnElement[_T]):
3562 """Wrap a column expression as it appears in a 'reference' context.
3564 This expression is any that includes an _order_by_label_element,
3565 which is a Label, or a DESC / ASC construct wrapping a Label.
3567 The production of _label_reference() should occur when an expression
3568 is added to this context; this includes the ORDER BY or GROUP BY of a
3569 SELECT statement, as well as a few other places, such as the ORDER BY
3570 within an OVER clause.
3572 """
3574 __visit_name__ = "label_reference"
3576 _traverse_internals: _TraverseInternalsType = [
3577 ("element", InternalTraversal.dp_clauseelement)
3578 ]
3580 element: ColumnElement[_T]
3582 def __init__(self, element: ColumnElement[_T]):
3583 self.element = element
3585 @util.ro_non_memoized_property
3586 def _from_objects(self) -> List[FromClause]:
3587 return []
3590class _textual_label_reference(ColumnElement[Any]):
3591 __visit_name__ = "textual_label_reference"
3593 _traverse_internals: _TraverseInternalsType = [
3594 ("element", InternalTraversal.dp_string)
3595 ]
3597 def __init__(self, element: str):
3598 self.element = element
3600 @util.memoized_property
3601 def _text_clause(self) -> TextClause:
3602 return TextClause(self.element)
3605class UnaryExpression(ColumnElement[_T]):
3606 """Define a 'unary' expression.
3608 A unary expression has a single column expression
3609 and an operator. The operator can be placed on the left
3610 (where it is called the 'operator') or right (where it is called the
3611 'modifier') of the column expression.
3613 :class:`.UnaryExpression` is the basis for several unary operators
3614 including those used by :func:`.desc`, :func:`.asc`, :func:`.distinct`,
3615 :func:`.nulls_first` and :func:`.nulls_last`.
3617 """
3619 __visit_name__ = "unary"
3621 _traverse_internals: _TraverseInternalsType = [
3622 ("element", InternalTraversal.dp_clauseelement),
3623 ("operator", InternalTraversal.dp_operator),
3624 ("modifier", InternalTraversal.dp_operator),
3625 ]
3627 element: ColumnElement[Any]
3628 operator: Optional[OperatorType]
3629 modifier: Optional[OperatorType]
3631 def __init__(
3632 self,
3633 element: ColumnElement[Any],
3634 *,
3635 operator: Optional[OperatorType] = None,
3636 modifier: Optional[OperatorType] = None,
3637 type_: Optional[_TypeEngineArgument[_T]] = None,
3638 wraps_column_expression: bool = False, # legacy, not used as of 2.0.42
3639 ):
3640 self.operator = operator
3641 self.modifier = modifier
3642 self._propagate_attrs = element._propagate_attrs
3643 self.element = element.self_group(
3644 against=self.operator or self.modifier
3645 )
3647 # if type is None, we get NULLTYPE, which is our _T. But I don't
3648 # know how to get the overloads to express that correctly
3649 self.type = type_api.to_instance(type_) # type: ignore
3651 def _wraps_unnamed_column(self):
3652 ungrouped = self.element._ungroup()
3653 return (
3654 not isinstance(ungrouped, NamedColumn)
3655 or ungrouped._non_anon_label is None
3656 )
3658 @classmethod
3659 def _create_nulls_first(
3660 cls,
3661 column: _ColumnExpressionArgument[_T],
3662 ) -> UnaryExpression[_T]:
3663 return UnaryExpression(
3664 coercions.expect(roles.ByOfRole, column),
3665 modifier=operators.nulls_first_op,
3666 )
3668 @classmethod
3669 def _create_nulls_last(
3670 cls,
3671 column: _ColumnExpressionArgument[_T],
3672 ) -> UnaryExpression[_T]:
3673 return UnaryExpression(
3674 coercions.expect(roles.ByOfRole, column),
3675 modifier=operators.nulls_last_op,
3676 )
3678 @classmethod
3679 def _create_desc(
3680 cls, column: _ColumnExpressionOrStrLabelArgument[_T]
3681 ) -> UnaryExpression[_T]:
3682 return UnaryExpression(
3683 coercions.expect(roles.ByOfRole, column),
3684 modifier=operators.desc_op,
3685 )
3687 @classmethod
3688 def _create_asc(
3689 cls,
3690 column: _ColumnExpressionOrStrLabelArgument[_T],
3691 ) -> UnaryExpression[_T]:
3692 return UnaryExpression(
3693 coercions.expect(roles.ByOfRole, column),
3694 modifier=operators.asc_op,
3695 )
3697 @classmethod
3698 def _create_distinct(
3699 cls,
3700 expr: _ColumnExpressionArgument[_T],
3701 ) -> UnaryExpression[_T]:
3702 col_expr: ColumnElement[_T] = coercions.expect(
3703 roles.ExpressionElementRole, expr
3704 )
3705 return UnaryExpression(
3706 col_expr,
3707 operator=operators.distinct_op,
3708 type_=col_expr.type,
3709 )
3711 @classmethod
3712 def _create_bitwise_not(
3713 cls,
3714 expr: _ColumnExpressionArgument[_T],
3715 ) -> UnaryExpression[_T]:
3716 col_expr: ColumnElement[_T] = coercions.expect(
3717 roles.ExpressionElementRole, expr
3718 )
3719 return UnaryExpression(
3720 col_expr,
3721 operator=operators.bitwise_not_op,
3722 type_=col_expr.type,
3723 )
3725 @property
3726 def _order_by_label_element(self) -> Optional[Label[Any]]:
3727 if operators.is_order_by_modifier(self.modifier):
3728 return self.element._order_by_label_element
3729 else:
3730 return None
3732 @util.ro_non_memoized_property
3733 def _from_objects(self) -> List[FromClause]:
3734 return self.element._from_objects
3736 def _negate(self) -> ColumnElement[Any]:
3737 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
3738 return UnaryExpression(
3739 self.self_group(against=operators.inv),
3740 operator=operators.inv,
3741 type_=type_api.BOOLEANTYPE,
3742 )
3743 else:
3744 return ColumnElement._negate(self)
3746 def self_group(
3747 self, against: Optional[OperatorType] = None
3748 ) -> Union[Self, Grouping[_T]]:
3749 if self.operator and operators.is_precedent(self.operator, against):
3750 return Grouping(self)
3751 else:
3752 return self
3755class CollectionAggregate(UnaryExpression[_T]):
3756 """Forms the basis for right-hand collection operator modifiers
3757 ANY and ALL.
3759 The ANY and ALL keywords are available in different ways on different
3760 backends. On PostgreSQL, they only work for an ARRAY type. On
3761 MySQL, they only work for subqueries.
3763 """
3765 inherit_cache = True
3766 _is_collection_aggregate = True
3768 @classmethod
3769 def _create_any(
3770 cls, expr: _ColumnExpressionArgument[_T]
3771 ) -> CollectionAggregate[bool]:
3772 """create CollectionAggregate for the legacy
3773 ARRAY.Comparator.any() method"""
3774 col_expr: ColumnElement[_T] = coercions.expect(
3775 roles.ExpressionElementRole,
3776 expr,
3777 )
3778 col_expr = col_expr.self_group()
3779 return CollectionAggregate(
3780 col_expr,
3781 operator=operators.any_op,
3782 type_=type_api.BOOLEANTYPE,
3783 )
3785 @classmethod
3786 def _create_all(
3787 cls, expr: _ColumnExpressionArgument[_T]
3788 ) -> CollectionAggregate[bool]:
3789 """create CollectionAggregate for the legacy
3790 ARRAY.Comparator.all() method"""
3791 col_expr: ColumnElement[_T] = coercions.expect(
3792 roles.ExpressionElementRole,
3793 expr,
3794 )
3795 col_expr = col_expr.self_group()
3796 return CollectionAggregate(
3797 col_expr,
3798 operator=operators.all_op,
3799 type_=type_api.BOOLEANTYPE,
3800 )
3802 @util.preload_module("sqlalchemy.sql.sqltypes")
3803 def _bind_param(
3804 self,
3805 operator: operators.OperatorType,
3806 obj: Any,
3807 type_: Optional[TypeEngine[_T]] = None,
3808 expanding: bool = False,
3809 ) -> BindParameter[_T]:
3810 """For new style any_(), all_(), ensure compared literal value
3811 receives appropriate bound parameter type."""
3813 # a CollectionAggregate is specific to ARRAY or int
3814 # only. So for ARRAY case, make sure we use correct element type
3815 sqltypes = util.preloaded.sql_sqltypes
3816 if self.element.type._type_affinity is sqltypes.ARRAY:
3817 compared_to_type = cast(
3818 sqltypes.ARRAY[Any], self.element.type
3819 ).item_type
3820 else:
3821 compared_to_type = self.element.type
3823 return BindParameter(
3824 None,
3825 obj,
3826 _compared_to_operator=operator,
3827 type_=type_,
3828 _compared_to_type=compared_to_type,
3829 unique=True,
3830 expanding=expanding,
3831 )
3833 # operate and reverse_operate are hardwired to
3834 # dispatch onto the type comparator directly, so that we can
3835 # ensure "reversed" behavior.
3836 def operate(
3837 self, op: OperatorType, *other: Any, **kwargs: Any
3838 ) -> ColumnElement[_T]:
3839 if not operators.is_comparison(op):
3840 raise exc.ArgumentError(
3841 "Only comparison operators may be used with ANY/ALL"
3842 )
3843 kwargs["reverse"] = True
3844 return self.comparator.operate(operators.mirror(op), *other, **kwargs)
3846 def reverse_operate(
3847 self, op: OperatorType, other: Any, **kwargs: Any
3848 ) -> ColumnElement[_T]:
3849 # comparison operators should never call reverse_operate
3850 assert not operators.is_comparison(op)
3851 raise exc.ArgumentError(
3852 "Only comparison operators may be used with ANY/ALL"
3853 )
3856class AsBoolean(WrapsColumnExpression[bool], UnaryExpression[bool]):
3857 inherit_cache = True
3859 def __init__(self, element, operator, negate):
3860 self.element = element
3861 self.type = type_api.BOOLEANTYPE
3862 self.operator = operator
3863 self.negate = negate
3864 self.modifier = None
3865 self._is_implicitly_boolean = element._is_implicitly_boolean
3867 @property
3868 def wrapped_column_expression(self):
3869 return self.element
3871 def self_group(self, against: Optional[OperatorType] = None) -> Self:
3872 return self
3874 def _negate(self):
3875 if isinstance(self.element, (True_, False_)):
3876 return self.element._negate()
3877 else:
3878 return AsBoolean(self.element, self.negate, self.operator)
3881class BinaryExpression(OperatorExpression[_T]):
3882 """Represent an expression that is ``LEFT <operator> RIGHT``.
3884 A :class:`.BinaryExpression` is generated automatically
3885 whenever two column expressions are used in a Python binary expression:
3887 .. sourcecode:: pycon+sql
3889 >>> from sqlalchemy.sql import column
3890 >>> column("a") + column("b")
3891 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
3892 >>> print(column("a") + column("b"))
3893 {printsql}a + b
3895 """
3897 __visit_name__ = "binary"
3899 _traverse_internals: _TraverseInternalsType = [
3900 ("left", InternalTraversal.dp_clauseelement),
3901 ("right", InternalTraversal.dp_clauseelement),
3902 ("operator", InternalTraversal.dp_operator),
3903 ("negate", InternalTraversal.dp_operator),
3904 ("modifiers", InternalTraversal.dp_plain_dict),
3905 (
3906 "type",
3907 InternalTraversal.dp_type,
3908 ),
3909 ]
3911 _cache_key_traversal = [
3912 ("left", InternalTraversal.dp_clauseelement),
3913 ("right", InternalTraversal.dp_clauseelement),
3914 ("operator", InternalTraversal.dp_operator),
3915 ("modifiers", InternalTraversal.dp_plain_dict),
3916 # "type" affects JSON CAST operators, so while redundant in most cases,
3917 # is needed for that one
3918 (
3919 "type",
3920 InternalTraversal.dp_type,
3921 ),
3922 ]
3924 _is_implicitly_boolean = True
3925 """Indicates that any database will know this is a boolean expression
3926 even if the database does not have an explicit boolean datatype.
3928 """
3930 left: ColumnElement[Any]
3931 right: ColumnElement[Any]
3932 modifiers: Mapping[str, Any]
3934 def __init__(
3935 self,
3936 left: ColumnElement[Any],
3937 right: ColumnElement[Any],
3938 operator: OperatorType,
3939 type_: Optional[_TypeEngineArgument[_T]] = None,
3940 negate: Optional[OperatorType] = None,
3941 modifiers: Optional[Mapping[str, Any]] = None,
3942 ):
3943 # allow compatibility with libraries that
3944 # refer to BinaryExpression directly and pass strings
3945 if isinstance(operator, str):
3946 operator = operators.custom_op(operator)
3947 self._orig = (left.__hash__(), right.__hash__())
3948 self._propagate_attrs = left._propagate_attrs or right._propagate_attrs
3949 self.left = left.self_group(against=operator)
3950 self.right = right.self_group(against=operator)
3951 self.operator = operator
3953 # if type is None, we get NULLTYPE, which is our _T. But I don't
3954 # know how to get the overloads to express that correctly
3955 self.type = type_api.to_instance(type_) # type: ignore
3957 self.negate = negate
3958 self._is_implicitly_boolean = operators.is_boolean(operator)
3960 if modifiers is None:
3961 self.modifiers = {}
3962 else:
3963 self.modifiers = modifiers
3965 @property
3966 def _flattened_operator_clauses(
3967 self,
3968 ) -> typing_Tuple[ColumnElement[Any], ...]:
3969 return (self.left, self.right)
3971 def __bool__(self):
3972 """Implement Python-side "bool" for BinaryExpression as a
3973 simple "identity" check for the left and right attributes,
3974 if the operator is "eq" or "ne". Otherwise the expression
3975 continues to not support "bool" like all other column expressions.
3977 The rationale here is so that ColumnElement objects can be hashable.
3978 What? Well, suppose you do this::
3980 c1, c2 = column("x"), column("y")
3981 s1 = set([c1, c2])
3983 We do that **a lot**, columns inside of sets is an extremely basic
3984 thing all over the ORM for example.
3986 So what happens if we do this? ::
3988 c1 in s1
3990 Hashing means it will normally use ``__hash__()`` of the object,
3991 but in case of hash collision, it's going to also do ``c1 == c1``
3992 and/or ``c1 == c2`` inside. Those operations need to return a
3993 True/False value. But because we override ``==`` and ``!=``, they're
3994 going to get a BinaryExpression. Hence we implement ``__bool__`` here
3995 so that these comparisons behave in this particular context mostly
3996 like regular object comparisons. Thankfully Python is OK with
3997 that! Otherwise we'd have to use special set classes for columns
3998 (which we used to do, decades ago).
4000 """
4001 if self.operator in (operators.eq, operators.ne):
4002 # this is using the eq/ne operator given int hash values,
4003 # rather than Operator, so that "bool" can be based on
4004 # identity
4005 return self.operator(*self._orig) # type: ignore
4006 else:
4007 raise TypeError("Boolean value of this clause is not defined")
4009 if typing.TYPE_CHECKING:
4011 def __invert__(
4012 self: BinaryExpression[_T],
4013 ) -> BinaryExpression[_T]: ...
4015 @util.ro_non_memoized_property
4016 def _from_objects(self) -> List[FromClause]:
4017 return self.left._from_objects + self.right._from_objects
4019 def _negate(self):
4020 if self.negate is not None:
4021 return BinaryExpression(
4022 self.left,
4023 self.right._negate_in_binary(self.negate, self.operator),
4024 self.negate,
4025 negate=self.operator,
4026 type_=self.type,
4027 modifiers=self.modifiers,
4028 )
4029 else:
4030 return self.self_group()._negate()
4033class Slice(ColumnElement[Any]):
4034 """Represent SQL for a Python array-slice object.
4036 This is not a specific SQL construct at this level, but
4037 may be interpreted by specific dialects, e.g. PostgreSQL.
4039 """
4041 __visit_name__ = "slice"
4043 _traverse_internals: _TraverseInternalsType = [
4044 ("start", InternalTraversal.dp_clauseelement),
4045 ("stop", InternalTraversal.dp_clauseelement),
4046 ("step", InternalTraversal.dp_clauseelement),
4047 ]
4049 def __init__(self, start, stop, step, _name=None):
4050 self.start = coercions.expect(
4051 roles.ExpressionElementRole,
4052 start,
4053 name=_name,
4054 type_=type_api.INTEGERTYPE,
4055 )
4056 self.stop = coercions.expect(
4057 roles.ExpressionElementRole,
4058 stop,
4059 name=_name,
4060 type_=type_api.INTEGERTYPE,
4061 )
4062 self.step = coercions.expect(
4063 roles.ExpressionElementRole,
4064 step,
4065 name=_name,
4066 type_=type_api.INTEGERTYPE,
4067 )
4068 self.type = type_api.NULLTYPE
4070 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4071 assert against is operator.getitem
4072 return self
4075class IndexExpression(BinaryExpression[Any]):
4076 """Represent the class of expressions that are like an "index"
4077 operation."""
4079 inherit_cache = True
4082class GroupedElement(DQLDMLClauseElement):
4083 """Represent any parenthesized expression"""
4085 __visit_name__ = "grouping"
4087 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4088 return self
4090 def _ungroup(self) -> ClauseElement:
4091 raise NotImplementedError()
4094class Grouping(GroupedElement, ColumnElement[_T]):
4095 """Represent a grouping within a column expression"""
4097 _traverse_internals: _TraverseInternalsType = [
4098 ("element", InternalTraversal.dp_clauseelement),
4099 ("type", InternalTraversal.dp_type),
4100 ]
4102 _cache_key_traversal = [
4103 ("element", InternalTraversal.dp_clauseelement),
4104 ]
4106 element: Union[TextClause, ClauseList, ColumnElement[_T]]
4108 def __init__(
4109 self, element: Union[TextClause, ClauseList, ColumnElement[_T]]
4110 ):
4111 self.element = element
4113 # nulltype assignment issue
4114 self.type = getattr(element, "type", type_api.NULLTYPE) # type: ignore
4115 self._propagate_attrs = element._propagate_attrs
4117 def _with_binary_element_type(self, type_):
4118 return self.__class__(self.element._with_binary_element_type(type_))
4120 def _ungroup(self) -> ColumnElement[_T]:
4121 assert isinstance(self.element, ColumnElement)
4122 return self.element._ungroup()
4124 @util.memoized_property
4125 def _is_implicitly_boolean(self):
4126 return self.element._is_implicitly_boolean
4128 @util.non_memoized_property
4129 def _tq_label(self) -> Optional[str]:
4130 return (
4131 getattr(self.element, "_tq_label", None) or self._anon_name_label
4132 )
4134 @util.non_memoized_property
4135 def _proxies(self) -> List[ColumnElement[Any]]:
4136 if isinstance(self.element, ColumnElement):
4137 return [self.element]
4138 else:
4139 return []
4141 @util.ro_non_memoized_property
4142 def _from_objects(self) -> List[FromClause]:
4143 return self.element._from_objects
4145 def __getattr__(self, attr):
4146 return getattr(self.element, attr)
4148 def __getstate__(self):
4149 return {"element": self.element, "type": self.type}
4151 def __setstate__(self, state):
4152 self.element = state["element"]
4153 self.type = state["type"]
4155 if TYPE_CHECKING:
4157 def self_group(
4158 self, against: Optional[OperatorType] = None
4159 ) -> Self: ...
4162class _OverrideBinds(Grouping[_T]):
4163 """used by cache_key->_apply_params_to_element to allow compilation /
4164 execution of a SQL element that's been cached, using an alternate set of
4165 bound parameter values.
4167 This is used by the ORM to swap new parameter values into expressions
4168 that are embedded into loader options like with_expression(),
4169 selectinload(). Previously, this task was accomplished using the
4170 .params() method which would perform a deep-copy instead. This deep
4171 copy proved to be too expensive for more complex expressions.
4173 See #11085
4175 """
4177 __visit_name__ = "override_binds"
4179 def __init__(
4180 self,
4181 element: ColumnElement[_T],
4182 bindparams: Sequence[BindParameter[Any]],
4183 replaces_params: Sequence[BindParameter[Any]],
4184 ):
4185 self.element = element
4186 self.translate = {
4187 k.key: v.value for k, v in zip(replaces_params, bindparams)
4188 }
4190 def _gen_cache_key(
4191 self, anon_map: anon_map, bindparams: List[BindParameter[Any]]
4192 ) -> Optional[typing_Tuple[Any, ...]]:
4193 """generate a cache key for the given element, substituting its bind
4194 values for the translation values present."""
4196 existing_bps: List[BindParameter[Any]] = []
4197 ck = self.element._gen_cache_key(anon_map, existing_bps)
4199 bindparams.extend(
4200 (
4201 bp._with_value(
4202 self.translate[bp.key], maintain_key=True, required=False
4203 )
4204 if bp.key in self.translate
4205 else bp
4206 )
4207 for bp in existing_bps
4208 )
4210 return ck
4213class _OverRange(Enum):
4214 RANGE_UNBOUNDED = 0
4215 RANGE_CURRENT = 1
4218RANGE_UNBOUNDED = _OverRange.RANGE_UNBOUNDED
4219RANGE_CURRENT = _OverRange.RANGE_CURRENT
4221_IntOrRange = Union[int, _OverRange]
4224class Over(ColumnElement[_T]):
4225 """Represent an OVER clause.
4227 This is a special operator against a so-called
4228 "window" function, as well as any aggregate function,
4229 which produces results relative to the result set
4230 itself. Most modern SQL backends now support window functions.
4232 """
4234 __visit_name__ = "over"
4236 _traverse_internals: _TraverseInternalsType = [
4237 ("element", InternalTraversal.dp_clauseelement),
4238 ("order_by", InternalTraversal.dp_clauseelement),
4239 ("partition_by", InternalTraversal.dp_clauseelement),
4240 ("range_", InternalTraversal.dp_plain_obj),
4241 ("rows", InternalTraversal.dp_plain_obj),
4242 ("groups", InternalTraversal.dp_plain_obj),
4243 ]
4245 order_by: Optional[ClauseList] = None
4246 partition_by: Optional[ClauseList] = None
4248 element: ColumnElement[_T]
4249 """The underlying expression object to which this :class:`.Over`
4250 object refers."""
4252 range_: Optional[typing_Tuple[_IntOrRange, _IntOrRange]]
4253 rows: Optional[typing_Tuple[_IntOrRange, _IntOrRange]]
4254 groups: Optional[typing_Tuple[_IntOrRange, _IntOrRange]]
4256 def __init__(
4257 self,
4258 element: ColumnElement[_T],
4259 partition_by: Optional[_ByArgument] = None,
4260 order_by: Optional[_ByArgument] = None,
4261 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4262 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4263 groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4264 ):
4265 self.element = element
4266 if order_by is not None:
4267 self.order_by = ClauseList(
4268 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4269 )
4270 if partition_by is not None:
4271 self.partition_by = ClauseList(
4272 *util.to_list(partition_by),
4273 _literal_as_text_role=roles.ByOfRole,
4274 )
4276 if sum(bool(item) for item in (range_, rows, groups)) > 1:
4277 raise exc.ArgumentError(
4278 "only one of 'rows', 'range_', or 'groups' may be provided"
4279 )
4280 else:
4281 self.range_ = self._interpret_range(range_) if range_ else None
4282 self.rows = self._interpret_range(rows) if rows else None
4283 self.groups = self._interpret_range(groups) if groups else None
4285 def __reduce__(self):
4286 return self.__class__, (
4287 self.element,
4288 self.partition_by,
4289 self.order_by,
4290 self.range_,
4291 self.rows,
4292 self.groups,
4293 )
4295 def _interpret_range(
4296 self,
4297 range_: typing_Tuple[Optional[_IntOrRange], Optional[_IntOrRange]],
4298 ) -> typing_Tuple[_IntOrRange, _IntOrRange]:
4299 if not isinstance(range_, tuple) or len(range_) != 2:
4300 raise exc.ArgumentError("2-tuple expected for range/rows")
4302 r0, r1 = range_
4304 lower: _IntOrRange
4305 upper: _IntOrRange
4307 if r0 is None:
4308 lower = RANGE_UNBOUNDED
4309 elif isinstance(r0, _OverRange):
4310 lower = r0
4311 else:
4312 try:
4313 lower = int(r0)
4314 except ValueError as err:
4315 raise exc.ArgumentError(
4316 "Integer or None expected for range value"
4317 ) from err
4318 else:
4319 if lower == 0:
4320 lower = RANGE_CURRENT
4322 if r1 is None:
4323 upper = RANGE_UNBOUNDED
4324 elif isinstance(r1, _OverRange):
4325 upper = r1
4326 else:
4327 try:
4328 upper = int(r1)
4329 except ValueError as err:
4330 raise exc.ArgumentError(
4331 "Integer or None expected for range value"
4332 ) from err
4333 else:
4334 if upper == 0:
4335 upper = RANGE_CURRENT
4337 return lower, upper
4339 if not TYPE_CHECKING:
4341 @util.memoized_property
4342 def type(self) -> TypeEngine[_T]: # noqa: A001
4343 return self.element.type
4345 @util.ro_non_memoized_property
4346 def _from_objects(self) -> List[FromClause]:
4347 return list(
4348 itertools.chain(
4349 *[
4350 c._from_objects
4351 for c in (self.element, self.partition_by, self.order_by)
4352 if c is not None
4353 ]
4354 )
4355 )
4358class WithinGroup(ColumnElement[_T]):
4359 """Represent a WITHIN GROUP (ORDER BY) clause.
4361 This is a special operator against so-called
4362 "ordered set aggregate" and "hypothetical
4363 set aggregate" functions, including ``percentile_cont()``,
4364 ``rank()``, ``dense_rank()``, etc.
4366 It's supported only by certain database backends, such as PostgreSQL,
4367 Oracle Database and MS SQL Server.
4369 The :class:`.WithinGroup` construct extracts its type from the
4370 method :meth:`.FunctionElement.within_group_type`. If this returns
4371 ``None``, the function's ``.type`` is used.
4373 """
4375 __visit_name__ = "withingroup"
4377 _traverse_internals: _TraverseInternalsType = [
4378 ("element", InternalTraversal.dp_clauseelement),
4379 ("order_by", InternalTraversal.dp_clauseelement),
4380 ]
4382 order_by: Optional[ClauseList] = None
4384 def __init__(
4385 self,
4386 element: Union[FunctionElement[_T], FunctionFilter[_T]],
4387 *order_by: _ColumnExpressionArgument[Any],
4388 ):
4389 self.element = element
4390 if order_by is not None:
4391 self.order_by = ClauseList(
4392 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4393 )
4395 def __reduce__(self):
4396 return self.__class__, (self.element,) + (
4397 tuple(self.order_by) if self.order_by is not None else ()
4398 )
4400 def over(
4401 self,
4402 *,
4403 partition_by: Optional[_ByArgument] = None,
4404 order_by: Optional[_ByArgument] = None,
4405 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4406 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4407 groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4408 ) -> Over[_T]:
4409 """Produce an OVER clause against this :class:`.WithinGroup`
4410 construct.
4412 This function has the same signature as that of
4413 :meth:`.FunctionElement.over`.
4415 """
4416 return Over(
4417 self,
4418 partition_by=partition_by,
4419 order_by=order_by,
4420 range_=range_,
4421 rows=rows,
4422 groups=groups,
4423 )
4425 @overload
4426 def filter(self) -> Self: ...
4428 @overload
4429 def filter(
4430 self,
4431 __criterion0: _ColumnExpressionArgument[bool],
4432 *criterion: _ColumnExpressionArgument[bool],
4433 ) -> FunctionFilter[_T]: ...
4435 def filter(
4436 self, *criterion: _ColumnExpressionArgument[bool]
4437 ) -> Union[Self, FunctionFilter[_T]]:
4438 """Produce a FILTER clause against this function."""
4439 if not criterion:
4440 return self
4441 return FunctionFilter(self, *criterion)
4443 if not TYPE_CHECKING:
4445 @util.memoized_property
4446 def type(self) -> TypeEngine[_T]: # noqa: A001
4447 wgt = self.element.within_group_type(self)
4448 if wgt is not None:
4449 return wgt
4450 else:
4451 return self.element.type
4453 @util.ro_non_memoized_property
4454 def _from_objects(self) -> List[FromClause]:
4455 return list(
4456 itertools.chain(
4457 *[
4458 c._from_objects
4459 for c in (self.element, self.order_by)
4460 if c is not None
4461 ]
4462 )
4463 )
4466class FunctionFilter(Generative, ColumnElement[_T]):
4467 """Represent a function FILTER clause.
4469 This is a special operator against aggregate and window functions,
4470 which controls which rows are passed to it.
4471 It's supported only by certain database backends.
4473 Invocation of :class:`.FunctionFilter` is via
4474 :meth:`.FunctionElement.filter`::
4476 func.count(1).filter(True)
4478 .. seealso::
4480 :meth:`.FunctionElement.filter`
4482 """
4484 __visit_name__ = "funcfilter"
4486 _traverse_internals: _TraverseInternalsType = [
4487 ("func", InternalTraversal.dp_clauseelement),
4488 ("criterion", InternalTraversal.dp_clauseelement),
4489 ]
4491 criterion: Optional[ColumnElement[bool]] = None
4493 def __init__(
4494 self,
4495 func: Union[FunctionElement[_T], WithinGroup[_T]],
4496 *criterion: _ColumnExpressionArgument[bool],
4497 ):
4498 self.func = func
4499 self.filter.non_generative(self, *criterion) # type: ignore
4501 @_generative
4502 def filter(self, *criterion: _ColumnExpressionArgument[bool]) -> Self:
4503 """Produce an additional FILTER against the function.
4505 This method adds additional criteria to the initial criteria
4506 set up by :meth:`.FunctionElement.filter`.
4508 Multiple criteria are joined together at SQL render time
4509 via ``AND``.
4512 """
4514 for crit in list(criterion):
4515 crit = coercions.expect(roles.WhereHavingRole, crit)
4517 if self.criterion is not None:
4518 self.criterion = self.criterion & crit
4519 else:
4520 self.criterion = crit
4522 return self
4524 def over(
4525 self,
4526 partition_by: Optional[
4527 Union[
4528 Iterable[_ColumnExpressionArgument[Any]],
4529 _ColumnExpressionArgument[Any],
4530 ]
4531 ] = None,
4532 order_by: Optional[
4533 Union[
4534 Iterable[_ColumnExpressionArgument[Any]],
4535 _ColumnExpressionArgument[Any],
4536 ]
4537 ] = None,
4538 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4539 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4540 groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4541 ) -> Over[_T]:
4542 """Produce an OVER clause against this filtered function.
4544 Used against aggregate or so-called "window" functions,
4545 for database backends that support window functions.
4547 The expression::
4549 func.rank().filter(MyClass.y > 5).over(order_by="x")
4551 is shorthand for::
4553 from sqlalchemy import over, funcfilter
4555 over(funcfilter(func.rank(), MyClass.y > 5), order_by="x")
4557 See :func:`_expression.over` for a full description.
4559 """
4560 return Over(
4561 self,
4562 partition_by=partition_by,
4563 order_by=order_by,
4564 range_=range_,
4565 rows=rows,
4566 groups=groups,
4567 )
4569 def within_group(
4570 self, *order_by: _ColumnExpressionArgument[Any]
4571 ) -> WithinGroup[_T]:
4572 """Produce a WITHIN GROUP (ORDER BY expr) clause against
4573 this function.
4574 """
4575 return WithinGroup(self, *order_by)
4577 def within_group_type(
4578 self, within_group: WithinGroup[_T]
4579 ) -> Optional[TypeEngine[_T]]:
4580 return None
4582 def self_group(
4583 self, against: Optional[OperatorType] = None
4584 ) -> Union[Self, Grouping[_T]]:
4585 if operators.is_precedent(operators.filter_op, against):
4586 return Grouping(self)
4587 else:
4588 return self
4590 if not TYPE_CHECKING:
4592 @util.memoized_property
4593 def type(self) -> TypeEngine[_T]: # noqa: A001
4594 return self.func.type
4596 @util.ro_non_memoized_property
4597 def _from_objects(self) -> List[FromClause]:
4598 return list(
4599 itertools.chain(
4600 *[
4601 c._from_objects
4602 for c in (self.func, self.criterion)
4603 if c is not None
4604 ]
4605 )
4606 )
4609class NamedColumn(KeyedColumnElement[_T]):
4610 is_literal = False
4611 table: Optional[FromClause] = None
4612 name: str
4613 key: str
4615 def _compare_name_for_result(self, other):
4616 return (hasattr(other, "name") and self.name == other.name) or (
4617 hasattr(other, "_label") and self._label == other._label
4618 )
4620 @util.ro_memoized_property
4621 def description(self) -> str:
4622 return self.name
4624 @HasMemoized.memoized_attribute
4625 def _tq_key_label(self) -> Optional[str]:
4626 """table qualified label based on column key.
4628 for table-bound columns this is <tablename>_<column key/proxy key>;
4630 all other expressions it resolves to key/proxy key.
4632 """
4633 proxy_key = self._proxy_key
4634 if proxy_key and proxy_key != self.name:
4635 return self._gen_tq_label(proxy_key)
4636 else:
4637 return self._tq_label
4639 @HasMemoized.memoized_attribute
4640 def _tq_label(self) -> Optional[str]:
4641 """table qualified label based on column name.
4643 for table-bound columns this is <tablename>_<columnname>; all other
4644 expressions it resolves to .name.
4646 """
4647 return self._gen_tq_label(self.name)
4649 @HasMemoized.memoized_attribute
4650 def _render_label_in_columns_clause(self):
4651 return True
4653 @HasMemoized.memoized_attribute
4654 def _non_anon_label(self):
4655 return self.name
4657 def _gen_tq_label(
4658 self, name: str, dedupe_on_key: bool = True
4659 ) -> Optional[str]:
4660 return name
4662 def _bind_param(
4663 self,
4664 operator: OperatorType,
4665 obj: Any,
4666 type_: Optional[TypeEngine[_T]] = None,
4667 expanding: bool = False,
4668 ) -> BindParameter[_T]:
4669 return BindParameter(
4670 self.key,
4671 obj,
4672 _compared_to_operator=operator,
4673 _compared_to_type=self.type,
4674 type_=type_,
4675 unique=True,
4676 expanding=expanding,
4677 )
4679 def _make_proxy(
4680 self,
4681 selectable: FromClause,
4682 *,
4683 primary_key: ColumnSet,
4684 foreign_keys: Set[KeyedColumnElement[Any]],
4685 name: Optional[str] = None,
4686 key: Optional[str] = None,
4687 name_is_truncatable: bool = False,
4688 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
4689 disallow_is_literal: bool = False,
4690 **kw: Any,
4691 ) -> typing_Tuple[str, ColumnClause[_T]]:
4692 c = ColumnClause(
4693 (
4694 coercions.expect(roles.TruncatedLabelRole, name or self.name)
4695 if name_is_truncatable
4696 else (name or self.name)
4697 ),
4698 type_=self.type,
4699 _selectable=selectable,
4700 is_literal=False,
4701 )
4703 c._propagate_attrs = selectable._propagate_attrs
4704 if name is None:
4705 c.key = self.key
4706 if compound_select_cols:
4707 c._proxies = list(compound_select_cols)
4708 else:
4709 c._proxies = [self]
4711 if selectable._is_clone_of is not None:
4712 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
4713 return c.key, c
4716_PS = ParamSpec("_PS")
4719class Label(roles.LabeledColumnExprRole[_T], NamedColumn[_T]):
4720 """Represents a column label (AS).
4722 Represent a label, as typically applied to any column-level
4723 element using the ``AS`` sql keyword.
4725 """
4727 __visit_name__ = "label"
4729 _traverse_internals: _TraverseInternalsType = [
4730 ("name", InternalTraversal.dp_anon_name),
4731 ("type", InternalTraversal.dp_type),
4732 ("_element", InternalTraversal.dp_clauseelement),
4733 ]
4735 _cache_key_traversal = [
4736 ("name", InternalTraversal.dp_anon_name),
4737 ("_element", InternalTraversal.dp_clauseelement),
4738 ]
4740 _element: ColumnElement[_T]
4741 name: str
4743 def __init__(
4744 self,
4745 name: Optional[str],
4746 element: _ColumnExpressionArgument[_T],
4747 type_: Optional[_TypeEngineArgument[_T]] = None,
4748 ):
4749 orig_element = element
4750 element = coercions.expect(
4751 roles.ExpressionElementRole,
4752 element,
4753 apply_propagate_attrs=self,
4754 )
4755 while isinstance(element, Label):
4756 # TODO: this is only covered in test_text.py, but nothing
4757 # fails if it's removed. determine rationale
4758 element = element.element
4760 if name:
4761 self.name = name
4762 else:
4763 self.name = _anonymous_label.safe_construct(
4764 id(self), getattr(element, "name", "anon")
4765 )
4766 if isinstance(orig_element, Label):
4767 # TODO: no coverage for this block, again would be in
4768 # test_text.py where the resolve_label concept is important
4769 self._resolve_label = orig_element._label
4771 self.key = self._tq_label = self._tq_key_label = self.name
4772 self._element = element
4774 self.type = (
4775 type_api.to_instance(type_)
4776 if type_ is not None
4777 else self._element.type
4778 )
4780 self._proxies = [element]
4782 def __reduce__(self):
4783 return self.__class__, (self.name, self._element, self.type)
4785 @HasMemoized.memoized_attribute
4786 def _render_label_in_columns_clause(self):
4787 return True
4789 def _bind_param(self, operator, obj, type_=None, expanding=False):
4790 return BindParameter(
4791 None,
4792 obj,
4793 _compared_to_operator=operator,
4794 type_=type_,
4795 _compared_to_type=self.type,
4796 unique=True,
4797 expanding=expanding,
4798 )
4800 @util.memoized_property
4801 def _is_implicitly_boolean(self):
4802 return self.element._is_implicitly_boolean
4804 @HasMemoized.memoized_attribute
4805 def _allow_label_resolve(self):
4806 return self.element._allow_label_resolve
4808 @property
4809 def _order_by_label_element(self):
4810 return self
4812 @HasMemoized.memoized_attribute
4813 def element(self) -> ColumnElement[_T]:
4814 return self._element.self_group(against=operators.as_)
4816 def self_group(self, against: Optional[OperatorType] = None) -> Label[_T]:
4817 return self._apply_to_inner(self._element.self_group, against=against)
4819 def _negate(self):
4820 return self._apply_to_inner(self._element._negate)
4822 def _apply_to_inner(
4823 self,
4824 fn: Callable[_PS, ColumnElement[_T]],
4825 *arg: _PS.args,
4826 **kw: _PS.kwargs,
4827 ) -> Label[_T]:
4828 sub_element = fn(*arg, **kw)
4829 if sub_element is not self._element:
4830 return Label(self.name, sub_element, type_=self.type)
4831 else:
4832 return self
4834 @property
4835 def primary_key(self): # type: ignore[override]
4836 return self.element.primary_key
4838 @property
4839 def foreign_keys(self): # type: ignore[override]
4840 return self.element.foreign_keys
4842 def _copy_internals(
4843 self,
4844 *,
4845 clone: _CloneCallableType = _clone,
4846 anonymize_labels: bool = False,
4847 **kw: Any,
4848 ) -> None:
4849 self._reset_memoizations()
4850 self._element = clone(self._element, **kw)
4851 if anonymize_labels:
4852 self.name = _anonymous_label.safe_construct(
4853 id(self), getattr(self.element, "name", "anon")
4854 )
4855 self.key = self._tq_label = self._tq_key_label = self.name
4857 @util.ro_non_memoized_property
4858 def _from_objects(self) -> List[FromClause]:
4859 return self.element._from_objects
4861 def _make_proxy(
4862 self,
4863 selectable: FromClause,
4864 *,
4865 primary_key: ColumnSet,
4866 foreign_keys: Set[KeyedColumnElement[Any]],
4867 name: Optional[str] = None,
4868 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
4869 **kw: Any,
4870 ) -> typing_Tuple[str, ColumnClause[_T]]:
4871 name = self.name if not name else name
4873 key, e = self.element._make_proxy(
4874 selectable,
4875 name=name,
4876 disallow_is_literal=True,
4877 name_is_truncatable=isinstance(name, _truncated_label),
4878 compound_select_cols=compound_select_cols,
4879 primary_key=primary_key,
4880 foreign_keys=foreign_keys,
4881 )
4883 # there was a note here to remove this assertion, which was here
4884 # to determine if we later could support a use case where
4885 # the key and name of a label are separate. But I don't know what
4886 # that case was. For now, this is an unexpected case that occurs
4887 # when a label name conflicts with other columns and select()
4888 # is attempting to disambiguate an explicit label, which is not what
4889 # the user would want. See issue #6090.
4890 if key != self.name and not isinstance(self.name, _anonymous_label):
4891 raise exc.InvalidRequestError(
4892 "Label name %s is being renamed to an anonymous label due "
4893 "to disambiguation "
4894 "which is not supported right now. Please use unique names "
4895 "for explicit labels." % (self.name)
4896 )
4898 e._propagate_attrs = selectable._propagate_attrs
4899 e._proxies.append(self)
4900 if self.type is not None:
4901 e.type = self.type
4903 return self.key, e
4906class ColumnClause(
4907 roles.DDLReferredColumnRole,
4908 roles.LabeledColumnExprRole[_T],
4909 roles.StrAsPlainColumnRole,
4910 Immutable,
4911 NamedColumn[_T],
4912):
4913 """Represents a column expression from any textual string.
4915 The :class:`.ColumnClause`, a lightweight analogue to the
4916 :class:`_schema.Column` class, is typically invoked using the
4917 :func:`_expression.column` function, as in::
4919 from sqlalchemy import column
4921 id, name = column("id"), column("name")
4922 stmt = select(id, name).select_from("user")
4924 The above statement would produce SQL like:
4926 .. sourcecode:: sql
4928 SELECT id, name FROM user
4930 :class:`.ColumnClause` is the immediate superclass of the schema-specific
4931 :class:`_schema.Column` object. While the :class:`_schema.Column`
4932 class has all the
4933 same capabilities as :class:`.ColumnClause`, the :class:`.ColumnClause`
4934 class is usable by itself in those cases where behavioral requirements
4935 are limited to simple SQL expression generation. The object has none of
4936 the associations with schema-level metadata or with execution-time
4937 behavior that :class:`_schema.Column` does,
4938 so in that sense is a "lightweight"
4939 version of :class:`_schema.Column`.
4941 Full details on :class:`.ColumnClause` usage is at
4942 :func:`_expression.column`.
4944 .. seealso::
4946 :func:`_expression.column`
4948 :class:`_schema.Column`
4950 """
4952 table: Optional[FromClause]
4953 is_literal: bool
4955 __visit_name__ = "column"
4957 _traverse_internals: _TraverseInternalsType = [
4958 ("name", InternalTraversal.dp_anon_name),
4959 ("type", InternalTraversal.dp_type),
4960 ("table", InternalTraversal.dp_clauseelement),
4961 ("is_literal", InternalTraversal.dp_boolean),
4962 ]
4964 onupdate: Optional[DefaultGenerator] = None
4965 default: Optional[DefaultGenerator] = None
4966 server_default: Optional[FetchedValue] = None
4967 server_onupdate: Optional[FetchedValue] = None
4969 _is_multiparam_column = False
4971 @property
4972 def _is_star(self): # type: ignore[override]
4973 return self.is_literal and self.name == "*"
4975 def __init__(
4976 self,
4977 text: str,
4978 type_: Optional[_TypeEngineArgument[_T]] = None,
4979 is_literal: bool = False,
4980 _selectable: Optional[FromClause] = None,
4981 ):
4982 self.key = self.name = text
4983 self.table = _selectable
4985 # if type is None, we get NULLTYPE, which is our _T. But I don't
4986 # know how to get the overloads to express that correctly
4987 self.type = type_api.to_instance(type_) # type: ignore
4989 self.is_literal = is_literal
4991 def get_children(self, *, column_tables=False, **kw):
4992 # override base get_children() to not return the Table
4993 # or selectable that is parent to this column. Traversals
4994 # expect the columns of tables and subqueries to be leaf nodes.
4995 return []
4997 @property
4998 def entity_namespace(self):
4999 if self.table is not None:
5000 return self.table.entity_namespace
5001 else:
5002 return super().entity_namespace
5004 def _clone(self, detect_subquery_cols=False, **kw):
5005 if (
5006 detect_subquery_cols
5007 and self.table is not None
5008 and self.table._is_subquery
5009 ):
5010 clone = kw.pop("clone")
5011 table = clone(self.table, **kw)
5012 new = table.c.corresponding_column(self)
5013 return new
5015 return super()._clone(**kw)
5017 @HasMemoized_ro_memoized_attribute
5018 def _from_objects(self) -> List[FromClause]:
5019 t = self.table
5020 if t is not None:
5021 return [t]
5022 else:
5023 return []
5025 @HasMemoized.memoized_attribute
5026 def _render_label_in_columns_clause(self):
5027 return self.table is not None
5029 @property
5030 def _ddl_label(self):
5031 return self._gen_tq_label(self.name, dedupe_on_key=False)
5033 def _compare_name_for_result(self, other):
5034 if (
5035 self.is_literal
5036 or self.table is None
5037 or self.table._is_textual
5038 or not hasattr(other, "proxy_set")
5039 or (
5040 isinstance(other, ColumnClause)
5041 and (
5042 other.is_literal
5043 or other.table is None
5044 or other.table._is_textual
5045 )
5046 )
5047 ):
5048 return (hasattr(other, "name") and self.name == other.name) or (
5049 hasattr(other, "_tq_label")
5050 and self._tq_label == other._tq_label
5051 )
5052 else:
5053 return other.proxy_set.intersection(self.proxy_set)
5055 def _gen_tq_label(
5056 self, name: str, dedupe_on_key: bool = True
5057 ) -> Optional[str]:
5058 """generate table-qualified label
5060 for a table-bound column this is <tablename>_<columnname>.
5062 used primarily for LABEL_STYLE_TABLENAME_PLUS_COL
5063 as well as the .columns collection on a Join object.
5065 """
5066 label: str
5067 t = self.table
5068 if self.is_literal:
5069 return None
5070 elif t is not None and is_named_from_clause(t):
5071 if has_schema_attr(t) and t.schema:
5072 label = t.schema.replace(".", "_") + "_" + t.name + "_" + name
5073 else:
5074 assert not TYPE_CHECKING or isinstance(t, NamedFromClause)
5075 label = t.name + "_" + name
5077 # propagate name quoting rules for labels.
5078 if is_quoted_name(name) and name.quote is not None:
5079 if is_quoted_name(label):
5080 label.quote = name.quote
5081 else:
5082 label = quoted_name(label, name.quote)
5083 elif is_quoted_name(t.name) and t.name.quote is not None:
5084 # can't get this situation to occur, so let's
5085 # assert false on it for now
5086 assert not isinstance(label, quoted_name)
5087 label = quoted_name(label, t.name.quote)
5089 if dedupe_on_key:
5090 # ensure the label name doesn't conflict with that of an
5091 # existing column. note that this implies that any Column
5092 # must **not** set up its _label before its parent table has
5093 # all of its other Column objects set up. There are several
5094 # tables in the test suite which will fail otherwise; example:
5095 # table "owner" has columns "name" and "owner_name". Therefore
5096 # column owner.name cannot use the label "owner_name", it has
5097 # to be "owner_name_1".
5098 if label in t.c:
5099 _label = label
5100 counter = 1
5101 while _label in t.c:
5102 _label = label + "_" + str(counter)
5103 counter += 1
5104 label = _label
5106 return coercions.expect(roles.TruncatedLabelRole, label)
5108 else:
5109 return name
5111 def _make_proxy(
5112 self,
5113 selectable: FromClause,
5114 *,
5115 primary_key: ColumnSet,
5116 foreign_keys: Set[KeyedColumnElement[Any]],
5117 name: Optional[str] = None,
5118 key: Optional[str] = None,
5119 name_is_truncatable: bool = False,
5120 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5121 disallow_is_literal: bool = False,
5122 **kw: Any,
5123 ) -> typing_Tuple[str, ColumnClause[_T]]:
5124 # the "is_literal" flag normally should never be propagated; a proxied
5125 # column is always a SQL identifier and never the actual expression
5126 # being evaluated. however, there is a case where the "is_literal" flag
5127 # might be used to allow the given identifier to have a fixed quoting
5128 # pattern already, so maintain the flag for the proxy unless a
5129 # :class:`.Label` object is creating the proxy. See [ticket:4730].
5130 is_literal = (
5131 not disallow_is_literal
5132 and self.is_literal
5133 and (
5134 # note this does not accommodate for quoted_name differences
5135 # right now
5136 name is None
5137 or name == self.name
5138 )
5139 )
5140 c = self._constructor(
5141 (
5142 coercions.expect(roles.TruncatedLabelRole, name or self.name)
5143 if name_is_truncatable
5144 else (name or self.name)
5145 ),
5146 type_=self.type,
5147 _selectable=selectable,
5148 is_literal=is_literal,
5149 )
5150 c._propagate_attrs = selectable._propagate_attrs
5151 if name is None:
5152 c.key = self.key
5153 if compound_select_cols:
5154 c._proxies = list(compound_select_cols)
5155 else:
5156 c._proxies = [self]
5158 if selectable._is_clone_of is not None:
5159 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
5160 return c.key, c
5163class TableValuedColumn(NamedColumn[_T]):
5164 __visit_name__ = "table_valued_column"
5166 _traverse_internals: _TraverseInternalsType = [
5167 ("name", InternalTraversal.dp_anon_name),
5168 ("type", InternalTraversal.dp_type),
5169 ("scalar_alias", InternalTraversal.dp_clauseelement),
5170 ]
5172 def __init__(self, scalar_alias: NamedFromClause, type_: TypeEngine[_T]):
5173 self.scalar_alias = scalar_alias
5174 self.key = self.name = scalar_alias.name
5175 self.type = type_
5177 def _copy_internals(
5178 self, clone: _CloneCallableType = _clone, **kw: Any
5179 ) -> None:
5180 self.scalar_alias = clone(self.scalar_alias, **kw)
5181 self.key = self.name = self.scalar_alias.name
5183 @util.ro_non_memoized_property
5184 def _from_objects(self) -> List[FromClause]:
5185 return [self.scalar_alias]
5188class CollationClause(ColumnElement[str]):
5189 __visit_name__ = "collation"
5191 _traverse_internals: _TraverseInternalsType = [
5192 ("collation", InternalTraversal.dp_string)
5193 ]
5195 @classmethod
5196 @util.preload_module("sqlalchemy.sql.sqltypes")
5197 def _create_collation_expression(
5198 cls, expression: _ColumnExpressionArgument[str], collation: str
5199 ) -> BinaryExpression[str]:
5201 sqltypes = util.preloaded.sql_sqltypes
5203 expr = coercions.expect(roles.ExpressionElementRole[str], expression)
5205 if expr.type._type_affinity is sqltypes.String:
5206 collate_type = expr.type._with_collation(collation)
5207 else:
5208 collate_type = expr.type
5210 return BinaryExpression(
5211 expr,
5212 CollationClause(collation),
5213 operators.collate,
5214 type_=collate_type,
5215 )
5217 def __init__(self, collation):
5218 self.collation = collation
5221class _IdentifiedClause(Executable, ClauseElement):
5222 __visit_name__ = "identified"
5224 def __init__(self, ident):
5225 self.ident = ident
5228class SavepointClause(_IdentifiedClause):
5229 __visit_name__ = "savepoint"
5230 inherit_cache = False
5233class RollbackToSavepointClause(_IdentifiedClause):
5234 __visit_name__ = "rollback_to_savepoint"
5235 inherit_cache = False
5238class ReleaseSavepointClause(_IdentifiedClause):
5239 __visit_name__ = "release_savepoint"
5240 inherit_cache = False
5243class quoted_name(util.MemoizedSlots, str):
5244 """Represent a SQL identifier combined with quoting preferences.
5246 :class:`.quoted_name` is a Python unicode/str subclass which
5247 represents a particular identifier name along with a
5248 ``quote`` flag. This ``quote`` flag, when set to
5249 ``True`` or ``False``, overrides automatic quoting behavior
5250 for this identifier in order to either unconditionally quote
5251 or to not quote the name. If left at its default of ``None``,
5252 quoting behavior is applied to the identifier on a per-backend basis
5253 based on an examination of the token itself.
5255 A :class:`.quoted_name` object with ``quote=True`` is also
5256 prevented from being modified in the case of a so-called
5257 "name normalize" option. Certain database backends, such as
5258 Oracle Database, Firebird, and DB2 "normalize" case-insensitive names
5259 as uppercase. The SQLAlchemy dialects for these backends
5260 convert from SQLAlchemy's lower-case-means-insensitive convention
5261 to the upper-case-means-insensitive conventions of those backends.
5262 The ``quote=True`` flag here will prevent this conversion from occurring
5263 to support an identifier that's quoted as all lower case against
5264 such a backend.
5266 The :class:`.quoted_name` object is normally created automatically
5267 when specifying the name for key schema constructs such as
5268 :class:`_schema.Table`, :class:`_schema.Column`, and others.
5269 The class can also be
5270 passed explicitly as the name to any function that receives a name which
5271 can be quoted. Such as to use the :meth:`_engine.Engine.has_table`
5272 method with
5273 an unconditionally quoted name::
5275 from sqlalchemy import create_engine
5276 from sqlalchemy import inspect
5277 from sqlalchemy.sql import quoted_name
5279 engine = create_engine("oracle+oracledb://some_dsn")
5280 print(inspect(engine).has_table(quoted_name("some_table", True)))
5282 The above logic will run the "has table" logic against the Oracle Database
5283 backend, passing the name exactly as ``"some_table"`` without converting to
5284 upper case.
5286 .. versionchanged:: 1.2 The :class:`.quoted_name` construct is now
5287 importable from ``sqlalchemy.sql``, in addition to the previous
5288 location of ``sqlalchemy.sql.elements``.
5290 """
5292 __slots__ = "quote", "lower", "upper"
5294 quote: Optional[bool]
5296 @overload
5297 @classmethod
5298 def construct(cls, value: str, quote: Optional[bool]) -> quoted_name: ...
5300 @overload
5301 @classmethod
5302 def construct(cls, value: None, quote: Optional[bool]) -> None: ...
5304 @classmethod
5305 def construct(
5306 cls, value: Optional[str], quote: Optional[bool]
5307 ) -> Optional[quoted_name]:
5308 if value is None:
5309 return None
5310 else:
5311 return quoted_name(value, quote)
5313 def __new__(cls, value: str, quote: Optional[bool]) -> quoted_name:
5314 assert (
5315 value is not None
5316 ), "use quoted_name.construct() for None passthrough"
5317 if isinstance(value, cls) and (quote is None or value.quote == quote):
5318 return value
5319 self = super().__new__(cls, value)
5321 self.quote = quote
5322 return self
5324 def __reduce__(self):
5325 return quoted_name, (str(self), self.quote)
5327 def _memoized_method_lower(self):
5328 if self.quote:
5329 return self
5330 else:
5331 return str(self).lower()
5333 def _memoized_method_upper(self):
5334 if self.quote:
5335 return self
5336 else:
5337 return str(self).upper()
5340def _find_columns(clause: ClauseElement) -> Set[ColumnClause[Any]]:
5341 """locate Column objects within the given expression."""
5343 cols: Set[ColumnClause[Any]] = set()
5344 traverse(clause, {}, {"column": cols.add})
5345 return cols
5348def _type_from_args(args: Sequence[ColumnElement[_T]]) -> TypeEngine[_T]:
5349 for a in args:
5350 if not a.type._isnull:
5351 return a.type
5352 else:
5353 return type_api.NULLTYPE # type: ignore
5356def _corresponding_column_or_error(fromclause, column, require_embedded=False):
5357 c = fromclause.corresponding_column(
5358 column, require_embedded=require_embedded
5359 )
5360 if c is None:
5361 raise exc.InvalidRequestError(
5362 "Given column '%s', attached to table '%s', "
5363 "failed to locate a corresponding column from table '%s'"
5364 % (column, getattr(column, "table", None), fromclause.description)
5365 )
5366 return c
5369class _memoized_property_but_not_nulltype(
5370 util.memoized_property["TypeEngine[_T]"]
5371):
5372 """memoized property, but dont memoize NullType"""
5374 def __get__(self, obj, cls):
5375 if obj is None:
5376 return self
5377 result = self.fget(obj)
5378 if not result._isnull:
5379 obj.__dict__[self.__name__] = result
5380 return result
5383class AnnotatedColumnElement(Annotated):
5384 _Annotated__element: ColumnElement[Any]
5386 def __init__(self, element, values):
5387 Annotated.__init__(self, element, values)
5388 for attr in (
5389 "comparator",
5390 "_proxy_key",
5391 "_tq_key_label",
5392 "_tq_label",
5393 "_non_anon_label",
5394 "type",
5395 ):
5396 self.__dict__.pop(attr, None)
5397 for attr in ("name", "key", "table"):
5398 if self.__dict__.get(attr, False) is None:
5399 self.__dict__.pop(attr)
5401 def _with_annotations(self, values):
5402 clone = super()._with_annotations(values)
5403 for attr in (
5404 "comparator",
5405 "_proxy_key",
5406 "_tq_key_label",
5407 "_tq_label",
5408 "_non_anon_label",
5409 ):
5410 clone.__dict__.pop(attr, None)
5411 return clone
5413 @util.memoized_property
5414 def name(self):
5415 """pull 'name' from parent, if not present"""
5416 return self._Annotated__element.name
5418 @_memoized_property_but_not_nulltype
5419 def type(self):
5420 """pull 'type' from parent and don't cache if null.
5422 type is routinely changed on existing columns within the
5423 mapped_column() initialization process, and "type" is also consulted
5424 during the creation of SQL expressions. Therefore it can change after
5425 it was already retrieved. At the same time we don't want annotated
5426 objects having overhead when expressions are produced, so continue
5427 to memoize, but only when we have a non-null type.
5429 """
5430 return self._Annotated__element.type
5432 @util.memoized_property
5433 def table(self):
5434 """pull 'table' from parent, if not present"""
5435 return self._Annotated__element.table
5437 @util.memoized_property
5438 def key(self):
5439 """pull 'key' from parent, if not present"""
5440 return self._Annotated__element.key
5442 @util.memoized_property
5443 def info(self) -> _InfoType:
5444 if TYPE_CHECKING:
5445 assert isinstance(self._Annotated__element, Column)
5446 return self._Annotated__element.info
5448 @util.memoized_property
5449 def _anon_name_label(self) -> str:
5450 return self._Annotated__element._anon_name_label
5453class _truncated_label(quoted_name):
5454 """A unicode subclass used to identify symbolic "
5455 "names that may require truncation."""
5457 __slots__ = ()
5459 def __new__(cls, value: str, quote: Optional[bool] = None) -> Any:
5460 quote = getattr(value, "quote", quote)
5461 # return super(_truncated_label, cls).__new__(cls, value, quote, True)
5462 return super().__new__(cls, value, quote)
5464 def __reduce__(self) -> Any:
5465 return self.__class__, (str(self), self.quote)
5467 def apply_map(self, map_: Mapping[str, Any]) -> str:
5468 return self
5471class conv(_truncated_label):
5472 """Mark a string indicating that a name has already been converted
5473 by a naming convention.
5475 This is a string subclass that indicates a name that should not be
5476 subject to any further naming conventions.
5478 E.g. when we create a :class:`.Constraint` using a naming convention
5479 as follows::
5481 m = MetaData(
5482 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5483 )
5484 t = Table(
5485 "t", m, Column("x", Integer), CheckConstraint("x > 5", name="x5")
5486 )
5488 The name of the above constraint will be rendered as ``"ck_t_x5"``.
5489 That is, the existing name ``x5`` is used in the naming convention as the
5490 ``constraint_name`` token.
5492 In some situations, such as in migration scripts, we may be rendering
5493 the above :class:`.CheckConstraint` with a name that's already been
5494 converted. In order to make sure the name isn't double-modified, the
5495 new name is applied using the :func:`_schema.conv` marker. We can
5496 use this explicitly as follows::
5499 m = MetaData(
5500 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5501 )
5502 t = Table(
5503 "t",
5504 m,
5505 Column("x", Integer),
5506 CheckConstraint("x > 5", name=conv("ck_t_x5")),
5507 )
5509 Where above, the :func:`_schema.conv` marker indicates that the constraint
5510 name here is final, and the name will render as ``"ck_t_x5"`` and not
5511 ``"ck_t_ck_t_x5"``
5513 .. seealso::
5515 :ref:`constraint_naming_conventions`
5517 """
5519 __slots__ = ()
5522# for backwards compatibility in case
5523# someone is re-implementing the
5524# _truncated_identifier() sequence in a custom
5525# compiler
5526_generated_label = _truncated_label
5529class _anonymous_label(_truncated_label):
5530 """A unicode subclass used to identify anonymously
5531 generated names."""
5533 __slots__ = ()
5535 @classmethod
5536 def safe_construct(
5537 cls,
5538 seed: int,
5539 body: str,
5540 enclosing_label: Optional[str] = None,
5541 sanitize_key: bool = False,
5542 ) -> _anonymous_label:
5543 # need to escape chars that interfere with format
5544 # strings in any case, issue #8724
5545 body = re.sub(r"[%\(\) \$]+", "_", body)
5547 if sanitize_key:
5548 # sanitize_key is then an extra step used by BindParameter
5549 body = body.strip("_")
5551 label = "%%(%d %s)s" % (seed, body.replace("%", "%%"))
5552 if enclosing_label:
5553 label = "%s%s" % (enclosing_label, label)
5555 return _anonymous_label(label)
5557 def __add__(self, other):
5558 if "%" in other and not isinstance(other, _anonymous_label):
5559 other = str(other).replace("%", "%%")
5560 else:
5561 other = str(other)
5563 return _anonymous_label(
5564 quoted_name(
5565 str.__add__(self, other),
5566 self.quote,
5567 )
5568 )
5570 def __radd__(self, other):
5571 if "%" in other and not isinstance(other, _anonymous_label):
5572 other = str(other).replace("%", "%%")
5573 else:
5574 other = str(other)
5576 return _anonymous_label(
5577 quoted_name(
5578 str.__add__(other, self),
5579 self.quote,
5580 )
5581 )
5583 def apply_map(self, map_):
5584 if self.quote is not None:
5585 # preserve quoting only if necessary
5586 return quoted_name(self % map_, self.quote)
5587 else:
5588 # else skip the constructor call
5589 return self % map_