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
82from ..util.typing import TupleAny
83from ..util.typing import Unpack
86if typing.TYPE_CHECKING:
87 from ._typing import _ByArgument
88 from ._typing import _ColumnExpressionArgument
89 from ._typing import _ColumnExpressionOrStrLabelArgument
90 from ._typing import _HasDialect
91 from ._typing import _InfoType
92 from ._typing import _PropagateAttrsType
93 from ._typing import _TypeEngineArgument
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 @property
506 def entity_namespace(self):
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[Unpack[TupleAny]]:
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 /,
575 **kwargs: Any,
576 ) -> Self:
577 """Return a copy with :func:`_expression.bindparam` elements
578 replaced.
580 Same functionality as :meth:`_expression.ClauseElement.params`,
581 except adds `unique=True`
582 to affected bind parameters so that multiple statements can be
583 used.
585 """
586 return self._replace_params(True, __optionaldict, kwargs)
588 def params(
589 self,
590 __optionaldict: Optional[Mapping[str, Any]] = None,
591 /,
592 **kwargs: Any,
593 ) -> Self:
594 """Return a copy with :func:`_expression.bindparam` elements
595 replaced.
597 Returns a copy of this ClauseElement with
598 :func:`_expression.bindparam`
599 elements replaced with values taken from the given dictionary::
601 >>> clause = column("x") + bindparam("foo")
602 >>> print(clause.compile().params)
603 {'foo':None}
604 >>> print(clause.params({"foo": 7}).compile().params)
605 {'foo':7}
607 """
608 return self._replace_params(False, __optionaldict, kwargs)
610 def _replace_params(
611 self,
612 unique: bool,
613 optionaldict: Optional[Mapping[str, Any]],
614 kwargs: Dict[str, Any],
615 ) -> Self:
616 if optionaldict:
617 kwargs.update(optionaldict)
619 def visit_bindparam(bind: BindParameter[Any]) -> None:
620 if bind.key in kwargs:
621 bind.value = kwargs[bind.key]
622 bind.required = False
623 if unique:
624 bind._convert_to_unique()
626 return cloned_traverse(
627 self,
628 {"maintain_key": True, "detect_subquery_cols": True},
629 {"bindparam": visit_bindparam},
630 )
632 def compare(self, other: ClauseElement, **kw: Any) -> bool:
633 r"""Compare this :class:`_expression.ClauseElement` to
634 the given :class:`_expression.ClauseElement`.
636 Subclasses should override the default behavior, which is a
637 straight identity comparison.
639 \**kw are arguments consumed by subclass ``compare()`` methods and
640 may be used to modify the criteria for comparison
641 (see :class:`_expression.ColumnElement`).
643 """
644 return traversals.compare(self, other, **kw)
646 def self_group(
647 self, against: Optional[OperatorType] = None
648 ) -> ClauseElement:
649 """Apply a 'grouping' to this :class:`_expression.ClauseElement`.
651 This method is overridden by subclasses to return a "grouping"
652 construct, i.e. parenthesis. In particular it's used by "binary"
653 expressions to provide a grouping around themselves when placed into a
654 larger expression, as well as by :func:`_expression.select`
655 constructs when placed into the FROM clause of another
656 :func:`_expression.select`. (Note that subqueries should be
657 normally created using the :meth:`_expression.Select.alias` method,
658 as many
659 platforms require nested SELECT statements to be named).
661 As expressions are composed together, the application of
662 :meth:`self_group` is automatic - end-user code should never
663 need to use this method directly. Note that SQLAlchemy's
664 clause constructs take operator precedence into account -
665 so parenthesis might not be needed, for example, in
666 an expression like ``x OR (y AND z)`` - AND takes precedence
667 over OR.
669 The base :meth:`self_group` method of
670 :class:`_expression.ClauseElement`
671 just returns self.
672 """
673 return self
675 def _ungroup(self) -> ClauseElement:
676 """Return this :class:`_expression.ClauseElement`
677 without any groupings.
678 """
680 return self
682 def _compile_w_cache(
683 self,
684 dialect: Dialect,
685 *,
686 compiled_cache: Optional[CompiledCacheType],
687 column_keys: List[str],
688 for_executemany: bool = False,
689 schema_translate_map: Optional[SchemaTranslateMapType] = None,
690 **kw: Any,
691 ) -> typing_Tuple[
692 Compiled, Optional[Sequence[BindParameter[Any]]], CacheStats
693 ]:
694 elem_cache_key: Optional[CacheKey]
696 if compiled_cache is not None and dialect._supports_statement_cache:
697 elem_cache_key = self._generate_cache_key()
698 else:
699 elem_cache_key = None
701 extracted_params: Optional[Sequence[BindParameter[Any]]]
702 if elem_cache_key is not None:
703 if TYPE_CHECKING:
704 assert compiled_cache is not None
706 cache_key, extracted_params = elem_cache_key
707 key = (
708 dialect,
709 cache_key,
710 tuple(column_keys),
711 bool(schema_translate_map),
712 for_executemany,
713 )
714 compiled_sql = compiled_cache.get(key)
716 if compiled_sql is None:
717 cache_hit = dialect.CACHE_MISS
718 compiled_sql = self._compiler(
719 dialect,
720 cache_key=elem_cache_key,
721 column_keys=column_keys,
722 for_executemany=for_executemany,
723 schema_translate_map=schema_translate_map,
724 **kw,
725 )
726 compiled_cache[key] = compiled_sql
727 else:
728 cache_hit = dialect.CACHE_HIT
729 else:
730 extracted_params = None
731 compiled_sql = self._compiler(
732 dialect,
733 cache_key=elem_cache_key,
734 column_keys=column_keys,
735 for_executemany=for_executemany,
736 schema_translate_map=schema_translate_map,
737 **kw,
738 )
740 if not dialect._supports_statement_cache:
741 cache_hit = dialect.NO_DIALECT_SUPPORT
742 elif compiled_cache is None:
743 cache_hit = dialect.CACHING_DISABLED
744 else:
745 cache_hit = dialect.NO_CACHE_KEY
747 return compiled_sql, extracted_params, cache_hit
749 def __invert__(self):
750 # undocumented element currently used by the ORM for
751 # relationship.contains()
752 if hasattr(self, "negation_clause"):
753 return self.negation_clause
754 else:
755 return self._negate()
757 def _negate(self) -> ClauseElement:
758 # TODO: this code is uncovered and in all likelihood is not included
759 # in any codepath. So this should raise NotImplementedError in 2.1
760 grouped = self.self_group(against=operators.inv)
761 assert isinstance(grouped, ColumnElement)
762 return UnaryExpression(grouped, operator=operators.inv)
764 def __bool__(self):
765 raise TypeError("Boolean value of this clause is not defined")
767 def __repr__(self):
768 friendly = self.description
769 if friendly is None:
770 return object.__repr__(self)
771 else:
772 return "<%s.%s at 0x%x; %s>" % (
773 self.__module__,
774 self.__class__.__name__,
775 id(self),
776 friendly,
777 )
780class DQLDMLClauseElement(ClauseElement):
781 """represents a :class:`.ClauseElement` that compiles to a DQL or DML
782 expression, not DDL.
784 .. versionadded:: 2.0
786 """
788 if typing.TYPE_CHECKING:
790 def _compiler(self, dialect: Dialect, **kw: Any) -> SQLCompiler:
791 """Return a compiler appropriate for this ClauseElement, given a
792 Dialect."""
793 ...
795 def compile( # noqa: A001
796 self,
797 bind: Optional[_HasDialect] = None,
798 dialect: Optional[Dialect] = None,
799 **kw: Any,
800 ) -> SQLCompiler: ...
803class CompilerColumnElement(
804 roles.DMLColumnRole,
805 roles.DDLConstraintColumnRole,
806 roles.ColumnsClauseRole,
807 CompilerElement,
808):
809 """A compiler-only column element used for ad-hoc string compilations.
811 .. versionadded:: 2.0
813 """
815 __slots__ = ()
817 _propagate_attrs = util.EMPTY_DICT
818 _is_collection_aggregate = False
821# SQLCoreOperations should be suiting the ExpressionElementRole
822# and ColumnsClauseRole. however the MRO issues become too elaborate
823# at the moment.
824class SQLCoreOperations(Generic[_T_co], ColumnOperators, TypingOnly):
825 __slots__ = ()
827 # annotations for comparison methods
828 # these are from operators->Operators / ColumnOperators,
829 # redefined with the specific types returned by ColumnElement hierarchies
830 if typing.TYPE_CHECKING:
832 @util.non_memoized_property
833 def _propagate_attrs(self) -> _PropagateAttrsType: ...
835 def operate(
836 self, op: OperatorType, *other: Any, **kwargs: Any
837 ) -> ColumnElement[Any]: ...
839 def reverse_operate(
840 self, op: OperatorType, other: Any, **kwargs: Any
841 ) -> ColumnElement[Any]: ...
843 @overload
844 def op(
845 self,
846 opstring: str,
847 precedence: int = ...,
848 is_comparison: bool = ...,
849 *,
850 return_type: _TypeEngineArgument[_OPT],
851 python_impl: Optional[Callable[..., Any]] = None,
852 ) -> Callable[[Any], BinaryExpression[_OPT]]: ...
854 @overload
855 def op(
856 self,
857 opstring: str,
858 precedence: int = ...,
859 is_comparison: bool = ...,
860 return_type: Optional[_TypeEngineArgument[Any]] = ...,
861 python_impl: Optional[Callable[..., Any]] = ...,
862 ) -> Callable[[Any], BinaryExpression[Any]]: ...
864 def op(
865 self,
866 opstring: str,
867 precedence: int = 0,
868 is_comparison: bool = False,
869 return_type: Optional[_TypeEngineArgument[Any]] = None,
870 python_impl: Optional[Callable[..., Any]] = None,
871 ) -> Callable[[Any], BinaryExpression[Any]]: ...
873 def bool_op(
874 self,
875 opstring: str,
876 precedence: int = 0,
877 python_impl: Optional[Callable[..., Any]] = None,
878 ) -> Callable[[Any], BinaryExpression[bool]]: ...
880 def __and__(self, other: Any) -> BooleanClauseList: ...
882 def __or__(self, other: Any) -> BooleanClauseList: ...
884 def __invert__(self) -> ColumnElement[_T_co]: ...
886 def __lt__(self, other: Any) -> ColumnElement[bool]: ...
888 def __le__(self, other: Any) -> ColumnElement[bool]: ...
890 # declare also that this class has an hash method otherwise
891 # it may be assumed to be None by type checkers since the
892 # object defines __eq__ and python sets it to None in that case:
893 # https://docs.python.org/3/reference/datamodel.html#object.__hash__
894 def __hash__(self) -> int: ...
896 def __eq__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
897 ...
899 def __ne__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
900 ...
902 def is_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
904 def is_not_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
906 def __gt__(self, other: Any) -> ColumnElement[bool]: ...
908 def __ge__(self, other: Any) -> ColumnElement[bool]: ...
910 def __neg__(self) -> UnaryExpression[_T_co]: ...
912 def __contains__(self, other: Any) -> ColumnElement[bool]: ...
914 def __getitem__(self, index: Any) -> ColumnElement[Any]: ...
916 @overload
917 def __lshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
919 @overload
920 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
922 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
924 @overload
925 def __rlshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
927 @overload
928 def __rlshift__(self, other: Any) -> ColumnElement[Any]: ...
930 def __rlshift__(self, other: Any) -> ColumnElement[Any]: ...
932 @overload
933 def __rshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
935 @overload
936 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
938 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
940 @overload
941 def __rrshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
943 @overload
944 def __rrshift__(self, other: Any) -> ColumnElement[Any]: ...
946 def __rrshift__(self, other: Any) -> ColumnElement[Any]: ...
948 def __matmul__(self, other: Any) -> ColumnElement[Any]: ...
950 def __rmatmul__(self, other: Any) -> ColumnElement[Any]: ...
952 @overload
953 def concat(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
955 @overload
956 def concat(self, other: Any) -> ColumnElement[Any]: ...
958 def concat(self, other: Any) -> ColumnElement[Any]: ...
960 def like(
961 self, other: Any, escape: Optional[str] = None
962 ) -> BinaryExpression[bool]: ...
964 def ilike(
965 self, other: Any, escape: Optional[str] = None
966 ) -> BinaryExpression[bool]: ...
968 def bitwise_xor(self, other: Any) -> BinaryExpression[Any]: ...
970 def bitwise_or(self, other: Any) -> BinaryExpression[Any]: ...
972 def bitwise_and(self, other: Any) -> BinaryExpression[Any]: ...
974 def bitwise_not(self) -> UnaryExpression[_T_co]: ...
976 def bitwise_lshift(self, other: Any) -> BinaryExpression[Any]: ...
978 def bitwise_rshift(self, other: Any) -> BinaryExpression[Any]: ...
980 def in_(
981 self,
982 other: Union[
983 Iterable[Any], BindParameter[Any], roles.InElementRole
984 ],
985 ) -> BinaryExpression[bool]: ...
987 def not_in(
988 self,
989 other: Union[
990 Iterable[Any], BindParameter[Any], roles.InElementRole
991 ],
992 ) -> BinaryExpression[bool]: ...
994 def notin_(
995 self,
996 other: Union[
997 Iterable[Any], BindParameter[Any], roles.InElementRole
998 ],
999 ) -> BinaryExpression[bool]: ...
1001 def not_like(
1002 self, other: Any, escape: Optional[str] = None
1003 ) -> BinaryExpression[bool]: ...
1005 def notlike(
1006 self, other: Any, escape: Optional[str] = None
1007 ) -> BinaryExpression[bool]: ...
1009 def not_ilike(
1010 self, other: Any, escape: Optional[str] = None
1011 ) -> BinaryExpression[bool]: ...
1013 def notilike(
1014 self, other: Any, escape: Optional[str] = None
1015 ) -> BinaryExpression[bool]: ...
1017 def is_(self, other: Any) -> BinaryExpression[bool]: ...
1019 def is_not(self, other: Any) -> BinaryExpression[bool]: ...
1021 def isnot(self, other: Any) -> BinaryExpression[bool]: ...
1023 def startswith(
1024 self,
1025 other: Any,
1026 escape: Optional[str] = None,
1027 autoescape: bool = False,
1028 ) -> ColumnElement[bool]: ...
1030 def istartswith(
1031 self,
1032 other: Any,
1033 escape: Optional[str] = None,
1034 autoescape: bool = False,
1035 ) -> ColumnElement[bool]: ...
1037 def endswith(
1038 self,
1039 other: Any,
1040 escape: Optional[str] = None,
1041 autoescape: bool = False,
1042 ) -> ColumnElement[bool]: ...
1044 def iendswith(
1045 self,
1046 other: Any,
1047 escape: Optional[str] = None,
1048 autoescape: bool = False,
1049 ) -> ColumnElement[bool]: ...
1051 def contains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1053 def icontains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1055 def match(self, other: Any, **kwargs: Any) -> ColumnElement[bool]: ...
1057 def regexp_match(
1058 self, pattern: Any, flags: Optional[str] = None
1059 ) -> ColumnElement[bool]: ...
1061 def regexp_replace(
1062 self, pattern: Any, replacement: Any, flags: Optional[str] = None
1063 ) -> ColumnElement[str]: ...
1065 def desc(self) -> UnaryExpression[_T_co]: ...
1067 def asc(self) -> UnaryExpression[_T_co]: ...
1069 def nulls_first(self) -> UnaryExpression[_T_co]: ...
1071 def nullsfirst(self) -> UnaryExpression[_T_co]: ...
1073 def nulls_last(self) -> UnaryExpression[_T_co]: ...
1075 def nullslast(self) -> UnaryExpression[_T_co]: ...
1077 def collate(self, collation: str) -> CollationClause: ...
1079 def between(
1080 self, cleft: Any, cright: Any, symmetric: bool = False
1081 ) -> BinaryExpression[bool]: ...
1083 def distinct(self: _SQO[_T_co]) -> UnaryExpression[_T_co]: ...
1085 def any_(self) -> CollectionAggregate[Any]: ...
1087 def all_(self) -> CollectionAggregate[Any]: ...
1089 # numeric overloads. These need more tweaking
1090 # in particular they all need to have a variant for Optiona[_T]
1091 # because Optional only applies to the data side, not the expression
1092 # side
1094 @overload
1095 def __add__(
1096 self: _SQO[_NMT],
1097 other: Any,
1098 ) -> ColumnElement[_NMT]: ...
1100 @overload
1101 def __add__(
1102 self: _SQO[str],
1103 other: Any,
1104 ) -> ColumnElement[str]: ...
1106 @overload
1107 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1109 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1111 @overload
1112 def __radd__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1114 @overload
1115 def __radd__(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
1117 def __radd__(self, other: Any) -> ColumnElement[Any]: ...
1119 @overload
1120 def __sub__(
1121 self: _SQO[_NMT],
1122 other: Any,
1123 ) -> ColumnElement[_NMT]: ...
1125 @overload
1126 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1128 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1130 @overload
1131 def __rsub__(
1132 self: _SQO[_NMT],
1133 other: Any,
1134 ) -> ColumnElement[_NMT]: ...
1136 @overload
1137 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1139 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1141 @overload
1142 def __mul__(
1143 self: _SQO[_NMT],
1144 other: Any,
1145 ) -> ColumnElement[_NMT]: ...
1147 @overload
1148 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1150 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1152 @overload
1153 def __rmul__(
1154 self: _SQO[_NMT],
1155 other: Any,
1156 ) -> ColumnElement[_NMT]: ...
1158 @overload
1159 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1161 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1163 @overload
1164 def __mod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1166 @overload
1167 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1169 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1171 @overload
1172 def __rmod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1174 @overload
1175 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1177 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1179 @overload
1180 def __truediv__(
1181 self: _SQO[int], other: Any
1182 ) -> ColumnElement[_NUMERIC]: ...
1184 @overload
1185 def __truediv__(self: _SQO[_NT], other: Any) -> ColumnElement[_NT]: ...
1187 @overload
1188 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1190 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1192 @overload
1193 def __rtruediv__(
1194 self: _SQO[_NMT], other: Any
1195 ) -> ColumnElement[_NUMERIC]: ...
1197 @overload
1198 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1200 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1202 @overload
1203 def __floordiv__(
1204 self: _SQO[_NMT], other: Any
1205 ) -> ColumnElement[_NMT]: ...
1207 @overload
1208 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1210 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1212 @overload
1213 def __rfloordiv__(
1214 self: _SQO[_NMT], other: Any
1215 ) -> ColumnElement[_NMT]: ...
1217 @overload
1218 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1220 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1223class SQLColumnExpression(
1224 SQLCoreOperations[_T_co], roles.ExpressionElementRole[_T_co], TypingOnly
1225):
1226 """A type that may be used to indicate any SQL column element or object
1227 that acts in place of one.
1229 :class:`.SQLColumnExpression` is a base of
1230 :class:`.ColumnElement`, as well as within the bases of ORM elements
1231 such as :class:`.InstrumentedAttribute`, and may be used in :pep:`484`
1232 typing to indicate arguments or return values that should behave
1233 as column expressions.
1235 .. versionadded:: 2.0.0b4
1238 """
1240 __slots__ = ()
1243_SQO = SQLCoreOperations
1246class ColumnElement(
1247 roles.ColumnArgumentOrKeyRole,
1248 roles.StatementOptionRole,
1249 roles.WhereHavingRole,
1250 roles.BinaryElementRole[_T],
1251 roles.OrderByRole,
1252 roles.ColumnsClauseRole,
1253 roles.LimitOffsetRole,
1254 roles.DMLColumnRole,
1255 roles.DDLConstraintColumnRole,
1256 roles.DDLExpressionRole,
1257 SQLColumnExpression[_T],
1258 DQLDMLClauseElement,
1259):
1260 """Represent a column-oriented SQL expression suitable for usage in the
1261 "columns" clause, WHERE clause etc. of a statement.
1263 While the most familiar kind of :class:`_expression.ColumnElement` is the
1264 :class:`_schema.Column` object, :class:`_expression.ColumnElement`
1265 serves as the basis
1266 for any unit that may be present in a SQL expression, including
1267 the expressions themselves, SQL functions, bound parameters,
1268 literal expressions, keywords such as ``NULL``, etc.
1269 :class:`_expression.ColumnElement`
1270 is the ultimate base class for all such elements.
1272 A wide variety of SQLAlchemy Core functions work at the SQL expression
1273 level, and are intended to accept instances of
1274 :class:`_expression.ColumnElement` as
1275 arguments. These functions will typically document that they accept a
1276 "SQL expression" as an argument. What this means in terms of SQLAlchemy
1277 usually refers to an input which is either already in the form of a
1278 :class:`_expression.ColumnElement` object,
1279 or a value which can be **coerced** into
1280 one. The coercion rules followed by most, but not all, SQLAlchemy Core
1281 functions with regards to SQL expressions are as follows:
1283 * a literal Python value, such as a string, integer or floating
1284 point value, boolean, datetime, ``Decimal`` object, or virtually
1285 any other Python object, will be coerced into a "literal bound
1286 value". This generally means that a :func:`.bindparam` will be
1287 produced featuring the given value embedded into the construct; the
1288 resulting :class:`.BindParameter` object is an instance of
1289 :class:`_expression.ColumnElement`.
1290 The Python value will ultimately be sent
1291 to the DBAPI at execution time as a parameterized argument to the
1292 ``execute()`` or ``executemany()`` methods, after SQLAlchemy
1293 type-specific converters (e.g. those provided by any associated
1294 :class:`.TypeEngine` objects) are applied to the value.
1296 * any special object value, typically ORM-level constructs, which
1297 feature an accessor called ``__clause_element__()``. The Core
1298 expression system looks for this method when an object of otherwise
1299 unknown type is passed to a function that is looking to coerce the
1300 argument into a :class:`_expression.ColumnElement` and sometimes a
1301 :class:`_expression.SelectBase` expression.
1302 It is used within the ORM to
1303 convert from ORM-specific objects like mapped classes and
1304 mapped attributes into Core expression objects.
1306 * The Python ``None`` value is typically interpreted as ``NULL``,
1307 which in SQLAlchemy Core produces an instance of :func:`.null`.
1309 A :class:`_expression.ColumnElement` provides the ability to generate new
1310 :class:`_expression.ColumnElement`
1311 objects using Python expressions. This means that Python operators
1312 such as ``==``, ``!=`` and ``<`` are overloaded to mimic SQL operations,
1313 and allow the instantiation of further :class:`_expression.ColumnElement`
1314 instances
1315 which are composed from other, more fundamental
1316 :class:`_expression.ColumnElement`
1317 objects. For example, two :class:`.ColumnClause` objects can be added
1318 together with the addition operator ``+`` to produce
1319 a :class:`.BinaryExpression`.
1320 Both :class:`.ColumnClause` and :class:`.BinaryExpression` are subclasses
1321 of :class:`_expression.ColumnElement`:
1323 .. sourcecode:: pycon+sql
1325 >>> from sqlalchemy.sql import column
1326 >>> column("a") + column("b")
1327 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
1328 >>> print(column("a") + column("b"))
1329 {printsql}a + b
1331 .. seealso::
1333 :class:`_schema.Column`
1335 :func:`_expression.column`
1337 """
1339 __visit_name__ = "column_element"
1341 primary_key: bool = False
1342 _is_clone_of: Optional[ColumnElement[_T]]
1343 _is_column_element = True
1344 _insert_sentinel: bool = False
1345 _omit_from_statements = False
1346 _is_collection_aggregate = False
1348 foreign_keys: AbstractSet[ForeignKey] = frozenset()
1350 @util.memoized_property
1351 def _proxies(self) -> List[ColumnElement[Any]]:
1352 return []
1354 @util.non_memoized_property
1355 def _tq_label(self) -> Optional[str]:
1356 """The named label that can be used to target
1357 this column in a result set in a "table qualified" context.
1359 This label is almost always the label used when
1360 rendering <expr> AS <label> in a SELECT statement when using
1361 the LABEL_STYLE_TABLENAME_PLUS_COL label style, which is what the
1362 legacy ORM ``Query`` object uses as well.
1364 For a regular Column bound to a Table, this is typically the label
1365 <tablename>_<columnname>. For other constructs, different rules
1366 may apply, such as anonymized labels and others.
1368 .. versionchanged:: 1.4.21 renamed from ``._label``
1370 """
1371 return None
1373 key: Optional[str] = None
1374 """The 'key' that in some circumstances refers to this object in a
1375 Python namespace.
1377 This typically refers to the "key" of the column as present in the
1378 ``.c`` collection of a selectable, e.g. ``sometable.c["somekey"]`` would
1379 return a :class:`_schema.Column` with a ``.key`` of "somekey".
1381 """
1383 @HasMemoized.memoized_attribute
1384 def _tq_key_label(self) -> Optional[str]:
1385 """A label-based version of 'key' that in some circumstances refers
1386 to this object in a Python namespace.
1389 _tq_key_label comes into play when a select() statement is constructed
1390 with apply_labels(); in this case, all Column objects in the ``.c``
1391 collection are rendered as <tablename>_<columnname> in SQL; this is
1392 essentially the value of ._label. But to locate those columns in the
1393 ``.c`` collection, the name is along the lines of <tablename>_<key>;
1394 that's the typical value of .key_label.
1396 .. versionchanged:: 1.4.21 renamed from ``._key_label``
1398 """
1399 return self._proxy_key
1401 @property
1402 def _key_label(self) -> Optional[str]:
1403 """legacy; renamed to _tq_key_label"""
1404 return self._tq_key_label
1406 @property
1407 def _label(self) -> Optional[str]:
1408 """legacy; renamed to _tq_label"""
1409 return self._tq_label
1411 @property
1412 def _non_anon_label(self) -> Optional[str]:
1413 """the 'name' that naturally applies this element when rendered in
1414 SQL.
1416 Concretely, this is the "name" of a column or a label in a
1417 SELECT statement; ``<columnname>`` and ``<labelname>`` below:
1419 .. sourcecode:: sql
1421 SELECT <columnmame> FROM table
1423 SELECT column AS <labelname> FROM table
1425 Above, the two names noted will be what's present in the DBAPI
1426 ``cursor.description`` as the names.
1428 If this attribute returns ``None``, it means that the SQL element as
1429 written does not have a 100% fully predictable "name" that would appear
1430 in the ``cursor.description``. Examples include SQL functions, CAST
1431 functions, etc. While such things do return names in
1432 ``cursor.description``, they are only predictable on a
1433 database-specific basis; e.g. an expression like ``MAX(table.col)`` may
1434 appear as the string ``max`` on one database (like PostgreSQL) or may
1435 appear as the whole expression ``max(table.col)`` on SQLite.
1437 The default implementation looks for a ``.name`` attribute on the
1438 object, as has been the precedent established in SQLAlchemy for many
1439 years. An exception is made on the ``FunctionElement`` subclass
1440 so that the return value is always ``None``.
1442 .. versionadded:: 1.4.21
1446 """
1447 return getattr(self, "name", None)
1449 _render_label_in_columns_clause = True
1450 """A flag used by select._columns_plus_names that helps to determine
1451 we are actually going to render in terms of "SELECT <col> AS <label>".
1452 This flag can be returned as False for some Column objects that want
1453 to be rendered as simple "SELECT <col>"; typically columns that don't have
1454 any parent table and are named the same as what the label would be
1455 in any case.
1457 """
1459 _allow_label_resolve = True
1460 """A flag that can be flipped to prevent a column from being resolvable
1461 by string label name.
1463 The joined eager loader strategy in the ORM uses this, for example.
1465 """
1467 _is_implicitly_boolean = False
1469 _alt_names: Sequence[str] = ()
1471 if TYPE_CHECKING:
1473 def _ungroup(self) -> ColumnElement[_T]: ...
1475 @overload
1476 def self_group(self, against: None = None) -> ColumnElement[_T]: ...
1478 @overload
1479 def self_group(
1480 self, against: Optional[OperatorType] = None
1481 ) -> ColumnElement[Any]: ...
1483 def self_group(
1484 self, against: Optional[OperatorType] = None
1485 ) -> ColumnElement[Any]:
1486 if (
1487 against in (operators.and_, operators.or_, operators._asbool)
1488 and self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity
1489 ):
1490 return AsBoolean(self, operators.is_true, operators.is_false)
1491 elif against in (operators.any_op, operators.all_op):
1492 return Grouping(self)
1493 else:
1494 return self
1496 @overload
1497 def _negate(self: ColumnElement[bool]) -> ColumnElement[bool]: ...
1499 @overload
1500 def _negate(self: ColumnElement[_T]) -> ColumnElement[_T]: ...
1502 def _negate(self) -> ColumnElement[Any]:
1503 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
1504 return AsBoolean(self, operators.is_false, operators.is_true)
1505 else:
1506 grouped = self.self_group(against=operators.inv)
1507 assert isinstance(grouped, ColumnElement)
1508 return UnaryExpression(
1509 grouped,
1510 operator=operators.inv,
1511 )
1513 type: TypeEngine[_T]
1515 if not TYPE_CHECKING:
1517 @util.memoized_property
1518 def type(self) -> TypeEngine[_T]: # noqa: A001
1519 # used for delayed setup of
1520 # type_api
1521 return type_api.NULLTYPE
1523 @HasMemoized.memoized_attribute
1524 def comparator(self) -> TypeEngine.Comparator[_T]:
1525 try:
1526 comparator_factory = self.type.comparator_factory
1527 except AttributeError as err:
1528 raise TypeError(
1529 "Object %r associated with '.type' attribute "
1530 "is not a TypeEngine class or object" % self.type
1531 ) from err
1532 else:
1533 return comparator_factory(self)
1535 def __setstate__(self, state):
1536 self.__dict__.update(state)
1538 def __getattr__(self, key: str) -> Any:
1539 try:
1540 return getattr(self.comparator, key)
1541 except AttributeError as err:
1542 raise AttributeError(
1543 "Neither %r object nor %r object has an attribute %r"
1544 % (
1545 type(self).__name__,
1546 type(self.comparator).__name__,
1547 key,
1548 )
1549 ) from err
1551 def operate(
1552 self,
1553 op: operators.OperatorType,
1554 *other: Any,
1555 **kwargs: Any,
1556 ) -> ColumnElement[Any]:
1557 return op(self.comparator, *other, **kwargs) # type: ignore[no-any-return] # noqa: E501
1559 def reverse_operate(
1560 self, op: operators.OperatorType, other: Any, **kwargs: Any
1561 ) -> ColumnElement[Any]:
1562 return op(other, self.comparator, **kwargs) # type: ignore[no-any-return] # noqa: E501
1564 def _bind_param(
1565 self,
1566 operator: operators.OperatorType,
1567 obj: Any,
1568 type_: Optional[TypeEngine[_T]] = None,
1569 expanding: bool = False,
1570 ) -> BindParameter[_T]:
1571 return BindParameter(
1572 None,
1573 obj,
1574 _compared_to_operator=operator,
1575 type_=type_,
1576 _compared_to_type=self.type,
1577 unique=True,
1578 expanding=expanding,
1579 )
1581 @property
1582 def expression(self) -> ColumnElement[Any]:
1583 """Return a column expression.
1585 Part of the inspection interface; returns self.
1587 """
1588 return self
1590 @property
1591 def _select_iterable(self) -> _SelectIterable:
1592 return (self,)
1594 @util.memoized_property
1595 def base_columns(self) -> FrozenSet[ColumnElement[Any]]:
1596 return frozenset(c for c in self.proxy_set if not c._proxies)
1598 @util.memoized_property
1599 def proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1600 """set of all columns we are proxying
1602 as of 2.0 this is explicitly deannotated columns. previously it was
1603 effectively deannotated columns but wasn't enforced. annotated
1604 columns should basically not go into sets if at all possible because
1605 their hashing behavior is very non-performant.
1607 """
1608 return frozenset([self._deannotate()]).union(
1609 itertools.chain(*[c.proxy_set for c in self._proxies])
1610 )
1612 @util.memoized_property
1613 def _expanded_proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1614 return frozenset(_expand_cloned(self.proxy_set))
1616 def _uncached_proxy_list(self) -> List[ColumnElement[Any]]:
1617 """An 'uncached' version of proxy set.
1619 This list includes annotated columns which perform very poorly in
1620 set operations.
1622 """
1624 return [self] + list(
1625 itertools.chain(*[c._uncached_proxy_list() for c in self._proxies])
1626 )
1628 def shares_lineage(self, othercolumn: ColumnElement[Any]) -> bool:
1629 """Return True if the given :class:`_expression.ColumnElement`
1630 has a common ancestor to this :class:`_expression.ColumnElement`."""
1632 return bool(self.proxy_set.intersection(othercolumn.proxy_set))
1634 def _compare_name_for_result(self, other: ColumnElement[Any]) -> bool:
1635 """Return True if the given column element compares to this one
1636 when targeting within a result row."""
1638 return (
1639 hasattr(other, "name")
1640 and hasattr(self, "name")
1641 and other.name == self.name
1642 )
1644 @HasMemoized.memoized_attribute
1645 def _proxy_key(self) -> Optional[str]:
1646 if self._annotations and "proxy_key" in self._annotations:
1647 return cast(str, self._annotations["proxy_key"])
1649 name = self.key
1650 if not name:
1651 # there's a bit of a seeming contradiction which is that the
1652 # "_non_anon_label" of a column can in fact be an
1653 # "_anonymous_label"; this is when it's on a column that is
1654 # proxying for an anonymous expression in a subquery.
1655 name = self._non_anon_label
1657 if isinstance(name, _anonymous_label):
1658 return None
1659 else:
1660 return name
1662 @HasMemoized.memoized_attribute
1663 def _expression_label(self) -> Optional[str]:
1664 """a suggested label to use in the case that the column has no name,
1665 which should be used if possible as the explicit 'AS <label>'
1666 where this expression would normally have an anon label.
1668 this is essentially mostly what _proxy_key does except it returns
1669 None if the column has a normal name that can be used.
1671 """
1673 if getattr(self, "name", None) is not None:
1674 return None
1675 elif self._annotations and "proxy_key" in self._annotations:
1676 return cast(str, self._annotations["proxy_key"])
1677 else:
1678 return None
1680 def _make_proxy(
1681 self,
1682 selectable: FromClause,
1683 *,
1684 primary_key: ColumnSet,
1685 foreign_keys: Set[KeyedColumnElement[Any]],
1686 name: Optional[str] = None,
1687 key: Optional[str] = None,
1688 name_is_truncatable: bool = False,
1689 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
1690 **kw: Any,
1691 ) -> typing_Tuple[str, ColumnClause[_T]]:
1692 """Create a new :class:`_expression.ColumnElement` representing this
1693 :class:`_expression.ColumnElement` as it appears in the select list of
1694 a descending selectable.
1696 """
1697 if name is None:
1698 name = self._anon_name_label
1699 if key is None:
1700 key = self._proxy_key
1701 else:
1702 key = name
1704 assert key is not None
1706 co: ColumnClause[_T] = ColumnClause(
1707 (
1708 coercions.expect(roles.TruncatedLabelRole, name)
1709 if name_is_truncatable
1710 else name
1711 ),
1712 type_=getattr(self, "type", None),
1713 _selectable=selectable,
1714 )
1716 co._propagate_attrs = selectable._propagate_attrs
1717 if compound_select_cols:
1718 co._proxies = list(compound_select_cols)
1719 else:
1720 co._proxies = [self]
1721 if selectable._is_clone_of is not None:
1722 co._is_clone_of = selectable._is_clone_of.columns.get(key)
1723 return key, co
1725 def cast(self, type_: _TypeEngineArgument[_OPT]) -> Cast[_OPT]:
1726 """Produce a type cast, i.e. ``CAST(<expression> AS <type>)``.
1728 This is a shortcut to the :func:`_expression.cast` function.
1730 .. seealso::
1732 :ref:`tutorial_casts`
1734 :func:`_expression.cast`
1736 :func:`_expression.type_coerce`
1738 """
1739 return Cast(self, type_)
1741 def label(self, name: Optional[str]) -> Label[_T]:
1742 """Produce a column label, i.e. ``<columnname> AS <name>``.
1744 This is a shortcut to the :func:`_expression.label` function.
1746 If 'name' is ``None``, an anonymous label name will be generated.
1748 """
1749 return Label(name, self, self.type)
1751 def _anon_label(
1752 self, seed: Optional[str], add_hash: Optional[int] = None
1753 ) -> _anonymous_label:
1754 while self._is_clone_of is not None:
1755 self = self._is_clone_of
1757 # as of 1.4 anonymous label for ColumnElement uses hash(), not id(),
1758 # as the identifier, because a column and its annotated version are
1759 # the same thing in a SQL statement
1760 hash_value = hash(self)
1762 if add_hash:
1763 # this path is used for disambiguating anon labels that would
1764 # otherwise be the same name for the same element repeated.
1765 # an additional numeric value is factored in for each label.
1767 # shift hash(self) (which is id(self), typically 8 byte integer)
1768 # 16 bits leftward. fill extra add_hash on right
1769 assert add_hash < (2 << 15)
1770 assert seed
1771 hash_value = (hash_value << 16) | add_hash
1773 # extra underscore is added for labels with extra hash
1774 # values, to isolate the "deduped anon" namespace from the
1775 # regular namespace. eliminates chance of these
1776 # manufactured hash values overlapping with regular ones for some
1777 # undefined python interpreter
1778 seed = seed + "_"
1780 if isinstance(seed, _anonymous_label):
1781 # NOTE: the space after the hash is required
1782 return _anonymous_label(f"{seed}%({hash_value} )s")
1784 return _anonymous_label.safe_construct(hash_value, seed or "anon")
1786 @util.memoized_property
1787 def _anon_name_label(self) -> str:
1788 """Provides a constant 'anonymous label' for this ColumnElement.
1790 This is a label() expression which will be named at compile time.
1791 The same label() is returned each time ``anon_label`` is called so
1792 that expressions can reference ``anon_label`` multiple times,
1793 producing the same label name at compile time.
1795 The compiler uses this function automatically at compile time
1796 for expressions that are known to be 'unnamed' like binary
1797 expressions and function calls.
1799 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1800 public and is renamed to _anon_name_label. anon_name exists
1801 for backwards compat
1803 """
1804 name = getattr(self, "name", None)
1805 return self._anon_label(name)
1807 @util.memoized_property
1808 def _anon_key_label(self) -> _anonymous_label:
1809 """Provides a constant 'anonymous key label' for this ColumnElement.
1811 Compare to ``anon_label``, except that the "key" of the column,
1812 if available, is used to generate the label.
1814 This is used when a deduplicating key is placed into the columns
1815 collection of a selectable.
1817 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1818 public and is renamed to _anon_key_label. anon_key_label exists
1819 for backwards compat
1821 """
1822 return self._anon_label(self._proxy_key)
1824 @property
1825 @util.deprecated(
1826 "1.4",
1827 "The :attr:`_expression.ColumnElement.anon_label` attribute is now "
1828 "private, and the public accessor is deprecated.",
1829 )
1830 def anon_label(self) -> str:
1831 return self._anon_name_label
1833 @property
1834 @util.deprecated(
1835 "1.4",
1836 "The :attr:`_expression.ColumnElement.anon_key_label` attribute is "
1837 "now private, and the public accessor is deprecated.",
1838 )
1839 def anon_key_label(self) -> str:
1840 return self._anon_key_label
1842 def _dedupe_anon_label_idx(self, idx: int) -> str:
1843 """label to apply to a column that is anon labeled, but repeated
1844 in the SELECT, so that we have to make an "extra anon" label that
1845 disambiguates it from the previous appearance.
1847 these labels come out like "foo_bar_id__1" and have double underscores
1848 in them.
1850 """
1851 label = getattr(self, "name", None)
1853 # current convention is that if the element doesn't have a
1854 # ".name" (usually because it is not NamedColumn), we try to
1855 # use a "table qualified" form for the "dedupe anon" label,
1856 # based on the notion that a label like
1857 # "CAST(casttest.v1 AS DECIMAL) AS casttest_v1__1" looks better than
1858 # "CAST(casttest.v1 AS DECIMAL) AS anon__1"
1860 if label is None:
1861 return self._dedupe_anon_tq_label_idx(idx)
1862 else:
1863 return self._anon_label(label, add_hash=idx)
1865 @util.memoized_property
1866 def _anon_tq_label(self) -> _anonymous_label:
1867 return self._anon_label(getattr(self, "_tq_label", None))
1869 @util.memoized_property
1870 def _anon_tq_key_label(self) -> _anonymous_label:
1871 return self._anon_label(getattr(self, "_tq_key_label", None))
1873 def _dedupe_anon_tq_label_idx(self, idx: int) -> _anonymous_label:
1874 label = getattr(self, "_tq_label", None) or "anon"
1876 return self._anon_label(label, add_hash=idx)
1879class KeyedColumnElement(ColumnElement[_T]):
1880 """ColumnElement where ``.key`` is non-None."""
1882 _is_keyed_column_element = True
1884 key: str
1887class WrapsColumnExpression(ColumnElement[_T]):
1888 """Mixin that defines a :class:`_expression.ColumnElement`
1889 as a wrapper with special
1890 labeling behavior for an expression that already has a name.
1892 .. versionadded:: 1.4
1894 .. seealso::
1896 :ref:`change_4449`
1899 """
1901 @property
1902 def wrapped_column_expression(self) -> ColumnElement[_T]:
1903 raise NotImplementedError()
1905 @util.non_memoized_property
1906 def _tq_label(self) -> Optional[str]:
1907 wce = self.wrapped_column_expression
1908 if hasattr(wce, "_tq_label"):
1909 return wce._tq_label
1910 else:
1911 return None
1913 @property
1914 def _label(self) -> Optional[str]:
1915 return self._tq_label
1917 @property
1918 def _non_anon_label(self) -> Optional[str]:
1919 return None
1921 @util.non_memoized_property
1922 def _anon_name_label(self) -> str:
1923 wce = self.wrapped_column_expression
1925 # this logic tries to get the WrappedColumnExpression to render
1926 # with "<expr> AS <name>", where "<name>" is the natural name
1927 # within the expression itself. e.g. "CAST(table.foo) AS foo".
1928 if not wce._is_text_clause:
1929 nal = wce._non_anon_label
1930 if nal:
1931 return nal
1932 elif hasattr(wce, "_anon_name_label"):
1933 return wce._anon_name_label
1934 return super()._anon_name_label
1936 def _dedupe_anon_label_idx(self, idx: int) -> str:
1937 wce = self.wrapped_column_expression
1938 nal = wce._non_anon_label
1939 if nal:
1940 return self._anon_label(nal + "_")
1941 else:
1942 return self._dedupe_anon_tq_label_idx(idx)
1944 @property
1945 def _proxy_key(self):
1946 wce = self.wrapped_column_expression
1948 if not wce._is_text_clause:
1949 return wce._proxy_key
1950 return super()._proxy_key
1953class BindParameter(roles.InElementRole, KeyedColumnElement[_T]):
1954 r"""Represent a "bound expression".
1956 :class:`.BindParameter` is invoked explicitly using the
1957 :func:`.bindparam` function, as in::
1959 from sqlalchemy import bindparam
1961 stmt = select(users_table).where(
1962 users_table.c.name == bindparam("username")
1963 )
1965 Detailed discussion of how :class:`.BindParameter` is used is
1966 at :func:`.bindparam`.
1968 .. seealso::
1970 :func:`.bindparam`
1972 """
1974 __visit_name__ = "bindparam"
1976 _traverse_internals: _TraverseInternalsType = [
1977 ("key", InternalTraversal.dp_anon_name),
1978 ("type", InternalTraversal.dp_type),
1979 ("callable", InternalTraversal.dp_plain_dict),
1980 ("value", InternalTraversal.dp_plain_obj),
1981 ("literal_execute", InternalTraversal.dp_boolean),
1982 ]
1984 key: str
1985 _anon_map_key: Optional[str] = None
1986 type: TypeEngine[_T]
1987 value: Optional[_T]
1989 _is_crud = False
1990 _is_bind_parameter = True
1992 # bindparam implements its own _gen_cache_key() method however
1993 # we check subclasses for this flag, else no cache key is generated
1994 inherit_cache = True
1996 def __init__(
1997 self,
1998 key: Optional[str],
1999 value: Any = _NoArg.NO_ARG,
2000 type_: Optional[_TypeEngineArgument[_T]] = None,
2001 unique: bool = False,
2002 required: Union[bool, Literal[_NoArg.NO_ARG]] = _NoArg.NO_ARG,
2003 quote: Optional[bool] = None,
2004 callable_: Optional[Callable[[], Any]] = None,
2005 expanding: bool = False,
2006 isoutparam: bool = False,
2007 literal_execute: bool = False,
2008 _compared_to_operator: Optional[OperatorType] = None,
2009 _compared_to_type: Optional[TypeEngine[Any]] = None,
2010 _is_crud: bool = False,
2011 ):
2012 if required is _NoArg.NO_ARG:
2013 required = value is _NoArg.NO_ARG and callable_ is None
2014 if value is _NoArg.NO_ARG:
2015 value = None
2017 if quote is not None:
2018 key = quoted_name.construct(key, quote)
2020 if unique:
2021 self.key, self._anon_map_key = (
2022 _anonymous_label.safe_construct_with_key(
2023 id(self),
2024 (
2025 key
2026 if key is not None
2027 and not isinstance(key, _anonymous_label)
2028 else "param"
2029 ),
2030 sanitize_key=True,
2031 )
2032 )
2033 elif key:
2034 self.key = key
2035 else:
2036 self.key, self._anon_map_key = (
2037 _anonymous_label.safe_construct_with_key(id(self), "param")
2038 )
2040 # identifying key that won't change across
2041 # clones, used to identify the bind's logical
2042 # identity
2043 self._identifying_key = self.key
2045 # key that was passed in the first place, used to
2046 # generate new keys
2047 self._orig_key = key or "param"
2049 self.unique = unique
2050 self.value = value
2051 self.callable = callable_
2052 self.isoutparam = isoutparam
2053 self.required = required
2055 # indicate an "expanding" parameter; the compiler sets this
2056 # automatically in the compiler _render_in_expr_w_bindparam method
2057 # for an IN expression
2058 self.expanding = expanding
2060 # this is another hint to help w/ expanding and is typically
2061 # set in the compiler _render_in_expr_w_bindparam method for an
2062 # IN expression
2063 self.expand_op = None
2065 self.literal_execute = literal_execute
2066 if _is_crud:
2067 self._is_crud = True
2069 if type_ is None:
2070 if expanding:
2071 if value:
2072 check_value = value[0]
2073 else:
2074 check_value = type_api._NO_VALUE_IN_LIST
2075 else:
2076 check_value = value
2077 if _compared_to_type is not None:
2078 self.type = _compared_to_type.coerce_compared_value(
2079 _compared_to_operator, check_value
2080 )
2081 else:
2082 self.type = type_api._resolve_value_to_type(check_value)
2083 elif isinstance(type_, type):
2084 self.type = type_()
2085 elif is_tuple_type(type_):
2086 if value:
2087 if expanding:
2088 check_value = value[0]
2089 else:
2090 check_value = value
2091 cast("BindParameter[TupleAny]", self).type = (
2092 type_._resolve_values_to_types(check_value)
2093 )
2094 else:
2095 cast("BindParameter[TupleAny]", self).type = type_
2096 else:
2097 self.type = type_
2099 def _with_value(self, value, maintain_key=False, required=NO_ARG):
2100 """Return a copy of this :class:`.BindParameter` with the given value
2101 set.
2102 """
2103 cloned = self._clone(maintain_key=maintain_key)
2104 cloned.value = value
2105 cloned.callable = None
2106 cloned.required = required if required is not NO_ARG else self.required
2107 if cloned.type is type_api.NULLTYPE:
2108 cloned.type = type_api._resolve_value_to_type(value)
2109 return cloned
2111 @property
2112 def effective_value(self) -> Optional[_T]:
2113 """Return the value of this bound parameter,
2114 taking into account if the ``callable`` parameter
2115 was set.
2117 The ``callable`` value will be evaluated
2118 and returned if present, else ``value``.
2120 """
2121 if self.callable:
2122 # TODO: set up protocol for bind parameter callable
2123 return self.callable() # type: ignore
2124 else:
2125 return self.value
2127 def render_literal_execute(self) -> Self:
2128 """Produce a copy of this bound parameter that will enable the
2129 :paramref:`_sql.BindParameter.literal_execute` flag.
2131 The :paramref:`_sql.BindParameter.literal_execute` flag will
2132 have the effect of the parameter rendered in the compiled SQL
2133 string using ``[POSTCOMPILE]`` form, which is a special form that
2134 is converted to be a rendering of the literal value of the parameter
2135 at SQL execution time. The rationale is to support caching
2136 of SQL statement strings that can embed per-statement literal values,
2137 such as LIMIT and OFFSET parameters, in the final SQL string that
2138 is passed to the DBAPI. Dialects in particular may want to use
2139 this method within custom compilation schemes.
2141 .. versionadded:: 1.4.5
2143 .. seealso::
2145 :ref:`engine_thirdparty_caching`
2147 """
2148 c: Self = ClauseElement._clone(self)
2149 c.literal_execute = True
2150 return c
2152 def _negate_in_binary(self, negated_op, original_op):
2153 if self.expand_op is original_op:
2154 bind = self._clone()
2155 bind.expand_op = negated_op
2156 return bind
2157 else:
2158 return self
2160 def _with_binary_element_type(self, type_: TypeEngine[Any]) -> Self:
2161 c: Self = ClauseElement._clone(self)
2162 c.type = type_
2163 return c
2165 def _clone(self, maintain_key: bool = False, **kw: Any) -> Self:
2166 c: Self = ClauseElement._clone(self, **kw)
2167 # ensure all the BindParameter objects stay in cloned set.
2168 # in #7823, we changed "clone" so that a clone only keeps a reference
2169 # to the "original" element, since for column correspondence, that's
2170 # all we need. However, for BindParam, _cloned_set is used by
2171 # the "cache key bind match" lookup, which means if any of those
2172 # interim BindParameter objects became part of a cache key in the
2173 # cache, we need it. So here, make sure all clones keep carrying
2174 # forward.
2175 c._cloned_set.update(self._cloned_set)
2176 if not maintain_key and self.unique:
2177 c.key, c._anon_map_key = _anonymous_label.safe_construct_with_key(
2178 id(c), c._orig_key or "param", sanitize_key=True
2179 )
2180 return c
2182 def _gen_cache_key(self, anon_map, bindparams):
2183 _gen_cache_ok = self.__class__.__dict__.get("inherit_cache", False)
2185 if not _gen_cache_ok:
2186 if anon_map is not None:
2187 anon_map[NO_CACHE] = True
2188 return None
2190 id_, found = anon_map.get_anon(self)
2191 if found:
2192 return (id_, self.__class__)
2194 if bindparams is not None:
2195 bindparams.append(self)
2197 return (
2198 id_,
2199 self.__class__,
2200 self.type._static_cache_key,
2201 (
2202 anon_map[self._anon_map_key]
2203 if self._anon_map_key is not None
2204 else self.key
2205 ),
2206 self.literal_execute,
2207 )
2209 def _convert_to_unique(self):
2210 if not self.unique:
2211 self.unique = True
2212 self.key, self._anon_map_key = (
2213 _anonymous_label.safe_construct_with_key(
2214 id(self), self._orig_key or "param", sanitize_key=True
2215 )
2216 )
2218 def __getstate__(self):
2219 """execute a deferred value for serialization purposes."""
2221 d = self.__dict__.copy()
2222 v = self.value
2223 if self.callable:
2224 v = self.callable()
2225 d["callable"] = None
2226 d["value"] = v
2227 return d
2229 def __setstate__(self, state):
2230 if state.get("unique", False):
2231 anon_and_key = _anonymous_label.safe_construct_with_key(
2232 id(self), state.get("_orig_key", "param"), sanitize_key=True
2233 )
2234 state["key"], state["_anon_map_key"] = anon_and_key
2235 self.__dict__.update(state)
2237 def __repr__(self):
2238 return "%s(%r, %r, type_=%r)" % (
2239 self.__class__.__name__,
2240 self.key,
2241 self.value,
2242 self.type,
2243 )
2246class TypeClause(DQLDMLClauseElement):
2247 """Handle a type keyword in a SQL statement.
2249 Used by the ``Case`` statement.
2251 """
2253 __visit_name__ = "typeclause"
2255 _traverse_internals: _TraverseInternalsType = [
2256 ("type", InternalTraversal.dp_type)
2257 ]
2258 type: TypeEngine[Any]
2260 def __init__(self, type_: TypeEngine[Any]):
2261 self.type = type_
2264class TextClause(
2265 roles.DDLConstraintColumnRole,
2266 roles.DDLExpressionRole,
2267 roles.StatementOptionRole,
2268 roles.WhereHavingRole,
2269 roles.OrderByRole,
2270 roles.FromClauseRole,
2271 roles.SelectStatementRole,
2272 roles.InElementRole,
2273 Generative,
2274 Executable,
2275 DQLDMLClauseElement,
2276 roles.BinaryElementRole[Any],
2277 inspection.Inspectable["TextClause"],
2278):
2279 """Represent a literal SQL text fragment.
2281 E.g.::
2283 from sqlalchemy import text
2285 t = text("SELECT * FROM users")
2286 result = connection.execute(t)
2288 The :class:`_expression.TextClause` construct is produced using the
2289 :func:`_expression.text`
2290 function; see that function for full documentation.
2292 .. seealso::
2294 :func:`_expression.text`
2296 """
2298 __visit_name__ = "textclause"
2300 _traverse_internals: _TraverseInternalsType = [
2301 ("_bindparams", InternalTraversal.dp_string_clauseelement_dict),
2302 ("text", InternalTraversal.dp_string),
2303 ]
2305 _is_text_clause = True
2307 _is_textual = True
2309 _bind_params_regex = re.compile(r"(?<![:\w\x5c]):(\w+)(?!:)", re.UNICODE)
2310 _is_implicitly_boolean = False
2312 _render_label_in_columns_clause = False
2314 _omit_from_statements = False
2316 _is_collection_aggregate = False
2318 @property
2319 def _hide_froms(self) -> Iterable[FromClause]:
2320 return ()
2322 def __and__(self, other):
2323 # support use in select.where(), query.filter()
2324 return and_(self, other)
2326 @property
2327 def _select_iterable(self) -> _SelectIterable:
2328 return (self,)
2330 # help in those cases where text() is
2331 # interpreted in a column expression situation
2332 key: Optional[str] = None
2333 _label: Optional[str] = None
2335 _allow_label_resolve = False
2337 @property
2338 def _is_star(self): # type: ignore[override]
2339 return self.text == "*"
2341 def __init__(self, text: str):
2342 self._bindparams: Dict[str, BindParameter[Any]] = {}
2344 def repl(m):
2345 self._bindparams[m.group(1)] = BindParameter(m.group(1))
2346 return ":%s" % m.group(1)
2348 # scan the string and search for bind parameter names, add them
2349 # to the list of bindparams
2350 self.text = self._bind_params_regex.sub(repl, text)
2352 @_generative
2353 def bindparams(
2354 self,
2355 *binds: BindParameter[Any],
2356 **names_to_values: Any,
2357 ) -> Self:
2358 """Establish the values and/or types of bound parameters within
2359 this :class:`_expression.TextClause` construct.
2361 Given a text construct such as::
2363 from sqlalchemy import text
2365 stmt = text(
2366 "SELECT id, name FROM user WHERE name=:name AND timestamp=:timestamp"
2367 )
2369 the :meth:`_expression.TextClause.bindparams`
2370 method can be used to establish
2371 the initial value of ``:name`` and ``:timestamp``,
2372 using simple keyword arguments::
2374 stmt = stmt.bindparams(
2375 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2376 )
2378 Where above, new :class:`.BindParameter` objects
2379 will be generated with the names ``name`` and ``timestamp``, and
2380 values of ``jack`` and ``datetime.datetime(2012, 10, 8, 15, 12, 5)``,
2381 respectively. The types will be
2382 inferred from the values given, in this case :class:`.String` and
2383 :class:`.DateTime`.
2385 When specific typing behavior is needed, the positional ``*binds``
2386 argument can be used in which to specify :func:`.bindparam` constructs
2387 directly. These constructs must include at least the ``key``
2388 argument, then an optional value and type::
2390 from sqlalchemy import bindparam
2392 stmt = stmt.bindparams(
2393 bindparam("name", value="jack", type_=String),
2394 bindparam("timestamp", type_=DateTime),
2395 )
2397 Above, we specified the type of :class:`.DateTime` for the
2398 ``timestamp`` bind, and the type of :class:`.String` for the ``name``
2399 bind. In the case of ``name`` we also set the default value of
2400 ``"jack"``.
2402 Additional bound parameters can be supplied at statement execution
2403 time, e.g.::
2405 result = connection.execute(
2406 stmt, timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2407 )
2409 The :meth:`_expression.TextClause.bindparams`
2410 method can be called repeatedly,
2411 where it will re-use existing :class:`.BindParameter` objects to add
2412 new information. For example, we can call
2413 :meth:`_expression.TextClause.bindparams`
2414 first with typing information, and a
2415 second time with value information, and it will be combined::
2417 stmt = text(
2418 "SELECT id, name FROM user WHERE name=:name "
2419 "AND timestamp=:timestamp"
2420 )
2421 stmt = stmt.bindparams(
2422 bindparam("name", type_=String), bindparam("timestamp", type_=DateTime)
2423 )
2424 stmt = stmt.bindparams(
2425 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2426 )
2428 The :meth:`_expression.TextClause.bindparams`
2429 method also supports the concept of
2430 **unique** bound parameters. These are parameters that are
2431 "uniquified" on name at statement compilation time, so that multiple
2432 :func:`_expression.text`
2433 constructs may be combined together without the names
2434 conflicting. To use this feature, specify the
2435 :paramref:`.BindParameter.unique` flag on each :func:`.bindparam`
2436 object::
2438 stmt1 = text("select id from table where name=:name").bindparams(
2439 bindparam("name", value="name1", unique=True)
2440 )
2441 stmt2 = text("select id from table where name=:name").bindparams(
2442 bindparam("name", value="name2", unique=True)
2443 )
2445 union = union_all(stmt1.columns(column("id")), stmt2.columns(column("id")))
2447 The above statement will render as:
2449 .. sourcecode:: sql
2451 select id from table where name=:name_1
2452 UNION ALL select id from table where name=:name_2
2454 """ # noqa: E501
2455 self._bindparams = new_params = self._bindparams.copy()
2457 for bind in binds:
2458 try:
2459 # the regex used for text() currently will not match
2460 # a unique/anonymous key in any case, so use the _orig_key
2461 # so that a text() construct can support unique parameters
2462 existing = new_params[bind._orig_key]
2463 except KeyError as err:
2464 raise exc.ArgumentError(
2465 "This text() construct doesn't define a "
2466 "bound parameter named %r" % bind._orig_key
2467 ) from err
2468 else:
2469 new_params[existing._orig_key] = bind
2471 for key, value in names_to_values.items():
2472 try:
2473 existing = new_params[key]
2474 except KeyError as err:
2475 raise exc.ArgumentError(
2476 "This text() construct doesn't define a "
2477 "bound parameter named %r" % key
2478 ) from err
2479 else:
2480 new_params[key] = existing._with_value(value, required=False)
2481 return self
2483 @util.preload_module("sqlalchemy.sql.selectable")
2484 def columns(
2485 self,
2486 *cols: _ColumnExpressionArgument[Any],
2487 **types: _TypeEngineArgument[Any],
2488 ) -> TextualSelect:
2489 r"""Turn this :class:`_expression.TextClause` object into a
2490 :class:`_expression.TextualSelect`
2491 object that serves the same role as a SELECT
2492 statement.
2494 The :class:`_expression.TextualSelect` is part of the
2495 :class:`_expression.SelectBase`
2496 hierarchy and can be embedded into another statement by using the
2497 :meth:`_expression.TextualSelect.subquery` method to produce a
2498 :class:`.Subquery`
2499 object, which can then be SELECTed from.
2501 This function essentially bridges the gap between an entirely
2502 textual SELECT statement and the SQL expression language concept
2503 of a "selectable"::
2505 from sqlalchemy.sql import column, text
2507 stmt = text("SELECT id, name FROM some_table")
2508 stmt = stmt.columns(column("id"), column("name")).subquery("st")
2510 stmt = (
2511 select(mytable)
2512 .select_from(mytable.join(stmt, mytable.c.name == stmt.c.name))
2513 .where(stmt.c.id > 5)
2514 )
2516 Above, we pass a series of :func:`_expression.column` elements to the
2517 :meth:`_expression.TextClause.columns` method positionally. These
2518 :func:`_expression.column`
2519 elements now become first class elements upon the
2520 :attr:`_expression.TextualSelect.selected_columns` column collection,
2521 which then
2522 become part of the :attr:`.Subquery.c` collection after
2523 :meth:`_expression.TextualSelect.subquery` is invoked.
2525 The column expressions we pass to
2526 :meth:`_expression.TextClause.columns` may
2527 also be typed; when we do so, these :class:`.TypeEngine` objects become
2528 the effective return type of the column, so that SQLAlchemy's
2529 result-set-processing systems may be used on the return values.
2530 This is often needed for types such as date or boolean types, as well
2531 as for unicode processing on some dialect configurations::
2533 stmt = text("SELECT id, name, timestamp FROM some_table")
2534 stmt = stmt.columns(
2535 column("id", Integer),
2536 column("name", Unicode),
2537 column("timestamp", DateTime),
2538 )
2540 for id, name, timestamp in connection.execute(stmt):
2541 print(id, name, timestamp)
2543 As a shortcut to the above syntax, keyword arguments referring to
2544 types alone may be used, if only type conversion is needed::
2546 stmt = text("SELECT id, name, timestamp FROM some_table")
2547 stmt = stmt.columns(id=Integer, name=Unicode, timestamp=DateTime)
2549 for id, name, timestamp in connection.execute(stmt):
2550 print(id, name, timestamp)
2552 The positional form of :meth:`_expression.TextClause.columns`
2553 also provides the
2554 unique feature of **positional column targeting**, which is
2555 particularly useful when using the ORM with complex textual queries. If
2556 we specify the columns from our model to
2557 :meth:`_expression.TextClause.columns`,
2558 the result set will match to those columns positionally, meaning the
2559 name or origin of the column in the textual SQL doesn't matter::
2561 stmt = text(
2562 "SELECT users.id, addresses.id, users.id, "
2563 "users.name, addresses.email_address AS email "
2564 "FROM users JOIN addresses ON users.id=addresses.user_id "
2565 "WHERE users.id = 1"
2566 ).columns(
2567 User.id,
2568 Address.id,
2569 Address.user_id,
2570 User.name,
2571 Address.email_address,
2572 )
2574 query = (
2575 session.query(User)
2576 .from_statement(stmt)
2577 .options(contains_eager(User.addresses))
2578 )
2580 The :meth:`_expression.TextClause.columns` method provides a direct
2581 route to calling :meth:`_expression.FromClause.subquery` as well as
2582 :meth:`_expression.SelectBase.cte`
2583 against a textual SELECT statement::
2585 stmt = stmt.columns(id=Integer, name=String).cte("st")
2587 stmt = select(sometable).where(sometable.c.id == stmt.c.id)
2589 :param \*cols: A series of :class:`_expression.ColumnElement` objects,
2590 typically
2591 :class:`_schema.Column` objects from a :class:`_schema.Table`
2592 or ORM level
2593 column-mapped attributes, representing a set of columns that this
2594 textual string will SELECT from.
2596 :param \**types: A mapping of string names to :class:`.TypeEngine`
2597 type objects indicating the datatypes to use for names that are
2598 SELECTed from the textual string. Prefer to use the ``*cols``
2599 argument as it also indicates positional ordering.
2601 """
2602 selectable = util.preloaded.sql_selectable
2604 input_cols: List[NamedColumn[Any]] = [
2605 coercions.expect(roles.LabeledColumnExprRole, col) for col in cols
2606 ]
2608 positional_input_cols = [
2609 (
2610 ColumnClause(col.key, types.pop(col.key))
2611 if col.key in types
2612 else col
2613 )
2614 for col in input_cols
2615 ]
2616 keyed_input_cols: List[NamedColumn[Any]] = [
2617 ColumnClause(key, type_) for key, type_ in types.items()
2618 ]
2620 elem = selectable.TextualSelect.__new__(selectable.TextualSelect)
2621 elem._init(
2622 self,
2623 positional_input_cols + keyed_input_cols,
2624 positional=bool(positional_input_cols) and not keyed_input_cols,
2625 )
2626 return elem
2628 @property
2629 def type(self) -> TypeEngine[Any]:
2630 return type_api.NULLTYPE
2632 @property
2633 def comparator(self):
2634 # TODO: this seems wrong, it seems like we might not
2635 # be using this method.
2636 return self.type.comparator_factory(self) # type: ignore
2638 def self_group(
2639 self, against: Optional[OperatorType] = None
2640 ) -> Union[Self, Grouping[Any]]:
2641 if against is operators.in_op:
2642 return Grouping(self)
2643 else:
2644 return self
2647class Null(SingletonConstant, roles.ConstExprRole[None], ColumnElement[None]):
2648 """Represent the NULL keyword in a SQL statement.
2650 :class:`.Null` is accessed as a constant via the
2651 :func:`.null` function.
2653 """
2655 __visit_name__ = "null"
2657 _traverse_internals: _TraverseInternalsType = []
2658 _singleton: Null
2660 if not TYPE_CHECKING:
2662 @util.memoized_property
2663 def type(self) -> TypeEngine[_T]: # noqa: A001
2664 return type_api.NULLTYPE
2666 @classmethod
2667 def _instance(cls) -> Null:
2668 """Return a constant :class:`.Null` construct."""
2670 return Null._singleton
2673Null._create_singleton()
2676class False_(
2677 SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]
2678):
2679 """Represent the ``false`` keyword, or equivalent, in a SQL statement.
2681 :class:`.False_` is accessed as a constant via the
2682 :func:`.false` function.
2684 """
2686 __visit_name__ = "false"
2687 _traverse_internals: _TraverseInternalsType = []
2688 _singleton: False_
2690 if not TYPE_CHECKING:
2692 @util.memoized_property
2693 def type(self) -> TypeEngine[_T]: # noqa: A001
2694 return type_api.BOOLEANTYPE
2696 def _negate(self) -> True_:
2697 return True_._singleton
2699 @classmethod
2700 def _instance(cls) -> False_:
2701 return False_._singleton
2704False_._create_singleton()
2707class True_(SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]):
2708 """Represent the ``true`` keyword, or equivalent, in a SQL statement.
2710 :class:`.True_` is accessed as a constant via the
2711 :func:`.true` function.
2713 """
2715 __visit_name__ = "true"
2717 _traverse_internals: _TraverseInternalsType = []
2718 _singleton: True_
2720 if not TYPE_CHECKING:
2722 @util.memoized_property
2723 def type(self) -> TypeEngine[_T]: # noqa: A001
2724 return type_api.BOOLEANTYPE
2726 def _negate(self) -> False_:
2727 return False_._singleton
2729 @classmethod
2730 def _ifnone(
2731 cls, other: Optional[ColumnElement[Any]]
2732 ) -> ColumnElement[Any]:
2733 if other is None:
2734 return cls._instance()
2735 else:
2736 return other
2738 @classmethod
2739 def _instance(cls) -> True_:
2740 return True_._singleton
2743True_._create_singleton()
2746class ElementList(DQLDMLClauseElement):
2747 """Describe a list of clauses that will be space separated.
2749 This is a minimal version of :class:`.ClauseList` which is used by
2750 the :class:`.HasSyntaxExtension` class. It does not do any coercions
2751 so should be used internally only.
2753 .. versionadded:: 2.1
2755 """
2757 __visit_name__ = "element_list"
2759 _traverse_internals: _TraverseInternalsType = [
2760 ("clauses", InternalTraversal.dp_clauseelement_tuple),
2761 ]
2763 clauses: typing_Tuple[ClauseElement, ...]
2765 def __init__(self, clauses: Sequence[ClauseElement]):
2766 self.clauses = tuple(clauses)
2769class ClauseList(
2770 roles.InElementRole,
2771 roles.OrderByRole,
2772 roles.ColumnsClauseRole,
2773 roles.DMLColumnRole,
2774 DQLDMLClauseElement,
2775):
2776 """Describe a list of clauses, separated by an operator.
2778 By default, is comma-separated, such as a column listing.
2780 """
2782 __visit_name__ = "clauselist"
2784 # this is used only by the ORM in a legacy use case for
2785 # composite attributes
2786 _is_clause_list = True
2788 _traverse_internals: _TraverseInternalsType = [
2789 ("clauses", InternalTraversal.dp_clauseelement_list),
2790 ("operator", InternalTraversal.dp_operator),
2791 ]
2793 clauses: List[ColumnElement[Any]]
2795 def __init__(
2796 self,
2797 *clauses: _ColumnExpressionArgument[Any],
2798 operator: OperatorType = operators.comma_op,
2799 group: bool = True,
2800 group_contents: bool = True,
2801 _literal_as_text_role: Type[roles.SQLRole] = roles.WhereHavingRole,
2802 ):
2803 self.operator = operator
2804 self.group = group
2805 self.group_contents = group_contents
2806 clauses_iterator: Iterable[_ColumnExpressionArgument[Any]] = clauses
2807 text_converter_role: Type[roles.SQLRole] = _literal_as_text_role
2808 self._text_converter_role = text_converter_role
2810 if self.group_contents:
2811 self.clauses = [
2812 coercions.expect(
2813 text_converter_role, clause, apply_propagate_attrs=self
2814 ).self_group(against=self.operator)
2815 for clause in clauses_iterator
2816 ]
2817 else:
2818 self.clauses = [
2819 coercions.expect(
2820 text_converter_role, clause, apply_propagate_attrs=self
2821 )
2822 for clause in clauses_iterator
2823 ]
2824 self._is_implicitly_boolean = operators.is_boolean(self.operator)
2826 @classmethod
2827 def _construct_raw(
2828 cls,
2829 operator: OperatorType,
2830 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
2831 ) -> ClauseList:
2832 self = cls.__new__(cls)
2833 self.clauses = list(clauses) if clauses else []
2834 self.group = True
2835 self.operator = operator
2836 self.group_contents = True
2837 self._is_implicitly_boolean = False
2838 return self
2840 def __iter__(self) -> Iterator[ColumnElement[Any]]:
2841 return iter(self.clauses)
2843 def __len__(self) -> int:
2844 return len(self.clauses)
2846 @property
2847 def _select_iterable(self) -> _SelectIterable:
2848 return itertools.chain.from_iterable(
2849 [elem._select_iterable for elem in self.clauses]
2850 )
2852 def append(self, clause):
2853 if self.group_contents:
2854 self.clauses.append(
2855 coercions.expect(self._text_converter_role, clause).self_group(
2856 against=self.operator
2857 )
2858 )
2859 else:
2860 self.clauses.append(
2861 coercions.expect(self._text_converter_role, clause)
2862 )
2864 @util.ro_non_memoized_property
2865 def _from_objects(self) -> List[FromClause]:
2866 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
2868 def self_group(
2869 self, against: Optional[OperatorType] = None
2870 ) -> Union[Self, Grouping[Any]]:
2871 if self.group and operators.is_precedent(self.operator, against):
2872 return Grouping(self)
2873 else:
2874 return self
2877class OperatorExpression(ColumnElement[_T]):
2878 """base for expressions that contain an operator and operands
2880 .. versionadded:: 2.0
2882 """
2884 operator: OperatorType
2885 type: TypeEngine[_T]
2887 group: bool = True
2889 @property
2890 def is_comparison(self):
2891 return operators.is_comparison(self.operator)
2893 def self_group(
2894 self, against: Optional[OperatorType] = None
2895 ) -> Union[Self, Grouping[_T]]:
2896 if (
2897 self.group
2898 and operators.is_precedent(self.operator, against)
2899 or (
2900 # a negate against a non-boolean operator
2901 # doesn't make too much sense but we should
2902 # group for that
2903 against is operators.inv
2904 and not operators.is_boolean(self.operator)
2905 )
2906 ):
2907 return Grouping(self)
2908 else:
2909 return self
2911 @property
2912 def _flattened_operator_clauses(
2913 self,
2914 ) -> typing_Tuple[ColumnElement[Any], ...]:
2915 raise NotImplementedError()
2917 @classmethod
2918 def _construct_for_op(
2919 cls,
2920 left: ColumnElement[Any],
2921 right: ColumnElement[Any],
2922 op: OperatorType,
2923 *,
2924 type_: TypeEngine[_T],
2925 negate: Optional[OperatorType] = None,
2926 modifiers: Optional[Mapping[str, Any]] = None,
2927 ) -> OperatorExpression[_T]:
2928 if operators.is_associative(op):
2929 assert (
2930 negate is None
2931 ), f"negate not supported for associative operator {op}"
2933 multi = False
2934 if getattr(
2935 left, "operator", None
2936 ) is op and type_._compare_type_affinity(left.type):
2937 multi = True
2938 left_flattened = left._flattened_operator_clauses
2939 else:
2940 left_flattened = (left,)
2942 if getattr(
2943 right, "operator", None
2944 ) is op and type_._compare_type_affinity(right.type):
2945 multi = True
2946 right_flattened = right._flattened_operator_clauses
2947 else:
2948 right_flattened = (right,)
2950 if multi:
2951 return ExpressionClauseList._construct_for_list(
2952 op,
2953 type_,
2954 *(left_flattened + right_flattened),
2955 )
2957 if right._is_collection_aggregate:
2958 negate = None
2960 return BinaryExpression(
2961 left, right, op, type_=type_, negate=negate, modifiers=modifiers
2962 )
2965class ExpressionClauseList(OperatorExpression[_T]):
2966 """Describe a list of clauses, separated by an operator,
2967 in a column expression context.
2969 :class:`.ExpressionClauseList` differs from :class:`.ClauseList` in that
2970 it represents a column-oriented DQL expression only, not an open ended
2971 list of anything comma separated.
2973 .. versionadded:: 2.0
2975 """
2977 __visit_name__ = "expression_clauselist"
2979 _traverse_internals: _TraverseInternalsType = [
2980 ("clauses", InternalTraversal.dp_clauseelement_tuple),
2981 ("operator", InternalTraversal.dp_operator),
2982 ]
2984 clauses: typing_Tuple[ColumnElement[Any], ...]
2986 group: bool
2988 def __init__(
2989 self,
2990 operator: OperatorType,
2991 *clauses: _ColumnExpressionArgument[Any],
2992 type_: Optional[_TypeEngineArgument[_T]] = None,
2993 ):
2994 self.operator = operator
2996 self.clauses = tuple(
2997 coercions.expect(
2998 roles.ExpressionElementRole, clause, apply_propagate_attrs=self
2999 )
3000 for clause in clauses
3001 )
3002 self._is_implicitly_boolean = operators.is_boolean(self.operator)
3003 self.type = type_api.to_instance(type_) # type: ignore
3005 @property
3006 def _flattened_operator_clauses(
3007 self,
3008 ) -> typing_Tuple[ColumnElement[Any], ...]:
3009 return self.clauses
3011 def __iter__(self) -> Iterator[ColumnElement[Any]]:
3012 return iter(self.clauses)
3014 def __len__(self) -> int:
3015 return len(self.clauses)
3017 @property
3018 def _select_iterable(self) -> _SelectIterable:
3019 return (self,)
3021 @util.ro_non_memoized_property
3022 def _from_objects(self) -> List[FromClause]:
3023 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
3025 def _append_inplace(self, clause: ColumnElement[Any]) -> None:
3026 self.clauses += (clause,)
3028 @classmethod
3029 def _construct_for_list(
3030 cls,
3031 operator: OperatorType,
3032 type_: TypeEngine[_T],
3033 *clauses: ColumnElement[Any],
3034 group: bool = True,
3035 ) -> ExpressionClauseList[_T]:
3036 self = cls.__new__(cls)
3037 self.group = group
3038 if group:
3039 self.clauses = tuple(
3040 c.self_group(against=operator) for c in clauses
3041 )
3042 else:
3043 self.clauses = clauses
3044 self.operator = operator
3045 self.type = type_
3046 for c in clauses:
3047 if c._propagate_attrs:
3048 self._propagate_attrs = c._propagate_attrs
3049 break
3050 return self
3052 def _negate(self) -> Any:
3053 grouped = self.self_group(against=operators.inv)
3054 assert isinstance(grouped, ColumnElement)
3055 return UnaryExpression(grouped, operator=operators.inv)
3058class BooleanClauseList(ExpressionClauseList[bool]):
3059 __visit_name__ = "expression_clauselist"
3060 inherit_cache = True
3062 def __init__(self, *arg, **kw):
3063 raise NotImplementedError(
3064 "BooleanClauseList has a private constructor"
3065 )
3067 @classmethod
3068 def _process_clauses_for_boolean(
3069 cls,
3070 operator: OperatorType,
3071 continue_on: Any,
3072 skip_on: Any,
3073 clauses: Iterable[ColumnElement[Any]],
3074 ) -> typing_Tuple[int, List[ColumnElement[Any]]]:
3075 has_continue_on = None
3077 convert_clauses = []
3079 against = operators._asbool
3080 lcc = 0
3082 for clause in clauses:
3083 if clause is continue_on:
3084 # instance of continue_on, like and_(x, y, True, z), store it
3085 # if we didn't find one already, we will use it if there
3086 # are no other expressions here.
3087 has_continue_on = clause
3088 elif clause is skip_on:
3089 # instance of skip_on, e.g. and_(x, y, False, z), cancels
3090 # the rest out
3091 convert_clauses = [clause]
3092 lcc = 1
3093 break
3094 else:
3095 if not lcc:
3096 lcc = 1
3097 else:
3098 against = operator
3099 # technically this would be len(convert_clauses) + 1
3100 # however this only needs to indicate "greater than one"
3101 lcc = 2
3102 convert_clauses.append(clause)
3104 if not convert_clauses and has_continue_on is not None:
3105 convert_clauses = [has_continue_on]
3106 lcc = 1
3108 return lcc, [c.self_group(against=against) for c in convert_clauses]
3110 @classmethod
3111 def _construct(
3112 cls,
3113 operator: OperatorType,
3114 continue_on: Any,
3115 skip_on: Any,
3116 initial_clause: Any = _NoArg.NO_ARG,
3117 *clauses: Any,
3118 **kw: Any,
3119 ) -> ColumnElement[Any]:
3120 if initial_clause is _NoArg.NO_ARG:
3121 # no elements period. deprecated use case. return an empty
3122 # ClauseList construct that generates nothing unless it has
3123 # elements added to it.
3124 name = operator.__name__
3126 util.warn_deprecated(
3127 f"Invoking {name}() without arguments is deprecated, and "
3128 f"will be disallowed in a future release. For an empty "
3129 f"""{name}() construct, use '{name}({
3130 'true()' if continue_on is True_._singleton else 'false()'
3131 }, *args)' """
3132 f"""or '{name}({
3133 'True' if continue_on is True_._singleton else 'False'
3134 }, *args)'.""",
3135 version="1.4",
3136 )
3137 return cls._construct_raw(operator)
3139 lcc, convert_clauses = cls._process_clauses_for_boolean(
3140 operator,
3141 continue_on,
3142 skip_on,
3143 [
3144 coercions.expect(roles.WhereHavingRole, clause)
3145 for clause in util.coerce_generator_arg(
3146 (initial_clause,) + clauses
3147 )
3148 ],
3149 )
3151 if lcc > 1:
3152 # multiple elements. Return regular BooleanClauseList
3153 # which will link elements against the operator.
3155 flattened_clauses = itertools.chain.from_iterable(
3156 (
3157 (c for c in to_flat._flattened_operator_clauses)
3158 if getattr(to_flat, "operator", None) is operator
3159 else (to_flat,)
3160 )
3161 for to_flat in convert_clauses
3162 )
3164 return cls._construct_raw(operator, flattened_clauses) # type: ignore # noqa: E501
3165 else:
3166 assert lcc
3167 # just one element. return it as a single boolean element,
3168 # not a list and discard the operator.
3169 return convert_clauses[0]
3171 @classmethod
3172 def _construct_for_whereclause(
3173 cls, clauses: Iterable[ColumnElement[Any]]
3174 ) -> Optional[ColumnElement[bool]]:
3175 operator, continue_on, skip_on = (
3176 operators.and_,
3177 True_._singleton,
3178 False_._singleton,
3179 )
3181 lcc, convert_clauses = cls._process_clauses_for_boolean(
3182 operator,
3183 continue_on,
3184 skip_on,
3185 clauses, # these are assumed to be coerced already
3186 )
3188 if lcc > 1:
3189 # multiple elements. Return regular BooleanClauseList
3190 # which will link elements against the operator.
3191 return cls._construct_raw(operator, convert_clauses)
3192 elif lcc == 1:
3193 # just one element. return it as a single boolean element,
3194 # not a list and discard the operator.
3195 return convert_clauses[0]
3196 else:
3197 return None
3199 @classmethod
3200 def _construct_raw(
3201 cls,
3202 operator: OperatorType,
3203 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
3204 ) -> BooleanClauseList:
3205 self = cls.__new__(cls)
3206 self.clauses = tuple(clauses) if clauses else ()
3207 self.group = True
3208 self.operator = operator
3209 self.type = type_api.BOOLEANTYPE
3210 self._is_implicitly_boolean = True
3211 return self
3213 @classmethod
3214 def and_(
3215 cls,
3216 initial_clause: Union[
3217 Literal[True], _ColumnExpressionArgument[bool], _NoArg
3218 ] = _NoArg.NO_ARG,
3219 *clauses: _ColumnExpressionArgument[bool],
3220 ) -> ColumnElement[bool]:
3221 r"""Produce a conjunction of expressions joined by ``AND``.
3223 See :func:`_sql.and_` for full documentation.
3224 """
3225 return cls._construct(
3226 operators.and_,
3227 True_._singleton,
3228 False_._singleton,
3229 initial_clause,
3230 *clauses,
3231 )
3233 @classmethod
3234 def or_(
3235 cls,
3236 initial_clause: Union[
3237 Literal[False], _ColumnExpressionArgument[bool], _NoArg
3238 ] = _NoArg.NO_ARG,
3239 *clauses: _ColumnExpressionArgument[bool],
3240 ) -> ColumnElement[bool]:
3241 """Produce a conjunction of expressions joined by ``OR``.
3243 See :func:`_sql.or_` for full documentation.
3244 """
3245 return cls._construct(
3246 operators.or_,
3247 False_._singleton,
3248 True_._singleton,
3249 initial_clause,
3250 *clauses,
3251 )
3253 @property
3254 def _select_iterable(self) -> _SelectIterable:
3255 return (self,)
3257 def self_group(
3258 self, against: Optional[OperatorType] = None
3259 ) -> Union[Self, Grouping[bool]]:
3260 if not self.clauses:
3261 return self
3262 else:
3263 return super().self_group(against=against)
3266and_ = BooleanClauseList.and_
3267or_ = BooleanClauseList.or_
3270class Tuple(ClauseList, ColumnElement[TupleAny]):
3271 """Represent a SQL tuple."""
3273 __visit_name__ = "tuple"
3275 _traverse_internals: _TraverseInternalsType = (
3276 ClauseList._traverse_internals + []
3277 )
3279 type: TupleType
3281 @util.preload_module("sqlalchemy.sql.sqltypes")
3282 def __init__(
3283 self,
3284 *clauses: _ColumnExpressionArgument[Any],
3285 types: Optional[Sequence[_TypeEngineArgument[Any]]] = None,
3286 ):
3287 sqltypes = util.preloaded.sql_sqltypes
3289 if types is None:
3290 init_clauses: List[ColumnElement[Any]] = [
3291 coercions.expect(roles.ExpressionElementRole, c)
3292 for c in clauses
3293 ]
3294 else:
3295 if len(types) != len(clauses):
3296 raise exc.ArgumentError(
3297 "Wrong number of elements for %d-tuple: %r "
3298 % (len(types), clauses)
3299 )
3300 init_clauses = [
3301 coercions.expect(
3302 roles.ExpressionElementRole,
3303 c,
3304 type_=typ if not typ._isnull else None,
3305 )
3306 for typ, c in zip(types, clauses)
3307 ]
3309 self.type = sqltypes.TupleType(*[arg.type for arg in init_clauses])
3310 super().__init__(*init_clauses)
3312 @property
3313 def _select_iterable(self) -> _SelectIterable:
3314 return (self,)
3316 def _bind_param(self, operator, obj, type_=None, expanding=False):
3317 if expanding:
3318 return BindParameter(
3319 None,
3320 value=obj,
3321 _compared_to_operator=operator,
3322 unique=True,
3323 expanding=True,
3324 type_=type_,
3325 _compared_to_type=self.type,
3326 )
3327 else:
3328 return Tuple(
3329 *[
3330 BindParameter(
3331 None,
3332 o,
3333 _compared_to_operator=operator,
3334 _compared_to_type=compared_to_type,
3335 unique=True,
3336 type_=type_,
3337 )
3338 for o, compared_to_type in zip(obj, self.type.types)
3339 ]
3340 )
3342 def self_group(self, against: Optional[OperatorType] = None) -> Self:
3343 # Tuple is parenthesized by definition.
3344 return self
3347class Case(ColumnElement[_T]):
3348 """Represent a ``CASE`` expression.
3350 :class:`.Case` is produced using the :func:`.case` factory function,
3351 as in::
3353 from sqlalchemy import case
3355 stmt = select(users_table).where(
3356 case(
3357 (users_table.c.name == "wendy", "W"),
3358 (users_table.c.name == "jack", "J"),
3359 else_="E",
3360 )
3361 )
3363 Details on :class:`.Case` usage is at :func:`.case`.
3365 .. seealso::
3367 :func:`.case`
3369 """
3371 __visit_name__ = "case"
3373 _traverse_internals: _TraverseInternalsType = [
3374 ("value", InternalTraversal.dp_clauseelement),
3375 ("whens", InternalTraversal.dp_clauseelement_tuples),
3376 ("else_", InternalTraversal.dp_clauseelement),
3377 ]
3379 # for case(), the type is derived from the whens. so for the moment
3380 # users would have to cast() the case to get a specific type
3382 whens: List[typing_Tuple[ColumnElement[bool], ColumnElement[_T]]]
3383 else_: Optional[ColumnElement[_T]]
3384 value: Optional[ColumnElement[Any]]
3386 def __init__(
3387 self,
3388 *whens: Union[
3389 typing_Tuple[_ColumnExpressionArgument[bool], Any],
3390 Mapping[Any, Any],
3391 ],
3392 value: Optional[Any] = None,
3393 else_: Optional[Any] = None,
3394 ):
3395 new_whens: Iterable[Any] = coercions._expression_collection_was_a_list(
3396 "whens", "case", whens
3397 )
3398 try:
3399 new_whens = util.dictlike_iteritems(new_whens)
3400 except TypeError:
3401 pass
3403 self.whens = [
3404 (
3405 coercions.expect(
3406 roles.ExpressionElementRole,
3407 c,
3408 apply_propagate_attrs=self,
3409 ).self_group(),
3410 coercions.expect(roles.ExpressionElementRole, r),
3411 )
3412 for (c, r) in new_whens
3413 ]
3415 if value is None:
3416 self.value = None
3417 else:
3418 self.value = coercions.expect(roles.ExpressionElementRole, value)
3420 if else_ is not None:
3421 self.else_ = coercions.expect(roles.ExpressionElementRole, else_)
3422 else:
3423 self.else_ = None
3425 type_ = next(
3426 (
3427 then.type
3428 # Iterate `whens` in reverse to match previous behaviour
3429 # where type of final element took priority
3430 for *_, then in reversed(self.whens)
3431 if not then.type._isnull
3432 ),
3433 self.else_.type if self.else_ is not None else type_api.NULLTYPE,
3434 )
3435 self.type = cast(_T, type_)
3437 @util.ro_non_memoized_property
3438 def _from_objects(self) -> List[FromClause]:
3439 return list(
3440 itertools.chain(*[x._from_objects for x in self.get_children()])
3441 )
3444class Cast(WrapsColumnExpression[_T]):
3445 """Represent a ``CAST`` expression.
3447 :class:`.Cast` is produced using the :func:`.cast` factory function,
3448 as in::
3450 from sqlalchemy import cast, Numeric
3452 stmt = select(cast(product_table.c.unit_price, Numeric(10, 4)))
3454 Details on :class:`.Cast` usage is at :func:`.cast`.
3456 .. seealso::
3458 :ref:`tutorial_casts`
3460 :func:`.cast`
3462 :func:`.try_cast`
3464 :func:`.type_coerce` - an alternative to CAST that coerces the type
3465 on the Python side only, which is often sufficient to generate the
3466 correct SQL and data coercion.
3468 """
3470 __visit_name__ = "cast"
3472 _traverse_internals: _TraverseInternalsType = [
3473 ("clause", InternalTraversal.dp_clauseelement),
3474 ("type", InternalTraversal.dp_type),
3475 ]
3477 clause: ColumnElement[Any]
3478 type: TypeEngine[_T]
3479 typeclause: TypeClause
3481 def __init__(
3482 self,
3483 expression: _ColumnExpressionArgument[Any],
3484 type_: _TypeEngineArgument[_T],
3485 ):
3486 self.type = type_api.to_instance(type_)
3487 self.clause = coercions.expect(
3488 roles.ExpressionElementRole,
3489 expression,
3490 type_=self.type,
3491 apply_propagate_attrs=self,
3492 )
3493 self.typeclause = TypeClause(self.type)
3495 @util.ro_non_memoized_property
3496 def _from_objects(self) -> List[FromClause]:
3497 return self.clause._from_objects
3499 @property
3500 def wrapped_column_expression(self):
3501 return self.clause
3504class TryCast(Cast[_T]):
3505 """Represent a TRY_CAST expression.
3507 Details on :class:`.TryCast` usage is at :func:`.try_cast`.
3509 .. seealso::
3511 :func:`.try_cast`
3513 :ref:`tutorial_casts`
3514 """
3516 __visit_name__ = "try_cast"
3517 inherit_cache = True
3520class TypeCoerce(WrapsColumnExpression[_T]):
3521 """Represent a Python-side type-coercion wrapper.
3523 :class:`.TypeCoerce` supplies the :func:`_expression.type_coerce`
3524 function; see that function for usage details.
3526 .. seealso::
3528 :func:`_expression.type_coerce`
3530 :func:`.cast`
3532 """
3534 __visit_name__ = "type_coerce"
3536 _traverse_internals: _TraverseInternalsType = [
3537 ("clause", InternalTraversal.dp_clauseelement),
3538 ("type", InternalTraversal.dp_type),
3539 ]
3541 clause: ColumnElement[Any]
3542 type: TypeEngine[_T]
3544 def __init__(
3545 self,
3546 expression: _ColumnExpressionArgument[Any],
3547 type_: _TypeEngineArgument[_T],
3548 ):
3549 self.type = type_api.to_instance(type_)
3550 self.clause = coercions.expect(
3551 roles.ExpressionElementRole,
3552 expression,
3553 type_=self.type,
3554 apply_propagate_attrs=self,
3555 )
3557 @util.ro_non_memoized_property
3558 def _from_objects(self) -> List[FromClause]:
3559 return self.clause._from_objects
3561 @HasMemoized.memoized_attribute
3562 def typed_expression(self):
3563 if isinstance(self.clause, BindParameter):
3564 bp = self.clause._clone()
3565 bp.type = self.type
3566 return bp
3567 else:
3568 return self.clause
3570 @property
3571 def wrapped_column_expression(self):
3572 return self.clause
3574 def self_group(
3575 self, against: Optional[OperatorType] = None
3576 ) -> TypeCoerce[_T]:
3577 grouped = self.clause.self_group(against=against)
3578 if grouped is not self.clause:
3579 return TypeCoerce(grouped, self.type)
3580 else:
3581 return self
3584class Extract(ColumnElement[int]):
3585 """Represent a SQL EXTRACT clause, ``extract(field FROM expr)``."""
3587 __visit_name__ = "extract"
3589 _traverse_internals: _TraverseInternalsType = [
3590 ("expr", InternalTraversal.dp_clauseelement),
3591 ("field", InternalTraversal.dp_string),
3592 ]
3594 expr: ColumnElement[Any]
3595 field: str
3597 def __init__(self, field: str, expr: _ColumnExpressionArgument[Any]):
3598 self.type = type_api.INTEGERTYPE
3599 self.field = field
3600 self.expr = coercions.expect(roles.ExpressionElementRole, expr)
3602 @util.ro_non_memoized_property
3603 def _from_objects(self) -> List[FromClause]:
3604 return self.expr._from_objects
3607class _label_reference(ColumnElement[_T]):
3608 """Wrap a column expression as it appears in a 'reference' context.
3610 This expression is any that includes an _order_by_label_element,
3611 which is a Label, or a DESC / ASC construct wrapping a Label.
3613 The production of _label_reference() should occur when an expression
3614 is added to this context; this includes the ORDER BY or GROUP BY of a
3615 SELECT statement, as well as a few other places, such as the ORDER BY
3616 within an OVER clause.
3618 """
3620 __visit_name__ = "label_reference"
3622 _traverse_internals: _TraverseInternalsType = [
3623 ("element", InternalTraversal.dp_clauseelement)
3624 ]
3626 element: ColumnElement[_T]
3628 def __init__(self, element: ColumnElement[_T]):
3629 self.element = element
3630 self._propagate_attrs = element._propagate_attrs
3632 @util.ro_non_memoized_property
3633 def _from_objects(self) -> List[FromClause]:
3634 return []
3637class _textual_label_reference(ColumnElement[Any]):
3638 __visit_name__ = "textual_label_reference"
3640 _traverse_internals: _TraverseInternalsType = [
3641 ("element", InternalTraversal.dp_string)
3642 ]
3644 def __init__(self, element: str):
3645 self.element = element
3647 @util.memoized_property
3648 def _text_clause(self) -> TextClause:
3649 return TextClause(self.element)
3652class UnaryExpression(ColumnElement[_T]):
3653 """Define a 'unary' expression.
3655 A unary expression has a single column expression
3656 and an operator. The operator can be placed on the left
3657 (where it is called the 'operator') or right (where it is called the
3658 'modifier') of the column expression.
3660 :class:`.UnaryExpression` is the basis for several unary operators
3661 including those used by :func:`.desc`, :func:`.asc`, :func:`.distinct`,
3662 :func:`.nulls_first` and :func:`.nulls_last`.
3664 """
3666 __visit_name__ = "unary"
3668 _traverse_internals: _TraverseInternalsType = [
3669 ("element", InternalTraversal.dp_clauseelement),
3670 ("operator", InternalTraversal.dp_operator),
3671 ("modifier", InternalTraversal.dp_operator),
3672 ]
3674 element: ColumnElement[Any]
3675 operator: Optional[OperatorType]
3676 modifier: Optional[OperatorType]
3678 def __init__(
3679 self,
3680 element: ColumnElement[Any],
3681 *,
3682 operator: Optional[OperatorType] = None,
3683 modifier: Optional[OperatorType] = None,
3684 type_: Optional[_TypeEngineArgument[_T]] = None,
3685 wraps_column_expression: bool = False, # legacy, not used as of 2.0.42
3686 ):
3687 self.operator = operator
3688 self.modifier = modifier
3689 self._propagate_attrs = element._propagate_attrs
3690 self.element = element.self_group(
3691 against=self.operator or self.modifier
3692 )
3694 # if type is None, we get NULLTYPE, which is our _T. But I don't
3695 # know how to get the overloads to express that correctly
3696 self.type = type_api.to_instance(type_) # type: ignore
3698 def _wraps_unnamed_column(self):
3699 ungrouped = self.element._ungroup()
3700 return (
3701 not isinstance(ungrouped, NamedColumn)
3702 or ungrouped._non_anon_label is None
3703 )
3705 @classmethod
3706 def _create_nulls_first(
3707 cls,
3708 column: _ColumnExpressionArgument[_T],
3709 ) -> UnaryExpression[_T]:
3710 return UnaryExpression(
3711 coercions.expect(roles.ByOfRole, column),
3712 modifier=operators.nulls_first_op,
3713 )
3715 @classmethod
3716 def _create_nulls_last(
3717 cls,
3718 column: _ColumnExpressionArgument[_T],
3719 ) -> UnaryExpression[_T]:
3720 return UnaryExpression(
3721 coercions.expect(roles.ByOfRole, column),
3722 modifier=operators.nulls_last_op,
3723 )
3725 @classmethod
3726 def _create_desc(
3727 cls, column: _ColumnExpressionOrStrLabelArgument[_T]
3728 ) -> UnaryExpression[_T]:
3729 return UnaryExpression(
3730 coercions.expect(roles.ByOfRole, column),
3731 modifier=operators.desc_op,
3732 )
3734 @classmethod
3735 def _create_asc(
3736 cls,
3737 column: _ColumnExpressionOrStrLabelArgument[_T],
3738 ) -> UnaryExpression[_T]:
3739 return UnaryExpression(
3740 coercions.expect(roles.ByOfRole, column),
3741 modifier=operators.asc_op,
3742 )
3744 @classmethod
3745 def _create_distinct(
3746 cls,
3747 expr: _ColumnExpressionArgument[_T],
3748 ) -> UnaryExpression[_T]:
3749 col_expr: ColumnElement[_T] = coercions.expect(
3750 roles.ExpressionElementRole, expr
3751 )
3752 return UnaryExpression(
3753 col_expr,
3754 operator=operators.distinct_op,
3755 type_=col_expr.type,
3756 )
3758 @classmethod
3759 def _create_bitwise_not(
3760 cls,
3761 expr: _ColumnExpressionArgument[_T],
3762 ) -> UnaryExpression[_T]:
3763 col_expr: ColumnElement[_T] = coercions.expect(
3764 roles.ExpressionElementRole, expr
3765 )
3766 return UnaryExpression(
3767 col_expr,
3768 operator=operators.bitwise_not_op,
3769 type_=col_expr.type,
3770 )
3772 @property
3773 def _order_by_label_element(self) -> Optional[Label[Any]]:
3774 if operators.is_order_by_modifier(self.modifier):
3775 return self.element._order_by_label_element
3776 else:
3777 return None
3779 @util.ro_non_memoized_property
3780 def _from_objects(self) -> List[FromClause]:
3781 return self.element._from_objects
3783 def _negate(self) -> ColumnElement[Any]:
3784 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
3785 return UnaryExpression(
3786 self.self_group(against=operators.inv),
3787 operator=operators.inv,
3788 type_=type_api.BOOLEANTYPE,
3789 )
3790 else:
3791 return ColumnElement._negate(self)
3793 def self_group(
3794 self, against: Optional[OperatorType] = None
3795 ) -> Union[Self, Grouping[_T]]:
3796 if self.operator and operators.is_precedent(self.operator, against):
3797 return Grouping(self)
3798 else:
3799 return self
3802class CollectionAggregate(UnaryExpression[_T]):
3803 """Forms the basis for right-hand collection operator modifiers
3804 ANY and ALL.
3806 The ANY and ALL keywords are available in different ways on different
3807 backends. On PostgreSQL, they only work for an ARRAY type. On
3808 MySQL, they only work for subqueries.
3810 """
3812 inherit_cache = True
3813 _is_collection_aggregate = True
3815 @classmethod
3816 def _create_any(
3817 cls, expr: _ColumnExpressionArgument[_T]
3818 ) -> CollectionAggregate[bool]:
3819 col_expr: ColumnElement[_T] = coercions.expect(
3820 roles.ExpressionElementRole,
3821 expr,
3822 )
3823 col_expr = col_expr.self_group()
3824 return CollectionAggregate(
3825 col_expr,
3826 operator=operators.any_op,
3827 type_=type_api.BOOLEANTYPE,
3828 )
3830 @classmethod
3831 def _create_all(
3832 cls, expr: _ColumnExpressionArgument[_T]
3833 ) -> CollectionAggregate[bool]:
3834 col_expr: ColumnElement[_T] = coercions.expect(
3835 roles.ExpressionElementRole,
3836 expr,
3837 )
3838 col_expr = col_expr.self_group()
3839 return CollectionAggregate(
3840 col_expr,
3841 operator=operators.all_op,
3842 type_=type_api.BOOLEANTYPE,
3843 )
3845 # operate and reverse_operate are hardwired to
3846 # dispatch onto the type comparator directly, so that we can
3847 # ensure "reversed" behavior.
3848 def operate(
3849 self, op: OperatorType, *other: Any, **kwargs: Any
3850 ) -> ColumnElement[_T]:
3851 if not operators.is_comparison(op):
3852 raise exc.ArgumentError(
3853 "Only comparison operators may be used with ANY/ALL"
3854 )
3855 kwargs["reverse"] = True
3856 return self.comparator.operate(operators.mirror(op), *other, **kwargs)
3858 def reverse_operate(
3859 self, op: OperatorType, other: Any, **kwargs: Any
3860 ) -> ColumnElement[_T]:
3861 # comparison operators should never call reverse_operate
3862 assert not operators.is_comparison(op)
3863 raise exc.ArgumentError(
3864 "Only comparison operators may be used with ANY/ALL"
3865 )
3868class AsBoolean(WrapsColumnExpression[bool], UnaryExpression[bool]):
3869 inherit_cache = True
3871 def __init__(self, element, operator, negate):
3872 self.element = element
3873 self.type = type_api.BOOLEANTYPE
3874 self.operator = operator
3875 self.negate = negate
3876 self.modifier = None
3877 self._is_implicitly_boolean = element._is_implicitly_boolean
3879 @property
3880 def wrapped_column_expression(self):
3881 return self.element
3883 def self_group(self, against: Optional[OperatorType] = None) -> Self:
3884 return self
3886 def _negate(self):
3887 if isinstance(self.element, (True_, False_)):
3888 return self.element._negate()
3889 else:
3890 return AsBoolean(self.element, self.negate, self.operator)
3893class BinaryExpression(OperatorExpression[_T]):
3894 """Represent an expression that is ``LEFT <operator> RIGHT``.
3896 A :class:`.BinaryExpression` is generated automatically
3897 whenever two column expressions are used in a Python binary expression:
3899 .. sourcecode:: pycon+sql
3901 >>> from sqlalchemy.sql import column
3902 >>> column("a") + column("b")
3903 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
3904 >>> print(column("a") + column("b"))
3905 {printsql}a + b
3907 """
3909 __visit_name__ = "binary"
3911 _traverse_internals: _TraverseInternalsType = [
3912 ("left", InternalTraversal.dp_clauseelement),
3913 ("right", InternalTraversal.dp_clauseelement),
3914 ("operator", InternalTraversal.dp_operator),
3915 ("negate", InternalTraversal.dp_operator),
3916 ("modifiers", InternalTraversal.dp_plain_dict),
3917 (
3918 "type",
3919 InternalTraversal.dp_type,
3920 ),
3921 ]
3923 _cache_key_traversal = [
3924 ("left", InternalTraversal.dp_clauseelement),
3925 ("right", InternalTraversal.dp_clauseelement),
3926 ("operator", InternalTraversal.dp_operator),
3927 ("modifiers", InternalTraversal.dp_plain_dict),
3928 # "type" affects JSON CAST operators, so while redundant in most cases,
3929 # is needed for that one
3930 (
3931 "type",
3932 InternalTraversal.dp_type,
3933 ),
3934 ]
3936 _is_implicitly_boolean = True
3937 """Indicates that any database will know this is a boolean expression
3938 even if the database does not have an explicit boolean datatype.
3940 """
3942 left: ColumnElement[Any]
3943 right: ColumnElement[Any]
3944 modifiers: Mapping[str, Any]
3946 def __init__(
3947 self,
3948 left: ColumnElement[Any],
3949 right: ColumnElement[Any],
3950 operator: OperatorType,
3951 type_: Optional[_TypeEngineArgument[_T]] = None,
3952 negate: Optional[OperatorType] = None,
3953 modifiers: Optional[Mapping[str, Any]] = None,
3954 ):
3955 # allow compatibility with libraries that
3956 # refer to BinaryExpression directly and pass strings
3957 if isinstance(operator, str):
3958 operator = operators.custom_op(operator)
3959 self._orig = (left.__hash__(), right.__hash__())
3960 self._propagate_attrs = left._propagate_attrs or right._propagate_attrs
3961 self.left = left.self_group(against=operator)
3962 self.right = right.self_group(against=operator)
3963 self.operator = operator
3965 # if type is None, we get NULLTYPE, which is our _T. But I don't
3966 # know how to get the overloads to express that correctly
3967 self.type = type_api.to_instance(type_) # type: ignore
3969 self.negate = negate
3970 self._is_implicitly_boolean = operators.is_boolean(operator)
3972 if modifiers is None:
3973 self.modifiers = {}
3974 else:
3975 self.modifiers = modifiers
3977 @property
3978 def _flattened_operator_clauses(
3979 self,
3980 ) -> typing_Tuple[ColumnElement[Any], ...]:
3981 return (self.left, self.right)
3983 def __bool__(self):
3984 """Implement Python-side "bool" for BinaryExpression as a
3985 simple "identity" check for the left and right attributes,
3986 if the operator is "eq" or "ne". Otherwise the expression
3987 continues to not support "bool" like all other column expressions.
3989 The rationale here is so that ColumnElement objects can be hashable.
3990 What? Well, suppose you do this::
3992 c1, c2 = column("x"), column("y")
3993 s1 = set([c1, c2])
3995 We do that **a lot**, columns inside of sets is an extremely basic
3996 thing all over the ORM for example.
3998 So what happens if we do this? ::
4000 c1 in s1
4002 Hashing means it will normally use ``__hash__()`` of the object,
4003 but in case of hash collision, it's going to also do ``c1 == c1``
4004 and/or ``c1 == c2`` inside. Those operations need to return a
4005 True/False value. But because we override ``==`` and ``!=``, they're
4006 going to get a BinaryExpression. Hence we implement ``__bool__`` here
4007 so that these comparisons behave in this particular context mostly
4008 like regular object comparisons. Thankfully Python is OK with
4009 that! Otherwise we'd have to use special set classes for columns
4010 (which we used to do, decades ago).
4012 """
4013 if self.operator in (operators.eq, operators.ne):
4014 # this is using the eq/ne operator given int hash values,
4015 # rather than Operator, so that "bool" can be based on
4016 # identity
4017 return self.operator(*self._orig) # type: ignore
4018 else:
4019 raise TypeError("Boolean value of this clause is not defined")
4021 if typing.TYPE_CHECKING:
4023 def __invert__(
4024 self: BinaryExpression[_T],
4025 ) -> BinaryExpression[_T]: ...
4027 @util.ro_non_memoized_property
4028 def _from_objects(self) -> List[FromClause]:
4029 return self.left._from_objects + self.right._from_objects
4031 def _negate(self):
4032 if self.negate is not None:
4033 return BinaryExpression(
4034 self.left,
4035 self.right._negate_in_binary(self.negate, self.operator),
4036 self.negate,
4037 negate=self.operator,
4038 type_=self.type,
4039 modifiers=self.modifiers,
4040 )
4041 else:
4042 return self.self_group()._negate()
4045class Slice(ColumnElement[Any]):
4046 """Represent SQL for a Python array-slice object.
4048 This is not a specific SQL construct at this level, but
4049 may be interpreted by specific dialects, e.g. PostgreSQL.
4051 """
4053 __visit_name__ = "slice"
4055 _traverse_internals: _TraverseInternalsType = [
4056 ("start", InternalTraversal.dp_clauseelement),
4057 ("stop", InternalTraversal.dp_clauseelement),
4058 ("step", InternalTraversal.dp_clauseelement),
4059 ]
4061 def __init__(self, start, stop, step, _name=None):
4062 self.start = coercions.expect(
4063 roles.ExpressionElementRole,
4064 start,
4065 name=_name,
4066 type_=type_api.INTEGERTYPE,
4067 )
4068 self.stop = coercions.expect(
4069 roles.ExpressionElementRole,
4070 stop,
4071 name=_name,
4072 type_=type_api.INTEGERTYPE,
4073 )
4074 self.step = coercions.expect(
4075 roles.ExpressionElementRole,
4076 step,
4077 name=_name,
4078 type_=type_api.INTEGERTYPE,
4079 )
4080 self.type = type_api.NULLTYPE
4082 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4083 assert against is operator.getitem
4084 return self
4087class IndexExpression(BinaryExpression[Any]):
4088 """Represent the class of expressions that are like an "index"
4089 operation."""
4091 inherit_cache = True
4094class GroupedElement(DQLDMLClauseElement):
4095 """Represent any parenthesized expression"""
4097 __visit_name__ = "grouping"
4099 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4100 return self
4102 def _ungroup(self) -> ClauseElement:
4103 raise NotImplementedError()
4106class Grouping(GroupedElement, ColumnElement[_T]):
4107 """Represent a grouping within a column expression"""
4109 _traverse_internals: _TraverseInternalsType = [
4110 ("element", InternalTraversal.dp_clauseelement),
4111 ("type", InternalTraversal.dp_type),
4112 ]
4114 _cache_key_traversal = [
4115 ("element", InternalTraversal.dp_clauseelement),
4116 ]
4118 element: Union[TextClause, ClauseList, ColumnElement[_T]]
4120 def __init__(
4121 self, element: Union[TextClause, ClauseList, ColumnElement[_T]]
4122 ):
4123 self.element = element
4125 # nulltype assignment issue
4126 self.type = getattr(element, "type", type_api.NULLTYPE) # type: ignore
4127 self._propagate_attrs = element._propagate_attrs
4129 def _with_binary_element_type(self, type_):
4130 return self.__class__(self.element._with_binary_element_type(type_))
4132 def _ungroup(self) -> ColumnElement[_T]:
4133 assert isinstance(self.element, ColumnElement)
4134 return self.element._ungroup()
4136 @util.memoized_property
4137 def _is_implicitly_boolean(self):
4138 return self.element._is_implicitly_boolean
4140 @util.non_memoized_property
4141 def _tq_label(self) -> Optional[str]:
4142 return (
4143 getattr(self.element, "_tq_label", None) or self._anon_name_label
4144 )
4146 @util.non_memoized_property
4147 def _proxies(self) -> List[ColumnElement[Any]]:
4148 if isinstance(self.element, ColumnElement):
4149 return [self.element]
4150 else:
4151 return []
4153 @util.ro_non_memoized_property
4154 def _from_objects(self) -> List[FromClause]:
4155 return self.element._from_objects
4157 def __getattr__(self, attr):
4158 return getattr(self.element, attr)
4160 def __getstate__(self):
4161 return {"element": self.element, "type": self.type}
4163 def __setstate__(self, state):
4164 self.element = state["element"]
4165 self.type = state["type"]
4167 if TYPE_CHECKING:
4169 def self_group(
4170 self, against: Optional[OperatorType] = None
4171 ) -> Self: ...
4174class _OverrideBinds(Grouping[_T]):
4175 """used by cache_key->_apply_params_to_element to allow compilation /
4176 execution of a SQL element that's been cached, using an alternate set of
4177 bound parameter values.
4179 This is used by the ORM to swap new parameter values into expressions
4180 that are embedded into loader options like with_expression(),
4181 selectinload(). Previously, this task was accomplished using the
4182 .params() method which would perform a deep-copy instead. This deep
4183 copy proved to be too expensive for more complex expressions.
4185 See #11085
4187 """
4189 __visit_name__ = "override_binds"
4191 def __init__(
4192 self,
4193 element: ColumnElement[_T],
4194 bindparams: Sequence[BindParameter[Any]],
4195 replaces_params: Sequence[BindParameter[Any]],
4196 ):
4197 self.element = element
4198 self.translate = {
4199 k.key: v.value for k, v in zip(replaces_params, bindparams)
4200 }
4202 def _gen_cache_key(
4203 self, anon_map: anon_map, bindparams: List[BindParameter[Any]]
4204 ) -> Optional[typing_Tuple[Any, ...]]:
4205 """generate a cache key for the given element, substituting its bind
4206 values for the translation values present."""
4208 existing_bps: List[BindParameter[Any]] = []
4209 ck = self.element._gen_cache_key(anon_map, existing_bps)
4211 bindparams.extend(
4212 (
4213 bp._with_value(
4214 self.translate[bp.key], maintain_key=True, required=False
4215 )
4216 if bp.key in self.translate
4217 else bp
4218 )
4219 for bp in existing_bps
4220 )
4222 return ck
4225class Over(ColumnElement[_T]):
4226 """Represent an OVER clause.
4228 This is a special operator against a so-called
4229 "window" function, as well as any aggregate function,
4230 which produces results relative to the result set
4231 itself. Most modern SQL backends now support window functions.
4233 """
4235 __visit_name__ = "over"
4237 _traverse_internals: _TraverseInternalsType = [
4238 ("element", InternalTraversal.dp_clauseelement),
4239 ("order_by", InternalTraversal.dp_clauseelement),
4240 ("partition_by", InternalTraversal.dp_clauseelement),
4241 ("range_", InternalTraversal.dp_clauseelement),
4242 ("rows", InternalTraversal.dp_clauseelement),
4243 ("groups", InternalTraversal.dp_clauseelement),
4244 ]
4246 order_by: Optional[ClauseList] = None
4247 partition_by: Optional[ClauseList] = None
4249 element: ColumnElement[_T]
4250 """The underlying expression object to which this :class:`.Over`
4251 object refers."""
4253 range_: Optional[_FrameClause]
4254 rows: Optional[_FrameClause]
4255 groups: Optional[_FrameClause]
4257 def __init__(
4258 self,
4259 element: ColumnElement[_T],
4260 partition_by: Optional[_ByArgument] = None,
4261 order_by: Optional[_ByArgument] = None,
4262 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4263 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4264 groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4265 ):
4266 self.element = element
4267 if order_by is not None:
4268 self.order_by = ClauseList(
4269 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4270 )
4271 if partition_by is not None:
4272 self.partition_by = ClauseList(
4273 *util.to_list(partition_by),
4274 _literal_as_text_role=roles.ByOfRole,
4275 )
4277 if sum(bool(item) for item in (range_, rows, groups)) > 1:
4278 raise exc.ArgumentError(
4279 "only one of 'rows', 'range_', or 'groups' may be provided"
4280 )
4281 else:
4282 self.range_ = _FrameClause(range_) if range_ else None
4283 self.rows = _FrameClause(rows) if rows else None
4284 self.groups = _FrameClause(groups) if groups else None
4286 if not TYPE_CHECKING:
4288 @util.memoized_property
4289 def type(self) -> TypeEngine[_T]: # noqa: A001
4290 return self.element.type
4292 @util.ro_non_memoized_property
4293 def _from_objects(self) -> List[FromClause]:
4294 return list(
4295 itertools.chain(
4296 *[
4297 c._from_objects
4298 for c in (self.element, self.partition_by, self.order_by)
4299 if c is not None
4300 ]
4301 )
4302 )
4305class _FrameClauseType(Enum):
4306 RANGE_UNBOUNDED = 0
4307 RANGE_CURRENT = 1
4308 RANGE_PRECEDING = 2
4309 RANGE_FOLLOWING = 3
4312class _FrameClause(ClauseElement):
4313 """indicate the 'rows' or 'range' field of a window function, e.g. using
4314 :class:`.Over`.
4316 .. versionadded:: 2.1
4318 """
4320 __visit_name__ = "frame_clause"
4322 _traverse_internals: _TraverseInternalsType = [
4323 ("lower_integer_bind", InternalTraversal.dp_clauseelement),
4324 ("upper_integer_bind", InternalTraversal.dp_clauseelement),
4325 ("lower_type", InternalTraversal.dp_plain_obj),
4326 ("upper_type", InternalTraversal.dp_plain_obj),
4327 ]
4329 def __init__(
4330 self,
4331 range_: typing_Tuple[Optional[int], Optional[int]],
4332 ):
4333 try:
4334 r0, r1 = range_
4335 except (ValueError, TypeError) as ve:
4336 raise exc.ArgumentError("2-tuple expected for range/rows") from ve
4338 if r0 is None:
4339 self.lower_type = _FrameClauseType.RANGE_UNBOUNDED
4340 self.lower_integer_bind = None
4341 else:
4342 try:
4343 lower_integer = int(r0)
4344 except ValueError as err:
4345 raise exc.ArgumentError(
4346 "Integer or None expected for range value"
4347 ) from err
4348 else:
4349 if lower_integer == 0:
4350 self.lower_type = _FrameClauseType.RANGE_CURRENT
4351 self.lower_integer_bind = None
4352 elif lower_integer < 0:
4353 self.lower_type = _FrameClauseType.RANGE_PRECEDING
4354 self.lower_integer_bind = literal(
4355 abs(lower_integer), type_api.INTEGERTYPE
4356 )
4357 else:
4358 self.lower_type = _FrameClauseType.RANGE_FOLLOWING
4359 self.lower_integer_bind = literal(
4360 lower_integer, type_api.INTEGERTYPE
4361 )
4363 if r1 is None:
4364 self.upper_type = _FrameClauseType.RANGE_UNBOUNDED
4365 self.upper_integer_bind = None
4366 else:
4367 try:
4368 upper_integer = int(r1)
4369 except ValueError as err:
4370 raise exc.ArgumentError(
4371 "Integer or None expected for range value"
4372 ) from err
4373 else:
4374 if upper_integer == 0:
4375 self.upper_type = _FrameClauseType.RANGE_CURRENT
4376 self.upper_integer_bind = None
4377 elif upper_integer < 0:
4378 self.upper_type = _FrameClauseType.RANGE_PRECEDING
4379 self.upper_integer_bind = literal(
4380 abs(upper_integer), type_api.INTEGERTYPE
4381 )
4382 else:
4383 self.upper_type = _FrameClauseType.RANGE_FOLLOWING
4384 self.upper_integer_bind = literal(
4385 upper_integer, type_api.INTEGERTYPE
4386 )
4389class WithinGroup(ColumnElement[_T]):
4390 """Represent a WITHIN GROUP (ORDER BY) clause.
4392 This is a special operator against so-called
4393 "ordered set aggregate" and "hypothetical
4394 set aggregate" functions, including ``percentile_cont()``,
4395 ``rank()``, ``dense_rank()``, etc.
4397 It's supported only by certain database backends, such as PostgreSQL,
4398 Oracle Database and MS SQL Server.
4400 The :class:`.WithinGroup` construct extracts its type from the
4401 method :meth:`.FunctionElement.within_group_type`. If this returns
4402 ``None``, the function's ``.type`` is used.
4404 """
4406 __visit_name__ = "withingroup"
4408 _traverse_internals: _TraverseInternalsType = [
4409 ("element", InternalTraversal.dp_clauseelement),
4410 ("order_by", InternalTraversal.dp_clauseelement),
4411 ]
4413 order_by: Optional[ClauseList] = None
4415 def __init__(
4416 self,
4417 element: Union[FunctionElement[_T], FunctionFilter[_T]],
4418 *order_by: _ColumnExpressionArgument[Any],
4419 ):
4420 self.element = element
4421 if order_by is not None:
4422 self.order_by = ClauseList(
4423 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4424 )
4426 def __reduce__(self):
4427 return self.__class__, (self.element,) + (
4428 tuple(self.order_by) if self.order_by is not None else ()
4429 )
4431 def over(
4432 self,
4433 *,
4434 partition_by: Optional[_ByArgument] = None,
4435 order_by: Optional[_ByArgument] = None,
4436 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4437 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4438 groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4439 ) -> Over[_T]:
4440 """Produce an OVER clause against this :class:`.WithinGroup`
4441 construct.
4443 This function has the same signature as that of
4444 :meth:`.FunctionElement.over`.
4446 """
4447 return Over(
4448 self,
4449 partition_by=partition_by,
4450 order_by=order_by,
4451 range_=range_,
4452 rows=rows,
4453 groups=groups,
4454 )
4456 @overload
4457 def filter(self) -> Self: ...
4459 @overload
4460 def filter(
4461 self,
4462 __criterion0: _ColumnExpressionArgument[bool],
4463 *criterion: _ColumnExpressionArgument[bool],
4464 ) -> FunctionFilter[_T]: ...
4466 def filter(
4467 self, *criterion: _ColumnExpressionArgument[bool]
4468 ) -> Union[Self, FunctionFilter[_T]]:
4469 """Produce a FILTER clause against this function."""
4470 if not criterion:
4471 return self
4472 return FunctionFilter(self, *criterion)
4474 if not TYPE_CHECKING:
4476 @util.memoized_property
4477 def type(self) -> TypeEngine[_T]: # noqa: A001
4478 wgt = self.element.within_group_type(self)
4479 if wgt is not None:
4480 return wgt
4481 else:
4482 return self.element.type
4484 @util.ro_non_memoized_property
4485 def _from_objects(self) -> List[FromClause]:
4486 return list(
4487 itertools.chain(
4488 *[
4489 c._from_objects
4490 for c in (self.element, self.order_by)
4491 if c is not None
4492 ]
4493 )
4494 )
4497class FunctionFilter(Generative, ColumnElement[_T]):
4498 """Represent a function FILTER clause.
4500 This is a special operator against aggregate and window functions,
4501 which controls which rows are passed to it.
4502 It's supported only by certain database backends.
4504 Invocation of :class:`.FunctionFilter` is via
4505 :meth:`.FunctionElement.filter`::
4507 func.count(1).filter(True)
4509 .. seealso::
4511 :meth:`.FunctionElement.filter`
4513 """
4515 __visit_name__ = "funcfilter"
4517 _traverse_internals: _TraverseInternalsType = [
4518 ("func", InternalTraversal.dp_clauseelement),
4519 ("criterion", InternalTraversal.dp_clauseelement),
4520 ]
4522 criterion: Optional[ColumnElement[bool]] = None
4524 def __init__(
4525 self,
4526 func: Union[FunctionElement[_T], WithinGroup[_T]],
4527 *criterion: _ColumnExpressionArgument[bool],
4528 ):
4529 self.func = func
4530 self.filter.non_generative(self, *criterion) # type: ignore
4532 @_generative
4533 def filter(self, *criterion: _ColumnExpressionArgument[bool]) -> Self:
4534 """Produce an additional FILTER against the function.
4536 This method adds additional criteria to the initial criteria
4537 set up by :meth:`.FunctionElement.filter`.
4539 Multiple criteria are joined together at SQL render time
4540 via ``AND``.
4543 """
4545 for crit in list(criterion):
4546 crit = coercions.expect(roles.WhereHavingRole, crit)
4548 if self.criterion is not None:
4549 self.criterion = self.criterion & crit
4550 else:
4551 self.criterion = crit
4553 return self
4555 def over(
4556 self,
4557 partition_by: Optional[
4558 Union[
4559 Iterable[_ColumnExpressionArgument[Any]],
4560 _ColumnExpressionArgument[Any],
4561 ]
4562 ] = None,
4563 order_by: Optional[
4564 Union[
4565 Iterable[_ColumnExpressionArgument[Any]],
4566 _ColumnExpressionArgument[Any],
4567 ]
4568 ] = None,
4569 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4570 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4571 groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4572 ) -> Over[_T]:
4573 """Produce an OVER clause against this filtered function.
4575 Used against aggregate or so-called "window" functions,
4576 for database backends that support window functions.
4578 The expression::
4580 func.rank().filter(MyClass.y > 5).over(order_by="x")
4582 is shorthand for::
4584 from sqlalchemy import over, funcfilter
4586 over(funcfilter(func.rank(), MyClass.y > 5), order_by="x")
4588 See :func:`_expression.over` for a full description.
4590 """
4591 return Over(
4592 self,
4593 partition_by=partition_by,
4594 order_by=order_by,
4595 range_=range_,
4596 rows=rows,
4597 groups=groups,
4598 )
4600 def within_group(
4601 self, *order_by: _ColumnExpressionArgument[Any]
4602 ) -> WithinGroup[_T]:
4603 """Produce a WITHIN GROUP (ORDER BY expr) clause against
4604 this function.
4605 """
4606 return WithinGroup(self, *order_by)
4608 def within_group_type(
4609 self, within_group: WithinGroup[_T]
4610 ) -> Optional[TypeEngine[_T]]:
4611 return None
4613 def self_group(
4614 self, against: Optional[OperatorType] = None
4615 ) -> Union[Self, Grouping[_T]]:
4616 if operators.is_precedent(operators.filter_op, against):
4617 return Grouping(self)
4618 else:
4619 return self
4621 if not TYPE_CHECKING:
4623 @util.memoized_property
4624 def type(self) -> TypeEngine[_T]: # noqa: A001
4625 return self.func.type
4627 @util.ro_non_memoized_property
4628 def _from_objects(self) -> List[FromClause]:
4629 return list(
4630 itertools.chain(
4631 *[
4632 c._from_objects
4633 for c in (self.func, self.criterion)
4634 if c is not None
4635 ]
4636 )
4637 )
4640class NamedColumn(KeyedColumnElement[_T]):
4641 is_literal = False
4642 table: Optional[FromClause] = None
4643 name: str
4644 key: str
4646 def _compare_name_for_result(self, other):
4647 return (hasattr(other, "name") and self.name == other.name) or (
4648 hasattr(other, "_label") and self._label == other._label
4649 )
4651 @util.ro_memoized_property
4652 def description(self) -> str:
4653 return self.name
4655 @HasMemoized.memoized_attribute
4656 def _tq_key_label(self) -> Optional[str]:
4657 """table qualified label based on column key.
4659 for table-bound columns this is <tablename>_<column key/proxy key>;
4661 all other expressions it resolves to key/proxy key.
4663 """
4664 proxy_key = self._proxy_key
4665 if proxy_key and proxy_key != self.name:
4666 return self._gen_tq_label(proxy_key)
4667 else:
4668 return self._tq_label
4670 @HasMemoized.memoized_attribute
4671 def _tq_label(self) -> Optional[str]:
4672 """table qualified label based on column name.
4674 for table-bound columns this is <tablename>_<columnname>; all other
4675 expressions it resolves to .name.
4677 """
4678 return self._gen_tq_label(self.name)
4680 @HasMemoized.memoized_attribute
4681 def _render_label_in_columns_clause(self):
4682 return True
4684 @HasMemoized.memoized_attribute
4685 def _non_anon_label(self):
4686 return self.name
4688 def _gen_tq_label(
4689 self, name: str, dedupe_on_key: bool = True
4690 ) -> Optional[str]:
4691 return name
4693 def _bind_param(
4694 self,
4695 operator: OperatorType,
4696 obj: Any,
4697 type_: Optional[TypeEngine[_T]] = None,
4698 expanding: bool = False,
4699 ) -> BindParameter[_T]:
4700 return BindParameter(
4701 self.key,
4702 obj,
4703 _compared_to_operator=operator,
4704 _compared_to_type=self.type,
4705 type_=type_,
4706 unique=True,
4707 expanding=expanding,
4708 )
4710 def _make_proxy(
4711 self,
4712 selectable: FromClause,
4713 *,
4714 primary_key: ColumnSet,
4715 foreign_keys: Set[KeyedColumnElement[Any]],
4716 name: Optional[str] = None,
4717 key: Optional[str] = None,
4718 name_is_truncatable: bool = False,
4719 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
4720 disallow_is_literal: bool = False,
4721 **kw: Any,
4722 ) -> typing_Tuple[str, ColumnClause[_T]]:
4723 c = ColumnClause(
4724 (
4725 coercions.expect(roles.TruncatedLabelRole, name or self.name)
4726 if name_is_truncatable
4727 else (name or self.name)
4728 ),
4729 type_=self.type,
4730 _selectable=selectable,
4731 is_literal=False,
4732 )
4734 c._propagate_attrs = selectable._propagate_attrs
4735 if name is None:
4736 c.key = self.key
4737 if compound_select_cols:
4738 c._proxies = list(compound_select_cols)
4739 else:
4740 c._proxies = [self]
4742 if selectable._is_clone_of is not None:
4743 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
4744 return c.key, c
4747_PS = ParamSpec("_PS")
4750class Label(roles.LabeledColumnExprRole[_T], NamedColumn[_T]):
4751 """Represents a column label (AS).
4753 Represent a label, as typically applied to any column-level
4754 element using the ``AS`` sql keyword.
4756 """
4758 __visit_name__ = "label"
4760 _traverse_internals: _TraverseInternalsType = [
4761 ("name", InternalTraversal.dp_anon_name),
4762 ("type", InternalTraversal.dp_type),
4763 ("_element", InternalTraversal.dp_clauseelement),
4764 ]
4766 _cache_key_traversal = [
4767 ("name", InternalTraversal.dp_anon_name),
4768 ("_element", InternalTraversal.dp_clauseelement),
4769 ]
4771 _element: ColumnElement[_T]
4772 name: str
4774 def __init__(
4775 self,
4776 name: Optional[str],
4777 element: _ColumnExpressionArgument[_T],
4778 type_: Optional[_TypeEngineArgument[_T]] = None,
4779 ):
4780 orig_element = element
4781 element = coercions.expect(
4782 roles.ExpressionElementRole,
4783 element,
4784 apply_propagate_attrs=self,
4785 )
4786 while isinstance(element, Label):
4787 # TODO: this is only covered in test_text.py, but nothing
4788 # fails if it's removed. determine rationale
4789 element = element.element
4791 if name:
4792 self.name = name
4793 else:
4794 self.name = _anonymous_label.safe_construct(
4795 id(self), getattr(element, "name", "anon")
4796 )
4797 if isinstance(orig_element, Label):
4798 # TODO: no coverage for this block, again would be in
4799 # test_text.py where the resolve_label concept is important
4800 self._resolve_label = orig_element._label
4802 self.key = self._tq_label = self._tq_key_label = self.name
4803 self._element = element
4805 self.type = (
4806 type_api.to_instance(type_)
4807 if type_ is not None
4808 else self._element.type
4809 )
4811 self._proxies = [element]
4813 def __reduce__(self):
4814 return self.__class__, (self.name, self._element, self.type)
4816 @HasMemoized.memoized_attribute
4817 def _render_label_in_columns_clause(self):
4818 return True
4820 def _bind_param(self, operator, obj, type_=None, expanding=False):
4821 return BindParameter(
4822 None,
4823 obj,
4824 _compared_to_operator=operator,
4825 type_=type_,
4826 _compared_to_type=self.type,
4827 unique=True,
4828 expanding=expanding,
4829 )
4831 @util.memoized_property
4832 def _is_implicitly_boolean(self):
4833 return self.element._is_implicitly_boolean
4835 @HasMemoized.memoized_attribute
4836 def _allow_label_resolve(self):
4837 return self.element._allow_label_resolve
4839 @property
4840 def _order_by_label_element(self):
4841 return self
4843 def as_reference(self) -> _label_reference[_T]:
4844 """refer to this labeled expression in a clause such as GROUP BY,
4845 ORDER BY etc. as the label name itself, without expanding
4846 into the full expression.
4848 .. versionadded:: 2.1
4850 """
4851 return _label_reference(self)
4853 @HasMemoized.memoized_attribute
4854 def element(self) -> ColumnElement[_T]:
4855 return self._element.self_group(against=operators.as_)
4857 def self_group(self, against: Optional[OperatorType] = None) -> Label[_T]:
4858 return self._apply_to_inner(self._element.self_group, against=against)
4860 def _negate(self):
4861 return self._apply_to_inner(self._element._negate)
4863 def _apply_to_inner(
4864 self,
4865 fn: Callable[_PS, ColumnElement[_T]],
4866 *arg: _PS.args,
4867 **kw: _PS.kwargs,
4868 ) -> Label[_T]:
4869 sub_element = fn(*arg, **kw)
4870 if sub_element is not self._element:
4871 return Label(self.name, sub_element, type_=self.type)
4872 else:
4873 return self
4875 @property
4876 def primary_key(self): # type: ignore[override]
4877 return self.element.primary_key
4879 @property
4880 def foreign_keys(self): # type: ignore[override]
4881 return self.element.foreign_keys
4883 def _copy_internals(
4884 self,
4885 *,
4886 clone: _CloneCallableType = _clone,
4887 anonymize_labels: bool = False,
4888 **kw: Any,
4889 ) -> None:
4890 self._reset_memoizations()
4891 self._element = clone(self._element, **kw)
4892 if anonymize_labels:
4893 self.name = _anonymous_label.safe_construct(
4894 id(self), getattr(self.element, "name", "anon")
4895 )
4896 self.key = self._tq_label = self._tq_key_label = self.name
4898 @util.ro_non_memoized_property
4899 def _from_objects(self) -> List[FromClause]:
4900 return self.element._from_objects
4902 def _make_proxy(
4903 self,
4904 selectable: FromClause,
4905 *,
4906 primary_key: ColumnSet,
4907 foreign_keys: Set[KeyedColumnElement[Any]],
4908 name: Optional[str] = None,
4909 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
4910 **kw: Any,
4911 ) -> typing_Tuple[str, ColumnClause[_T]]:
4912 name = self.name if not name else name
4914 key, e = self.element._make_proxy(
4915 selectable,
4916 name=name,
4917 disallow_is_literal=True,
4918 name_is_truncatable=isinstance(name, _truncated_label),
4919 compound_select_cols=compound_select_cols,
4920 primary_key=primary_key,
4921 foreign_keys=foreign_keys,
4922 )
4924 # there was a note here to remove this assertion, which was here
4925 # to determine if we later could support a use case where
4926 # the key and name of a label are separate. But I don't know what
4927 # that case was. For now, this is an unexpected case that occurs
4928 # when a label name conflicts with other columns and select()
4929 # is attempting to disambiguate an explicit label, which is not what
4930 # the user would want. See issue #6090.
4931 if key != self.name and not isinstance(self.name, _anonymous_label):
4932 raise exc.InvalidRequestError(
4933 "Label name %s is being renamed to an anonymous label due "
4934 "to disambiguation "
4935 "which is not supported right now. Please use unique names "
4936 "for explicit labels." % (self.name)
4937 )
4939 e._propagate_attrs = selectable._propagate_attrs
4940 e._proxies.append(self)
4941 if self.type is not None:
4942 e.type = self.type
4944 return self.key, e
4947class ColumnClause(
4948 roles.DDLReferredColumnRole,
4949 roles.LabeledColumnExprRole[_T],
4950 roles.StrAsPlainColumnRole,
4951 Immutable,
4952 NamedColumn[_T],
4953):
4954 """Represents a column expression from any textual string.
4956 The :class:`.ColumnClause`, a lightweight analogue to the
4957 :class:`_schema.Column` class, is typically invoked using the
4958 :func:`_expression.column` function, as in::
4960 from sqlalchemy import column
4962 id, name = column("id"), column("name")
4963 stmt = select(id, name).select_from("user")
4965 The above statement would produce SQL like:
4967 .. sourcecode:: sql
4969 SELECT id, name FROM user
4971 :class:`.ColumnClause` is the immediate superclass of the schema-specific
4972 :class:`_schema.Column` object. While the :class:`_schema.Column`
4973 class has all the
4974 same capabilities as :class:`.ColumnClause`, the :class:`.ColumnClause`
4975 class is usable by itself in those cases where behavioral requirements
4976 are limited to simple SQL expression generation. The object has none of
4977 the associations with schema-level metadata or with execution-time
4978 behavior that :class:`_schema.Column` does,
4979 so in that sense is a "lightweight"
4980 version of :class:`_schema.Column`.
4982 Full details on :class:`.ColumnClause` usage is at
4983 :func:`_expression.column`.
4985 .. seealso::
4987 :func:`_expression.column`
4989 :class:`_schema.Column`
4991 """
4993 table: Optional[FromClause]
4994 is_literal: bool
4996 __visit_name__ = "column"
4998 _traverse_internals: _TraverseInternalsType = [
4999 ("name", InternalTraversal.dp_anon_name),
5000 ("type", InternalTraversal.dp_type),
5001 ("table", InternalTraversal.dp_clauseelement),
5002 ("is_literal", InternalTraversal.dp_boolean),
5003 ]
5005 onupdate: Optional[DefaultGenerator] = None
5006 default: Optional[DefaultGenerator] = None
5007 server_default: Optional[FetchedValue] = None
5008 server_onupdate: Optional[FetchedValue] = None
5010 _is_multiparam_column = False
5012 @property
5013 def _is_star(self): # type: ignore[override]
5014 return self.is_literal and self.name == "*"
5016 def __init__(
5017 self,
5018 text: str,
5019 type_: Optional[_TypeEngineArgument[_T]] = None,
5020 is_literal: bool = False,
5021 _selectable: Optional[FromClause] = None,
5022 ):
5023 self.key = self.name = text
5024 self.table = _selectable
5026 # if type is None, we get NULLTYPE, which is our _T. But I don't
5027 # know how to get the overloads to express that correctly
5028 self.type = type_api.to_instance(type_) # type: ignore
5030 self.is_literal = is_literal
5032 def get_children(self, *, column_tables=False, **kw):
5033 # override base get_children() to not return the Table
5034 # or selectable that is parent to this column. Traversals
5035 # expect the columns of tables and subqueries to be leaf nodes.
5036 return []
5038 @property
5039 def entity_namespace(self):
5040 if self.table is not None:
5041 return self.table.entity_namespace
5042 else:
5043 return super().entity_namespace
5045 def _clone(self, detect_subquery_cols=False, **kw):
5046 if (
5047 detect_subquery_cols
5048 and self.table is not None
5049 and self.table._is_subquery
5050 ):
5051 clone = kw.pop("clone")
5052 table = clone(self.table, **kw)
5053 new = table.c.corresponding_column(self)
5054 return new
5056 return super()._clone(**kw)
5058 @HasMemoized_ro_memoized_attribute
5059 def _from_objects(self) -> List[FromClause]:
5060 t = self.table
5061 if t is not None:
5062 return [t]
5063 else:
5064 return []
5066 @HasMemoized.memoized_attribute
5067 def _render_label_in_columns_clause(self):
5068 return self.table is not None
5070 @property
5071 def _ddl_label(self):
5072 return self._gen_tq_label(self.name, dedupe_on_key=False)
5074 def _compare_name_for_result(self, other):
5075 if (
5076 self.is_literal
5077 or self.table is None
5078 or self.table._is_textual
5079 or not hasattr(other, "proxy_set")
5080 or (
5081 isinstance(other, ColumnClause)
5082 and (
5083 other.is_literal
5084 or other.table is None
5085 or other.table._is_textual
5086 )
5087 )
5088 ):
5089 return (hasattr(other, "name") and self.name == other.name) or (
5090 hasattr(other, "_tq_label")
5091 and self._tq_label == other._tq_label
5092 )
5093 else:
5094 return other.proxy_set.intersection(self.proxy_set)
5096 def _gen_tq_label(
5097 self, name: str, dedupe_on_key: bool = True
5098 ) -> Optional[str]:
5099 """generate table-qualified label
5101 for a table-bound column this is <tablename>_<columnname>.
5103 used primarily for LABEL_STYLE_TABLENAME_PLUS_COL
5104 as well as the .columns collection on a Join object.
5106 """
5107 label: str
5108 t = self.table
5109 if self.is_literal:
5110 return None
5111 elif t is not None and is_named_from_clause(t):
5112 if has_schema_attr(t) and t.schema:
5113 label = (
5114 t.schema.replace(".", "_") + "_" + t.name + ("_" + name)
5115 )
5116 else:
5117 assert not TYPE_CHECKING or isinstance(t, NamedFromClause)
5118 label = t.name + ("_" + name)
5120 # propagate name quoting rules for labels.
5121 if is_quoted_name(name) and name.quote is not None:
5122 if is_quoted_name(label):
5123 label.quote = name.quote
5124 else:
5125 label = quoted_name(label, name.quote)
5126 elif is_quoted_name(t.name) and t.name.quote is not None:
5127 # can't get this situation to occur, so let's
5128 # assert false on it for now
5129 assert not isinstance(label, quoted_name)
5130 label = quoted_name(label, t.name.quote)
5132 if dedupe_on_key:
5133 # ensure the label name doesn't conflict with that of an
5134 # existing column. note that this implies that any Column
5135 # must **not** set up its _label before its parent table has
5136 # all of its other Column objects set up. There are several
5137 # tables in the test suite which will fail otherwise; example:
5138 # table "owner" has columns "name" and "owner_name". Therefore
5139 # column owner.name cannot use the label "owner_name", it has
5140 # to be "owner_name_1".
5141 if label in t.c:
5142 _label = label
5143 counter = 1
5144 while _label in t.c:
5145 _label = label + f"_{counter}"
5146 counter += 1
5147 label = _label
5149 return coercions.expect(roles.TruncatedLabelRole, label)
5151 else:
5152 return name
5154 def _make_proxy(
5155 self,
5156 selectable: FromClause,
5157 *,
5158 primary_key: ColumnSet,
5159 foreign_keys: Set[KeyedColumnElement[Any]],
5160 name: Optional[str] = None,
5161 key: Optional[str] = None,
5162 name_is_truncatable: bool = False,
5163 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5164 disallow_is_literal: bool = False,
5165 **kw: Any,
5166 ) -> typing_Tuple[str, ColumnClause[_T]]:
5167 # the "is_literal" flag normally should never be propagated; a proxied
5168 # column is always a SQL identifier and never the actual expression
5169 # being evaluated. however, there is a case where the "is_literal" flag
5170 # might be used to allow the given identifier to have a fixed quoting
5171 # pattern already, so maintain the flag for the proxy unless a
5172 # :class:`.Label` object is creating the proxy. See [ticket:4730].
5173 is_literal = (
5174 not disallow_is_literal
5175 and self.is_literal
5176 and (
5177 # note this does not accommodate for quoted_name differences
5178 # right now
5179 name is None
5180 or name == self.name
5181 )
5182 )
5183 c = self._constructor(
5184 (
5185 coercions.expect(roles.TruncatedLabelRole, name or self.name)
5186 if name_is_truncatable
5187 else (name or self.name)
5188 ),
5189 type_=self.type,
5190 _selectable=selectable,
5191 is_literal=is_literal,
5192 )
5193 c._propagate_attrs = selectable._propagate_attrs
5194 if name is None:
5195 c.key = self.key
5196 if compound_select_cols:
5197 c._proxies = list(compound_select_cols)
5198 else:
5199 c._proxies = [self]
5201 if selectable._is_clone_of is not None:
5202 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
5203 return c.key, c
5206class TableValuedColumn(NamedColumn[_T]):
5207 __visit_name__ = "table_valued_column"
5209 _traverse_internals: _TraverseInternalsType = [
5210 ("name", InternalTraversal.dp_anon_name),
5211 ("type", InternalTraversal.dp_type),
5212 ("scalar_alias", InternalTraversal.dp_clauseelement),
5213 ]
5215 def __init__(self, scalar_alias: NamedFromClause, type_: TypeEngine[_T]):
5216 self.scalar_alias = scalar_alias
5217 self.key = self.name = scalar_alias.name
5218 self.type = type_
5220 def _copy_internals(
5221 self, clone: _CloneCallableType = _clone, **kw: Any
5222 ) -> None:
5223 self.scalar_alias = clone(self.scalar_alias, **kw)
5224 self.key = self.name = self.scalar_alias.name
5226 @util.ro_non_memoized_property
5227 def _from_objects(self) -> List[FromClause]:
5228 return [self.scalar_alias]
5231class CollationClause(ColumnElement[str]):
5232 __visit_name__ = "collation"
5234 _traverse_internals: _TraverseInternalsType = [
5235 ("collation", InternalTraversal.dp_string)
5236 ]
5238 @classmethod
5239 @util.preload_module("sqlalchemy.sql.sqltypes")
5240 def _create_collation_expression(
5241 cls, expression: _ColumnExpressionArgument[str], collation: str
5242 ) -> BinaryExpression[str]:
5244 sqltypes = util.preloaded.sql_sqltypes
5246 expr = coercions.expect(roles.ExpressionElementRole[str], expression)
5248 if expr.type._type_affinity is sqltypes.String:
5249 collate_type = expr.type._with_collation(collation)
5250 else:
5251 collate_type = expr.type
5253 return BinaryExpression(
5254 expr,
5255 CollationClause(collation),
5256 operators.collate,
5257 type_=collate_type,
5258 )
5260 def __init__(self, collation):
5261 self.collation = collation
5264class _IdentifiedClause(Executable, ClauseElement):
5265 __visit_name__ = "identified"
5267 def __init__(self, ident):
5268 self.ident = ident
5271class SavepointClause(_IdentifiedClause):
5272 __visit_name__ = "savepoint"
5273 inherit_cache = False
5276class RollbackToSavepointClause(_IdentifiedClause):
5277 __visit_name__ = "rollback_to_savepoint"
5278 inherit_cache = False
5281class ReleaseSavepointClause(_IdentifiedClause):
5282 __visit_name__ = "release_savepoint"
5283 inherit_cache = False
5286class quoted_name(util.MemoizedSlots, str):
5287 """Represent a SQL identifier combined with quoting preferences.
5289 :class:`.quoted_name` is a Python unicode/str subclass which
5290 represents a particular identifier name along with a
5291 ``quote`` flag. This ``quote`` flag, when set to
5292 ``True`` or ``False``, overrides automatic quoting behavior
5293 for this identifier in order to either unconditionally quote
5294 or to not quote the name. If left at its default of ``None``,
5295 quoting behavior is applied to the identifier on a per-backend basis
5296 based on an examination of the token itself.
5298 A :class:`.quoted_name` object with ``quote=True`` is also
5299 prevented from being modified in the case of a so-called
5300 "name normalize" option. Certain database backends, such as
5301 Oracle Database, Firebird, and DB2 "normalize" case-insensitive names
5302 as uppercase. The SQLAlchemy dialects for these backends
5303 convert from SQLAlchemy's lower-case-means-insensitive convention
5304 to the upper-case-means-insensitive conventions of those backends.
5305 The ``quote=True`` flag here will prevent this conversion from occurring
5306 to support an identifier that's quoted as all lower case against
5307 such a backend.
5309 The :class:`.quoted_name` object is normally created automatically
5310 when specifying the name for key schema constructs such as
5311 :class:`_schema.Table`, :class:`_schema.Column`, and others.
5312 The class can also be
5313 passed explicitly as the name to any function that receives a name which
5314 can be quoted. Such as to use the :meth:`_engine.Engine.has_table`
5315 method with
5316 an unconditionally quoted name::
5318 from sqlalchemy import create_engine
5319 from sqlalchemy import inspect
5320 from sqlalchemy.sql import quoted_name
5322 engine = create_engine("oracle+oracledb://some_dsn")
5323 print(inspect(engine).has_table(quoted_name("some_table", True)))
5325 The above logic will run the "has table" logic against the Oracle Database
5326 backend, passing the name exactly as ``"some_table"`` without converting to
5327 upper case.
5329 """
5331 __slots__ = "quote", "lower", "upper"
5333 quote: Optional[bool]
5335 @overload
5336 @classmethod
5337 def construct(cls, value: str, quote: Optional[bool]) -> quoted_name: ...
5339 @overload
5340 @classmethod
5341 def construct(cls, value: None, quote: Optional[bool]) -> None: ...
5343 @classmethod
5344 def construct(
5345 cls, value: Optional[str], quote: Optional[bool]
5346 ) -> Optional[quoted_name]:
5347 if value is None:
5348 return None
5349 else:
5350 return quoted_name(value, quote)
5352 def __new__(cls, value: str, quote: Optional[bool]) -> quoted_name:
5353 assert (
5354 value is not None
5355 ), "use quoted_name.construct() for None passthrough"
5356 if isinstance(value, cls) and (quote is None or value.quote == quote):
5357 return value
5358 self = super().__new__(cls, value)
5360 self.quote = quote
5361 return self
5363 def __reduce__(self):
5364 return quoted_name, (str(self), self.quote)
5366 def _memoized_method_lower(self):
5367 if self.quote:
5368 return self
5369 else:
5370 return str(self).lower()
5372 def _memoized_method_upper(self):
5373 if self.quote:
5374 return self
5375 else:
5376 return str(self).upper()
5379def _find_columns(clause: ClauseElement) -> Set[ColumnClause[Any]]:
5380 """locate Column objects within the given expression."""
5382 cols: Set[ColumnClause[Any]] = set()
5383 traverse(clause, {}, {"column": cols.add})
5384 return cols
5387def _type_from_args(args: Sequence[ColumnElement[_T]]) -> TypeEngine[_T]:
5388 for a in args:
5389 if not a.type._isnull:
5390 return a.type
5391 else:
5392 return type_api.NULLTYPE # type: ignore
5395def _corresponding_column_or_error(fromclause, column, require_embedded=False):
5396 c = fromclause.corresponding_column(
5397 column, require_embedded=require_embedded
5398 )
5399 if c is None:
5400 raise exc.InvalidRequestError(
5401 "Given column '%s', attached to table '%s', "
5402 "failed to locate a corresponding column from table '%s'"
5403 % (column, getattr(column, "table", None), fromclause.description)
5404 )
5405 return c
5408class _memoized_property_but_not_nulltype(
5409 util.memoized_property["TypeEngine[_T]"]
5410):
5411 """memoized property, but dont memoize NullType"""
5413 def __get__(self, obj, cls):
5414 if obj is None:
5415 return self
5416 result = self.fget(obj)
5417 if not result._isnull:
5418 obj.__dict__[self.__name__] = result
5419 return result
5422class AnnotatedColumnElement(Annotated):
5423 _Annotated__element: ColumnElement[Any]
5425 def __init__(self, element, values):
5426 Annotated.__init__(self, element, values)
5427 for attr in (
5428 "comparator",
5429 "_proxy_key",
5430 "_tq_key_label",
5431 "_tq_label",
5432 "_non_anon_label",
5433 "type",
5434 ):
5435 self.__dict__.pop(attr, None)
5436 for attr in ("name", "key", "table"):
5437 if self.__dict__.get(attr, False) is None:
5438 self.__dict__.pop(attr)
5440 def _with_annotations(self, values):
5441 clone = super()._with_annotations(values)
5442 for attr in (
5443 "comparator",
5444 "_proxy_key",
5445 "_tq_key_label",
5446 "_tq_label",
5447 "_non_anon_label",
5448 ):
5449 clone.__dict__.pop(attr, None)
5450 return clone
5452 @util.memoized_property
5453 def name(self):
5454 """pull 'name' from parent, if not present"""
5455 return self._Annotated__element.name
5457 @_memoized_property_but_not_nulltype
5458 def type(self):
5459 """pull 'type' from parent and don't cache if null.
5461 type is routinely changed on existing columns within the
5462 mapped_column() initialization process, and "type" is also consulted
5463 during the creation of SQL expressions. Therefore it can change after
5464 it was already retrieved. At the same time we don't want annotated
5465 objects having overhead when expressions are produced, so continue
5466 to memoize, but only when we have a non-null type.
5468 """
5469 return self._Annotated__element.type
5471 @util.memoized_property
5472 def table(self):
5473 """pull 'table' from parent, if not present"""
5474 return self._Annotated__element.table
5476 @util.memoized_property
5477 def key(self):
5478 """pull 'key' from parent, if not present"""
5479 return self._Annotated__element.key
5481 @util.memoized_property
5482 def info(self) -> _InfoType:
5483 if TYPE_CHECKING:
5484 assert isinstance(self._Annotated__element, Column)
5485 return self._Annotated__element.info
5487 @util.memoized_property
5488 def _anon_name_label(self) -> str:
5489 return self._Annotated__element._anon_name_label
5492class _truncated_label(quoted_name):
5493 """A unicode subclass used to identify symbolic "
5494 "names that may require truncation."""
5496 __slots__ = ()
5498 def __new__(cls, value: str, quote: Optional[bool] = None) -> Any:
5499 quote = getattr(value, "quote", quote)
5500 # return super(_truncated_label, cls).__new__(cls, value, quote, True)
5501 return super().__new__(cls, value, quote)
5503 def __reduce__(self) -> Any:
5504 return self.__class__, (str(self), self.quote)
5506 def apply_map(self, map_: Mapping[str, Any]) -> str:
5507 return self
5510class conv(_truncated_label):
5511 """Mark a string indicating that a name has already been converted
5512 by a naming convention.
5514 This is a string subclass that indicates a name that should not be
5515 subject to any further naming conventions.
5517 E.g. when we create a :class:`.Constraint` using a naming convention
5518 as follows::
5520 m = MetaData(
5521 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5522 )
5523 t = Table(
5524 "t", m, Column("x", Integer), CheckConstraint("x > 5", name="x5")
5525 )
5527 The name of the above constraint will be rendered as ``"ck_t_x5"``.
5528 That is, the existing name ``x5`` is used in the naming convention as the
5529 ``constraint_name`` token.
5531 In some situations, such as in migration scripts, we may be rendering
5532 the above :class:`.CheckConstraint` with a name that's already been
5533 converted. In order to make sure the name isn't double-modified, the
5534 new name is applied using the :func:`_schema.conv` marker. We can
5535 use this explicitly as follows::
5538 m = MetaData(
5539 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5540 )
5541 t = Table(
5542 "t",
5543 m,
5544 Column("x", Integer),
5545 CheckConstraint("x > 5", name=conv("ck_t_x5")),
5546 )
5548 Where above, the :func:`_schema.conv` marker indicates that the constraint
5549 name here is final, and the name will render as ``"ck_t_x5"`` and not
5550 ``"ck_t_ck_t_x5"``
5552 .. seealso::
5554 :ref:`constraint_naming_conventions`
5556 """
5558 __slots__ = ()
5561# for backwards compatibility in case
5562# someone is re-implementing the
5563# _truncated_identifier() sequence in a custom
5564# compiler
5565_generated_label = _truncated_label
5566_anonymous_label_escape = re.compile(r"[%\(\) \$]+")
5569class _anonymous_label(_truncated_label):
5570 """A unicode subclass used to identify anonymously
5571 generated names."""
5573 __slots__ = ()
5575 @classmethod
5576 def safe_construct_with_key(
5577 cls, seed: int, body: str, sanitize_key: bool = False
5578 ) -> typing_Tuple[_anonymous_label, str]:
5579 # need to escape chars that interfere with format
5580 # strings in any case, issue #8724
5581 body = _anonymous_label_escape.sub("_", body)
5583 if sanitize_key:
5584 # sanitize_key is then an extra step used by BindParameter
5585 body = body.strip("_")
5587 key = f"{seed} {body.replace('%', '%%')}"
5588 label = _anonymous_label(f"%({key})s")
5589 return label, key
5591 @classmethod
5592 def safe_construct(
5593 cls, seed: int, body: str, sanitize_key: bool = False
5594 ) -> _anonymous_label:
5595 # need to escape chars that interfere with format
5596 # strings in any case, issue #8724
5597 body = _anonymous_label_escape.sub("_", body)
5599 if sanitize_key:
5600 # sanitize_key is then an extra step used by BindParameter
5601 body = body.strip("_")
5603 return _anonymous_label(f"%({seed} {body.replace('%', '%%')})s")
5605 def __add__(self, other: str) -> _anonymous_label:
5606 if "%" in other and not isinstance(other, _anonymous_label):
5607 other = str(other).replace("%", "%%")
5608 else:
5609 other = str(other)
5611 return _anonymous_label(
5612 quoted_name(
5613 str.__add__(self, other),
5614 self.quote,
5615 )
5616 )
5618 def __radd__(self, other: str) -> _anonymous_label:
5619 if "%" in other and not isinstance(other, _anonymous_label):
5620 other = str(other).replace("%", "%%")
5621 else:
5622 other = str(other)
5624 return _anonymous_label(
5625 quoted_name(
5626 str.__add__(other, self),
5627 self.quote,
5628 )
5629 )
5631 def apply_map(self, map_: Mapping[str, Any]) -> str:
5632 if self.quote is not None:
5633 # preserve quoting only if necessary
5634 return quoted_name(self % map_, self.quote)
5635 else:
5636 # else skip the constructor call
5637 return self % map_