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 _EntityNamespace
95 from .base import ColumnSet
96 from .cache_key import _CacheKeyTraversalType
97 from .cache_key import CacheKey
98 from .compiler import Compiled
99 from .compiler import SQLCompiler
100 from .functions import FunctionElement
101 from .operators import OperatorType
102 from .schema import Column
103 from .schema import DefaultGenerator
104 from .schema import FetchedValue
105 from .schema import ForeignKey
106 from .selectable import _SelectIterable
107 from .selectable import FromClause
108 from .selectable import NamedFromClause
109 from .selectable import TextualSelect
110 from .sqltypes import TupleType
111 from .type_api import TypeEngine
112 from .visitors import _CloneCallableType
113 from .visitors import _TraverseInternalsType
114 from .visitors import anon_map
115 from ..engine import Connection
116 from ..engine import Dialect
117 from ..engine.interfaces import _CoreMultiExecuteParams
118 from ..engine.interfaces import CacheStats
119 from ..engine.interfaces import CompiledCacheType
120 from ..engine.interfaces import CoreExecuteOptionsParameter
121 from ..engine.interfaces import SchemaTranslateMapType
122 from ..engine.result import Result
125_NUMERIC = Union[float, Decimal]
126_NUMBER = Union[float, int, Decimal]
128_T = TypeVar("_T", bound="Any")
129_T_co = TypeVar("_T_co", bound=Any, covariant=True)
130_OPT = TypeVar("_OPT", bound="Any")
131_NT = TypeVar("_NT", bound="_NUMERIC")
133_NMT = TypeVar("_NMT", bound="_NUMBER")
136@overload
137def literal(
138 value: Any,
139 type_: _TypeEngineArgument[_T],
140 literal_execute: bool = False,
141) -> BindParameter[_T]: ...
144@overload
145def literal(
146 value: _T,
147 type_: None = None,
148 literal_execute: bool = False,
149) -> BindParameter[_T]: ...
152@overload
153def literal(
154 value: Any,
155 type_: Optional[_TypeEngineArgument[Any]] = None,
156 literal_execute: bool = False,
157) -> BindParameter[Any]: ...
160def literal(
161 value: Any,
162 type_: Optional[_TypeEngineArgument[Any]] = None,
163 literal_execute: bool = False,
164) -> BindParameter[Any]:
165 r"""Return a literal clause, bound to a bind parameter.
167 Literal clauses are created automatically when non-
168 :class:`_expression.ClauseElement` objects (such as strings, ints, dates,
169 etc.) are
170 used in a comparison operation with a :class:`_expression.ColumnElement`
171 subclass,
172 such as a :class:`~sqlalchemy.schema.Column` object. Use this function
173 to force the generation of a literal clause, which will be created as a
174 :class:`BindParameter` with a bound value.
176 :param value: the value to be bound. Can be any Python object supported by
177 the underlying DB-API, or is translatable via the given type argument.
179 :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine` which will
180 provide bind-parameter translation for this literal.
182 :param literal_execute: optional bool, when True, the SQL engine will
183 attempt to render the bound value directly in the SQL statement at
184 execution time rather than providing as a parameter value.
186 .. versionadded:: 2.0
188 """
189 return coercions.expect(
190 roles.LiteralValueRole,
191 value,
192 type_=type_,
193 literal_execute=literal_execute,
194 )
197def literal_column(
198 text: str, type_: Optional[_TypeEngineArgument[_T]] = None
199) -> ColumnClause[_T]:
200 r"""Produce a :class:`.ColumnClause` object that has the
201 :paramref:`_expression.column.is_literal` flag set to True.
203 :func:`_expression.literal_column` is similar to
204 :func:`_expression.column`, except that
205 it is more often used as a "standalone" column expression that renders
206 exactly as stated; while :func:`_expression.column`
207 stores a string name that
208 will be assumed to be part of a table and may be quoted as such,
209 :func:`_expression.literal_column` can be that,
210 or any other arbitrary column-oriented
211 expression.
213 :param text: the text of the expression; can be any SQL expression.
214 Quoting rules will not be applied. To specify a column-name expression
215 which should be subject to quoting rules, use the :func:`column`
216 function.
218 :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine`
219 object which will
220 provide result-set translation and additional expression semantics for
221 this column. If left as ``None`` the type will be :class:`.NullType`.
223 .. seealso::
225 :func:`_expression.column`
227 :func:`_expression.text`
229 :ref:`tutorial_select_arbitrary_text`
231 """
232 return ColumnClause(text, type_=type_, is_literal=True)
235class CompilerElement(Visitable):
236 """base class for SQL elements that can be compiled to produce a
237 SQL string.
239 .. versionadded:: 2.0
241 """
243 __slots__ = ()
244 __visit_name__ = "compiler_element"
246 supports_execution = False
248 stringify_dialect = "default"
250 @util.preload_module("sqlalchemy.engine.default")
251 @util.preload_module("sqlalchemy.engine.url")
252 def compile(
253 self,
254 bind: Optional[_HasDialect] = None,
255 dialect: Optional[Dialect] = None,
256 **kw: Any,
257 ) -> Compiled:
258 """Compile this SQL expression.
260 The return value is a :class:`~.Compiled` object.
261 Calling ``str()`` or ``unicode()`` on the returned value will yield a
262 string representation of the result. The
263 :class:`~.Compiled` object also can return a
264 dictionary of bind parameter names and values
265 using the ``params`` accessor.
267 :param bind: An :class:`.Connection` or :class:`.Engine` which
268 can provide a :class:`.Dialect` in order to generate a
269 :class:`.Compiled` object. If the ``bind`` and
270 ``dialect`` parameters are both omitted, a default SQL compiler
271 is used.
273 :param column_keys: Used for INSERT and UPDATE statements, a list of
274 column names which should be present in the VALUES clause of the
275 compiled statement. If ``None``, all columns from the target table
276 object are rendered.
278 :param dialect: A :class:`.Dialect` instance which can generate
279 a :class:`.Compiled` object. This argument takes precedence over
280 the ``bind`` argument.
282 :param compile_kwargs: optional dictionary of additional parameters
283 that will be passed through to the compiler within all "visit"
284 methods. This allows any custom flag to be passed through to
285 a custom compilation construct, for example. It is also used
286 for the case of passing the ``literal_binds`` flag through::
288 from sqlalchemy.sql import table, column, select
290 t = table("t", column("x"))
292 s = select(t).where(t.c.x == 5)
294 print(s.compile(compile_kwargs={"literal_binds": True}))
296 .. seealso::
298 :ref:`faq_sql_expression_string`
300 """
302 if dialect is None:
303 if bind:
304 dialect = bind.dialect
305 elif self.stringify_dialect == "default":
306 dialect = self._default_dialect()
307 else:
308 url = util.preloaded.engine_url
309 dialect = url.URL.create(
310 self.stringify_dialect
311 ).get_dialect()()
313 return self._compiler(dialect, **kw)
315 def _default_dialect(self):
316 default = util.preloaded.engine_default
317 return default.StrCompileDialect()
319 def _compiler(self, dialect: Dialect, **kw: Any) -> Compiled:
320 """Return a compiler appropriate for this ClauseElement, given a
321 Dialect."""
323 if TYPE_CHECKING:
324 assert isinstance(self, ClauseElement)
325 return dialect.statement_compiler(dialect, self, **kw)
327 def __str__(self) -> str:
328 return str(self.compile())
331@inspection._self_inspects
332class ClauseElement(
333 SupportsWrappingAnnotations,
334 MemoizedHasCacheKey,
335 HasCopyInternals,
336 ExternallyTraversible,
337 CompilerElement,
338):
339 """Base class for elements of a programmatically constructed SQL
340 expression.
342 """
344 __visit_name__ = "clause"
346 if TYPE_CHECKING:
348 @util.memoized_property
349 def _propagate_attrs(self) -> _PropagateAttrsType:
350 """like annotations, however these propagate outwards liberally
351 as SQL constructs are built, and are set up at construction time.
353 """
354 ...
356 else:
357 _propagate_attrs = util.EMPTY_DICT
359 @util.ro_memoized_property
360 def description(self) -> Optional[str]:
361 return None
363 _is_clone_of: Optional[Self] = None
365 is_clause_element = True
366 is_selectable = False
367 is_dml = False
368 _is_column_element = False
369 _is_keyed_column_element = False
370 _is_table = False
371 _gen_static_annotations_cache_key = False
372 _is_textual = False
373 _is_from_clause = False
374 _is_returns_rows = False
375 _is_text_clause = False
376 _is_from_container = False
377 _is_select_container = False
378 _is_select_base = False
379 _is_select_statement = False
380 _is_bind_parameter = False
381 _is_clause_list = False
382 _is_lambda_element = False
383 _is_singleton_constant = False
384 _is_immutable = False
385 _is_star = False
387 @property
388 def _order_by_label_element(self) -> Optional[Label[Any]]:
389 return None
391 _cache_key_traversal: _CacheKeyTraversalType = None
393 negation_clause: ColumnElement[bool]
395 if typing.TYPE_CHECKING:
397 def get_children(
398 self, *, omit_attrs: typing_Tuple[str, ...] = ..., **kw: Any
399 ) -> Iterable[ClauseElement]: ...
401 @util.ro_non_memoized_property
402 def _from_objects(self) -> List[FromClause]:
403 return []
405 def _set_propagate_attrs(self, values: Mapping[str, Any]) -> Self:
406 # usually, self._propagate_attrs is empty here. one case where it's
407 # not is a subquery against ORM select, that is then pulled as a
408 # property of an aliased class. should all be good
410 # assert not self._propagate_attrs
412 self._propagate_attrs = util.immutabledict(values)
413 return self
415 def _default_compiler(self) -> SQLCompiler:
416 dialect = self._default_dialect()
417 return dialect.statement_compiler(dialect, self) # type: ignore
419 def _clone(self, **kw: Any) -> Self:
420 """Create a shallow copy of this ClauseElement.
422 This method may be used by a generative API. Its also used as
423 part of the "deep" copy afforded by a traversal that combines
424 the _copy_internals() method.
426 """
428 skip = self._memoized_keys
429 c = self.__class__.__new__(self.__class__)
431 if skip:
432 # ensure this iteration remains atomic
433 c.__dict__ = {
434 k: v for k, v in self.__dict__.copy().items() if k not in skip
435 }
436 else:
437 c.__dict__ = self.__dict__.copy()
439 # this is a marker that helps to "equate" clauses to each other
440 # when a Select returns its list of FROM clauses. the cloning
441 # process leaves around a lot of remnants of the previous clause
442 # typically in the form of column expressions still attached to the
443 # old table.
444 cc = self._is_clone_of
445 c._is_clone_of = cc if cc is not None else self
446 return c
448 def _negate_in_binary(self, negated_op, original_op):
449 """a hook to allow the right side of a binary expression to respond
450 to a negation of the binary expression.
452 Used for the special case of expanding bind parameter with IN.
454 """
455 return self
457 def _with_binary_element_type(self, type_):
458 """in the context of binary expression, convert the type of this
459 object to the one given.
461 applies only to :class:`_expression.ColumnElement` classes.
463 """
464 return self
466 @property
467 def _constructor(self): # type: ignore[override]
468 """return the 'constructor' for this ClauseElement.
470 This is for the purposes for creating a new object of
471 this type. Usually, its just the element's __class__.
472 However, the "Annotated" version of the object overrides
473 to return the class of its proxied element.
475 """
476 return self.__class__
478 @HasMemoized.memoized_attribute
479 def _cloned_set(self):
480 """Return the set consisting all cloned ancestors of this
481 ClauseElement.
483 Includes this ClauseElement. This accessor tends to be used for
484 FromClause objects to identify 'equivalent' FROM clauses, regardless
485 of transformative operations.
487 """
488 s = util.column_set()
489 f: Optional[ClauseElement] = self
491 # note this creates a cycle, asserted in test_memusage. however,
492 # turning this into a plain @property adds tends of thousands of method
493 # calls to Core / ORM performance tests, so the small overhead
494 # introduced by the relatively small amount of short term cycles
495 # produced here is preferable
496 while f is not None:
497 s.add(f)
498 f = f._is_clone_of
499 return s
501 def _de_clone(self):
502 while self._is_clone_of is not None:
503 self = self._is_clone_of
504 return self
506 @util.ro_non_memoized_property
507 def entity_namespace(self) -> _EntityNamespace:
508 raise AttributeError(
509 "This SQL expression has no entity namespace "
510 "with which to filter from."
511 )
513 def __getstate__(self):
514 d = self.__dict__.copy()
515 d.pop("_is_clone_of", None)
516 d.pop("_generate_cache_key", None)
517 return d
519 def _execute_on_connection(
520 self,
521 connection: Connection,
522 distilled_params: _CoreMultiExecuteParams,
523 execution_options: CoreExecuteOptionsParameter,
524 ) -> Result[Unpack[TupleAny]]:
525 if self.supports_execution:
526 if TYPE_CHECKING:
527 assert isinstance(self, Executable)
528 return connection._execute_clauseelement(
529 self, distilled_params, execution_options
530 )
531 else:
532 raise exc.ObjectNotExecutableError(self)
534 def _execute_on_scalar(
535 self,
536 connection: Connection,
537 distilled_params: _CoreMultiExecuteParams,
538 execution_options: CoreExecuteOptionsParameter,
539 ) -> Any:
540 """an additional hook for subclasses to provide a different
541 implementation for connection.scalar() vs. connection.execute().
543 .. versionadded:: 2.0
545 """
546 return self._execute_on_connection(
547 connection, distilled_params, execution_options
548 ).scalar()
550 def _get_embedded_bindparams(self) -> Sequence[BindParameter[Any]]:
551 """Return the list of :class:`.BindParameter` objects embedded in the
552 object.
554 This accomplishes the same purpose as ``visitors.traverse()`` or
555 similar would provide, however by making use of the cache key
556 it takes advantage of memoization of the key to result in fewer
557 net method calls, assuming the statement is also going to be
558 executed.
560 """
562 key = self._generate_cache_key()
563 if key is None:
564 bindparams: List[BindParameter[Any]] = []
566 traverse(self, {}, {"bindparam": bindparams.append})
567 return bindparams
569 else:
570 return key.bindparams
572 def unique_params(
573 self,
574 __optionaldict: Optional[Dict[str, Any]] = None,
575 /,
576 **kwargs: Any,
577 ) -> Self:
578 """Return a copy with :func:`_expression.bindparam` elements
579 replaced.
581 Same functionality as :meth:`_expression.ClauseElement.params`,
582 except adds `unique=True`
583 to affected bind parameters so that multiple statements can be
584 used.
586 """
587 return self._replace_params(True, __optionaldict, kwargs)
589 def params(
590 self,
591 __optionaldict: Optional[Mapping[str, Any]] = None,
592 /,
593 **kwargs: Any,
594 ) -> Self:
595 """Return a copy with :func:`_expression.bindparam` elements
596 replaced.
598 Returns a copy of this ClauseElement with
599 :func:`_expression.bindparam`
600 elements replaced with values taken from the given dictionary::
602 >>> clause = column("x") + bindparam("foo")
603 >>> print(clause.compile().params)
604 {'foo':None}
605 >>> print(clause.params({"foo": 7}).compile().params)
606 {'foo':7}
608 """
609 return self._replace_params(False, __optionaldict, kwargs)
611 def _replace_params(
612 self,
613 unique: bool,
614 optionaldict: Optional[Mapping[str, Any]],
615 kwargs: Dict[str, Any],
616 ) -> Self:
617 if optionaldict:
618 kwargs.update(optionaldict)
620 def visit_bindparam(bind: BindParameter[Any]) -> None:
621 if bind.key in kwargs:
622 bind.value = kwargs[bind.key]
623 bind.required = False
624 if unique:
625 bind._convert_to_unique()
627 return cloned_traverse(
628 self,
629 {"maintain_key": True, "detect_subquery_cols": True},
630 {"bindparam": visit_bindparam},
631 )
633 def compare(self, other: ClauseElement, **kw: Any) -> bool:
634 r"""Compare this :class:`_expression.ClauseElement` to
635 the given :class:`_expression.ClauseElement`.
637 Subclasses should override the default behavior, which is a
638 straight identity comparison.
640 \**kw are arguments consumed by subclass ``compare()`` methods and
641 may be used to modify the criteria for comparison
642 (see :class:`_expression.ColumnElement`).
644 """
645 return traversals.compare(self, other, **kw)
647 def self_group(
648 self, against: Optional[OperatorType] = None
649 ) -> ClauseElement:
650 """Apply a 'grouping' to this :class:`_expression.ClauseElement`.
652 This method is overridden by subclasses to return a "grouping"
653 construct, i.e. parenthesis. In particular it's used by "binary"
654 expressions to provide a grouping around themselves when placed into a
655 larger expression, as well as by :func:`_expression.select`
656 constructs when placed into the FROM clause of another
657 :func:`_expression.select`. (Note that subqueries should be
658 normally created using the :meth:`_expression.Select.alias` method,
659 as many
660 platforms require nested SELECT statements to be named).
662 As expressions are composed together, the application of
663 :meth:`self_group` is automatic - end-user code should never
664 need to use this method directly. Note that SQLAlchemy's
665 clause constructs take operator precedence into account -
666 so parenthesis might not be needed, for example, in
667 an expression like ``x OR (y AND z)`` - AND takes precedence
668 over OR.
670 The base :meth:`self_group` method of
671 :class:`_expression.ClauseElement`
672 just returns self.
673 """
674 return self
676 def _ungroup(self) -> ClauseElement:
677 """Return this :class:`_expression.ClauseElement`
678 without any groupings.
679 """
681 return self
683 def _compile_w_cache(
684 self,
685 dialect: Dialect,
686 *,
687 compiled_cache: Optional[CompiledCacheType],
688 column_keys: List[str],
689 for_executemany: bool = False,
690 schema_translate_map: Optional[SchemaTranslateMapType] = None,
691 **kw: Any,
692 ) -> typing_Tuple[
693 Compiled, Optional[Sequence[BindParameter[Any]]], CacheStats
694 ]:
695 elem_cache_key: Optional[CacheKey]
697 if compiled_cache is not None and dialect._supports_statement_cache:
698 elem_cache_key = self._generate_cache_key()
699 else:
700 elem_cache_key = None
702 extracted_params: Optional[Sequence[BindParameter[Any]]]
703 if elem_cache_key is not None:
704 if TYPE_CHECKING:
705 assert compiled_cache is not None
707 cache_key, extracted_params = elem_cache_key
708 key = (
709 dialect,
710 cache_key,
711 tuple(column_keys),
712 bool(schema_translate_map),
713 for_executemany,
714 )
715 compiled_sql = compiled_cache.get(key)
717 if compiled_sql is None:
718 cache_hit = dialect.CACHE_MISS
719 compiled_sql = self._compiler(
720 dialect,
721 cache_key=elem_cache_key,
722 column_keys=column_keys,
723 for_executemany=for_executemany,
724 schema_translate_map=schema_translate_map,
725 **kw,
726 )
727 compiled_cache[key] = compiled_sql
728 else:
729 cache_hit = dialect.CACHE_HIT
730 else:
731 extracted_params = None
732 compiled_sql = self._compiler(
733 dialect,
734 cache_key=elem_cache_key,
735 column_keys=column_keys,
736 for_executemany=for_executemany,
737 schema_translate_map=schema_translate_map,
738 **kw,
739 )
741 if not dialect._supports_statement_cache:
742 cache_hit = dialect.NO_DIALECT_SUPPORT
743 elif compiled_cache is None:
744 cache_hit = dialect.CACHING_DISABLED
745 else:
746 cache_hit = dialect.NO_CACHE_KEY
748 return compiled_sql, extracted_params, cache_hit
750 def __invert__(self):
751 # undocumented element currently used by the ORM for
752 # relationship.contains()
753 if hasattr(self, "negation_clause"):
754 return self.negation_clause
755 else:
756 return self._negate()
758 def _negate(self) -> ClauseElement:
759 # TODO: this code is uncovered and in all likelihood is not included
760 # in any codepath. So this should raise NotImplementedError in 2.1
761 grouped = self.self_group(against=operators.inv)
762 assert isinstance(grouped, ColumnElement)
763 return UnaryExpression(grouped, operator=operators.inv)
765 def __bool__(self):
766 raise TypeError("Boolean value of this clause is not defined")
768 def __repr__(self):
769 friendly = self.description
770 if friendly is None:
771 return object.__repr__(self)
772 else:
773 return "<%s.%s at 0x%x; %s>" % (
774 self.__module__,
775 self.__class__.__name__,
776 id(self),
777 friendly,
778 )
781class DQLDMLClauseElement(ClauseElement):
782 """represents a :class:`.ClauseElement` that compiles to a DQL or DML
783 expression, not DDL.
785 .. versionadded:: 2.0
787 """
789 if typing.TYPE_CHECKING:
791 def _compiler(self, dialect: Dialect, **kw: Any) -> SQLCompiler:
792 """Return a compiler appropriate for this ClauseElement, given a
793 Dialect."""
794 ...
796 def compile( # noqa: A001
797 self,
798 bind: Optional[_HasDialect] = None,
799 dialect: Optional[Dialect] = None,
800 **kw: Any,
801 ) -> SQLCompiler: ...
804class CompilerColumnElement(
805 roles.DMLColumnRole,
806 roles.DDLConstraintColumnRole,
807 roles.ColumnsClauseRole,
808 CompilerElement,
809):
810 """A compiler-only column element used for ad-hoc string compilations.
812 .. versionadded:: 2.0
814 """
816 __slots__ = ()
818 _propagate_attrs = util.EMPTY_DICT
819 _is_collection_aggregate = False
822# SQLCoreOperations should be suiting the ExpressionElementRole
823# and ColumnsClauseRole. however the MRO issues become too elaborate
824# at the moment.
825class SQLCoreOperations(Generic[_T_co], ColumnOperators, TypingOnly):
826 __slots__ = ()
828 # annotations for comparison methods
829 # these are from operators->Operators / ColumnOperators,
830 # redefined with the specific types returned by ColumnElement hierarchies
831 if typing.TYPE_CHECKING:
833 @util.non_memoized_property
834 def _propagate_attrs(self) -> _PropagateAttrsType: ...
836 def operate(
837 self, op: OperatorType, *other: Any, **kwargs: Any
838 ) -> ColumnElement[Any]: ...
840 def reverse_operate(
841 self, op: OperatorType, other: Any, **kwargs: Any
842 ) -> ColumnElement[Any]: ...
844 @overload
845 def op(
846 self,
847 opstring: str,
848 precedence: int = ...,
849 is_comparison: bool = ...,
850 *,
851 return_type: _TypeEngineArgument[_OPT],
852 python_impl: Optional[Callable[..., Any]] = None,
853 ) -> Callable[[Any], BinaryExpression[_OPT]]: ...
855 @overload
856 def op(
857 self,
858 opstring: str,
859 precedence: int = ...,
860 is_comparison: bool = ...,
861 return_type: Optional[_TypeEngineArgument[Any]] = ...,
862 python_impl: Optional[Callable[..., Any]] = ...,
863 ) -> Callable[[Any], BinaryExpression[Any]]: ...
865 def op(
866 self,
867 opstring: str,
868 precedence: int = 0,
869 is_comparison: bool = False,
870 return_type: Optional[_TypeEngineArgument[Any]] = None,
871 python_impl: Optional[Callable[..., Any]] = None,
872 ) -> Callable[[Any], BinaryExpression[Any]]: ...
874 def bool_op(
875 self,
876 opstring: str,
877 precedence: int = 0,
878 python_impl: Optional[Callable[..., Any]] = None,
879 ) -> Callable[[Any], BinaryExpression[bool]]: ...
881 def __and__(self, other: Any) -> BooleanClauseList: ...
883 def __or__(self, other: Any) -> BooleanClauseList: ...
885 def __invert__(self) -> ColumnElement[_T_co]: ...
887 def __lt__(self, other: Any) -> ColumnElement[bool]: ...
889 def __le__(self, other: Any) -> ColumnElement[bool]: ...
891 # declare also that this class has an hash method otherwise
892 # it may be assumed to be None by type checkers since the
893 # object defines __eq__ and python sets it to None in that case:
894 # https://docs.python.org/3/reference/datamodel.html#object.__hash__
895 def __hash__(self) -> int: ...
897 def __eq__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
898 ...
900 def __ne__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
901 ...
903 def is_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
905 def is_not_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
907 def __gt__(self, other: Any) -> ColumnElement[bool]: ...
909 def __ge__(self, other: Any) -> ColumnElement[bool]: ...
911 def __neg__(self) -> UnaryExpression[_T_co]: ...
913 def __contains__(self, other: Any) -> ColumnElement[bool]: ...
915 def __getitem__(self, index: Any) -> ColumnElement[Any]: ...
917 @overload
918 def __lshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
920 @overload
921 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
923 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
925 @overload
926 def __rlshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
928 @overload
929 def __rlshift__(self, other: Any) -> ColumnElement[Any]: ...
931 def __rlshift__(self, other: Any) -> ColumnElement[Any]: ...
933 @overload
934 def __rshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
936 @overload
937 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
939 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
941 @overload
942 def __rrshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
944 @overload
945 def __rrshift__(self, other: Any) -> ColumnElement[Any]: ...
947 def __rrshift__(self, other: Any) -> ColumnElement[Any]: ...
949 def __matmul__(self, other: Any) -> ColumnElement[Any]: ...
951 def __rmatmul__(self, other: Any) -> ColumnElement[Any]: ...
953 @overload
954 def concat(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
956 @overload
957 def concat(self, other: Any) -> ColumnElement[Any]: ...
959 def concat(self, other: Any) -> ColumnElement[Any]: ...
961 def like(
962 self, other: Any, escape: Optional[str] = None
963 ) -> BinaryExpression[bool]: ...
965 def ilike(
966 self, other: Any, escape: Optional[str] = None
967 ) -> BinaryExpression[bool]: ...
969 def bitwise_xor(self, other: Any) -> BinaryExpression[Any]: ...
971 def bitwise_or(self, other: Any) -> BinaryExpression[Any]: ...
973 def bitwise_and(self, other: Any) -> BinaryExpression[Any]: ...
975 def bitwise_not(self) -> UnaryExpression[_T_co]: ...
977 def bitwise_lshift(self, other: Any) -> BinaryExpression[Any]: ...
979 def bitwise_rshift(self, other: Any) -> BinaryExpression[Any]: ...
981 def in_(
982 self,
983 other: Union[
984 Iterable[Any], BindParameter[Any], roles.InElementRole
985 ],
986 ) -> BinaryExpression[bool]: ...
988 def not_in(
989 self,
990 other: Union[
991 Iterable[Any], BindParameter[Any], roles.InElementRole
992 ],
993 ) -> BinaryExpression[bool]: ...
995 def notin_(
996 self,
997 other: Union[
998 Iterable[Any], BindParameter[Any], roles.InElementRole
999 ],
1000 ) -> BinaryExpression[bool]: ...
1002 def not_like(
1003 self, other: Any, escape: Optional[str] = None
1004 ) -> BinaryExpression[bool]: ...
1006 def notlike(
1007 self, other: Any, escape: Optional[str] = None
1008 ) -> BinaryExpression[bool]: ...
1010 def not_ilike(
1011 self, other: Any, escape: Optional[str] = None
1012 ) -> BinaryExpression[bool]: ...
1014 def notilike(
1015 self, other: Any, escape: Optional[str] = None
1016 ) -> BinaryExpression[bool]: ...
1018 def is_(self, other: Any) -> BinaryExpression[bool]: ...
1020 def is_not(self, other: Any) -> BinaryExpression[bool]: ...
1022 def isnot(self, other: Any) -> BinaryExpression[bool]: ...
1024 def startswith(
1025 self,
1026 other: Any,
1027 escape: Optional[str] = None,
1028 autoescape: bool = False,
1029 ) -> ColumnElement[bool]: ...
1031 def istartswith(
1032 self,
1033 other: Any,
1034 escape: Optional[str] = None,
1035 autoescape: bool = False,
1036 ) -> ColumnElement[bool]: ...
1038 def endswith(
1039 self,
1040 other: Any,
1041 escape: Optional[str] = None,
1042 autoescape: bool = False,
1043 ) -> ColumnElement[bool]: ...
1045 def iendswith(
1046 self,
1047 other: Any,
1048 escape: Optional[str] = None,
1049 autoescape: bool = False,
1050 ) -> ColumnElement[bool]: ...
1052 def contains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1054 def icontains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1056 def match(self, other: Any, **kwargs: Any) -> ColumnElement[bool]: ...
1058 def regexp_match(
1059 self, pattern: Any, flags: Optional[str] = None
1060 ) -> ColumnElement[bool]: ...
1062 def regexp_replace(
1063 self, pattern: Any, replacement: Any, flags: Optional[str] = None
1064 ) -> ColumnElement[str]: ...
1066 def desc(self) -> UnaryExpression[_T_co]: ...
1068 def asc(self) -> UnaryExpression[_T_co]: ...
1070 def nulls_first(self) -> UnaryExpression[_T_co]: ...
1072 def nullsfirst(self) -> UnaryExpression[_T_co]: ...
1074 def nulls_last(self) -> UnaryExpression[_T_co]: ...
1076 def nullslast(self) -> UnaryExpression[_T_co]: ...
1078 def collate(self, collation: str) -> CollationClause: ...
1080 def between(
1081 self, cleft: Any, cright: Any, symmetric: bool = False
1082 ) -> BinaryExpression[bool]: ...
1084 def distinct(self: _SQO[_T_co]) -> UnaryExpression[_T_co]: ...
1086 def any_(self) -> CollectionAggregate[Any]: ...
1088 def all_(self) -> CollectionAggregate[Any]: ...
1090 # numeric overloads. These need more tweaking
1091 # in particular they all need to have a variant for Optiona[_T]
1092 # because Optional only applies to the data side, not the expression
1093 # side
1095 @overload
1096 def __add__(
1097 self: _SQO[_NMT],
1098 other: Any,
1099 ) -> ColumnElement[_NMT]: ...
1101 @overload
1102 def __add__(
1103 self: _SQO[str],
1104 other: Any,
1105 ) -> ColumnElement[str]: ...
1107 @overload
1108 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1110 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1112 @overload
1113 def __radd__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1115 @overload
1116 def __radd__(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
1118 def __radd__(self, other: Any) -> ColumnElement[Any]: ...
1120 @overload
1121 def __sub__(
1122 self: _SQO[_NMT],
1123 other: Any,
1124 ) -> ColumnElement[_NMT]: ...
1126 @overload
1127 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1129 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1131 @overload
1132 def __rsub__(
1133 self: _SQO[_NMT],
1134 other: Any,
1135 ) -> ColumnElement[_NMT]: ...
1137 @overload
1138 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1140 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1142 @overload
1143 def __mul__(
1144 self: _SQO[_NMT],
1145 other: Any,
1146 ) -> ColumnElement[_NMT]: ...
1148 @overload
1149 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1151 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1153 @overload
1154 def __rmul__(
1155 self: _SQO[_NMT],
1156 other: Any,
1157 ) -> ColumnElement[_NMT]: ...
1159 @overload
1160 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1162 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1164 @overload
1165 def __mod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1167 @overload
1168 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1170 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1172 @overload
1173 def __rmod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1175 @overload
1176 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1178 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1180 @overload
1181 def __truediv__(
1182 self: _SQO[int], other: Any
1183 ) -> ColumnElement[_NUMERIC]: ...
1185 @overload
1186 def __truediv__(self: _SQO[_NT], other: Any) -> ColumnElement[_NT]: ...
1188 @overload
1189 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1191 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1193 @overload
1194 def __rtruediv__(
1195 self: _SQO[_NMT], other: Any
1196 ) -> ColumnElement[_NUMERIC]: ...
1198 @overload
1199 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1201 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1203 @overload
1204 def __floordiv__(
1205 self: _SQO[_NMT], other: Any
1206 ) -> ColumnElement[_NMT]: ...
1208 @overload
1209 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1211 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1213 @overload
1214 def __rfloordiv__(
1215 self: _SQO[_NMT], other: Any
1216 ) -> ColumnElement[_NMT]: ...
1218 @overload
1219 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1221 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1224class SQLColumnExpression(
1225 SQLCoreOperations[_T_co], roles.ExpressionElementRole[_T_co], TypingOnly
1226):
1227 """A type that may be used to indicate any SQL column element or object
1228 that acts in place of one.
1230 :class:`.SQLColumnExpression` is a base of
1231 :class:`.ColumnElement`, as well as within the bases of ORM elements
1232 such as :class:`.InstrumentedAttribute`, and may be used in :pep:`484`
1233 typing to indicate arguments or return values that should behave
1234 as column expressions.
1236 .. versionadded:: 2.0.0b4
1239 """
1241 __slots__ = ()
1244_SQO = SQLCoreOperations
1247class ColumnElement(
1248 roles.ColumnArgumentOrKeyRole,
1249 roles.StatementOptionRole,
1250 roles.WhereHavingRole,
1251 roles.BinaryElementRole[_T],
1252 roles.OrderByRole,
1253 roles.ColumnsClauseRole,
1254 roles.LimitOffsetRole,
1255 roles.DMLColumnRole,
1256 roles.DDLConstraintColumnRole,
1257 roles.DDLExpressionRole,
1258 SQLColumnExpression[_T],
1259 DQLDMLClauseElement,
1260):
1261 """Represent a column-oriented SQL expression suitable for usage in the
1262 "columns" clause, WHERE clause etc. of a statement.
1264 While the most familiar kind of :class:`_expression.ColumnElement` is the
1265 :class:`_schema.Column` object, :class:`_expression.ColumnElement`
1266 serves as the basis
1267 for any unit that may be present in a SQL expression, including
1268 the expressions themselves, SQL functions, bound parameters,
1269 literal expressions, keywords such as ``NULL``, etc.
1270 :class:`_expression.ColumnElement`
1271 is the ultimate base class for all such elements.
1273 A wide variety of SQLAlchemy Core functions work at the SQL expression
1274 level, and are intended to accept instances of
1275 :class:`_expression.ColumnElement` as
1276 arguments. These functions will typically document that they accept a
1277 "SQL expression" as an argument. What this means in terms of SQLAlchemy
1278 usually refers to an input which is either already in the form of a
1279 :class:`_expression.ColumnElement` object,
1280 or a value which can be **coerced** into
1281 one. The coercion rules followed by most, but not all, SQLAlchemy Core
1282 functions with regards to SQL expressions are as follows:
1284 * a literal Python value, such as a string, integer or floating
1285 point value, boolean, datetime, ``Decimal`` object, or virtually
1286 any other Python object, will be coerced into a "literal bound
1287 value". This generally means that a :func:`.bindparam` will be
1288 produced featuring the given value embedded into the construct; the
1289 resulting :class:`.BindParameter` object is an instance of
1290 :class:`_expression.ColumnElement`.
1291 The Python value will ultimately be sent
1292 to the DBAPI at execution time as a parameterized argument to the
1293 ``execute()`` or ``executemany()`` methods, after SQLAlchemy
1294 type-specific converters (e.g. those provided by any associated
1295 :class:`.TypeEngine` objects) are applied to the value.
1297 * any special object value, typically ORM-level constructs, which
1298 feature an accessor called ``__clause_element__()``. The Core
1299 expression system looks for this method when an object of otherwise
1300 unknown type is passed to a function that is looking to coerce the
1301 argument into a :class:`_expression.ColumnElement` and sometimes a
1302 :class:`_expression.SelectBase` expression.
1303 It is used within the ORM to
1304 convert from ORM-specific objects like mapped classes and
1305 mapped attributes into Core expression objects.
1307 * The Python ``None`` value is typically interpreted as ``NULL``,
1308 which in SQLAlchemy Core produces an instance of :func:`.null`.
1310 A :class:`_expression.ColumnElement` provides the ability to generate new
1311 :class:`_expression.ColumnElement`
1312 objects using Python expressions. This means that Python operators
1313 such as ``==``, ``!=`` and ``<`` are overloaded to mimic SQL operations,
1314 and allow the instantiation of further :class:`_expression.ColumnElement`
1315 instances
1316 which are composed from other, more fundamental
1317 :class:`_expression.ColumnElement`
1318 objects. For example, two :class:`.ColumnClause` objects can be added
1319 together with the addition operator ``+`` to produce
1320 a :class:`.BinaryExpression`.
1321 Both :class:`.ColumnClause` and :class:`.BinaryExpression` are subclasses
1322 of :class:`_expression.ColumnElement`:
1324 .. sourcecode:: pycon+sql
1326 >>> from sqlalchemy.sql import column
1327 >>> column("a") + column("b")
1328 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
1329 >>> print(column("a") + column("b"))
1330 {printsql}a + b
1332 .. seealso::
1334 :class:`_schema.Column`
1336 :func:`_expression.column`
1338 """
1340 __visit_name__ = "column_element"
1342 primary_key: bool = False
1343 _is_clone_of: Optional[ColumnElement[_T]]
1344 _is_column_element = True
1345 _insert_sentinel: bool = False
1346 _omit_from_statements = False
1347 _is_collection_aggregate = False
1349 foreign_keys: AbstractSet[ForeignKey] = frozenset()
1351 @util.memoized_property
1352 def _proxies(self) -> List[ColumnElement[Any]]:
1353 return []
1355 @util.non_memoized_property
1356 def _tq_label(self) -> Optional[str]:
1357 """The named label that can be used to target
1358 this column in a result set in a "table qualified" context.
1360 This label is almost always the label used when
1361 rendering <expr> AS <label> in a SELECT statement when using
1362 the LABEL_STYLE_TABLENAME_PLUS_COL label style, which is what the
1363 legacy ORM ``Query`` object uses as well.
1365 For a regular Column bound to a Table, this is typically the label
1366 <tablename>_<columnname>. For other constructs, different rules
1367 may apply, such as anonymized labels and others.
1369 .. versionchanged:: 1.4.21 renamed from ``._label``
1371 """
1372 return None
1374 key: Optional[str] = None
1375 """The 'key' that in some circumstances refers to this object in a
1376 Python namespace.
1378 This typically refers to the "key" of the column as present in the
1379 ``.c`` collection of a selectable, e.g. ``sometable.c["somekey"]`` would
1380 return a :class:`_schema.Column` with a ``.key`` of "somekey".
1382 """
1384 @HasMemoized.memoized_attribute
1385 def _tq_key_label(self) -> Optional[str]:
1386 """A label-based version of 'key' that in some circumstances refers
1387 to this object in a Python namespace.
1390 _tq_key_label comes into play when a select() statement is constructed
1391 with apply_labels(); in this case, all Column objects in the ``.c``
1392 collection are rendered as <tablename>_<columnname> in SQL; this is
1393 essentially the value of ._label. But to locate those columns in the
1394 ``.c`` collection, the name is along the lines of <tablename>_<key>;
1395 that's the typical value of .key_label.
1397 .. versionchanged:: 1.4.21 renamed from ``._key_label``
1399 """
1400 return self._proxy_key
1402 @property
1403 def _key_label(self) -> Optional[str]:
1404 """legacy; renamed to _tq_key_label"""
1405 return self._tq_key_label
1407 @property
1408 def _label(self) -> Optional[str]:
1409 """legacy; renamed to _tq_label"""
1410 return self._tq_label
1412 @property
1413 def _non_anon_label(self) -> Optional[str]:
1414 """the 'name' that naturally applies this element when rendered in
1415 SQL.
1417 Concretely, this is the "name" of a column or a label in a
1418 SELECT statement; ``<columnname>`` and ``<labelname>`` below:
1420 .. sourcecode:: sql
1422 SELECT <columnmame> FROM table
1424 SELECT column AS <labelname> FROM table
1426 Above, the two names noted will be what's present in the DBAPI
1427 ``cursor.description`` as the names.
1429 If this attribute returns ``None``, it means that the SQL element as
1430 written does not have a 100% fully predictable "name" that would appear
1431 in the ``cursor.description``. Examples include SQL functions, CAST
1432 functions, etc. While such things do return names in
1433 ``cursor.description``, they are only predictable on a
1434 database-specific basis; e.g. an expression like ``MAX(table.col)`` may
1435 appear as the string ``max`` on one database (like PostgreSQL) or may
1436 appear as the whole expression ``max(table.col)`` on SQLite.
1438 The default implementation looks for a ``.name`` attribute on the
1439 object, as has been the precedent established in SQLAlchemy for many
1440 years. An exception is made on the ``FunctionElement`` subclass
1441 so that the return value is always ``None``.
1443 .. versionadded:: 1.4.21
1447 """
1448 return getattr(self, "name", None)
1450 _render_label_in_columns_clause = True
1451 """A flag used by select._columns_plus_names that helps to determine
1452 we are actually going to render in terms of "SELECT <col> AS <label>".
1453 This flag can be returned as False for some Column objects that want
1454 to be rendered as simple "SELECT <col>"; typically columns that don't have
1455 any parent table and are named the same as what the label would be
1456 in any case.
1458 """
1460 _allow_label_resolve = True
1461 """A flag that can be flipped to prevent a column from being resolvable
1462 by string label name.
1464 The joined eager loader strategy in the ORM uses this, for example.
1466 """
1468 _is_implicitly_boolean = False
1470 _alt_names: Sequence[str] = ()
1472 if TYPE_CHECKING:
1474 def _ungroup(self) -> ColumnElement[_T]: ...
1476 @overload
1477 def self_group(self, against: None = None) -> ColumnElement[_T]: ...
1479 @overload
1480 def self_group(
1481 self, against: Optional[OperatorType] = None
1482 ) -> ColumnElement[Any]: ...
1484 def self_group(
1485 self, against: Optional[OperatorType] = None
1486 ) -> ColumnElement[Any]:
1487 if (
1488 against in (operators.and_, operators.or_, operators._asbool)
1489 and self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity
1490 ):
1491 return AsBoolean(self, operators.is_true, operators.is_false)
1492 elif against in (operators.any_op, operators.all_op):
1493 return Grouping(self)
1494 else:
1495 return self
1497 @overload
1498 def _negate(self: ColumnElement[bool]) -> ColumnElement[bool]: ...
1500 @overload
1501 def _negate(self: ColumnElement[_T]) -> ColumnElement[_T]: ...
1503 def _negate(self) -> ColumnElement[Any]:
1504 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
1505 return AsBoolean(self, operators.is_false, operators.is_true)
1506 else:
1507 grouped = self.self_group(against=operators.inv)
1508 assert isinstance(grouped, ColumnElement)
1509 return UnaryExpression(
1510 grouped,
1511 operator=operators.inv,
1512 )
1514 type: TypeEngine[_T]
1516 if not TYPE_CHECKING:
1518 @util.memoized_property
1519 def type(self) -> TypeEngine[_T]: # noqa: A001
1520 # used for delayed setup of
1521 # type_api
1522 return type_api.NULLTYPE
1524 @HasMemoized.memoized_attribute
1525 def comparator(self) -> TypeEngine.Comparator[_T]:
1526 try:
1527 comparator_factory = self.type.comparator_factory
1528 except AttributeError as err:
1529 raise TypeError(
1530 "Object %r associated with '.type' attribute "
1531 "is not a TypeEngine class or object" % self.type
1532 ) from err
1533 else:
1534 return comparator_factory(self)
1536 def __setstate__(self, state):
1537 self.__dict__.update(state)
1539 def __getattr__(self, key: str) -> Any:
1540 try:
1541 return getattr(self.comparator, key)
1542 except AttributeError as err:
1543 raise AttributeError(
1544 "Neither %r object nor %r object has an attribute %r"
1545 % (
1546 type(self).__name__,
1547 type(self.comparator).__name__,
1548 key,
1549 )
1550 ) from err
1552 def operate(
1553 self,
1554 op: operators.OperatorType,
1555 *other: Any,
1556 **kwargs: Any,
1557 ) -> ColumnElement[Any]:
1558 return op(self.comparator, *other, **kwargs) # type: ignore[no-any-return] # noqa: E501
1560 def reverse_operate(
1561 self, op: operators.OperatorType, other: Any, **kwargs: Any
1562 ) -> ColumnElement[Any]:
1563 return op(other, self.comparator, **kwargs) # type: ignore[no-any-return] # noqa: E501
1565 def _bind_param(
1566 self,
1567 operator: operators.OperatorType,
1568 obj: Any,
1569 type_: Optional[TypeEngine[_T]] = None,
1570 expanding: bool = False,
1571 ) -> BindParameter[_T]:
1572 return BindParameter(
1573 None,
1574 obj,
1575 _compared_to_operator=operator,
1576 type_=type_,
1577 _compared_to_type=self.type,
1578 unique=True,
1579 expanding=expanding,
1580 )
1582 @property
1583 def expression(self) -> ColumnElement[Any]:
1584 """Return a column expression.
1586 Part of the inspection interface; returns self.
1588 """
1589 return self
1591 @property
1592 def _select_iterable(self) -> _SelectIterable:
1593 return (self,)
1595 @util.memoized_property
1596 def base_columns(self) -> FrozenSet[ColumnElement[Any]]:
1597 return frozenset(c for c in self.proxy_set if not c._proxies)
1599 @util.memoized_property
1600 def proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1601 """set of all columns we are proxying
1603 as of 2.0 this is explicitly deannotated columns. previously it was
1604 effectively deannotated columns but wasn't enforced. annotated
1605 columns should basically not go into sets if at all possible because
1606 their hashing behavior is very non-performant.
1608 """
1609 return frozenset([self._deannotate()]).union(
1610 itertools.chain(*[c.proxy_set for c in self._proxies])
1611 )
1613 @util.memoized_property
1614 def _expanded_proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1615 return frozenset(_expand_cloned(self.proxy_set))
1617 def _uncached_proxy_list(self) -> List[ColumnElement[Any]]:
1618 """An 'uncached' version of proxy set.
1620 This list includes annotated columns which perform very poorly in
1621 set operations.
1623 """
1625 return [self] + list(
1626 itertools.chain(*[c._uncached_proxy_list() for c in self._proxies])
1627 )
1629 def shares_lineage(self, othercolumn: ColumnElement[Any]) -> bool:
1630 """Return True if the given :class:`_expression.ColumnElement`
1631 has a common ancestor to this :class:`_expression.ColumnElement`."""
1633 return bool(self.proxy_set.intersection(othercolumn.proxy_set))
1635 def _compare_name_for_result(self, other: ColumnElement[Any]) -> bool:
1636 """Return True if the given column element compares to this one
1637 when targeting within a result row."""
1639 return (
1640 hasattr(other, "name")
1641 and hasattr(self, "name")
1642 and other.name == self.name
1643 )
1645 @HasMemoized.memoized_attribute
1646 def _proxy_key(self) -> Optional[str]:
1647 if self._annotations and "proxy_key" in self._annotations:
1648 return cast(str, self._annotations["proxy_key"])
1650 name = self.key
1651 if not name:
1652 # there's a bit of a seeming contradiction which is that the
1653 # "_non_anon_label" of a column can in fact be an
1654 # "_anonymous_label"; this is when it's on a column that is
1655 # proxying for an anonymous expression in a subquery.
1656 name = self._non_anon_label
1658 if isinstance(name, _anonymous_label):
1659 return None
1660 else:
1661 return name
1663 @HasMemoized.memoized_attribute
1664 def _expression_label(self) -> Optional[str]:
1665 """a suggested label to use in the case that the column has no name,
1666 which should be used if possible as the explicit 'AS <label>'
1667 where this expression would normally have an anon label.
1669 this is essentially mostly what _proxy_key does except it returns
1670 None if the column has a normal name that can be used.
1672 """
1674 if getattr(self, "name", None) is not None:
1675 return None
1676 elif self._annotations and "proxy_key" in self._annotations:
1677 return cast(str, self._annotations["proxy_key"])
1678 else:
1679 return None
1681 def _make_proxy(
1682 self,
1683 selectable: FromClause,
1684 *,
1685 primary_key: ColumnSet,
1686 foreign_keys: Set[KeyedColumnElement[Any]],
1687 name: Optional[str] = None,
1688 key: Optional[str] = None,
1689 name_is_truncatable: bool = False,
1690 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
1691 **kw: Any,
1692 ) -> typing_Tuple[str, ColumnClause[_T]]:
1693 """Create a new :class:`_expression.ColumnElement` representing this
1694 :class:`_expression.ColumnElement` as it appears in the select list of
1695 a descending selectable.
1697 """
1698 if name is None:
1699 name = self._anon_name_label
1700 if key is None:
1701 key = self._proxy_key
1702 else:
1703 key = name
1705 assert key is not None
1707 co: ColumnClause[_T] = ColumnClause(
1708 (
1709 coercions.expect(roles.TruncatedLabelRole, name)
1710 if name_is_truncatable
1711 else name
1712 ),
1713 type_=getattr(self, "type", None),
1714 _selectable=selectable,
1715 )
1717 co._propagate_attrs = selectable._propagate_attrs
1718 if compound_select_cols:
1719 co._proxies = list(compound_select_cols)
1720 else:
1721 co._proxies = [self]
1722 if selectable._is_clone_of is not None:
1723 co._is_clone_of = selectable._is_clone_of.columns.get(key)
1724 return key, co
1726 def cast(self, type_: _TypeEngineArgument[_OPT]) -> Cast[_OPT]:
1727 """Produce a type cast, i.e. ``CAST(<expression> AS <type>)``.
1729 This is a shortcut to the :func:`_expression.cast` function.
1731 .. seealso::
1733 :ref:`tutorial_casts`
1735 :func:`_expression.cast`
1737 :func:`_expression.type_coerce`
1739 """
1740 return Cast(self, type_)
1742 def label(self, name: Optional[str]) -> Label[_T]:
1743 """Produce a column label, i.e. ``<columnname> AS <name>``.
1745 This is a shortcut to the :func:`_expression.label` function.
1747 If 'name' is ``None``, an anonymous label name will be generated.
1749 """
1750 return Label(name, self, self.type)
1752 def _anon_label(
1753 self, seed: Optional[str], add_hash: Optional[int] = None
1754 ) -> _anonymous_label:
1755 while self._is_clone_of is not None:
1756 self = self._is_clone_of
1758 # as of 1.4 anonymous label for ColumnElement uses hash(), not id(),
1759 # as the identifier, because a column and its annotated version are
1760 # the same thing in a SQL statement
1761 hash_value = hash(self)
1763 if add_hash:
1764 # this path is used for disambiguating anon labels that would
1765 # otherwise be the same name for the same element repeated.
1766 # an additional numeric value is factored in for each label.
1768 # shift hash(self) (which is id(self), typically 8 byte integer)
1769 # 16 bits leftward. fill extra add_hash on right
1770 assert add_hash < (2 << 15)
1771 assert seed
1772 hash_value = (hash_value << 16) | add_hash
1774 # extra underscore is added for labels with extra hash
1775 # values, to isolate the "deduped anon" namespace from the
1776 # regular namespace. eliminates chance of these
1777 # manufactured hash values overlapping with regular ones for some
1778 # undefined python interpreter
1779 seed = seed + "_"
1781 if isinstance(seed, _anonymous_label):
1782 # NOTE: the space after the hash is required
1783 return _anonymous_label(f"{seed}%({hash_value} )s")
1785 return _anonymous_label.safe_construct(hash_value, seed or "anon")
1787 @util.memoized_property
1788 def _anon_name_label(self) -> str:
1789 """Provides a constant 'anonymous label' for this ColumnElement.
1791 This is a label() expression which will be named at compile time.
1792 The same label() is returned each time ``anon_label`` is called so
1793 that expressions can reference ``anon_label`` multiple times,
1794 producing the same label name at compile time.
1796 The compiler uses this function automatically at compile time
1797 for expressions that are known to be 'unnamed' like binary
1798 expressions and function calls.
1800 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1801 public and is renamed to _anon_name_label. anon_name exists
1802 for backwards compat
1804 """
1805 name = getattr(self, "name", None)
1806 return self._anon_label(name)
1808 @util.memoized_property
1809 def _anon_key_label(self) -> _anonymous_label:
1810 """Provides a constant 'anonymous key label' for this ColumnElement.
1812 Compare to ``anon_label``, except that the "key" of the column,
1813 if available, is used to generate the label.
1815 This is used when a deduplicating key is placed into the columns
1816 collection of a selectable.
1818 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1819 public and is renamed to _anon_key_label. anon_key_label exists
1820 for backwards compat
1822 """
1823 return self._anon_label(self._proxy_key)
1825 @property
1826 @util.deprecated(
1827 "1.4",
1828 "The :attr:`_expression.ColumnElement.anon_label` attribute is now "
1829 "private, and the public accessor is deprecated.",
1830 )
1831 def anon_label(self) -> str:
1832 return self._anon_name_label
1834 @property
1835 @util.deprecated(
1836 "1.4",
1837 "The :attr:`_expression.ColumnElement.anon_key_label` attribute is "
1838 "now private, and the public accessor is deprecated.",
1839 )
1840 def anon_key_label(self) -> str:
1841 return self._anon_key_label
1843 def _dedupe_anon_label_idx(self, idx: int) -> str:
1844 """label to apply to a column that is anon labeled, but repeated
1845 in the SELECT, so that we have to make an "extra anon" label that
1846 disambiguates it from the previous appearance.
1848 these labels come out like "foo_bar_id__1" and have double underscores
1849 in them.
1851 """
1852 label = getattr(self, "name", None)
1854 # current convention is that if the element doesn't have a
1855 # ".name" (usually because it is not NamedColumn), we try to
1856 # use a "table qualified" form for the "dedupe anon" label,
1857 # based on the notion that a label like
1858 # "CAST(casttest.v1 AS DECIMAL) AS casttest_v1__1" looks better than
1859 # "CAST(casttest.v1 AS DECIMAL) AS anon__1"
1861 if label is None:
1862 return self._dedupe_anon_tq_label_idx(idx)
1863 else:
1864 return self._anon_label(label, add_hash=idx)
1866 @util.memoized_property
1867 def _anon_tq_label(self) -> _anonymous_label:
1868 return self._anon_label(getattr(self, "_tq_label", None))
1870 @util.memoized_property
1871 def _anon_tq_key_label(self) -> _anonymous_label:
1872 return self._anon_label(getattr(self, "_tq_key_label", None))
1874 def _dedupe_anon_tq_label_idx(self, idx: int) -> _anonymous_label:
1875 label = getattr(self, "_tq_label", None) or "anon"
1877 return self._anon_label(label, add_hash=idx)
1880class KeyedColumnElement(ColumnElement[_T]):
1881 """ColumnElement where ``.key`` is non-None."""
1883 _is_keyed_column_element = True
1885 key: str
1888class WrapsColumnExpression(ColumnElement[_T]):
1889 """Mixin that defines a :class:`_expression.ColumnElement`
1890 as a wrapper with special
1891 labeling behavior for an expression that already has a name.
1893 .. versionadded:: 1.4
1895 .. seealso::
1897 :ref:`change_4449`
1900 """
1902 @property
1903 def wrapped_column_expression(self) -> ColumnElement[_T]:
1904 raise NotImplementedError()
1906 @util.non_memoized_property
1907 def _tq_label(self) -> Optional[str]:
1908 wce = self.wrapped_column_expression
1909 if hasattr(wce, "_tq_label"):
1910 return wce._tq_label
1911 else:
1912 return None
1914 @property
1915 def _label(self) -> Optional[str]:
1916 return self._tq_label
1918 @property
1919 def _non_anon_label(self) -> Optional[str]:
1920 return None
1922 @util.non_memoized_property
1923 def _anon_name_label(self) -> str:
1924 wce = self.wrapped_column_expression
1926 # this logic tries to get the WrappedColumnExpression to render
1927 # with "<expr> AS <name>", where "<name>" is the natural name
1928 # within the expression itself. e.g. "CAST(table.foo) AS foo".
1929 if not wce._is_text_clause:
1930 nal = wce._non_anon_label
1931 if nal:
1932 return nal
1933 elif hasattr(wce, "_anon_name_label"):
1934 return wce._anon_name_label
1935 return super()._anon_name_label
1937 def _dedupe_anon_label_idx(self, idx: int) -> str:
1938 wce = self.wrapped_column_expression
1939 nal = wce._non_anon_label
1940 if nal:
1941 return self._anon_label(nal + "_")
1942 else:
1943 return self._dedupe_anon_tq_label_idx(idx)
1945 @property
1946 def _proxy_key(self):
1947 wce = self.wrapped_column_expression
1949 if not wce._is_text_clause:
1950 return wce._proxy_key
1951 return super()._proxy_key
1954class BindParameter(roles.InElementRole, KeyedColumnElement[_T]):
1955 r"""Represent a "bound expression".
1957 :class:`.BindParameter` is invoked explicitly using the
1958 :func:`.bindparam` function, as in::
1960 from sqlalchemy import bindparam
1962 stmt = select(users_table).where(
1963 users_table.c.name == bindparam("username")
1964 )
1966 Detailed discussion of how :class:`.BindParameter` is used is
1967 at :func:`.bindparam`.
1969 .. seealso::
1971 :func:`.bindparam`
1973 """
1975 __visit_name__ = "bindparam"
1977 _traverse_internals: _TraverseInternalsType = [
1978 ("key", InternalTraversal.dp_anon_name),
1979 ("type", InternalTraversal.dp_type),
1980 ("callable", InternalTraversal.dp_plain_dict),
1981 ("value", InternalTraversal.dp_plain_obj),
1982 ("literal_execute", InternalTraversal.dp_boolean),
1983 ]
1985 key: str
1986 _anon_map_key: Optional[str] = None
1987 type: TypeEngine[_T]
1988 value: Optional[_T]
1990 _is_crud = False
1991 _is_bind_parameter = True
1993 # bindparam implements its own _gen_cache_key() method however
1994 # we check subclasses for this flag, else no cache key is generated
1995 inherit_cache = True
1997 def __init__(
1998 self,
1999 key: Optional[str],
2000 value: Any = _NoArg.NO_ARG,
2001 type_: Optional[_TypeEngineArgument[_T]] = None,
2002 unique: bool = False,
2003 required: Union[bool, Literal[_NoArg.NO_ARG]] = _NoArg.NO_ARG,
2004 quote: Optional[bool] = None,
2005 callable_: Optional[Callable[[], Any]] = None,
2006 expanding: bool = False,
2007 isoutparam: bool = False,
2008 literal_execute: bool = False,
2009 _compared_to_operator: Optional[OperatorType] = None,
2010 _compared_to_type: Optional[TypeEngine[Any]] = None,
2011 _is_crud: bool = False,
2012 ):
2013 if required is _NoArg.NO_ARG:
2014 required = value is _NoArg.NO_ARG and callable_ is None
2015 if value is _NoArg.NO_ARG:
2016 value = None
2018 if quote is not None:
2019 key = quoted_name.construct(key, quote)
2021 if unique:
2022 self.key, self._anon_map_key = (
2023 _anonymous_label.safe_construct_with_key(
2024 id(self),
2025 (
2026 key
2027 if key is not None
2028 and not isinstance(key, _anonymous_label)
2029 else "param"
2030 ),
2031 sanitize_key=True,
2032 )
2033 )
2034 elif key:
2035 self.key = key
2036 else:
2037 self.key, self._anon_map_key = (
2038 _anonymous_label.safe_construct_with_key(id(self), "param")
2039 )
2041 # identifying key that won't change across
2042 # clones, used to identify the bind's logical
2043 # identity
2044 self._identifying_key = self.key
2046 # key that was passed in the first place, used to
2047 # generate new keys
2048 self._orig_key = key or "param"
2050 self.unique = unique
2051 self.value = value
2052 self.callable = callable_
2053 self.isoutparam = isoutparam
2054 self.required = required
2056 # indicate an "expanding" parameter; the compiler sets this
2057 # automatically in the compiler _render_in_expr_w_bindparam method
2058 # for an IN expression
2059 self.expanding = expanding
2061 # this is another hint to help w/ expanding and is typically
2062 # set in the compiler _render_in_expr_w_bindparam method for an
2063 # IN expression
2064 self.expand_op = None
2066 self.literal_execute = literal_execute
2067 if _is_crud:
2068 self._is_crud = True
2070 if type_ is None:
2071 if expanding:
2072 if value:
2073 check_value = value[0]
2074 else:
2075 check_value = type_api._NO_VALUE_IN_LIST
2076 else:
2077 check_value = value
2078 if _compared_to_type is not None:
2079 self.type = _compared_to_type.coerce_compared_value(
2080 _compared_to_operator, check_value
2081 )
2082 else:
2083 self.type = type_api._resolve_value_to_type(check_value)
2084 elif isinstance(type_, type):
2085 self.type = type_()
2086 elif is_tuple_type(type_):
2087 if value:
2088 if expanding:
2089 check_value = value[0]
2090 else:
2091 check_value = value
2092 cast("BindParameter[TupleAny]", self).type = (
2093 type_._resolve_values_to_types(check_value)
2094 )
2095 else:
2096 cast("BindParameter[TupleAny]", self).type = type_
2097 else:
2098 self.type = type_
2100 def _with_value(self, value, maintain_key=False, required=NO_ARG):
2101 """Return a copy of this :class:`.BindParameter` with the given value
2102 set.
2103 """
2104 cloned = self._clone(maintain_key=maintain_key)
2105 cloned.value = value
2106 cloned.callable = None
2107 cloned.required = required if required is not NO_ARG else self.required
2108 if cloned.type is type_api.NULLTYPE:
2109 cloned.type = type_api._resolve_value_to_type(value)
2110 return cloned
2112 @property
2113 def effective_value(self) -> Optional[_T]:
2114 """Return the value of this bound parameter,
2115 taking into account if the ``callable`` parameter
2116 was set.
2118 The ``callable`` value will be evaluated
2119 and returned if present, else ``value``.
2121 """
2122 if self.callable:
2123 # TODO: set up protocol for bind parameter callable
2124 return self.callable() # type: ignore
2125 else:
2126 return self.value
2128 def render_literal_execute(self) -> Self:
2129 """Produce a copy of this bound parameter that will enable the
2130 :paramref:`_sql.BindParameter.literal_execute` flag.
2132 The :paramref:`_sql.BindParameter.literal_execute` flag will
2133 have the effect of the parameter rendered in the compiled SQL
2134 string using ``[POSTCOMPILE]`` form, which is a special form that
2135 is converted to be a rendering of the literal value of the parameter
2136 at SQL execution time. The rationale is to support caching
2137 of SQL statement strings that can embed per-statement literal values,
2138 such as LIMIT and OFFSET parameters, in the final SQL string that
2139 is passed to the DBAPI. Dialects in particular may want to use
2140 this method within custom compilation schemes.
2142 .. versionadded:: 1.4.5
2144 .. seealso::
2146 :ref:`engine_thirdparty_caching`
2148 """
2149 c: Self = ClauseElement._clone(self)
2150 c.literal_execute = True
2151 return c
2153 def _negate_in_binary(self, negated_op, original_op):
2154 if self.expand_op is original_op:
2155 bind = self._clone()
2156 bind.expand_op = negated_op
2157 return bind
2158 else:
2159 return self
2161 def _with_binary_element_type(self, type_: TypeEngine[Any]) -> Self:
2162 c: Self = ClauseElement._clone(self)
2163 c.type = type_
2164 return c
2166 def _clone(self, maintain_key: bool = False, **kw: Any) -> Self:
2167 c: Self = ClauseElement._clone(self, **kw)
2168 # ensure all the BindParameter objects stay in cloned set.
2169 # in #7823, we changed "clone" so that a clone only keeps a reference
2170 # to the "original" element, since for column correspondence, that's
2171 # all we need. However, for BindParam, _cloned_set is used by
2172 # the "cache key bind match" lookup, which means if any of those
2173 # interim BindParameter objects became part of a cache key in the
2174 # cache, we need it. So here, make sure all clones keep carrying
2175 # forward.
2176 c._cloned_set.update(self._cloned_set)
2177 if not maintain_key and self.unique:
2178 c.key, c._anon_map_key = _anonymous_label.safe_construct_with_key(
2179 id(c), c._orig_key or "param", sanitize_key=True
2180 )
2181 return c
2183 def _gen_cache_key(self, anon_map, bindparams):
2184 _gen_cache_ok = self.__class__.__dict__.get("inherit_cache", False)
2186 if not _gen_cache_ok:
2187 if anon_map is not None:
2188 anon_map[NO_CACHE] = True
2189 return None
2191 id_, found = anon_map.get_anon(self)
2192 if found:
2193 return (id_, self.__class__)
2195 if bindparams is not None:
2196 bindparams.append(self)
2198 return (
2199 id_,
2200 self.__class__,
2201 self.type._static_cache_key,
2202 (
2203 anon_map[self._anon_map_key]
2204 if self._anon_map_key is not None
2205 else self.key
2206 ),
2207 self.literal_execute,
2208 )
2210 def _convert_to_unique(self):
2211 if not self.unique:
2212 self.unique = True
2213 self.key, self._anon_map_key = (
2214 _anonymous_label.safe_construct_with_key(
2215 id(self), self._orig_key or "param", sanitize_key=True
2216 )
2217 )
2219 def __getstate__(self):
2220 """execute a deferred value for serialization purposes."""
2222 d = self.__dict__.copy()
2223 v = self.value
2224 if self.callable:
2225 v = self.callable()
2226 d["callable"] = None
2227 d["value"] = v
2228 return d
2230 def __setstate__(self, state):
2231 if state.get("unique", False):
2232 anon_and_key = _anonymous_label.safe_construct_with_key(
2233 id(self), state.get("_orig_key", "param"), sanitize_key=True
2234 )
2235 state["key"], state["_anon_map_key"] = anon_and_key
2236 self.__dict__.update(state)
2238 def __repr__(self):
2239 return "%s(%r, %r, type_=%r)" % (
2240 self.__class__.__name__,
2241 self.key,
2242 self.value,
2243 self.type,
2244 )
2247class TypeClause(DQLDMLClauseElement):
2248 """Handle a type keyword in a SQL statement.
2250 Used by the ``Case`` statement.
2252 """
2254 __visit_name__ = "typeclause"
2256 _traverse_internals: _TraverseInternalsType = [
2257 ("type", InternalTraversal.dp_type)
2258 ]
2259 type: TypeEngine[Any]
2261 def __init__(self, type_: TypeEngine[Any]):
2262 self.type = type_
2265class TextClause(
2266 roles.DDLConstraintColumnRole,
2267 roles.DDLExpressionRole,
2268 roles.StatementOptionRole,
2269 roles.WhereHavingRole,
2270 roles.OrderByRole,
2271 roles.FromClauseRole,
2272 roles.SelectStatementRole,
2273 roles.InElementRole,
2274 Generative,
2275 Executable,
2276 DQLDMLClauseElement,
2277 roles.BinaryElementRole[Any],
2278 inspection.Inspectable["TextClause"],
2279):
2280 """Represent a literal SQL text fragment.
2282 E.g.::
2284 from sqlalchemy import text
2286 t = text("SELECT * FROM users")
2287 result = connection.execute(t)
2289 The :class:`_expression.TextClause` construct is produced using the
2290 :func:`_expression.text`
2291 function; see that function for full documentation.
2293 .. seealso::
2295 :func:`_expression.text`
2297 """
2299 __visit_name__ = "textclause"
2301 _traverse_internals: _TraverseInternalsType = [
2302 ("_bindparams", InternalTraversal.dp_string_clauseelement_dict),
2303 ("text", InternalTraversal.dp_string),
2304 ]
2306 _is_text_clause = True
2308 _is_textual = True
2310 _bind_params_regex = re.compile(r"(?<![:\w\x5c]):(\w+)(?!:)", re.UNICODE)
2311 _is_implicitly_boolean = False
2313 _render_label_in_columns_clause = False
2315 _omit_from_statements = False
2317 _is_collection_aggregate = False
2319 @property
2320 def _hide_froms(self) -> Iterable[FromClause]:
2321 return ()
2323 def __and__(self, other):
2324 # support use in select.where(), query.filter()
2325 return and_(self, other)
2327 @property
2328 def _select_iterable(self) -> _SelectIterable:
2329 return (self,)
2331 # help in those cases where text() is
2332 # interpreted in a column expression situation
2333 key: Optional[str] = None
2334 _label: Optional[str] = None
2336 _allow_label_resolve = False
2338 @property
2339 def _is_star(self): # type: ignore[override]
2340 return self.text == "*"
2342 def __init__(self, text: str):
2343 self._bindparams: Dict[str, BindParameter[Any]] = {}
2345 def repl(m):
2346 self._bindparams[m.group(1)] = BindParameter(m.group(1))
2347 return ":%s" % m.group(1)
2349 # scan the string and search for bind parameter names, add them
2350 # to the list of bindparams
2351 self.text = self._bind_params_regex.sub(repl, text)
2353 @_generative
2354 def bindparams(
2355 self,
2356 *binds: BindParameter[Any],
2357 **names_to_values: Any,
2358 ) -> Self:
2359 """Establish the values and/or types of bound parameters within
2360 this :class:`_expression.TextClause` construct.
2362 Given a text construct such as::
2364 from sqlalchemy import text
2366 stmt = text(
2367 "SELECT id, name FROM user WHERE name=:name AND timestamp=:timestamp"
2368 )
2370 the :meth:`_expression.TextClause.bindparams`
2371 method can be used to establish
2372 the initial value of ``:name`` and ``:timestamp``,
2373 using simple keyword arguments::
2375 stmt = stmt.bindparams(
2376 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2377 )
2379 Where above, new :class:`.BindParameter` objects
2380 will be generated with the names ``name`` and ``timestamp``, and
2381 values of ``jack`` and ``datetime.datetime(2012, 10, 8, 15, 12, 5)``,
2382 respectively. The types will be
2383 inferred from the values given, in this case :class:`.String` and
2384 :class:`.DateTime`.
2386 When specific typing behavior is needed, the positional ``*binds``
2387 argument can be used in which to specify :func:`.bindparam` constructs
2388 directly. These constructs must include at least the ``key``
2389 argument, then an optional value and type::
2391 from sqlalchemy import bindparam
2393 stmt = stmt.bindparams(
2394 bindparam("name", value="jack", type_=String),
2395 bindparam("timestamp", type_=DateTime),
2396 )
2398 Above, we specified the type of :class:`.DateTime` for the
2399 ``timestamp`` bind, and the type of :class:`.String` for the ``name``
2400 bind. In the case of ``name`` we also set the default value of
2401 ``"jack"``.
2403 Additional bound parameters can be supplied at statement execution
2404 time, e.g.::
2406 result = connection.execute(
2407 stmt, timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2408 )
2410 The :meth:`_expression.TextClause.bindparams`
2411 method can be called repeatedly,
2412 where it will re-use existing :class:`.BindParameter` objects to add
2413 new information. For example, we can call
2414 :meth:`_expression.TextClause.bindparams`
2415 first with typing information, and a
2416 second time with value information, and it will be combined::
2418 stmt = text(
2419 "SELECT id, name FROM user WHERE name=:name "
2420 "AND timestamp=:timestamp"
2421 )
2422 stmt = stmt.bindparams(
2423 bindparam("name", type_=String), bindparam("timestamp", type_=DateTime)
2424 )
2425 stmt = stmt.bindparams(
2426 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2427 )
2429 The :meth:`_expression.TextClause.bindparams`
2430 method also supports the concept of
2431 **unique** bound parameters. These are parameters that are
2432 "uniquified" on name at statement compilation time, so that multiple
2433 :func:`_expression.text`
2434 constructs may be combined together without the names
2435 conflicting. To use this feature, specify the
2436 :paramref:`.BindParameter.unique` flag on each :func:`.bindparam`
2437 object::
2439 stmt1 = text("select id from table where name=:name").bindparams(
2440 bindparam("name", value="name1", unique=True)
2441 )
2442 stmt2 = text("select id from table where name=:name").bindparams(
2443 bindparam("name", value="name2", unique=True)
2444 )
2446 union = union_all(stmt1.columns(column("id")), stmt2.columns(column("id")))
2448 The above statement will render as:
2450 .. sourcecode:: sql
2452 select id from table where name=:name_1
2453 UNION ALL select id from table where name=:name_2
2455 """ # noqa: E501
2456 self._bindparams = new_params = self._bindparams.copy()
2458 for bind in binds:
2459 try:
2460 # the regex used for text() currently will not match
2461 # a unique/anonymous key in any case, so use the _orig_key
2462 # so that a text() construct can support unique parameters
2463 existing = new_params[bind._orig_key]
2464 except KeyError as err:
2465 raise exc.ArgumentError(
2466 "This text() construct doesn't define a "
2467 "bound parameter named %r" % bind._orig_key
2468 ) from err
2469 else:
2470 new_params[existing._orig_key] = bind
2472 for key, value in names_to_values.items():
2473 try:
2474 existing = new_params[key]
2475 except KeyError as err:
2476 raise exc.ArgumentError(
2477 "This text() construct doesn't define a "
2478 "bound parameter named %r" % key
2479 ) from err
2480 else:
2481 new_params[key] = existing._with_value(value, required=False)
2482 return self
2484 @util.preload_module("sqlalchemy.sql.selectable")
2485 def columns(
2486 self,
2487 *cols: _ColumnExpressionArgument[Any],
2488 **types: _TypeEngineArgument[Any],
2489 ) -> TextualSelect:
2490 r"""Turn this :class:`_expression.TextClause` object into a
2491 :class:`_expression.TextualSelect`
2492 object that serves the same role as a SELECT
2493 statement.
2495 The :class:`_expression.TextualSelect` is part of the
2496 :class:`_expression.SelectBase`
2497 hierarchy and can be embedded into another statement by using the
2498 :meth:`_expression.TextualSelect.subquery` method to produce a
2499 :class:`.Subquery`
2500 object, which can then be SELECTed from.
2502 This function essentially bridges the gap between an entirely
2503 textual SELECT statement and the SQL expression language concept
2504 of a "selectable"::
2506 from sqlalchemy.sql import column, text
2508 stmt = text("SELECT id, name FROM some_table")
2509 stmt = stmt.columns(column("id"), column("name")).subquery("st")
2511 stmt = (
2512 select(mytable)
2513 .select_from(mytable.join(stmt, mytable.c.name == stmt.c.name))
2514 .where(stmt.c.id > 5)
2515 )
2517 Above, we pass a series of :func:`_expression.column` elements to the
2518 :meth:`_expression.TextClause.columns` method positionally. These
2519 :func:`_expression.column`
2520 elements now become first class elements upon the
2521 :attr:`_expression.TextualSelect.selected_columns` column collection,
2522 which then
2523 become part of the :attr:`.Subquery.c` collection after
2524 :meth:`_expression.TextualSelect.subquery` is invoked.
2526 The column expressions we pass to
2527 :meth:`_expression.TextClause.columns` may
2528 also be typed; when we do so, these :class:`.TypeEngine` objects become
2529 the effective return type of the column, so that SQLAlchemy's
2530 result-set-processing systems may be used on the return values.
2531 This is often needed for types such as date or boolean types, as well
2532 as for unicode processing on some dialect configurations::
2534 stmt = text("SELECT id, name, timestamp FROM some_table")
2535 stmt = stmt.columns(
2536 column("id", Integer),
2537 column("name", Unicode),
2538 column("timestamp", DateTime),
2539 )
2541 for id, name, timestamp in connection.execute(stmt):
2542 print(id, name, timestamp)
2544 As a shortcut to the above syntax, keyword arguments referring to
2545 types alone may be used, if only type conversion is needed::
2547 stmt = text("SELECT id, name, timestamp FROM some_table")
2548 stmt = stmt.columns(id=Integer, name=Unicode, timestamp=DateTime)
2550 for id, name, timestamp in connection.execute(stmt):
2551 print(id, name, timestamp)
2553 The positional form of :meth:`_expression.TextClause.columns`
2554 also provides the
2555 unique feature of **positional column targeting**, which is
2556 particularly useful when using the ORM with complex textual queries. If
2557 we specify the columns from our model to
2558 :meth:`_expression.TextClause.columns`,
2559 the result set will match to those columns positionally, meaning the
2560 name or origin of the column in the textual SQL doesn't matter::
2562 stmt = text(
2563 "SELECT users.id, addresses.id, users.id, "
2564 "users.name, addresses.email_address AS email "
2565 "FROM users JOIN addresses ON users.id=addresses.user_id "
2566 "WHERE users.id = 1"
2567 ).columns(
2568 User.id,
2569 Address.id,
2570 Address.user_id,
2571 User.name,
2572 Address.email_address,
2573 )
2575 query = (
2576 session.query(User)
2577 .from_statement(stmt)
2578 .options(contains_eager(User.addresses))
2579 )
2581 The :meth:`_expression.TextClause.columns` method provides a direct
2582 route to calling :meth:`_expression.FromClause.subquery` as well as
2583 :meth:`_expression.SelectBase.cte`
2584 against a textual SELECT statement::
2586 stmt = stmt.columns(id=Integer, name=String).cte("st")
2588 stmt = select(sometable).where(sometable.c.id == stmt.c.id)
2590 :param \*cols: A series of :class:`_expression.ColumnElement` objects,
2591 typically
2592 :class:`_schema.Column` objects from a :class:`_schema.Table`
2593 or ORM level
2594 column-mapped attributes, representing a set of columns that this
2595 textual string will SELECT from.
2597 :param \**types: A mapping of string names to :class:`.TypeEngine`
2598 type objects indicating the datatypes to use for names that are
2599 SELECTed from the textual string. Prefer to use the ``*cols``
2600 argument as it also indicates positional ordering.
2602 """
2603 selectable = util.preloaded.sql_selectable
2605 input_cols: List[NamedColumn[Any]] = [
2606 coercions.expect(roles.LabeledColumnExprRole, col) for col in cols
2607 ]
2609 positional_input_cols = [
2610 (
2611 ColumnClause(col.key, types.pop(col.key))
2612 if col.key in types
2613 else col
2614 )
2615 for col in input_cols
2616 ]
2617 keyed_input_cols: List[NamedColumn[Any]] = [
2618 ColumnClause(key, type_) for key, type_ in types.items()
2619 ]
2621 elem = selectable.TextualSelect.__new__(selectable.TextualSelect)
2622 elem._init(
2623 self,
2624 positional_input_cols + keyed_input_cols,
2625 positional=bool(positional_input_cols) and not keyed_input_cols,
2626 )
2627 return elem
2629 @property
2630 def type(self) -> TypeEngine[Any]:
2631 return type_api.NULLTYPE
2633 @property
2634 def comparator(self):
2635 # TODO: this seems wrong, it seems like we might not
2636 # be using this method.
2637 return self.type.comparator_factory(self) # type: ignore
2639 def self_group(
2640 self, against: Optional[OperatorType] = None
2641 ) -> Union[Self, Grouping[Any]]:
2642 if against is operators.in_op:
2643 return Grouping(self)
2644 else:
2645 return self
2648class Null(SingletonConstant, roles.ConstExprRole[None], ColumnElement[None]):
2649 """Represent the NULL keyword in a SQL statement.
2651 :class:`.Null` is accessed as a constant via the
2652 :func:`.null` function.
2654 """
2656 __visit_name__ = "null"
2658 _traverse_internals: _TraverseInternalsType = []
2659 _singleton: Null
2661 if not TYPE_CHECKING:
2663 @util.memoized_property
2664 def type(self) -> TypeEngine[_T]: # noqa: A001
2665 return type_api.NULLTYPE
2667 @classmethod
2668 def _instance(cls) -> Null:
2669 """Return a constant :class:`.Null` construct."""
2671 return Null._singleton
2674Null._create_singleton()
2677class False_(
2678 SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]
2679):
2680 """Represent the ``false`` keyword, or equivalent, in a SQL statement.
2682 :class:`.False_` is accessed as a constant via the
2683 :func:`.false` function.
2685 """
2687 __visit_name__ = "false"
2688 _traverse_internals: _TraverseInternalsType = []
2689 _singleton: False_
2691 if not TYPE_CHECKING:
2693 @util.memoized_property
2694 def type(self) -> TypeEngine[_T]: # noqa: A001
2695 return type_api.BOOLEANTYPE
2697 def _negate(self) -> True_:
2698 return True_._singleton
2700 @classmethod
2701 def _instance(cls) -> False_:
2702 return False_._singleton
2705False_._create_singleton()
2708class True_(SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]):
2709 """Represent the ``true`` keyword, or equivalent, in a SQL statement.
2711 :class:`.True_` is accessed as a constant via the
2712 :func:`.true` function.
2714 """
2716 __visit_name__ = "true"
2718 _traverse_internals: _TraverseInternalsType = []
2719 _singleton: True_
2721 if not TYPE_CHECKING:
2723 @util.memoized_property
2724 def type(self) -> TypeEngine[_T]: # noqa: A001
2725 return type_api.BOOLEANTYPE
2727 def _negate(self) -> False_:
2728 return False_._singleton
2730 @classmethod
2731 def _ifnone(
2732 cls, other: Optional[ColumnElement[Any]]
2733 ) -> ColumnElement[Any]:
2734 if other is None:
2735 return cls._instance()
2736 else:
2737 return other
2739 @classmethod
2740 def _instance(cls) -> True_:
2741 return True_._singleton
2744True_._create_singleton()
2747class ElementList(DQLDMLClauseElement):
2748 """Describe a list of clauses that will be space separated.
2750 This is a minimal version of :class:`.ClauseList` which is used by
2751 the :class:`.HasSyntaxExtension` class. It does not do any coercions
2752 so should be used internally only.
2754 .. versionadded:: 2.1
2756 """
2758 __visit_name__ = "element_list"
2760 _traverse_internals: _TraverseInternalsType = [
2761 ("clauses", InternalTraversal.dp_clauseelement_tuple),
2762 ]
2764 clauses: typing_Tuple[ClauseElement, ...]
2766 def __init__(self, clauses: Sequence[ClauseElement]):
2767 self.clauses = tuple(clauses)
2770class ClauseList(
2771 roles.InElementRole,
2772 roles.OrderByRole,
2773 roles.ColumnsClauseRole,
2774 roles.DMLColumnRole,
2775 DQLDMLClauseElement,
2776):
2777 """Describe a list of clauses, separated by an operator.
2779 By default, is comma-separated, such as a column listing.
2781 """
2783 __visit_name__ = "clauselist"
2785 # this is used only by the ORM in a legacy use case for
2786 # composite attributes
2787 _is_clause_list = True
2789 _traverse_internals: _TraverseInternalsType = [
2790 ("clauses", InternalTraversal.dp_clauseelement_list),
2791 ("operator", InternalTraversal.dp_operator),
2792 ]
2794 clauses: List[ColumnElement[Any]]
2796 def __init__(
2797 self,
2798 *clauses: _ColumnExpressionArgument[Any],
2799 operator: OperatorType = operators.comma_op,
2800 group: bool = True,
2801 group_contents: bool = True,
2802 _literal_as_text_role: Type[roles.SQLRole] = roles.WhereHavingRole,
2803 ):
2804 self.operator = operator
2805 self.group = group
2806 self.group_contents = group_contents
2807 clauses_iterator: Iterable[_ColumnExpressionArgument[Any]] = clauses
2808 text_converter_role: Type[roles.SQLRole] = _literal_as_text_role
2809 self._text_converter_role = text_converter_role
2811 if self.group_contents:
2812 self.clauses = [
2813 coercions.expect(
2814 text_converter_role, clause, apply_propagate_attrs=self
2815 ).self_group(against=self.operator)
2816 for clause in clauses_iterator
2817 ]
2818 else:
2819 self.clauses = [
2820 coercions.expect(
2821 text_converter_role, clause, apply_propagate_attrs=self
2822 )
2823 for clause in clauses_iterator
2824 ]
2825 self._is_implicitly_boolean = operators.is_boolean(self.operator)
2827 @classmethod
2828 def _construct_raw(
2829 cls,
2830 operator: OperatorType,
2831 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
2832 ) -> ClauseList:
2833 self = cls.__new__(cls)
2834 self.clauses = list(clauses) if clauses else []
2835 self.group = True
2836 self.operator = operator
2837 self.group_contents = True
2838 self._is_implicitly_boolean = False
2839 return self
2841 def __iter__(self) -> Iterator[ColumnElement[Any]]:
2842 return iter(self.clauses)
2844 def __len__(self) -> int:
2845 return len(self.clauses)
2847 @property
2848 def _select_iterable(self) -> _SelectIterable:
2849 return itertools.chain.from_iterable(
2850 [elem._select_iterable for elem in self.clauses]
2851 )
2853 def append(self, clause):
2854 if self.group_contents:
2855 self.clauses.append(
2856 coercions.expect(self._text_converter_role, clause).self_group(
2857 against=self.operator
2858 )
2859 )
2860 else:
2861 self.clauses.append(
2862 coercions.expect(self._text_converter_role, clause)
2863 )
2865 @util.ro_non_memoized_property
2866 def _from_objects(self) -> List[FromClause]:
2867 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
2869 def self_group(
2870 self, against: Optional[OperatorType] = None
2871 ) -> Union[Self, Grouping[Any]]:
2872 if self.group and operators.is_precedent(self.operator, against):
2873 return Grouping(self)
2874 else:
2875 return self
2878class OperatorExpression(ColumnElement[_T]):
2879 """base for expressions that contain an operator and operands
2881 .. versionadded:: 2.0
2883 """
2885 operator: OperatorType
2886 type: TypeEngine[_T]
2888 group: bool = True
2890 @property
2891 def is_comparison(self):
2892 return operators.is_comparison(self.operator)
2894 def self_group(
2895 self, against: Optional[OperatorType] = None
2896 ) -> Union[Self, Grouping[_T]]:
2897 if (
2898 self.group
2899 and operators.is_precedent(self.operator, against)
2900 or (
2901 # a negate against a non-boolean operator
2902 # doesn't make too much sense but we should
2903 # group for that
2904 against is operators.inv
2905 and not operators.is_boolean(self.operator)
2906 )
2907 ):
2908 return Grouping(self)
2909 else:
2910 return self
2912 @property
2913 def _flattened_operator_clauses(
2914 self,
2915 ) -> typing_Tuple[ColumnElement[Any], ...]:
2916 raise NotImplementedError()
2918 @classmethod
2919 def _construct_for_op(
2920 cls,
2921 left: ColumnElement[Any],
2922 right: ColumnElement[Any],
2923 op: OperatorType,
2924 *,
2925 type_: TypeEngine[_T],
2926 negate: Optional[OperatorType] = None,
2927 modifiers: Optional[Mapping[str, Any]] = None,
2928 ) -> OperatorExpression[_T]:
2929 if operators.is_associative(op):
2930 assert (
2931 negate is None
2932 ), f"negate not supported for associative operator {op}"
2934 multi = False
2935 if getattr(
2936 left, "operator", None
2937 ) is op and type_._compare_type_affinity(left.type):
2938 multi = True
2939 left_flattened = left._flattened_operator_clauses
2940 else:
2941 left_flattened = (left,)
2943 if getattr(
2944 right, "operator", None
2945 ) is op and type_._compare_type_affinity(right.type):
2946 multi = True
2947 right_flattened = right._flattened_operator_clauses
2948 else:
2949 right_flattened = (right,)
2951 if multi:
2952 return ExpressionClauseList._construct_for_list(
2953 op,
2954 type_,
2955 *(left_flattened + right_flattened),
2956 )
2958 if right._is_collection_aggregate:
2959 negate = None
2961 return BinaryExpression(
2962 left, right, op, type_=type_, negate=negate, modifiers=modifiers
2963 )
2966class ExpressionClauseList(OperatorExpression[_T]):
2967 """Describe a list of clauses, separated by an operator,
2968 in a column expression context.
2970 :class:`.ExpressionClauseList` differs from :class:`.ClauseList` in that
2971 it represents a column-oriented DQL expression only, not an open ended
2972 list of anything comma separated.
2974 .. versionadded:: 2.0
2976 """
2978 __visit_name__ = "expression_clauselist"
2980 _traverse_internals: _TraverseInternalsType = [
2981 ("clauses", InternalTraversal.dp_clauseelement_tuple),
2982 ("operator", InternalTraversal.dp_operator),
2983 ]
2985 clauses: typing_Tuple[ColumnElement[Any], ...]
2987 group: bool
2989 def __init__(
2990 self,
2991 operator: OperatorType,
2992 *clauses: _ColumnExpressionArgument[Any],
2993 type_: Optional[_TypeEngineArgument[_T]] = None,
2994 ):
2995 self.operator = operator
2997 self.clauses = tuple(
2998 coercions.expect(
2999 roles.ExpressionElementRole, clause, apply_propagate_attrs=self
3000 )
3001 for clause in clauses
3002 )
3003 self._is_implicitly_boolean = operators.is_boolean(self.operator)
3004 self.type = type_api.to_instance(type_) # type: ignore
3006 @property
3007 def _flattened_operator_clauses(
3008 self,
3009 ) -> typing_Tuple[ColumnElement[Any], ...]:
3010 return self.clauses
3012 def __iter__(self) -> Iterator[ColumnElement[Any]]:
3013 return iter(self.clauses)
3015 def __len__(self) -> int:
3016 return len(self.clauses)
3018 @property
3019 def _select_iterable(self) -> _SelectIterable:
3020 return (self,)
3022 @util.ro_non_memoized_property
3023 def _from_objects(self) -> List[FromClause]:
3024 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
3026 def _append_inplace(self, clause: ColumnElement[Any]) -> None:
3027 self.clauses += (clause,)
3029 @classmethod
3030 def _construct_for_list(
3031 cls,
3032 operator: OperatorType,
3033 type_: TypeEngine[_T],
3034 *clauses: ColumnElement[Any],
3035 group: bool = True,
3036 ) -> ExpressionClauseList[_T]:
3037 self = cls.__new__(cls)
3038 self.group = group
3039 if group:
3040 self.clauses = tuple(
3041 c.self_group(against=operator) for c in clauses
3042 )
3043 else:
3044 self.clauses = clauses
3045 self.operator = operator
3046 self.type = type_
3047 for c in clauses:
3048 if c._propagate_attrs:
3049 self._propagate_attrs = c._propagate_attrs
3050 break
3051 return self
3053 def _negate(self) -> Any:
3054 grouped = self.self_group(against=operators.inv)
3055 assert isinstance(grouped, ColumnElement)
3056 return UnaryExpression(grouped, operator=operators.inv)
3059class BooleanClauseList(ExpressionClauseList[bool]):
3060 __visit_name__ = "expression_clauselist"
3061 inherit_cache = True
3063 def __init__(self, *arg, **kw):
3064 raise NotImplementedError(
3065 "BooleanClauseList has a private constructor"
3066 )
3068 @classmethod
3069 def _process_clauses_for_boolean(
3070 cls,
3071 operator: OperatorType,
3072 continue_on: Any,
3073 skip_on: Any,
3074 clauses: Iterable[ColumnElement[Any]],
3075 ) -> typing_Tuple[int, List[ColumnElement[Any]]]:
3076 has_continue_on = None
3078 convert_clauses = []
3080 against = operators._asbool
3081 lcc = 0
3083 for clause in clauses:
3084 if clause is continue_on:
3085 # instance of continue_on, like and_(x, y, True, z), store it
3086 # if we didn't find one already, we will use it if there
3087 # are no other expressions here.
3088 has_continue_on = clause
3089 elif clause is skip_on:
3090 # instance of skip_on, e.g. and_(x, y, False, z), cancels
3091 # the rest out
3092 convert_clauses = [clause]
3093 lcc = 1
3094 break
3095 else:
3096 if not lcc:
3097 lcc = 1
3098 else:
3099 against = operator
3100 # technically this would be len(convert_clauses) + 1
3101 # however this only needs to indicate "greater than one"
3102 lcc = 2
3103 convert_clauses.append(clause)
3105 if not convert_clauses and has_continue_on is not None:
3106 convert_clauses = [has_continue_on]
3107 lcc = 1
3109 return lcc, [c.self_group(against=against) for c in convert_clauses]
3111 @classmethod
3112 def _construct(
3113 cls,
3114 operator: OperatorType,
3115 continue_on: Any,
3116 skip_on: Any,
3117 initial_clause: Any = _NoArg.NO_ARG,
3118 *clauses: Any,
3119 **kw: Any,
3120 ) -> ColumnElement[Any]:
3121 if initial_clause is _NoArg.NO_ARG:
3122 # no elements period. deprecated use case. return an empty
3123 # ClauseList construct that generates nothing unless it has
3124 # elements added to it.
3125 name = operator.__name__
3127 util.warn_deprecated(
3128 f"Invoking {name}() without arguments is deprecated, and "
3129 f"will be disallowed in a future release. For an empty "
3130 f"""{name}() construct, use '{name}({
3131 'true()' if continue_on is True_._singleton else 'false()'
3132 }, *args)' """
3133 f"""or '{name}({
3134 'True' if continue_on is True_._singleton else 'False'
3135 }, *args)'.""",
3136 version="1.4",
3137 )
3138 return cls._construct_raw(operator)
3140 lcc, convert_clauses = cls._process_clauses_for_boolean(
3141 operator,
3142 continue_on,
3143 skip_on,
3144 [
3145 coercions.expect(roles.WhereHavingRole, clause)
3146 for clause in util.coerce_generator_arg(
3147 (initial_clause,) + clauses
3148 )
3149 ],
3150 )
3152 if lcc > 1:
3153 # multiple elements. Return regular BooleanClauseList
3154 # which will link elements against the operator.
3156 flattened_clauses = itertools.chain.from_iterable(
3157 (
3158 (c for c in to_flat._flattened_operator_clauses)
3159 if getattr(to_flat, "operator", None) is operator
3160 else (to_flat,)
3161 )
3162 for to_flat in convert_clauses
3163 )
3165 return cls._construct_raw(operator, flattened_clauses) # type: ignore # noqa: E501
3166 else:
3167 assert lcc
3168 # just one element. return it as a single boolean element,
3169 # not a list and discard the operator.
3170 return convert_clauses[0]
3172 @classmethod
3173 def _construct_for_whereclause(
3174 cls, clauses: Iterable[ColumnElement[Any]]
3175 ) -> Optional[ColumnElement[bool]]:
3176 operator, continue_on, skip_on = (
3177 operators.and_,
3178 True_._singleton,
3179 False_._singleton,
3180 )
3182 lcc, convert_clauses = cls._process_clauses_for_boolean(
3183 operator,
3184 continue_on,
3185 skip_on,
3186 clauses, # these are assumed to be coerced already
3187 )
3189 if lcc > 1:
3190 # multiple elements. Return regular BooleanClauseList
3191 # which will link elements against the operator.
3192 return cls._construct_raw(operator, convert_clauses)
3193 elif lcc == 1:
3194 # just one element. return it as a single boolean element,
3195 # not a list and discard the operator.
3196 return convert_clauses[0]
3197 else:
3198 return None
3200 @classmethod
3201 def _construct_raw(
3202 cls,
3203 operator: OperatorType,
3204 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
3205 ) -> BooleanClauseList:
3206 self = cls.__new__(cls)
3207 self.clauses = tuple(clauses) if clauses else ()
3208 self.group = True
3209 self.operator = operator
3210 self.type = type_api.BOOLEANTYPE
3211 self._is_implicitly_boolean = True
3212 return self
3214 @classmethod
3215 def and_(
3216 cls,
3217 initial_clause: Union[
3218 Literal[True], _ColumnExpressionArgument[bool], _NoArg
3219 ] = _NoArg.NO_ARG,
3220 *clauses: _ColumnExpressionArgument[bool],
3221 ) -> ColumnElement[bool]:
3222 r"""Produce a conjunction of expressions joined by ``AND``.
3224 See :func:`_sql.and_` for full documentation.
3225 """
3226 return cls._construct(
3227 operators.and_,
3228 True_._singleton,
3229 False_._singleton,
3230 initial_clause,
3231 *clauses,
3232 )
3234 @classmethod
3235 def or_(
3236 cls,
3237 initial_clause: Union[
3238 Literal[False], _ColumnExpressionArgument[bool], _NoArg
3239 ] = _NoArg.NO_ARG,
3240 *clauses: _ColumnExpressionArgument[bool],
3241 ) -> ColumnElement[bool]:
3242 """Produce a conjunction of expressions joined by ``OR``.
3244 See :func:`_sql.or_` for full documentation.
3245 """
3246 return cls._construct(
3247 operators.or_,
3248 False_._singleton,
3249 True_._singleton,
3250 initial_clause,
3251 *clauses,
3252 )
3254 @property
3255 def _select_iterable(self) -> _SelectIterable:
3256 return (self,)
3258 def self_group(
3259 self, against: Optional[OperatorType] = None
3260 ) -> Union[Self, Grouping[bool]]:
3261 if not self.clauses:
3262 return self
3263 else:
3264 return super().self_group(against=against)
3267and_ = BooleanClauseList.and_
3268or_ = BooleanClauseList.or_
3271class Tuple(ClauseList, ColumnElement[TupleAny]):
3272 """Represent a SQL tuple."""
3274 __visit_name__ = "tuple"
3276 _traverse_internals: _TraverseInternalsType = (
3277 ClauseList._traverse_internals + []
3278 )
3280 type: TupleType
3282 @util.preload_module("sqlalchemy.sql.sqltypes")
3283 def __init__(
3284 self,
3285 *clauses: _ColumnExpressionArgument[Any],
3286 types: Optional[Sequence[_TypeEngineArgument[Any]]] = None,
3287 ):
3288 sqltypes = util.preloaded.sql_sqltypes
3290 if types is None:
3291 init_clauses: List[ColumnElement[Any]] = [
3292 coercions.expect(roles.ExpressionElementRole, c)
3293 for c in clauses
3294 ]
3295 else:
3296 if len(types) != len(clauses):
3297 raise exc.ArgumentError(
3298 "Wrong number of elements for %d-tuple: %r "
3299 % (len(types), clauses)
3300 )
3301 init_clauses = [
3302 coercions.expect(
3303 roles.ExpressionElementRole,
3304 c,
3305 type_=typ if not typ._isnull else None,
3306 )
3307 for typ, c in zip(types, clauses)
3308 ]
3310 self.type = sqltypes.TupleType(*[arg.type for arg in init_clauses])
3311 super().__init__(*init_clauses)
3313 @property
3314 def _select_iterable(self) -> _SelectIterable:
3315 return (self,)
3317 def _bind_param(self, operator, obj, type_=None, expanding=False):
3318 if expanding:
3319 return BindParameter(
3320 None,
3321 value=obj,
3322 _compared_to_operator=operator,
3323 unique=True,
3324 expanding=True,
3325 type_=type_,
3326 _compared_to_type=self.type,
3327 )
3328 else:
3329 return Tuple(
3330 *[
3331 BindParameter(
3332 None,
3333 o,
3334 _compared_to_operator=operator,
3335 _compared_to_type=compared_to_type,
3336 unique=True,
3337 type_=type_,
3338 )
3339 for o, compared_to_type in zip(obj, self.type.types)
3340 ]
3341 )
3343 def self_group(self, against: Optional[OperatorType] = None) -> Self:
3344 # Tuple is parenthesized by definition.
3345 return self
3348class Case(ColumnElement[_T]):
3349 """Represent a ``CASE`` expression.
3351 :class:`.Case` is produced using the :func:`.case` factory function,
3352 as in::
3354 from sqlalchemy import case
3356 stmt = select(users_table).where(
3357 case(
3358 (users_table.c.name == "wendy", "W"),
3359 (users_table.c.name == "jack", "J"),
3360 else_="E",
3361 )
3362 )
3364 Details on :class:`.Case` usage is at :func:`.case`.
3366 .. seealso::
3368 :func:`.case`
3370 """
3372 __visit_name__ = "case"
3374 _traverse_internals: _TraverseInternalsType = [
3375 ("value", InternalTraversal.dp_clauseelement),
3376 ("whens", InternalTraversal.dp_clauseelement_tuples),
3377 ("else_", InternalTraversal.dp_clauseelement),
3378 ]
3380 # for case(), the type is derived from the whens. so for the moment
3381 # users would have to cast() the case to get a specific type
3383 whens: List[typing_Tuple[ColumnElement[bool], ColumnElement[_T]]]
3384 else_: Optional[ColumnElement[_T]]
3385 value: Optional[ColumnElement[Any]]
3387 def __init__(
3388 self,
3389 *whens: Union[
3390 typing_Tuple[_ColumnExpressionArgument[bool], Any],
3391 Mapping[Any, Any],
3392 ],
3393 value: Optional[Any] = None,
3394 else_: Optional[Any] = None,
3395 ):
3396 new_whens: Iterable[Any] = coercions._expression_collection_was_a_list(
3397 "whens", "case", whens
3398 )
3399 try:
3400 new_whens = util.dictlike_iteritems(new_whens)
3401 except TypeError:
3402 pass
3404 self.whens = [
3405 (
3406 coercions.expect(
3407 roles.ExpressionElementRole,
3408 c,
3409 apply_propagate_attrs=self,
3410 ).self_group(),
3411 coercions.expect(roles.ExpressionElementRole, r),
3412 )
3413 for (c, r) in new_whens
3414 ]
3416 if value is None:
3417 self.value = None
3418 else:
3419 self.value = coercions.expect(roles.ExpressionElementRole, value)
3421 if else_ is not None:
3422 self.else_ = coercions.expect(roles.ExpressionElementRole, else_)
3423 else:
3424 self.else_ = None
3426 type_ = next(
3427 (
3428 then.type
3429 # Iterate `whens` in reverse to match previous behaviour
3430 # where type of final element took priority
3431 for *_, then in reversed(self.whens)
3432 if not then.type._isnull
3433 ),
3434 self.else_.type if self.else_ is not None else type_api.NULLTYPE,
3435 )
3436 self.type = cast(_T, type_)
3438 @util.ro_non_memoized_property
3439 def _from_objects(self) -> List[FromClause]:
3440 return list(
3441 itertools.chain(*[x._from_objects for x in self.get_children()])
3442 )
3445class Cast(WrapsColumnExpression[_T]):
3446 """Represent a ``CAST`` expression.
3448 :class:`.Cast` is produced using the :func:`.cast` factory function,
3449 as in::
3451 from sqlalchemy import cast, Numeric
3453 stmt = select(cast(product_table.c.unit_price, Numeric(10, 4)))
3455 Details on :class:`.Cast` usage is at :func:`.cast`.
3457 .. seealso::
3459 :ref:`tutorial_casts`
3461 :func:`.cast`
3463 :func:`.try_cast`
3465 :func:`.type_coerce` - an alternative to CAST that coerces the type
3466 on the Python side only, which is often sufficient to generate the
3467 correct SQL and data coercion.
3469 """
3471 __visit_name__ = "cast"
3473 _traverse_internals: _TraverseInternalsType = [
3474 ("clause", InternalTraversal.dp_clauseelement),
3475 ("type", InternalTraversal.dp_type),
3476 ]
3478 clause: ColumnElement[Any]
3479 type: TypeEngine[_T]
3480 typeclause: TypeClause
3482 def __init__(
3483 self,
3484 expression: _ColumnExpressionArgument[Any],
3485 type_: _TypeEngineArgument[_T],
3486 ):
3487 self.type = type_api.to_instance(type_)
3488 self.clause = coercions.expect(
3489 roles.ExpressionElementRole,
3490 expression,
3491 type_=self.type,
3492 apply_propagate_attrs=self,
3493 )
3494 self.typeclause = TypeClause(self.type)
3496 @util.ro_non_memoized_property
3497 def _from_objects(self) -> List[FromClause]:
3498 return self.clause._from_objects
3500 @property
3501 def wrapped_column_expression(self):
3502 return self.clause
3505class TryCast(Cast[_T]):
3506 """Represent a TRY_CAST expression.
3508 Details on :class:`.TryCast` usage is at :func:`.try_cast`.
3510 .. seealso::
3512 :func:`.try_cast`
3514 :ref:`tutorial_casts`
3515 """
3517 __visit_name__ = "try_cast"
3518 inherit_cache = True
3521class TypeCoerce(WrapsColumnExpression[_T]):
3522 """Represent a Python-side type-coercion wrapper.
3524 :class:`.TypeCoerce` supplies the :func:`_expression.type_coerce`
3525 function; see that function for usage details.
3527 .. seealso::
3529 :func:`_expression.type_coerce`
3531 :func:`.cast`
3533 """
3535 __visit_name__ = "type_coerce"
3537 _traverse_internals: _TraverseInternalsType = [
3538 ("clause", InternalTraversal.dp_clauseelement),
3539 ("type", InternalTraversal.dp_type),
3540 ]
3542 clause: ColumnElement[Any]
3543 type: TypeEngine[_T]
3545 def __init__(
3546 self,
3547 expression: _ColumnExpressionArgument[Any],
3548 type_: _TypeEngineArgument[_T],
3549 ):
3550 self.type = type_api.to_instance(type_)
3551 self.clause = coercions.expect(
3552 roles.ExpressionElementRole,
3553 expression,
3554 type_=self.type,
3555 apply_propagate_attrs=self,
3556 )
3558 @util.ro_non_memoized_property
3559 def _from_objects(self) -> List[FromClause]:
3560 return self.clause._from_objects
3562 @HasMemoized.memoized_attribute
3563 def typed_expression(self):
3564 if isinstance(self.clause, BindParameter):
3565 bp = self.clause._clone()
3566 bp.type = self.type
3567 return bp
3568 else:
3569 return self.clause
3571 @property
3572 def wrapped_column_expression(self):
3573 return self.clause
3575 def self_group(
3576 self, against: Optional[OperatorType] = None
3577 ) -> TypeCoerce[_T]:
3578 grouped = self.clause.self_group(against=against)
3579 if grouped is not self.clause:
3580 return TypeCoerce(grouped, self.type)
3581 else:
3582 return self
3585class Extract(ColumnElement[int]):
3586 """Represent a SQL EXTRACT clause, ``extract(field FROM expr)``."""
3588 __visit_name__ = "extract"
3590 _traverse_internals: _TraverseInternalsType = [
3591 ("expr", InternalTraversal.dp_clauseelement),
3592 ("field", InternalTraversal.dp_string),
3593 ]
3595 expr: ColumnElement[Any]
3596 field: str
3598 def __init__(self, field: str, expr: _ColumnExpressionArgument[Any]):
3599 self.type = type_api.INTEGERTYPE
3600 self.field = field
3601 self.expr = coercions.expect(roles.ExpressionElementRole, expr)
3603 @util.ro_non_memoized_property
3604 def _from_objects(self) -> List[FromClause]:
3605 return self.expr._from_objects
3608class _label_reference(ColumnElement[_T]):
3609 """Wrap a column expression as it appears in a 'reference' context.
3611 This expression is any that includes an _order_by_label_element,
3612 which is a Label, or a DESC / ASC construct wrapping a Label.
3614 The production of _label_reference() should occur when an expression
3615 is added to this context; this includes the ORDER BY or GROUP BY of a
3616 SELECT statement, as well as a few other places, such as the ORDER BY
3617 within an OVER clause.
3619 """
3621 __visit_name__ = "label_reference"
3623 _traverse_internals: _TraverseInternalsType = [
3624 ("element", InternalTraversal.dp_clauseelement)
3625 ]
3627 element: ColumnElement[_T]
3629 def __init__(self, element: ColumnElement[_T]):
3630 self.element = element
3631 self._propagate_attrs = element._propagate_attrs
3633 @util.ro_non_memoized_property
3634 def _from_objects(self) -> List[FromClause]:
3635 return []
3638class _textual_label_reference(ColumnElement[Any]):
3639 __visit_name__ = "textual_label_reference"
3641 _traverse_internals: _TraverseInternalsType = [
3642 ("element", InternalTraversal.dp_string)
3643 ]
3645 def __init__(self, element: str):
3646 self.element = element
3648 @util.memoized_property
3649 def _text_clause(self) -> TextClause:
3650 return TextClause(self.element)
3653class UnaryExpression(ColumnElement[_T]):
3654 """Define a 'unary' expression.
3656 A unary expression has a single column expression
3657 and an operator. The operator can be placed on the left
3658 (where it is called the 'operator') or right (where it is called the
3659 'modifier') of the column expression.
3661 :class:`.UnaryExpression` is the basis for several unary operators
3662 including those used by :func:`.desc`, :func:`.asc`, :func:`.distinct`,
3663 :func:`.nulls_first` and :func:`.nulls_last`.
3665 """
3667 __visit_name__ = "unary"
3669 _traverse_internals: _TraverseInternalsType = [
3670 ("element", InternalTraversal.dp_clauseelement),
3671 ("operator", InternalTraversal.dp_operator),
3672 ("modifier", InternalTraversal.dp_operator),
3673 ]
3675 element: ColumnElement[Any]
3676 operator: Optional[OperatorType]
3677 modifier: Optional[OperatorType]
3679 def __init__(
3680 self,
3681 element: ColumnElement[Any],
3682 *,
3683 operator: Optional[OperatorType] = None,
3684 modifier: Optional[OperatorType] = None,
3685 type_: Optional[_TypeEngineArgument[_T]] = None,
3686 wraps_column_expression: bool = False, # legacy, not used as of 2.0.42
3687 ):
3688 self.operator = operator
3689 self.modifier = modifier
3690 self._propagate_attrs = element._propagate_attrs
3691 self.element = element.self_group(
3692 against=self.operator or self.modifier
3693 )
3695 # if type is None, we get NULLTYPE, which is our _T. But I don't
3696 # know how to get the overloads to express that correctly
3697 self.type = type_api.to_instance(type_) # type: ignore
3699 def _wraps_unnamed_column(self):
3700 ungrouped = self.element._ungroup()
3701 return (
3702 not isinstance(ungrouped, NamedColumn)
3703 or ungrouped._non_anon_label is None
3704 )
3706 @classmethod
3707 def _create_nulls_first(
3708 cls,
3709 column: _ColumnExpressionArgument[_T],
3710 ) -> UnaryExpression[_T]:
3711 return UnaryExpression(
3712 coercions.expect(roles.ByOfRole, column),
3713 modifier=operators.nulls_first_op,
3714 )
3716 @classmethod
3717 def _create_nulls_last(
3718 cls,
3719 column: _ColumnExpressionArgument[_T],
3720 ) -> UnaryExpression[_T]:
3721 return UnaryExpression(
3722 coercions.expect(roles.ByOfRole, column),
3723 modifier=operators.nulls_last_op,
3724 )
3726 @classmethod
3727 def _create_desc(
3728 cls, column: _ColumnExpressionOrStrLabelArgument[_T]
3729 ) -> UnaryExpression[_T]:
3730 return UnaryExpression(
3731 coercions.expect(roles.ByOfRole, column),
3732 modifier=operators.desc_op,
3733 )
3735 @classmethod
3736 def _create_asc(
3737 cls,
3738 column: _ColumnExpressionOrStrLabelArgument[_T],
3739 ) -> UnaryExpression[_T]:
3740 return UnaryExpression(
3741 coercions.expect(roles.ByOfRole, column),
3742 modifier=operators.asc_op,
3743 )
3745 @classmethod
3746 def _create_distinct(
3747 cls,
3748 expr: _ColumnExpressionArgument[_T],
3749 ) -> UnaryExpression[_T]:
3750 col_expr: ColumnElement[_T] = coercions.expect(
3751 roles.ExpressionElementRole, expr
3752 )
3753 return UnaryExpression(
3754 col_expr,
3755 operator=operators.distinct_op,
3756 type_=col_expr.type,
3757 )
3759 @classmethod
3760 def _create_bitwise_not(
3761 cls,
3762 expr: _ColumnExpressionArgument[_T],
3763 ) -> UnaryExpression[_T]:
3764 col_expr: ColumnElement[_T] = coercions.expect(
3765 roles.ExpressionElementRole, expr
3766 )
3767 return UnaryExpression(
3768 col_expr,
3769 operator=operators.bitwise_not_op,
3770 type_=col_expr.type,
3771 )
3773 @property
3774 def _order_by_label_element(self) -> Optional[Label[Any]]:
3775 if operators.is_order_by_modifier(self.modifier):
3776 return self.element._order_by_label_element
3777 else:
3778 return None
3780 @util.ro_non_memoized_property
3781 def _from_objects(self) -> List[FromClause]:
3782 return self.element._from_objects
3784 def _negate(self) -> ColumnElement[Any]:
3785 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
3786 return UnaryExpression(
3787 self.self_group(against=operators.inv),
3788 operator=operators.inv,
3789 type_=type_api.BOOLEANTYPE,
3790 )
3791 else:
3792 return ColumnElement._negate(self)
3794 def self_group(
3795 self, against: Optional[OperatorType] = None
3796 ) -> Union[Self, Grouping[_T]]:
3797 if self.operator and operators.is_precedent(self.operator, against):
3798 return Grouping(self)
3799 else:
3800 return self
3803class CollectionAggregate(UnaryExpression[_T]):
3804 """Forms the basis for right-hand collection operator modifiers
3805 ANY and ALL.
3807 The ANY and ALL keywords are available in different ways on different
3808 backends. On PostgreSQL, they only work for an ARRAY type. On
3809 MySQL, they only work for subqueries.
3811 """
3813 inherit_cache = True
3814 _is_collection_aggregate = True
3816 @classmethod
3817 def _create_any(
3818 cls, expr: _ColumnExpressionArgument[_T]
3819 ) -> CollectionAggregate[bool]:
3820 col_expr: ColumnElement[_T] = coercions.expect(
3821 roles.ExpressionElementRole,
3822 expr,
3823 )
3824 col_expr = col_expr.self_group()
3825 return CollectionAggregate(
3826 col_expr,
3827 operator=operators.any_op,
3828 type_=type_api.BOOLEANTYPE,
3829 )
3831 @classmethod
3832 def _create_all(
3833 cls, expr: _ColumnExpressionArgument[_T]
3834 ) -> CollectionAggregate[bool]:
3835 col_expr: ColumnElement[_T] = coercions.expect(
3836 roles.ExpressionElementRole,
3837 expr,
3838 )
3839 col_expr = col_expr.self_group()
3840 return CollectionAggregate(
3841 col_expr,
3842 operator=operators.all_op,
3843 type_=type_api.BOOLEANTYPE,
3844 )
3846 # operate and reverse_operate are hardwired to
3847 # dispatch onto the type comparator directly, so that we can
3848 # ensure "reversed" behavior.
3849 def operate(
3850 self, op: OperatorType, *other: Any, **kwargs: Any
3851 ) -> ColumnElement[_T]:
3852 if not operators.is_comparison(op):
3853 raise exc.ArgumentError(
3854 "Only comparison operators may be used with ANY/ALL"
3855 )
3856 kwargs["reverse"] = True
3857 return self.comparator.operate(operators.mirror(op), *other, **kwargs)
3859 def reverse_operate(
3860 self, op: OperatorType, other: Any, **kwargs: Any
3861 ) -> ColumnElement[_T]:
3862 # comparison operators should never call reverse_operate
3863 assert not operators.is_comparison(op)
3864 raise exc.ArgumentError(
3865 "Only comparison operators may be used with ANY/ALL"
3866 )
3869class AsBoolean(WrapsColumnExpression[bool], UnaryExpression[bool]):
3870 inherit_cache = True
3872 def __init__(self, element, operator, negate):
3873 self.element = element
3874 self.type = type_api.BOOLEANTYPE
3875 self.operator = operator
3876 self.negate = negate
3877 self.modifier = None
3878 self._is_implicitly_boolean = element._is_implicitly_boolean
3880 @property
3881 def wrapped_column_expression(self):
3882 return self.element
3884 def self_group(self, against: Optional[OperatorType] = None) -> Self:
3885 return self
3887 def _negate(self):
3888 if isinstance(self.element, (True_, False_)):
3889 return self.element._negate()
3890 else:
3891 return AsBoolean(self.element, self.negate, self.operator)
3894class BinaryExpression(OperatorExpression[_T]):
3895 """Represent an expression that is ``LEFT <operator> RIGHT``.
3897 A :class:`.BinaryExpression` is generated automatically
3898 whenever two column expressions are used in a Python binary expression:
3900 .. sourcecode:: pycon+sql
3902 >>> from sqlalchemy.sql import column
3903 >>> column("a") + column("b")
3904 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
3905 >>> print(column("a") + column("b"))
3906 {printsql}a + b
3908 """
3910 __visit_name__ = "binary"
3912 _traverse_internals: _TraverseInternalsType = [
3913 ("left", InternalTraversal.dp_clauseelement),
3914 ("right", InternalTraversal.dp_clauseelement),
3915 ("operator", InternalTraversal.dp_operator),
3916 ("negate", InternalTraversal.dp_operator),
3917 ("modifiers", InternalTraversal.dp_plain_dict),
3918 (
3919 "type",
3920 InternalTraversal.dp_type,
3921 ),
3922 ]
3924 _cache_key_traversal = [
3925 ("left", InternalTraversal.dp_clauseelement),
3926 ("right", InternalTraversal.dp_clauseelement),
3927 ("operator", InternalTraversal.dp_operator),
3928 ("modifiers", InternalTraversal.dp_plain_dict),
3929 # "type" affects JSON CAST operators, so while redundant in most cases,
3930 # is needed for that one
3931 (
3932 "type",
3933 InternalTraversal.dp_type,
3934 ),
3935 ]
3937 _is_implicitly_boolean = True
3938 """Indicates that any database will know this is a boolean expression
3939 even if the database does not have an explicit boolean datatype.
3941 """
3943 left: ColumnElement[Any]
3944 right: ColumnElement[Any]
3945 modifiers: Mapping[str, Any]
3947 def __init__(
3948 self,
3949 left: ColumnElement[Any],
3950 right: ColumnElement[Any],
3951 operator: OperatorType,
3952 type_: Optional[_TypeEngineArgument[_T]] = None,
3953 negate: Optional[OperatorType] = None,
3954 modifiers: Optional[Mapping[str, Any]] = None,
3955 ):
3956 # allow compatibility with libraries that
3957 # refer to BinaryExpression directly and pass strings
3958 if isinstance(operator, str):
3959 operator = operators.custom_op(operator)
3960 self._orig = (left.__hash__(), right.__hash__())
3961 self._propagate_attrs = left._propagate_attrs or right._propagate_attrs
3962 self.left = left.self_group(against=operator)
3963 self.right = right.self_group(against=operator)
3964 self.operator = operator
3966 # if type is None, we get NULLTYPE, which is our _T. But I don't
3967 # know how to get the overloads to express that correctly
3968 self.type = type_api.to_instance(type_) # type: ignore
3970 self.negate = negate
3971 self._is_implicitly_boolean = operators.is_boolean(operator)
3973 if modifiers is None:
3974 self.modifiers = {}
3975 else:
3976 self.modifiers = modifiers
3978 @property
3979 def _flattened_operator_clauses(
3980 self,
3981 ) -> typing_Tuple[ColumnElement[Any], ...]:
3982 return (self.left, self.right)
3984 def __bool__(self):
3985 """Implement Python-side "bool" for BinaryExpression as a
3986 simple "identity" check for the left and right attributes,
3987 if the operator is "eq" or "ne". Otherwise the expression
3988 continues to not support "bool" like all other column expressions.
3990 The rationale here is so that ColumnElement objects can be hashable.
3991 What? Well, suppose you do this::
3993 c1, c2 = column("x"), column("y")
3994 s1 = set([c1, c2])
3996 We do that **a lot**, columns inside of sets is an extremely basic
3997 thing all over the ORM for example.
3999 So what happens if we do this? ::
4001 c1 in s1
4003 Hashing means it will normally use ``__hash__()`` of the object,
4004 but in case of hash collision, it's going to also do ``c1 == c1``
4005 and/or ``c1 == c2`` inside. Those operations need to return a
4006 True/False value. But because we override ``==`` and ``!=``, they're
4007 going to get a BinaryExpression. Hence we implement ``__bool__`` here
4008 so that these comparisons behave in this particular context mostly
4009 like regular object comparisons. Thankfully Python is OK with
4010 that! Otherwise we'd have to use special set classes for columns
4011 (which we used to do, decades ago).
4013 """
4014 if self.operator in (operators.eq, operators.ne):
4015 # this is using the eq/ne operator given int hash values,
4016 # rather than Operator, so that "bool" can be based on
4017 # identity
4018 return self.operator(*self._orig) # type: ignore
4019 else:
4020 raise TypeError("Boolean value of this clause is not defined")
4022 if typing.TYPE_CHECKING:
4024 def __invert__(
4025 self: BinaryExpression[_T],
4026 ) -> BinaryExpression[_T]: ...
4028 @util.ro_non_memoized_property
4029 def _from_objects(self) -> List[FromClause]:
4030 return self.left._from_objects + self.right._from_objects
4032 def _negate(self):
4033 if self.negate is not None:
4034 return BinaryExpression(
4035 self.left,
4036 self.right._negate_in_binary(self.negate, self.operator),
4037 self.negate,
4038 negate=self.operator,
4039 type_=self.type,
4040 modifiers=self.modifiers,
4041 )
4042 else:
4043 return self.self_group()._negate()
4046class Slice(ColumnElement[Any]):
4047 """Represent SQL for a Python array-slice object.
4049 This is not a specific SQL construct at this level, but
4050 may be interpreted by specific dialects, e.g. PostgreSQL.
4052 """
4054 __visit_name__ = "slice"
4056 _traverse_internals: _TraverseInternalsType = [
4057 ("start", InternalTraversal.dp_clauseelement),
4058 ("stop", InternalTraversal.dp_clauseelement),
4059 ("step", InternalTraversal.dp_clauseelement),
4060 ]
4062 def __init__(self, start, stop, step, _name=None):
4063 self.start = coercions.expect(
4064 roles.ExpressionElementRole,
4065 start,
4066 name=_name,
4067 type_=type_api.INTEGERTYPE,
4068 )
4069 self.stop = coercions.expect(
4070 roles.ExpressionElementRole,
4071 stop,
4072 name=_name,
4073 type_=type_api.INTEGERTYPE,
4074 )
4075 self.step = coercions.expect(
4076 roles.ExpressionElementRole,
4077 step,
4078 name=_name,
4079 type_=type_api.INTEGERTYPE,
4080 )
4081 self.type = type_api.NULLTYPE
4083 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4084 assert against is operator.getitem
4085 return self
4088class IndexExpression(BinaryExpression[Any]):
4089 """Represent the class of expressions that are like an "index"
4090 operation."""
4092 inherit_cache = True
4095class GroupedElement(DQLDMLClauseElement):
4096 """Represent any parenthesized expression"""
4098 __visit_name__ = "grouping"
4100 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4101 return self
4103 def _ungroup(self) -> ClauseElement:
4104 raise NotImplementedError()
4107class Grouping(GroupedElement, ColumnElement[_T]):
4108 """Represent a grouping within a column expression"""
4110 _traverse_internals: _TraverseInternalsType = [
4111 ("element", InternalTraversal.dp_clauseelement),
4112 ("type", InternalTraversal.dp_type),
4113 ]
4115 _cache_key_traversal = [
4116 ("element", InternalTraversal.dp_clauseelement),
4117 ]
4119 element: Union[TextClause, ClauseList, ColumnElement[_T]]
4121 def __init__(
4122 self, element: Union[TextClause, ClauseList, ColumnElement[_T]]
4123 ):
4124 self.element = element
4126 # nulltype assignment issue
4127 self.type = getattr(element, "type", type_api.NULLTYPE) # type: ignore
4128 self._propagate_attrs = element._propagate_attrs
4130 def _with_binary_element_type(self, type_):
4131 return self.__class__(self.element._with_binary_element_type(type_))
4133 def _ungroup(self) -> ColumnElement[_T]:
4134 assert isinstance(self.element, ColumnElement)
4135 return self.element._ungroup()
4137 @util.memoized_property
4138 def _is_implicitly_boolean(self):
4139 return self.element._is_implicitly_boolean
4141 @util.non_memoized_property
4142 def _tq_label(self) -> Optional[str]:
4143 return (
4144 getattr(self.element, "_tq_label", None) or self._anon_name_label
4145 )
4147 @util.non_memoized_property
4148 def _proxies(self) -> List[ColumnElement[Any]]:
4149 if isinstance(self.element, ColumnElement):
4150 return [self.element]
4151 else:
4152 return []
4154 @util.ro_non_memoized_property
4155 def _from_objects(self) -> List[FromClause]:
4156 return self.element._from_objects
4158 def __getattr__(self, attr):
4159 return getattr(self.element, attr)
4161 def __getstate__(self):
4162 return {"element": self.element, "type": self.type}
4164 def __setstate__(self, state):
4165 self.element = state["element"]
4166 self.type = state["type"]
4168 if TYPE_CHECKING:
4170 def self_group(
4171 self, against: Optional[OperatorType] = None
4172 ) -> Self: ...
4175class _OverrideBinds(Grouping[_T]):
4176 """used by cache_key->_apply_params_to_element to allow compilation /
4177 execution of a SQL element that's been cached, using an alternate set of
4178 bound parameter values.
4180 This is used by the ORM to swap new parameter values into expressions
4181 that are embedded into loader options like with_expression(),
4182 selectinload(). Previously, this task was accomplished using the
4183 .params() method which would perform a deep-copy instead. This deep
4184 copy proved to be too expensive for more complex expressions.
4186 See #11085
4188 """
4190 __visit_name__ = "override_binds"
4192 def __init__(
4193 self,
4194 element: ColumnElement[_T],
4195 bindparams: Sequence[BindParameter[Any]],
4196 replaces_params: Sequence[BindParameter[Any]],
4197 ):
4198 self.element = element
4199 self.translate = {
4200 k.key: v.value for k, v in zip(replaces_params, bindparams)
4201 }
4203 def _gen_cache_key(
4204 self, anon_map: anon_map, bindparams: List[BindParameter[Any]]
4205 ) -> Optional[typing_Tuple[Any, ...]]:
4206 """generate a cache key for the given element, substituting its bind
4207 values for the translation values present."""
4209 existing_bps: List[BindParameter[Any]] = []
4210 ck = self.element._gen_cache_key(anon_map, existing_bps)
4212 bindparams.extend(
4213 (
4214 bp._with_value(
4215 self.translate[bp.key], maintain_key=True, required=False
4216 )
4217 if bp.key in self.translate
4218 else bp
4219 )
4220 for bp in existing_bps
4221 )
4223 return ck
4226class Over(ColumnElement[_T]):
4227 """Represent an OVER clause.
4229 This is a special operator against a so-called
4230 "window" function, as well as any aggregate function,
4231 which produces results relative to the result set
4232 itself. Most modern SQL backends now support window functions.
4234 """
4236 __visit_name__ = "over"
4238 _traverse_internals: _TraverseInternalsType = [
4239 ("element", InternalTraversal.dp_clauseelement),
4240 ("order_by", InternalTraversal.dp_clauseelement),
4241 ("partition_by", InternalTraversal.dp_clauseelement),
4242 ("range_", InternalTraversal.dp_clauseelement),
4243 ("rows", InternalTraversal.dp_clauseelement),
4244 ("groups", InternalTraversal.dp_clauseelement),
4245 ]
4247 order_by: Optional[ClauseList] = None
4248 partition_by: Optional[ClauseList] = None
4250 element: ColumnElement[_T]
4251 """The underlying expression object to which this :class:`.Over`
4252 object refers."""
4254 range_: Optional[_FrameClause]
4255 rows: Optional[_FrameClause]
4256 groups: Optional[_FrameClause]
4258 def __init__(
4259 self,
4260 element: ColumnElement[_T],
4261 partition_by: Optional[_ByArgument] = None,
4262 order_by: Optional[_ByArgument] = None,
4263 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4264 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4265 groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4266 ):
4267 self.element = element
4268 if order_by is not None:
4269 self.order_by = ClauseList(
4270 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4271 )
4272 if partition_by is not None:
4273 self.partition_by = ClauseList(
4274 *util.to_list(partition_by),
4275 _literal_as_text_role=roles.ByOfRole,
4276 )
4278 if sum(bool(item) for item in (range_, rows, groups)) > 1:
4279 raise exc.ArgumentError(
4280 "only one of 'rows', 'range_', or 'groups' may be provided"
4281 )
4282 else:
4283 self.range_ = _FrameClause(range_) if range_ else None
4284 self.rows = _FrameClause(rows) if rows else None
4285 self.groups = _FrameClause(groups) if groups else None
4287 if not TYPE_CHECKING:
4289 @util.memoized_property
4290 def type(self) -> TypeEngine[_T]: # noqa: A001
4291 return self.element.type
4293 @util.ro_non_memoized_property
4294 def _from_objects(self) -> List[FromClause]:
4295 return list(
4296 itertools.chain(
4297 *[
4298 c._from_objects
4299 for c in (self.element, self.partition_by, self.order_by)
4300 if c is not None
4301 ]
4302 )
4303 )
4306class _FrameClauseType(Enum):
4307 RANGE_UNBOUNDED = 0
4308 RANGE_CURRENT = 1
4309 RANGE_PRECEDING = 2
4310 RANGE_FOLLOWING = 3
4313class _FrameClause(ClauseElement):
4314 """indicate the 'rows' or 'range' field of a window function, e.g. using
4315 :class:`.Over`.
4317 .. versionadded:: 2.1
4319 """
4321 __visit_name__ = "frame_clause"
4323 _traverse_internals: _TraverseInternalsType = [
4324 ("lower_integer_bind", InternalTraversal.dp_clauseelement),
4325 ("upper_integer_bind", InternalTraversal.dp_clauseelement),
4326 ("lower_type", InternalTraversal.dp_plain_obj),
4327 ("upper_type", InternalTraversal.dp_plain_obj),
4328 ]
4330 def __init__(
4331 self,
4332 range_: typing_Tuple[Optional[int], Optional[int]],
4333 ):
4334 try:
4335 r0, r1 = range_
4336 except (ValueError, TypeError) as ve:
4337 raise exc.ArgumentError("2-tuple expected for range/rows") from ve
4339 if r0 is None:
4340 self.lower_type = _FrameClauseType.RANGE_UNBOUNDED
4341 self.lower_integer_bind = None
4342 else:
4343 try:
4344 lower_integer = int(r0)
4345 except ValueError as err:
4346 raise exc.ArgumentError(
4347 "Integer or None expected for range value"
4348 ) from err
4349 else:
4350 if lower_integer == 0:
4351 self.lower_type = _FrameClauseType.RANGE_CURRENT
4352 self.lower_integer_bind = None
4353 elif lower_integer < 0:
4354 self.lower_type = _FrameClauseType.RANGE_PRECEDING
4355 self.lower_integer_bind = literal(
4356 abs(lower_integer), type_api.INTEGERTYPE
4357 )
4358 else:
4359 self.lower_type = _FrameClauseType.RANGE_FOLLOWING
4360 self.lower_integer_bind = literal(
4361 lower_integer, type_api.INTEGERTYPE
4362 )
4364 if r1 is None:
4365 self.upper_type = _FrameClauseType.RANGE_UNBOUNDED
4366 self.upper_integer_bind = None
4367 else:
4368 try:
4369 upper_integer = int(r1)
4370 except ValueError as err:
4371 raise exc.ArgumentError(
4372 "Integer or None expected for range value"
4373 ) from err
4374 else:
4375 if upper_integer == 0:
4376 self.upper_type = _FrameClauseType.RANGE_CURRENT
4377 self.upper_integer_bind = None
4378 elif upper_integer < 0:
4379 self.upper_type = _FrameClauseType.RANGE_PRECEDING
4380 self.upper_integer_bind = literal(
4381 abs(upper_integer), type_api.INTEGERTYPE
4382 )
4383 else:
4384 self.upper_type = _FrameClauseType.RANGE_FOLLOWING
4385 self.upper_integer_bind = literal(
4386 upper_integer, type_api.INTEGERTYPE
4387 )
4390class WithinGroup(ColumnElement[_T]):
4391 """Represent a WITHIN GROUP (ORDER BY) clause.
4393 This is a special operator against so-called
4394 "ordered set aggregate" and "hypothetical
4395 set aggregate" functions, including ``percentile_cont()``,
4396 ``rank()``, ``dense_rank()``, etc.
4398 It's supported only by certain database backends, such as PostgreSQL,
4399 Oracle Database and MS SQL Server.
4401 The :class:`.WithinGroup` construct extracts its type from the
4402 method :meth:`.FunctionElement.within_group_type`. If this returns
4403 ``None``, the function's ``.type`` is used.
4405 """
4407 __visit_name__ = "withingroup"
4409 _traverse_internals: _TraverseInternalsType = [
4410 ("element", InternalTraversal.dp_clauseelement),
4411 ("order_by", InternalTraversal.dp_clauseelement),
4412 ]
4414 order_by: Optional[ClauseList] = None
4416 def __init__(
4417 self,
4418 element: Union[FunctionElement[_T], FunctionFilter[_T]],
4419 *order_by: _ColumnExpressionArgument[Any],
4420 ):
4421 self.element = element
4422 if order_by is not None:
4423 self.order_by = ClauseList(
4424 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4425 )
4427 def __reduce__(self):
4428 return self.__class__, (self.element,) + (
4429 tuple(self.order_by) if self.order_by is not None else ()
4430 )
4432 def over(
4433 self,
4434 *,
4435 partition_by: Optional[_ByArgument] = None,
4436 order_by: Optional[_ByArgument] = None,
4437 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4438 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4439 groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4440 ) -> Over[_T]:
4441 """Produce an OVER clause against this :class:`.WithinGroup`
4442 construct.
4444 This function has the same signature as that of
4445 :meth:`.FunctionElement.over`.
4447 """
4448 return Over(
4449 self,
4450 partition_by=partition_by,
4451 order_by=order_by,
4452 range_=range_,
4453 rows=rows,
4454 groups=groups,
4455 )
4457 @overload
4458 def filter(self) -> Self: ...
4460 @overload
4461 def filter(
4462 self,
4463 __criterion0: _ColumnExpressionArgument[bool],
4464 *criterion: _ColumnExpressionArgument[bool],
4465 ) -> FunctionFilter[_T]: ...
4467 def filter(
4468 self, *criterion: _ColumnExpressionArgument[bool]
4469 ) -> Union[Self, FunctionFilter[_T]]:
4470 """Produce a FILTER clause against this function."""
4471 if not criterion:
4472 return self
4473 return FunctionFilter(self, *criterion)
4475 if not TYPE_CHECKING:
4477 @util.memoized_property
4478 def type(self) -> TypeEngine[_T]: # noqa: A001
4479 wgt = self.element.within_group_type(self)
4480 if wgt is not None:
4481 return wgt
4482 else:
4483 return self.element.type
4485 @util.ro_non_memoized_property
4486 def _from_objects(self) -> List[FromClause]:
4487 return list(
4488 itertools.chain(
4489 *[
4490 c._from_objects
4491 for c in (self.element, self.order_by)
4492 if c is not None
4493 ]
4494 )
4495 )
4498class FunctionFilter(Generative, ColumnElement[_T]):
4499 """Represent a function FILTER clause.
4501 This is a special operator against aggregate and window functions,
4502 which controls which rows are passed to it.
4503 It's supported only by certain database backends.
4505 Invocation of :class:`.FunctionFilter` is via
4506 :meth:`.FunctionElement.filter`::
4508 func.count(1).filter(True)
4510 .. seealso::
4512 :meth:`.FunctionElement.filter`
4514 """
4516 __visit_name__ = "funcfilter"
4518 _traverse_internals: _TraverseInternalsType = [
4519 ("func", InternalTraversal.dp_clauseelement),
4520 ("criterion", InternalTraversal.dp_clauseelement),
4521 ]
4523 criterion: Optional[ColumnElement[bool]] = None
4525 def __init__(
4526 self,
4527 func: Union[FunctionElement[_T], WithinGroup[_T]],
4528 *criterion: _ColumnExpressionArgument[bool],
4529 ):
4530 self.func = func
4531 self.filter.non_generative(self, *criterion) # type: ignore
4533 @_generative
4534 def filter(self, *criterion: _ColumnExpressionArgument[bool]) -> Self:
4535 """Produce an additional FILTER against the function.
4537 This method adds additional criteria to the initial criteria
4538 set up by :meth:`.FunctionElement.filter`.
4540 Multiple criteria are joined together at SQL render time
4541 via ``AND``.
4544 """
4546 for crit in list(criterion):
4547 crit = coercions.expect(roles.WhereHavingRole, crit)
4549 if self.criterion is not None:
4550 self.criterion = self.criterion & crit
4551 else:
4552 self.criterion = crit
4554 return self
4556 def over(
4557 self,
4558 partition_by: Optional[
4559 Union[
4560 Iterable[_ColumnExpressionArgument[Any]],
4561 _ColumnExpressionArgument[Any],
4562 ]
4563 ] = None,
4564 order_by: Optional[
4565 Union[
4566 Iterable[_ColumnExpressionArgument[Any]],
4567 _ColumnExpressionArgument[Any],
4568 ]
4569 ] = None,
4570 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4571 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4572 groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4573 ) -> Over[_T]:
4574 """Produce an OVER clause against this filtered function.
4576 Used against aggregate or so-called "window" functions,
4577 for database backends that support window functions.
4579 The expression::
4581 func.rank().filter(MyClass.y > 5).over(order_by="x")
4583 is shorthand for::
4585 from sqlalchemy import over, funcfilter
4587 over(funcfilter(func.rank(), MyClass.y > 5), order_by="x")
4589 See :func:`_expression.over` for a full description.
4591 """
4592 return Over(
4593 self,
4594 partition_by=partition_by,
4595 order_by=order_by,
4596 range_=range_,
4597 rows=rows,
4598 groups=groups,
4599 )
4601 def within_group(
4602 self, *order_by: _ColumnExpressionArgument[Any]
4603 ) -> WithinGroup[_T]:
4604 """Produce a WITHIN GROUP (ORDER BY expr) clause against
4605 this function.
4606 """
4607 return WithinGroup(self, *order_by)
4609 def within_group_type(
4610 self, within_group: WithinGroup[_T]
4611 ) -> Optional[TypeEngine[_T]]:
4612 return None
4614 def self_group(
4615 self, against: Optional[OperatorType] = None
4616 ) -> Union[Self, Grouping[_T]]:
4617 if operators.is_precedent(operators.filter_op, against):
4618 return Grouping(self)
4619 else:
4620 return self
4622 if not TYPE_CHECKING:
4624 @util.memoized_property
4625 def type(self) -> TypeEngine[_T]: # noqa: A001
4626 return self.func.type
4628 @util.ro_non_memoized_property
4629 def _from_objects(self) -> List[FromClause]:
4630 return list(
4631 itertools.chain(
4632 *[
4633 c._from_objects
4634 for c in (self.func, self.criterion)
4635 if c is not None
4636 ]
4637 )
4638 )
4641class NamedColumn(KeyedColumnElement[_T]):
4642 is_literal = False
4643 table: Optional[FromClause] = None
4644 name: str
4645 key: str
4647 def _compare_name_for_result(self, other):
4648 return (hasattr(other, "name") and self.name == other.name) or (
4649 hasattr(other, "_label") and self._label == other._label
4650 )
4652 @util.ro_memoized_property
4653 def description(self) -> str:
4654 return self.name
4656 @HasMemoized.memoized_attribute
4657 def _tq_key_label(self) -> Optional[str]:
4658 """table qualified label based on column key.
4660 for table-bound columns this is <tablename>_<column key/proxy key>;
4662 all other expressions it resolves to key/proxy key.
4664 """
4665 proxy_key = self._proxy_key
4666 if proxy_key and proxy_key != self.name:
4667 return self._gen_tq_label(proxy_key)
4668 else:
4669 return self._tq_label
4671 @HasMemoized.memoized_attribute
4672 def _tq_label(self) -> Optional[str]:
4673 """table qualified label based on column name.
4675 for table-bound columns this is <tablename>_<columnname>; all other
4676 expressions it resolves to .name.
4678 """
4679 return self._gen_tq_label(self.name)
4681 @HasMemoized.memoized_attribute
4682 def _render_label_in_columns_clause(self):
4683 return True
4685 @HasMemoized.memoized_attribute
4686 def _non_anon_label(self):
4687 return self.name
4689 def _gen_tq_label(
4690 self, name: str, dedupe_on_key: bool = True
4691 ) -> Optional[str]:
4692 return name
4694 def _bind_param(
4695 self,
4696 operator: OperatorType,
4697 obj: Any,
4698 type_: Optional[TypeEngine[_T]] = None,
4699 expanding: bool = False,
4700 ) -> BindParameter[_T]:
4701 return BindParameter(
4702 self.key,
4703 obj,
4704 _compared_to_operator=operator,
4705 _compared_to_type=self.type,
4706 type_=type_,
4707 unique=True,
4708 expanding=expanding,
4709 )
4711 def _make_proxy(
4712 self,
4713 selectable: FromClause,
4714 *,
4715 primary_key: ColumnSet,
4716 foreign_keys: Set[KeyedColumnElement[Any]],
4717 name: Optional[str] = None,
4718 key: Optional[str] = None,
4719 name_is_truncatable: bool = False,
4720 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
4721 disallow_is_literal: bool = False,
4722 **kw: Any,
4723 ) -> typing_Tuple[str, ColumnClause[_T]]:
4724 c = ColumnClause(
4725 (
4726 coercions.expect(roles.TruncatedLabelRole, name or self.name)
4727 if name_is_truncatable
4728 else (name or self.name)
4729 ),
4730 type_=self.type,
4731 _selectable=selectable,
4732 is_literal=False,
4733 )
4735 c._propagate_attrs = selectable._propagate_attrs
4736 if name is None:
4737 c.key = self.key
4738 if compound_select_cols:
4739 c._proxies = list(compound_select_cols)
4740 else:
4741 c._proxies = [self]
4743 if selectable._is_clone_of is not None:
4744 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
4745 return c.key, c
4748_PS = ParamSpec("_PS")
4751class Label(roles.LabeledColumnExprRole[_T], NamedColumn[_T]):
4752 """Represents a column label (AS).
4754 Represent a label, as typically applied to any column-level
4755 element using the ``AS`` sql keyword.
4757 """
4759 __visit_name__ = "label"
4761 _traverse_internals: _TraverseInternalsType = [
4762 ("name", InternalTraversal.dp_anon_name),
4763 ("type", InternalTraversal.dp_type),
4764 ("_element", InternalTraversal.dp_clauseelement),
4765 ]
4767 _cache_key_traversal = [
4768 ("name", InternalTraversal.dp_anon_name),
4769 ("_element", InternalTraversal.dp_clauseelement),
4770 ]
4772 _element: ColumnElement[_T]
4773 name: str
4775 def __init__(
4776 self,
4777 name: Optional[str],
4778 element: _ColumnExpressionArgument[_T],
4779 type_: Optional[_TypeEngineArgument[_T]] = None,
4780 ):
4781 orig_element = element
4782 element = coercions.expect(
4783 roles.ExpressionElementRole,
4784 element,
4785 apply_propagate_attrs=self,
4786 )
4787 while isinstance(element, Label):
4788 # TODO: this is only covered in test_text.py, but nothing
4789 # fails if it's removed. determine rationale
4790 element = element.element
4792 if name:
4793 self.name = name
4794 else:
4795 self.name = _anonymous_label.safe_construct(
4796 id(self), getattr(element, "name", "anon")
4797 )
4798 if isinstance(orig_element, Label):
4799 # TODO: no coverage for this block, again would be in
4800 # test_text.py where the resolve_label concept is important
4801 self._resolve_label = orig_element._label
4803 self.key = self._tq_label = self._tq_key_label = self.name
4804 self._element = element
4806 self.type = (
4807 type_api.to_instance(type_)
4808 if type_ is not None
4809 else self._element.type
4810 )
4812 self._proxies = [element]
4814 def __reduce__(self):
4815 return self.__class__, (self.name, self._element, self.type)
4817 @HasMemoized.memoized_attribute
4818 def _render_label_in_columns_clause(self):
4819 return True
4821 def _bind_param(self, operator, obj, type_=None, expanding=False):
4822 return BindParameter(
4823 None,
4824 obj,
4825 _compared_to_operator=operator,
4826 type_=type_,
4827 _compared_to_type=self.type,
4828 unique=True,
4829 expanding=expanding,
4830 )
4832 @util.memoized_property
4833 def _is_implicitly_boolean(self):
4834 return self.element._is_implicitly_boolean
4836 @HasMemoized.memoized_attribute
4837 def _allow_label_resolve(self):
4838 return self.element._allow_label_resolve
4840 @property
4841 def _order_by_label_element(self):
4842 return self
4844 def as_reference(self) -> _label_reference[_T]:
4845 """refer to this labeled expression in a clause such as GROUP BY,
4846 ORDER BY etc. as the label name itself, without expanding
4847 into the full expression.
4849 .. versionadded:: 2.1
4851 """
4852 return _label_reference(self)
4854 @HasMemoized.memoized_attribute
4855 def element(self) -> ColumnElement[_T]:
4856 return self._element.self_group(against=operators.as_)
4858 def self_group(self, against: Optional[OperatorType] = None) -> Label[_T]:
4859 return self._apply_to_inner(self._element.self_group, against=against)
4861 def _negate(self):
4862 return self._apply_to_inner(self._element._negate)
4864 def _apply_to_inner(
4865 self,
4866 fn: Callable[_PS, ColumnElement[_T]],
4867 *arg: _PS.args,
4868 **kw: _PS.kwargs,
4869 ) -> Label[_T]:
4870 sub_element = fn(*arg, **kw)
4871 if sub_element is not self._element:
4872 return Label(self.name, sub_element, type_=self.type)
4873 else:
4874 return self
4876 @property
4877 def primary_key(self): # type: ignore[override]
4878 return self.element.primary_key
4880 @property
4881 def foreign_keys(self): # type: ignore[override]
4882 return self.element.foreign_keys
4884 def _copy_internals(
4885 self,
4886 *,
4887 clone: _CloneCallableType = _clone,
4888 anonymize_labels: bool = False,
4889 **kw: Any,
4890 ) -> None:
4891 self._reset_memoizations()
4892 self._element = clone(self._element, **kw)
4893 if anonymize_labels:
4894 self.name = _anonymous_label.safe_construct(
4895 id(self), getattr(self.element, "name", "anon")
4896 )
4897 self.key = self._tq_label = self._tq_key_label = self.name
4899 @util.ro_non_memoized_property
4900 def _from_objects(self) -> List[FromClause]:
4901 return self.element._from_objects
4903 def _make_proxy(
4904 self,
4905 selectable: FromClause,
4906 *,
4907 primary_key: ColumnSet,
4908 foreign_keys: Set[KeyedColumnElement[Any]],
4909 name: Optional[str] = None,
4910 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
4911 **kw: Any,
4912 ) -> typing_Tuple[str, ColumnClause[_T]]:
4913 name = self.name if not name else name
4915 key, e = self.element._make_proxy(
4916 selectable,
4917 name=name,
4918 disallow_is_literal=True,
4919 name_is_truncatable=isinstance(name, _truncated_label),
4920 compound_select_cols=compound_select_cols,
4921 primary_key=primary_key,
4922 foreign_keys=foreign_keys,
4923 )
4925 # there was a note here to remove this assertion, which was here
4926 # to determine if we later could support a use case where
4927 # the key and name of a label are separate. But I don't know what
4928 # that case was. For now, this is an unexpected case that occurs
4929 # when a label name conflicts with other columns and select()
4930 # is attempting to disambiguate an explicit label, which is not what
4931 # the user would want. See issue #6090.
4932 if key != self.name and not isinstance(self.name, _anonymous_label):
4933 raise exc.InvalidRequestError(
4934 "Label name %s is being renamed to an anonymous label due "
4935 "to disambiguation "
4936 "which is not supported right now. Please use unique names "
4937 "for explicit labels." % (self.name)
4938 )
4940 e._propagate_attrs = selectable._propagate_attrs
4941 e._proxies.append(self)
4942 if self.type is not None:
4943 e.type = self.type
4945 return self.key, e
4948class ColumnClause(
4949 roles.DDLReferredColumnRole,
4950 roles.LabeledColumnExprRole[_T],
4951 roles.StrAsPlainColumnRole,
4952 Immutable,
4953 NamedColumn[_T],
4954):
4955 """Represents a column expression from any textual string.
4957 The :class:`.ColumnClause`, a lightweight analogue to the
4958 :class:`_schema.Column` class, is typically invoked using the
4959 :func:`_expression.column` function, as in::
4961 from sqlalchemy import column
4963 id, name = column("id"), column("name")
4964 stmt = select(id, name).select_from("user")
4966 The above statement would produce SQL like:
4968 .. sourcecode:: sql
4970 SELECT id, name FROM user
4972 :class:`.ColumnClause` is the immediate superclass of the schema-specific
4973 :class:`_schema.Column` object. While the :class:`_schema.Column`
4974 class has all the
4975 same capabilities as :class:`.ColumnClause`, the :class:`.ColumnClause`
4976 class is usable by itself in those cases where behavioral requirements
4977 are limited to simple SQL expression generation. The object has none of
4978 the associations with schema-level metadata or with execution-time
4979 behavior that :class:`_schema.Column` does,
4980 so in that sense is a "lightweight"
4981 version of :class:`_schema.Column`.
4983 Full details on :class:`.ColumnClause` usage is at
4984 :func:`_expression.column`.
4986 .. seealso::
4988 :func:`_expression.column`
4990 :class:`_schema.Column`
4992 """
4994 table: Optional[FromClause]
4995 is_literal: bool
4997 __visit_name__ = "column"
4999 _traverse_internals: _TraverseInternalsType = [
5000 ("name", InternalTraversal.dp_anon_name),
5001 ("type", InternalTraversal.dp_type),
5002 ("table", InternalTraversal.dp_clauseelement),
5003 ("is_literal", InternalTraversal.dp_boolean),
5004 ]
5006 onupdate: Optional[DefaultGenerator] = None
5007 default: Optional[DefaultGenerator] = None
5008 server_default: Optional[FetchedValue] = None
5009 server_onupdate: Optional[FetchedValue] = None
5011 _is_multiparam_column = False
5013 @property
5014 def _is_star(self): # type: ignore[override]
5015 return self.is_literal and self.name == "*"
5017 def __init__(
5018 self,
5019 text: str,
5020 type_: Optional[_TypeEngineArgument[_T]] = None,
5021 is_literal: bool = False,
5022 _selectable: Optional[FromClause] = None,
5023 ):
5024 self.key = self.name = text
5025 self.table = _selectable
5027 # if type is None, we get NULLTYPE, which is our _T. But I don't
5028 # know how to get the overloads to express that correctly
5029 self.type = type_api.to_instance(type_) # type: ignore
5031 self.is_literal = is_literal
5033 def get_children(self, *, column_tables=False, **kw):
5034 # override base get_children() to not return the Table
5035 # or selectable that is parent to this column. Traversals
5036 # expect the columns of tables and subqueries to be leaf nodes.
5037 return []
5039 @property
5040 def entity_namespace(self):
5041 if self.table is not None:
5042 return self.table.entity_namespace
5043 else:
5044 return super().entity_namespace
5046 def _clone(self, detect_subquery_cols=False, **kw):
5047 if (
5048 detect_subquery_cols
5049 and self.table is not None
5050 and self.table._is_subquery
5051 ):
5052 clone = kw.pop("clone")
5053 table = clone(self.table, **kw)
5054 new = table.c.corresponding_column(self)
5055 return new
5057 return super()._clone(**kw)
5059 @HasMemoized_ro_memoized_attribute
5060 def _from_objects(self) -> List[FromClause]:
5061 t = self.table
5062 if t is not None:
5063 return [t]
5064 else:
5065 return []
5067 @HasMemoized.memoized_attribute
5068 def _render_label_in_columns_clause(self):
5069 return self.table is not None
5071 @property
5072 def _ddl_label(self):
5073 return self._gen_tq_label(self.name, dedupe_on_key=False)
5075 def _compare_name_for_result(self, other):
5076 if (
5077 self.is_literal
5078 or self.table is None
5079 or self.table._is_textual
5080 or not hasattr(other, "proxy_set")
5081 or (
5082 isinstance(other, ColumnClause)
5083 and (
5084 other.is_literal
5085 or other.table is None
5086 or other.table._is_textual
5087 )
5088 )
5089 ):
5090 return (hasattr(other, "name") and self.name == other.name) or (
5091 hasattr(other, "_tq_label")
5092 and self._tq_label == other._tq_label
5093 )
5094 else:
5095 return other.proxy_set.intersection(self.proxy_set)
5097 def _gen_tq_label(
5098 self, name: str, dedupe_on_key: bool = True
5099 ) -> Optional[str]:
5100 """generate table-qualified label
5102 for a table-bound column this is <tablename>_<columnname>.
5104 used primarily for LABEL_STYLE_TABLENAME_PLUS_COL
5105 as well as the .columns collection on a Join object.
5107 """
5108 label: str
5109 t = self.table
5110 if self.is_literal:
5111 return None
5112 elif t is not None and is_named_from_clause(t):
5113 if has_schema_attr(t) and t.schema:
5114 label = (
5115 t.schema.replace(".", "_") + "_" + t.name + ("_" + name)
5116 )
5117 else:
5118 assert not TYPE_CHECKING or isinstance(t, NamedFromClause)
5119 label = t.name + ("_" + name)
5121 # propagate name quoting rules for labels.
5122 if is_quoted_name(name) and name.quote is not None:
5123 if is_quoted_name(label):
5124 label.quote = name.quote
5125 else:
5126 label = quoted_name(label, name.quote)
5127 elif is_quoted_name(t.name) and t.name.quote is not None:
5128 # can't get this situation to occur, so let's
5129 # assert false on it for now
5130 assert not isinstance(label, quoted_name)
5131 label = quoted_name(label, t.name.quote)
5133 if dedupe_on_key:
5134 # ensure the label name doesn't conflict with that of an
5135 # existing column. note that this implies that any Column
5136 # must **not** set up its _label before its parent table has
5137 # all of its other Column objects set up. There are several
5138 # tables in the test suite which will fail otherwise; example:
5139 # table "owner" has columns "name" and "owner_name". Therefore
5140 # column owner.name cannot use the label "owner_name", it has
5141 # to be "owner_name_1".
5142 if label in t.c:
5143 _label = label
5144 counter = 1
5145 while _label in t.c:
5146 _label = label + f"_{counter}"
5147 counter += 1
5148 label = _label
5150 return coercions.expect(roles.TruncatedLabelRole, label)
5152 else:
5153 return name
5155 def _make_proxy(
5156 self,
5157 selectable: FromClause,
5158 *,
5159 primary_key: ColumnSet,
5160 foreign_keys: Set[KeyedColumnElement[Any]],
5161 name: Optional[str] = None,
5162 key: Optional[str] = None,
5163 name_is_truncatable: bool = False,
5164 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5165 disallow_is_literal: bool = False,
5166 **kw: Any,
5167 ) -> typing_Tuple[str, ColumnClause[_T]]:
5168 # the "is_literal" flag normally should never be propagated; a proxied
5169 # column is always a SQL identifier and never the actual expression
5170 # being evaluated. however, there is a case where the "is_literal" flag
5171 # might be used to allow the given identifier to have a fixed quoting
5172 # pattern already, so maintain the flag for the proxy unless a
5173 # :class:`.Label` object is creating the proxy. See [ticket:4730].
5174 is_literal = (
5175 not disallow_is_literal
5176 and self.is_literal
5177 and (
5178 # note this does not accommodate for quoted_name differences
5179 # right now
5180 name is None
5181 or name == self.name
5182 )
5183 )
5184 c = self._constructor(
5185 (
5186 coercions.expect(roles.TruncatedLabelRole, name or self.name)
5187 if name_is_truncatable
5188 else (name or self.name)
5189 ),
5190 type_=self.type,
5191 _selectable=selectable,
5192 is_literal=is_literal,
5193 )
5194 c._propagate_attrs = selectable._propagate_attrs
5195 if name is None:
5196 c.key = self.key
5197 if compound_select_cols:
5198 c._proxies = list(compound_select_cols)
5199 else:
5200 c._proxies = [self]
5202 if selectable._is_clone_of is not None:
5203 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
5204 return c.key, c
5207class TableValuedColumn(NamedColumn[_T]):
5208 __visit_name__ = "table_valued_column"
5210 _traverse_internals: _TraverseInternalsType = [
5211 ("name", InternalTraversal.dp_anon_name),
5212 ("type", InternalTraversal.dp_type),
5213 ("scalar_alias", InternalTraversal.dp_clauseelement),
5214 ]
5216 def __init__(self, scalar_alias: NamedFromClause, type_: TypeEngine[_T]):
5217 self.scalar_alias = scalar_alias
5218 self.key = self.name = scalar_alias.name
5219 self.type = type_
5221 def _copy_internals(
5222 self, clone: _CloneCallableType = _clone, **kw: Any
5223 ) -> None:
5224 self.scalar_alias = clone(self.scalar_alias, **kw)
5225 self.key = self.name = self.scalar_alias.name
5227 @util.ro_non_memoized_property
5228 def _from_objects(self) -> List[FromClause]:
5229 return [self.scalar_alias]
5232class CollationClause(ColumnElement[str]):
5233 __visit_name__ = "collation"
5235 _traverse_internals: _TraverseInternalsType = [
5236 ("collation", InternalTraversal.dp_string)
5237 ]
5239 @classmethod
5240 @util.preload_module("sqlalchemy.sql.sqltypes")
5241 def _create_collation_expression(
5242 cls, expression: _ColumnExpressionArgument[str], collation: str
5243 ) -> BinaryExpression[str]:
5245 sqltypes = util.preloaded.sql_sqltypes
5247 expr = coercions.expect(roles.ExpressionElementRole[str], expression)
5249 if expr.type._type_affinity is sqltypes.String:
5250 collate_type = expr.type._with_collation(collation)
5251 else:
5252 collate_type = expr.type
5254 return BinaryExpression(
5255 expr,
5256 CollationClause(collation),
5257 operators.collate,
5258 type_=collate_type,
5259 )
5261 def __init__(self, collation):
5262 self.collation = collation
5265class _IdentifiedClause(Executable, ClauseElement):
5266 __visit_name__ = "identified"
5268 def __init__(self, ident):
5269 self.ident = ident
5272class SavepointClause(_IdentifiedClause):
5273 __visit_name__ = "savepoint"
5274 inherit_cache = False
5277class RollbackToSavepointClause(_IdentifiedClause):
5278 __visit_name__ = "rollback_to_savepoint"
5279 inherit_cache = False
5282class ReleaseSavepointClause(_IdentifiedClause):
5283 __visit_name__ = "release_savepoint"
5284 inherit_cache = False
5287class quoted_name(util.MemoizedSlots, str):
5288 """Represent a SQL identifier combined with quoting preferences.
5290 :class:`.quoted_name` is a Python unicode/str subclass which
5291 represents a particular identifier name along with a
5292 ``quote`` flag. This ``quote`` flag, when set to
5293 ``True`` or ``False``, overrides automatic quoting behavior
5294 for this identifier in order to either unconditionally quote
5295 or to not quote the name. If left at its default of ``None``,
5296 quoting behavior is applied to the identifier on a per-backend basis
5297 based on an examination of the token itself.
5299 A :class:`.quoted_name` object with ``quote=True`` is also
5300 prevented from being modified in the case of a so-called
5301 "name normalize" option. Certain database backends, such as
5302 Oracle Database, Firebird, and DB2 "normalize" case-insensitive names
5303 as uppercase. The SQLAlchemy dialects for these backends
5304 convert from SQLAlchemy's lower-case-means-insensitive convention
5305 to the upper-case-means-insensitive conventions of those backends.
5306 The ``quote=True`` flag here will prevent this conversion from occurring
5307 to support an identifier that's quoted as all lower case against
5308 such a backend.
5310 The :class:`.quoted_name` object is normally created automatically
5311 when specifying the name for key schema constructs such as
5312 :class:`_schema.Table`, :class:`_schema.Column`, and others.
5313 The class can also be
5314 passed explicitly as the name to any function that receives a name which
5315 can be quoted. Such as to use the :meth:`_engine.Engine.has_table`
5316 method with
5317 an unconditionally quoted name::
5319 from sqlalchemy import create_engine
5320 from sqlalchemy import inspect
5321 from sqlalchemy.sql import quoted_name
5323 engine = create_engine("oracle+oracledb://some_dsn")
5324 print(inspect(engine).has_table(quoted_name("some_table", True)))
5326 The above logic will run the "has table" logic against the Oracle Database
5327 backend, passing the name exactly as ``"some_table"`` without converting to
5328 upper case.
5330 """
5332 __slots__ = "quote", "lower", "upper"
5334 quote: Optional[bool]
5336 @overload
5337 @classmethod
5338 def construct(cls, value: str, quote: Optional[bool]) -> quoted_name: ...
5340 @overload
5341 @classmethod
5342 def construct(cls, value: None, quote: Optional[bool]) -> None: ...
5344 @classmethod
5345 def construct(
5346 cls, value: Optional[str], quote: Optional[bool]
5347 ) -> Optional[quoted_name]:
5348 if value is None:
5349 return None
5350 else:
5351 return quoted_name(value, quote)
5353 def __new__(cls, value: str, quote: Optional[bool]) -> quoted_name:
5354 assert (
5355 value is not None
5356 ), "use quoted_name.construct() for None passthrough"
5357 if isinstance(value, cls) and (quote is None or value.quote == quote):
5358 return value
5359 self = super().__new__(cls, value)
5361 self.quote = quote
5362 return self
5364 def __reduce__(self):
5365 return quoted_name, (str(self), self.quote)
5367 def _memoized_method_lower(self):
5368 if self.quote:
5369 return self
5370 else:
5371 return str(self).lower()
5373 def _memoized_method_upper(self):
5374 if self.quote:
5375 return self
5376 else:
5377 return str(self).upper()
5380def _find_columns(clause: ClauseElement) -> Set[ColumnClause[Any]]:
5381 """locate Column objects within the given expression."""
5383 cols: Set[ColumnClause[Any]] = set()
5384 traverse(clause, {}, {"column": cols.add})
5385 return cols
5388def _type_from_args(args: Sequence[ColumnElement[_T]]) -> TypeEngine[_T]:
5389 for a in args:
5390 if not a.type._isnull:
5391 return a.type
5392 else:
5393 return type_api.NULLTYPE # type: ignore
5396def _corresponding_column_or_error(fromclause, column, require_embedded=False):
5397 c = fromclause.corresponding_column(
5398 column, require_embedded=require_embedded
5399 )
5400 if c is None:
5401 raise exc.InvalidRequestError(
5402 "Given column '%s', attached to table '%s', "
5403 "failed to locate a corresponding column from table '%s'"
5404 % (column, getattr(column, "table", None), fromclause.description)
5405 )
5406 return c
5409class _memoized_property_but_not_nulltype(
5410 util.memoized_property["TypeEngine[_T]"]
5411):
5412 """memoized property, but dont memoize NullType"""
5414 def __get__(self, obj, cls):
5415 if obj is None:
5416 return self
5417 result = self.fget(obj)
5418 if not result._isnull:
5419 obj.__dict__[self.__name__] = result
5420 return result
5423class AnnotatedColumnElement(Annotated):
5424 _Annotated__element: ColumnElement[Any]
5426 def __init__(self, element, values):
5427 Annotated.__init__(self, element, values)
5428 for attr in (
5429 "comparator",
5430 "_proxy_key",
5431 "_tq_key_label",
5432 "_tq_label",
5433 "_non_anon_label",
5434 "type",
5435 ):
5436 self.__dict__.pop(attr, None)
5437 for attr in ("name", "key", "table"):
5438 if self.__dict__.get(attr, False) is None:
5439 self.__dict__.pop(attr)
5441 def _with_annotations(self, values):
5442 clone = super()._with_annotations(values)
5443 for attr in (
5444 "comparator",
5445 "_proxy_key",
5446 "_tq_key_label",
5447 "_tq_label",
5448 "_non_anon_label",
5449 ):
5450 clone.__dict__.pop(attr, None)
5451 return clone
5453 @util.memoized_property
5454 def name(self):
5455 """pull 'name' from parent, if not present"""
5456 return self._Annotated__element.name
5458 @_memoized_property_but_not_nulltype
5459 def type(self):
5460 """pull 'type' from parent and don't cache if null.
5462 type is routinely changed on existing columns within the
5463 mapped_column() initialization process, and "type" is also consulted
5464 during the creation of SQL expressions. Therefore it can change after
5465 it was already retrieved. At the same time we don't want annotated
5466 objects having overhead when expressions are produced, so continue
5467 to memoize, but only when we have a non-null type.
5469 """
5470 return self._Annotated__element.type
5472 @util.memoized_property
5473 def table(self):
5474 """pull 'table' from parent, if not present"""
5475 return self._Annotated__element.table
5477 @util.memoized_property
5478 def key(self):
5479 """pull 'key' from parent, if not present"""
5480 return self._Annotated__element.key
5482 @util.memoized_property
5483 def info(self) -> _InfoType:
5484 if TYPE_CHECKING:
5485 assert isinstance(self._Annotated__element, Column)
5486 return self._Annotated__element.info
5488 @util.memoized_property
5489 def _anon_name_label(self) -> str:
5490 return self._Annotated__element._anon_name_label
5493class _truncated_label(quoted_name):
5494 """A unicode subclass used to identify symbolic "
5495 "names that may require truncation."""
5497 __slots__ = ()
5499 def __new__(cls, value: str, quote: Optional[bool] = None) -> Any:
5500 quote = getattr(value, "quote", quote)
5501 # return super(_truncated_label, cls).__new__(cls, value, quote, True)
5502 return super().__new__(cls, value, quote)
5504 def __reduce__(self) -> Any:
5505 return self.__class__, (str(self), self.quote)
5507 def apply_map(self, map_: Mapping[str, Any]) -> str:
5508 return self
5511class conv(_truncated_label):
5512 """Mark a string indicating that a name has already been converted
5513 by a naming convention.
5515 This is a string subclass that indicates a name that should not be
5516 subject to any further naming conventions.
5518 E.g. when we create a :class:`.Constraint` using a naming convention
5519 as follows::
5521 m = MetaData(
5522 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5523 )
5524 t = Table(
5525 "t", m, Column("x", Integer), CheckConstraint("x > 5", name="x5")
5526 )
5528 The name of the above constraint will be rendered as ``"ck_t_x5"``.
5529 That is, the existing name ``x5`` is used in the naming convention as the
5530 ``constraint_name`` token.
5532 In some situations, such as in migration scripts, we may be rendering
5533 the above :class:`.CheckConstraint` with a name that's already been
5534 converted. In order to make sure the name isn't double-modified, the
5535 new name is applied using the :func:`_schema.conv` marker. We can
5536 use this explicitly as follows::
5539 m = MetaData(
5540 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5541 )
5542 t = Table(
5543 "t",
5544 m,
5545 Column("x", Integer),
5546 CheckConstraint("x > 5", name=conv("ck_t_x5")),
5547 )
5549 Where above, the :func:`_schema.conv` marker indicates that the constraint
5550 name here is final, and the name will render as ``"ck_t_x5"`` and not
5551 ``"ck_t_ck_t_x5"``
5553 .. seealso::
5555 :ref:`constraint_naming_conventions`
5557 """
5559 __slots__ = ()
5562# for backwards compatibility in case
5563# someone is re-implementing the
5564# _truncated_identifier() sequence in a custom
5565# compiler
5566_generated_label = _truncated_label
5567_anonymous_label_escape = re.compile(r"[%\(\) \$]+")
5570class _anonymous_label(_truncated_label):
5571 """A unicode subclass used to identify anonymously
5572 generated names."""
5574 __slots__ = ()
5576 @classmethod
5577 def safe_construct_with_key(
5578 cls, seed: int, body: str, sanitize_key: bool = False
5579 ) -> typing_Tuple[_anonymous_label, str]:
5580 # need to escape chars that interfere with format
5581 # strings in any case, issue #8724
5582 body = _anonymous_label_escape.sub("_", body)
5584 if sanitize_key:
5585 # sanitize_key is then an extra step used by BindParameter
5586 body = body.strip("_")
5588 key = f"{seed} {body.replace('%', '%%')}"
5589 label = _anonymous_label(f"%({key})s")
5590 return label, key
5592 @classmethod
5593 def safe_construct(
5594 cls, seed: int, body: str, sanitize_key: bool = False
5595 ) -> _anonymous_label:
5596 # need to escape chars that interfere with format
5597 # strings in any case, issue #8724
5598 body = _anonymous_label_escape.sub("_", body)
5600 if sanitize_key:
5601 # sanitize_key is then an extra step used by BindParameter
5602 body = body.strip("_")
5604 return _anonymous_label(f"%({seed} {body.replace('%', '%%')})s")
5606 def __add__(self, other: str) -> _anonymous_label:
5607 if "%" in other and not isinstance(other, _anonymous_label):
5608 other = str(other).replace("%", "%%")
5609 else:
5610 other = str(other)
5612 return _anonymous_label(
5613 quoted_name(
5614 str.__add__(self, other),
5615 self.quote,
5616 )
5617 )
5619 def __radd__(self, other: str) -> _anonymous_label:
5620 if "%" in other and not isinstance(other, _anonymous_label):
5621 other = str(other).replace("%", "%%")
5622 else:
5623 other = str(other)
5625 return _anonymous_label(
5626 quoted_name(
5627 str.__add__(other, self),
5628 self.quote,
5629 )
5630 )
5632 def apply_map(self, map_: Mapping[str, Any]) -> str:
5633 if self.quote is not None:
5634 # preserve quoting only if necessary
5635 return quoted_name(self % map_, self.quote)
5636 else:
5637 # else skip the constructor call
5638 return self % map_