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 _DMLOnlyColumnArgument
91 from ._typing import _HasDialect
92 from ._typing import _InfoType
93 from ._typing import _PropagateAttrsType
94 from ._typing import _TypeEngineArgument
95 from .base import _EntityNamespace
96 from .base import ColumnSet
97 from .cache_key import _CacheKeyTraversalType
98 from .cache_key import CacheKey
99 from .compiler import Compiled
100 from .compiler import SQLCompiler
101 from .functions import FunctionElement
102 from .operators import OperatorType
103 from .schema import Column
104 from .schema import DefaultGenerator
105 from .schema import FetchedValue
106 from .schema import ForeignKey
107 from .selectable import _SelectIterable
108 from .selectable import FromClause
109 from .selectable import NamedFromClause
110 from .selectable import TextualSelect
111 from .sqltypes import TupleType
112 from .type_api import TypeEngine
113 from .visitors import _CloneCallableType
114 from .visitors import _TraverseInternalsType
115 from .visitors import anon_map
116 from ..engine import Connection
117 from ..engine import Dialect
118 from ..engine.interfaces import _CoreMultiExecuteParams
119 from ..engine.interfaces import CacheStats
120 from ..engine.interfaces import CompiledCacheType
121 from ..engine.interfaces import CoreExecuteOptionsParameter
122 from ..engine.interfaces import SchemaTranslateMapType
123 from ..engine.result import Result
126_NUMERIC = Union[float, Decimal]
127_NUMBER = Union[float, int, Decimal]
129_T = TypeVar("_T", bound="Any")
130_T_co = TypeVar("_T_co", bound=Any, covariant=True)
131_OPT = TypeVar("_OPT", bound="Any")
132_NT = TypeVar("_NT", bound="_NUMERIC")
134_NMT = TypeVar("_NMT", bound="_NUMBER")
137@overload
138def literal(
139 value: Any,
140 type_: _TypeEngineArgument[_T],
141 literal_execute: bool = False,
142) -> BindParameter[_T]: ...
145@overload
146def literal(
147 value: _T,
148 type_: None = None,
149 literal_execute: bool = False,
150) -> BindParameter[_T]: ...
153@overload
154def literal(
155 value: Any,
156 type_: Optional[_TypeEngineArgument[Any]] = None,
157 literal_execute: bool = False,
158) -> BindParameter[Any]: ...
161def literal(
162 value: Any,
163 type_: Optional[_TypeEngineArgument[Any]] = None,
164 literal_execute: bool = False,
165) -> BindParameter[Any]:
166 r"""Return a literal clause, bound to a bind parameter.
168 Literal clauses are created automatically when non-
169 :class:`_expression.ClauseElement` objects (such as strings, ints, dates,
170 etc.) are
171 used in a comparison operation with a :class:`_expression.ColumnElement`
172 subclass,
173 such as a :class:`~sqlalchemy.schema.Column` object. Use this function
174 to force the generation of a literal clause, which will be created as a
175 :class:`BindParameter` with a bound value.
177 :param value: the value to be bound. Can be any Python object supported by
178 the underlying DB-API, or is translatable via the given type argument.
180 :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine` which will
181 provide bind-parameter translation for this literal.
183 :param literal_execute: optional bool, when True, the SQL engine will
184 attempt to render the bound value directly in the SQL statement at
185 execution time rather than providing as a parameter value.
187 .. versionadded:: 2.0
189 """
190 return coercions.expect(
191 roles.LiteralValueRole,
192 value,
193 type_=type_,
194 literal_execute=literal_execute,
195 )
198def literal_column(
199 text: str, type_: Optional[_TypeEngineArgument[_T]] = None
200) -> ColumnClause[_T]:
201 r"""Produce a :class:`.ColumnClause` object that has the
202 :paramref:`_expression.column.is_literal` flag set to True.
204 :func:`_expression.literal_column` is similar to
205 :func:`_expression.column`, except that
206 it is more often used as a "standalone" column expression that renders
207 exactly as stated; while :func:`_expression.column`
208 stores a string name that
209 will be assumed to be part of a table and may be quoted as such,
210 :func:`_expression.literal_column` can be that,
211 or any other arbitrary column-oriented
212 expression.
214 :param text: the text of the expression; can be any SQL expression.
215 Quoting rules will not be applied. To specify a column-name expression
216 which should be subject to quoting rules, use the :func:`column`
217 function.
219 :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine`
220 object which will
221 provide result-set translation and additional expression semantics for
222 this column. If left as ``None`` the type will be :class:`.NullType`.
224 .. seealso::
226 :func:`_expression.column`
228 :func:`_expression.text`
230 :ref:`tutorial_select_arbitrary_text`
232 """
233 return ColumnClause(text, type_=type_, is_literal=True)
236class CompilerElement(Visitable):
237 """base class for SQL elements that can be compiled to produce a
238 SQL string.
240 .. versionadded:: 2.0
242 """
244 __slots__ = ()
245 __visit_name__ = "compiler_element"
247 supports_execution = False
249 stringify_dialect = "default"
251 @util.preload_module("sqlalchemy.engine.default")
252 @util.preload_module("sqlalchemy.engine.url")
253 def compile(
254 self,
255 bind: Optional[_HasDialect] = None,
256 dialect: Optional[Dialect] = None,
257 **kw: Any,
258 ) -> Compiled:
259 """Compile this SQL expression.
261 The return value is a :class:`~.Compiled` object.
262 Calling ``str()`` or ``unicode()`` on the returned value will yield a
263 string representation of the result. The
264 :class:`~.Compiled` object also can return a
265 dictionary of bind parameter names and values
266 using the ``params`` accessor.
268 :param bind: An :class:`.Connection` or :class:`.Engine` which
269 can provide a :class:`.Dialect` in order to generate a
270 :class:`.Compiled` object. If the ``bind`` and
271 ``dialect`` parameters are both omitted, a default SQL compiler
272 is used.
274 :param column_keys: Used for INSERT and UPDATE statements, a list of
275 column names which should be present in the VALUES clause of the
276 compiled statement. If ``None``, all columns from the target table
277 object are rendered.
279 :param dialect: A :class:`.Dialect` instance which can generate
280 a :class:`.Compiled` object. This argument takes precedence over
281 the ``bind`` argument.
283 :param compile_kwargs: optional dictionary of additional parameters
284 that will be passed through to the compiler within all "visit"
285 methods. This allows any custom flag to be passed through to
286 a custom compilation construct, for example. It is also used
287 for the case of passing the ``literal_binds`` flag through::
289 from sqlalchemy.sql import table, column, select
291 t = table("t", column("x"))
293 s = select(t).where(t.c.x == 5)
295 print(s.compile(compile_kwargs={"literal_binds": True}))
297 .. seealso::
299 :ref:`faq_sql_expression_string`
301 """
303 if dialect is None:
304 if bind:
305 dialect = bind.dialect
306 elif self.stringify_dialect == "default":
307 dialect = self._default_dialect()
308 else:
309 url = util.preloaded.engine_url
310 dialect = url.URL.create(
311 self.stringify_dialect
312 ).get_dialect()()
314 return self._compiler(dialect, **kw)
316 def _default_dialect(self):
317 default = util.preloaded.engine_default
318 return default.StrCompileDialect()
320 def _compiler(self, dialect: Dialect, **kw: Any) -> Compiled:
321 """Return a compiler appropriate for this ClauseElement, given a
322 Dialect."""
324 if TYPE_CHECKING:
325 assert isinstance(self, ClauseElement)
326 return dialect.statement_compiler(dialect, self, **kw)
328 def __str__(self) -> str:
329 return str(self.compile())
332@inspection._self_inspects
333class ClauseElement(
334 SupportsWrappingAnnotations,
335 MemoizedHasCacheKey,
336 HasCopyInternals,
337 ExternallyTraversible,
338 CompilerElement,
339):
340 """Base class for elements of a programmatically constructed SQL
341 expression.
343 """
345 __visit_name__ = "clause"
347 if TYPE_CHECKING:
349 @util.memoized_property
350 def _propagate_attrs(self) -> _PropagateAttrsType:
351 """like annotations, however these propagate outwards liberally
352 as SQL constructs are built, and are set up at construction time.
354 """
355 ...
357 else:
358 _propagate_attrs = util.EMPTY_DICT
360 @util.ro_memoized_property
361 def description(self) -> Optional[str]:
362 return None
364 _is_clone_of: Optional[Self] = None
366 is_clause_element = True
367 is_selectable = False
368 is_dml = False
369 _is_column_element = False
370 _is_keyed_column_element = False
371 _is_table = False
372 _gen_static_annotations_cache_key = False
373 _is_textual = False
374 _is_from_clause = False
375 _is_returns_rows = False
376 _is_text_clause = False
377 _is_from_container = False
378 _is_select_container = False
379 _is_select_base = False
380 _is_select_statement = False
381 _is_bind_parameter = False
382 _is_clause_list = False
383 _is_lambda_element = False
384 _is_singleton_constant = False
385 _is_immutable = False
386 _is_star = False
388 @property
389 def _order_by_label_element(self) -> Optional[Label[Any]]:
390 return None
392 _cache_key_traversal: _CacheKeyTraversalType = None
394 negation_clause: ColumnElement[bool]
396 if typing.TYPE_CHECKING:
398 def get_children(
399 self, *, omit_attrs: typing_Tuple[str, ...] = ..., **kw: Any
400 ) -> Iterable[ClauseElement]: ...
402 @util.ro_non_memoized_property
403 def _from_objects(self) -> List[FromClause]:
404 return []
406 def _set_propagate_attrs(self, values: Mapping[str, Any]) -> Self:
407 # usually, self._propagate_attrs is empty here. one case where it's
408 # not is a subquery against ORM select, that is then pulled as a
409 # property of an aliased class. should all be good
411 # assert not self._propagate_attrs
413 self._propagate_attrs = util.immutabledict(values)
414 return self
416 def _default_compiler(self) -> SQLCompiler:
417 dialect = self._default_dialect()
418 return dialect.statement_compiler(dialect, self) # type: ignore
420 def _clone(self, **kw: Any) -> Self:
421 """Create a shallow copy of this ClauseElement.
423 This method may be used by a generative API. Its also used as
424 part of the "deep" copy afforded by a traversal that combines
425 the _copy_internals() method.
427 """
429 skip = self._memoized_keys
430 c = self.__class__.__new__(self.__class__)
432 if skip:
433 # ensure this iteration remains atomic
434 c.__dict__ = {
435 k: v for k, v in self.__dict__.copy().items() if k not in skip
436 }
437 else:
438 c.__dict__ = self.__dict__.copy()
440 # this is a marker that helps to "equate" clauses to each other
441 # when a Select returns its list of FROM clauses. the cloning
442 # process leaves around a lot of remnants of the previous clause
443 # typically in the form of column expressions still attached to the
444 # old table.
445 cc = self._is_clone_of
446 c._is_clone_of = cc if cc is not None else self
447 return c
449 def _negate_in_binary(self, negated_op, original_op):
450 """a hook to allow the right side of a binary expression to respond
451 to a negation of the binary expression.
453 Used for the special case of expanding bind parameter with IN.
455 """
456 return self
458 def _with_binary_element_type(self, type_):
459 """in the context of binary expression, convert the type of this
460 object to the one given.
462 applies only to :class:`_expression.ColumnElement` classes.
464 """
465 return self
467 @property
468 def _constructor(self): # type: ignore[override]
469 """return the 'constructor' for this ClauseElement.
471 This is for the purposes for creating a new object of
472 this type. Usually, its just the element's __class__.
473 However, the "Annotated" version of the object overrides
474 to return the class of its proxied element.
476 """
477 return self.__class__
479 @HasMemoized.memoized_attribute
480 def _cloned_set(self):
481 """Return the set consisting all cloned ancestors of this
482 ClauseElement.
484 Includes this ClauseElement. This accessor tends to be used for
485 FromClause objects to identify 'equivalent' FROM clauses, regardless
486 of transformative operations.
488 """
489 s = util.column_set()
490 f: Optional[ClauseElement] = self
492 # note this creates a cycle, asserted in test_memusage. however,
493 # turning this into a plain @property adds tends of thousands of method
494 # calls to Core / ORM performance tests, so the small overhead
495 # introduced by the relatively small amount of short term cycles
496 # produced here is preferable
497 while f is not None:
498 s.add(f)
499 f = f._is_clone_of
500 return s
502 def _de_clone(self):
503 while self._is_clone_of is not None:
504 self = self._is_clone_of
505 return self
507 @util.ro_non_memoized_property
508 def entity_namespace(self) -> _EntityNamespace:
509 raise AttributeError(
510 "This SQL expression has no entity namespace "
511 "with which to filter from."
512 )
514 def __getstate__(self):
515 d = self.__dict__.copy()
516 d.pop("_is_clone_of", None)
517 d.pop("_generate_cache_key", None)
518 return d
520 def _execute_on_connection(
521 self,
522 connection: Connection,
523 distilled_params: _CoreMultiExecuteParams,
524 execution_options: CoreExecuteOptionsParameter,
525 ) -> Result[Unpack[TupleAny]]:
526 if self.supports_execution:
527 if TYPE_CHECKING:
528 assert isinstance(self, Executable)
529 return connection._execute_clauseelement(
530 self, distilled_params, execution_options
531 )
532 else:
533 raise exc.ObjectNotExecutableError(self)
535 def _execute_on_scalar(
536 self,
537 connection: Connection,
538 distilled_params: _CoreMultiExecuteParams,
539 execution_options: CoreExecuteOptionsParameter,
540 ) -> Any:
541 """an additional hook for subclasses to provide a different
542 implementation for connection.scalar() vs. connection.execute().
544 .. versionadded:: 2.0
546 """
547 return self._execute_on_connection(
548 connection, distilled_params, execution_options
549 ).scalar()
551 def _get_embedded_bindparams(self) -> Sequence[BindParameter[Any]]:
552 """Return the list of :class:`.BindParameter` objects embedded in the
553 object.
555 This accomplishes the same purpose as ``visitors.traverse()`` or
556 similar would provide, however by making use of the cache key
557 it takes advantage of memoization of the key to result in fewer
558 net method calls, assuming the statement is also going to be
559 executed.
561 """
563 key = self._generate_cache_key()
564 if key is None:
565 bindparams: List[BindParameter[Any]] = []
567 traverse(self, {}, {"bindparam": bindparams.append})
568 return bindparams
570 else:
571 return key.bindparams
573 def unique_params(
574 self,
575 __optionaldict: Optional[Dict[str, Any]] = None,
576 /,
577 **kwargs: Any,
578 ) -> Self:
579 """Return a copy with :func:`_expression.bindparam` elements
580 replaced.
582 Same functionality as :meth:`_expression.ClauseElement.params`,
583 except adds `unique=True`
584 to affected bind parameters so that multiple statements can be
585 used.
587 """
588 return self._replace_params(True, __optionaldict, kwargs)
590 def params(
591 self,
592 __optionaldict: Optional[Mapping[str, Any]] = None,
593 /,
594 **kwargs: Any,
595 ) -> Self:
596 """Return a copy with :func:`_expression.bindparam` elements
597 replaced.
599 Returns a copy of this ClauseElement with
600 :func:`_expression.bindparam`
601 elements replaced with values taken from the given dictionary::
603 >>> clause = column("x") + bindparam("foo")
604 >>> print(clause.compile().params)
605 {'foo':None}
606 >>> print(clause.params({"foo": 7}).compile().params)
607 {'foo':7}
609 """
610 return self._replace_params(False, __optionaldict, kwargs)
612 def _replace_params(
613 self,
614 unique: bool,
615 optionaldict: Optional[Mapping[str, Any]],
616 kwargs: Dict[str, Any],
617 ) -> Self:
618 if optionaldict:
619 kwargs.update(optionaldict)
621 def visit_bindparam(bind: BindParameter[Any]) -> None:
622 if bind.key in kwargs:
623 bind.value = kwargs[bind.key]
624 bind.required = False
625 if unique:
626 bind._convert_to_unique()
628 return cloned_traverse(
629 self,
630 {"maintain_key": True, "detect_subquery_cols": True},
631 {"bindparam": visit_bindparam},
632 )
634 def compare(self, other: ClauseElement, **kw: Any) -> bool:
635 r"""Compare this :class:`_expression.ClauseElement` to
636 the given :class:`_expression.ClauseElement`.
638 Subclasses should override the default behavior, which is a
639 straight identity comparison.
641 \**kw are arguments consumed by subclass ``compare()`` methods and
642 may be used to modify the criteria for comparison
643 (see :class:`_expression.ColumnElement`).
645 """
646 return traversals.compare(self, other, **kw)
648 def self_group(
649 self, against: Optional[OperatorType] = None
650 ) -> ClauseElement:
651 """Apply a 'grouping' to this :class:`_expression.ClauseElement`.
653 This method is overridden by subclasses to return a "grouping"
654 construct, i.e. parenthesis. In particular it's used by "binary"
655 expressions to provide a grouping around themselves when placed into a
656 larger expression, as well as by :func:`_expression.select`
657 constructs when placed into the FROM clause of another
658 :func:`_expression.select`. (Note that subqueries should be
659 normally created using the :meth:`_expression.Select.alias` method,
660 as many
661 platforms require nested SELECT statements to be named).
663 As expressions are composed together, the application of
664 :meth:`self_group` is automatic - end-user code should never
665 need to use this method directly. Note that SQLAlchemy's
666 clause constructs take operator precedence into account -
667 so parenthesis might not be needed, for example, in
668 an expression like ``x OR (y AND z)`` - AND takes precedence
669 over OR.
671 The base :meth:`self_group` method of
672 :class:`_expression.ClauseElement`
673 just returns self.
674 """
675 return self
677 def _ungroup(self) -> ClauseElement:
678 """Return this :class:`_expression.ClauseElement`
679 without any groupings.
680 """
682 return self
684 def _compile_w_cache(
685 self,
686 dialect: Dialect,
687 *,
688 compiled_cache: Optional[CompiledCacheType],
689 column_keys: List[str],
690 for_executemany: bool = False,
691 schema_translate_map: Optional[SchemaTranslateMapType] = None,
692 **kw: Any,
693 ) -> typing_Tuple[
694 Compiled, Optional[Sequence[BindParameter[Any]]], CacheStats
695 ]:
696 elem_cache_key: Optional[CacheKey]
698 if compiled_cache is not None and dialect._supports_statement_cache:
699 elem_cache_key = self._generate_cache_key()
700 else:
701 elem_cache_key = None
703 extracted_params: Optional[Sequence[BindParameter[Any]]]
704 if elem_cache_key is not None:
705 if TYPE_CHECKING:
706 assert compiled_cache is not None
708 cache_key, extracted_params = elem_cache_key
709 key = (
710 dialect,
711 cache_key,
712 tuple(column_keys),
713 bool(schema_translate_map),
714 for_executemany,
715 )
716 compiled_sql = compiled_cache.get(key)
718 if compiled_sql is None:
719 cache_hit = dialect.CACHE_MISS
720 compiled_sql = self._compiler(
721 dialect,
722 cache_key=elem_cache_key,
723 column_keys=column_keys,
724 for_executemany=for_executemany,
725 schema_translate_map=schema_translate_map,
726 **kw,
727 )
728 compiled_cache[key] = compiled_sql
729 else:
730 cache_hit = dialect.CACHE_HIT
731 else:
732 extracted_params = None
733 compiled_sql = self._compiler(
734 dialect,
735 cache_key=elem_cache_key,
736 column_keys=column_keys,
737 for_executemany=for_executemany,
738 schema_translate_map=schema_translate_map,
739 **kw,
740 )
742 if not dialect._supports_statement_cache:
743 cache_hit = dialect.NO_DIALECT_SUPPORT
744 elif compiled_cache is None:
745 cache_hit = dialect.CACHING_DISABLED
746 else:
747 cache_hit = dialect.NO_CACHE_KEY
749 return compiled_sql, extracted_params, cache_hit
751 def __invert__(self):
752 # undocumented element currently used by the ORM for
753 # relationship.contains()
754 if hasattr(self, "negation_clause"):
755 return self.negation_clause
756 else:
757 return self._negate()
759 def _negate(self) -> ClauseElement:
760 # TODO: this code is uncovered and in all likelihood is not included
761 # in any codepath. So this should raise NotImplementedError in 2.1
762 grouped = self.self_group(against=operators.inv)
763 assert isinstance(grouped, ColumnElement)
764 return UnaryExpression(grouped, operator=operators.inv)
766 def __bool__(self):
767 raise TypeError("Boolean value of this clause is not defined")
769 def __repr__(self):
770 friendly = self.description
771 if friendly is None:
772 return object.__repr__(self)
773 else:
774 return "<%s.%s at 0x%x; %s>" % (
775 self.__module__,
776 self.__class__.__name__,
777 id(self),
778 friendly,
779 )
782class DQLDMLClauseElement(ClauseElement):
783 """represents a :class:`.ClauseElement` that compiles to a DQL or DML
784 expression, not DDL.
786 .. versionadded:: 2.0
788 """
790 if typing.TYPE_CHECKING:
792 def _compiler(self, dialect: Dialect, **kw: Any) -> SQLCompiler:
793 """Return a compiler appropriate for this ClauseElement, given a
794 Dialect."""
795 ...
797 def compile( # noqa: A001
798 self,
799 bind: Optional[_HasDialect] = None,
800 dialect: Optional[Dialect] = None,
801 **kw: Any,
802 ) -> SQLCompiler: ...
805class CompilerColumnElement(
806 roles.DMLColumnRole,
807 roles.DDLConstraintColumnRole,
808 roles.ColumnsClauseRole,
809 CompilerElement,
810):
811 """A compiler-only column element used for ad-hoc string compilations.
813 .. versionadded:: 2.0
815 """
817 __slots__ = ()
819 _propagate_attrs = util.EMPTY_DICT
820 _is_collection_aggregate = False
823# SQLCoreOperations should be suiting the ExpressionElementRole
824# and ColumnsClauseRole. however the MRO issues become too elaborate
825# at the moment.
826class SQLCoreOperations(Generic[_T_co], ColumnOperators, TypingOnly):
827 __slots__ = ()
829 # annotations for comparison methods
830 # these are from operators->Operators / ColumnOperators,
831 # redefined with the specific types returned by ColumnElement hierarchies
832 if typing.TYPE_CHECKING:
834 @util.non_memoized_property
835 def _propagate_attrs(self) -> _PropagateAttrsType: ...
837 def operate(
838 self, op: OperatorType, *other: Any, **kwargs: Any
839 ) -> ColumnElement[Any]: ...
841 def reverse_operate(
842 self, op: OperatorType, other: Any, **kwargs: Any
843 ) -> ColumnElement[Any]: ...
845 @overload
846 def op(
847 self,
848 opstring: str,
849 precedence: int = ...,
850 is_comparison: bool = ...,
851 *,
852 return_type: _TypeEngineArgument[_OPT],
853 python_impl: Optional[Callable[..., Any]] = None,
854 ) -> Callable[[Any], BinaryExpression[_OPT]]: ...
856 @overload
857 def op(
858 self,
859 opstring: str,
860 precedence: int = ...,
861 is_comparison: bool = ...,
862 return_type: Optional[_TypeEngineArgument[Any]] = ...,
863 python_impl: Optional[Callable[..., Any]] = ...,
864 ) -> Callable[[Any], BinaryExpression[Any]]: ...
866 def op(
867 self,
868 opstring: str,
869 precedence: int = 0,
870 is_comparison: bool = False,
871 return_type: Optional[_TypeEngineArgument[Any]] = None,
872 python_impl: Optional[Callable[..., Any]] = None,
873 ) -> Callable[[Any], BinaryExpression[Any]]: ...
875 def bool_op(
876 self,
877 opstring: str,
878 precedence: int = 0,
879 python_impl: Optional[Callable[..., Any]] = None,
880 ) -> Callable[[Any], BinaryExpression[bool]]: ...
882 def __and__(self, other: Any) -> BooleanClauseList: ...
884 def __or__(self, other: Any) -> BooleanClauseList: ...
886 def __invert__(self) -> ColumnElement[_T_co]: ...
888 def __lt__(self, other: Any) -> ColumnElement[bool]: ...
890 def __le__(self, other: Any) -> ColumnElement[bool]: ...
892 # declare also that this class has an hash method otherwise
893 # it may be assumed to be None by type checkers since the
894 # object defines __eq__ and python sets it to None in that case:
895 # https://docs.python.org/3/reference/datamodel.html#object.__hash__
896 def __hash__(self) -> int: ...
898 def __eq__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
899 ...
901 def __ne__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
902 ...
904 def is_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
906 def is_not_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
908 def __gt__(self, other: Any) -> ColumnElement[bool]: ...
910 def __ge__(self, other: Any) -> ColumnElement[bool]: ...
912 def __neg__(self) -> UnaryExpression[_T_co]: ...
914 def __contains__(self, other: Any) -> ColumnElement[bool]: ...
916 def __getitem__(self, index: Any) -> ColumnElement[Any]: ...
918 @overload
919 def __lshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
921 @overload
922 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
924 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
926 @overload
927 def __rlshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
929 @overload
930 def __rlshift__(self, other: Any) -> ColumnElement[Any]: ...
932 def __rlshift__(self, other: Any) -> ColumnElement[Any]: ...
934 @overload
935 def __rshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
937 @overload
938 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
940 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
942 @overload
943 def __rrshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
945 @overload
946 def __rrshift__(self, other: Any) -> ColumnElement[Any]: ...
948 def __rrshift__(self, other: Any) -> ColumnElement[Any]: ...
950 def __matmul__(self, other: Any) -> ColumnElement[Any]: ...
952 def __rmatmul__(self, other: Any) -> ColumnElement[Any]: ...
954 @overload
955 def concat(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
957 @overload
958 def concat(self, other: Any) -> ColumnElement[Any]: ...
960 def concat(self, other: Any) -> ColumnElement[Any]: ...
962 def like(
963 self, other: Any, escape: Optional[str] = None
964 ) -> BinaryExpression[bool]: ...
966 def ilike(
967 self, other: Any, escape: Optional[str] = None
968 ) -> BinaryExpression[bool]: ...
970 def bitwise_xor(self, other: Any) -> BinaryExpression[Any]: ...
972 def bitwise_or(self, other: Any) -> BinaryExpression[Any]: ...
974 def bitwise_and(self, other: Any) -> BinaryExpression[Any]: ...
976 def bitwise_not(self) -> UnaryExpression[_T_co]: ...
978 def bitwise_lshift(self, other: Any) -> BinaryExpression[Any]: ...
980 def bitwise_rshift(self, other: Any) -> BinaryExpression[Any]: ...
982 def in_(
983 self,
984 other: Union[
985 Iterable[Any], BindParameter[Any], roles.InElementRole
986 ],
987 ) -> BinaryExpression[bool]: ...
989 def not_in(
990 self,
991 other: Union[
992 Iterable[Any], BindParameter[Any], roles.InElementRole
993 ],
994 ) -> BinaryExpression[bool]: ...
996 def notin_(
997 self,
998 other: Union[
999 Iterable[Any], BindParameter[Any], roles.InElementRole
1000 ],
1001 ) -> BinaryExpression[bool]: ...
1003 def not_like(
1004 self, other: Any, escape: Optional[str] = None
1005 ) -> BinaryExpression[bool]: ...
1007 def notlike(
1008 self, other: Any, escape: Optional[str] = None
1009 ) -> BinaryExpression[bool]: ...
1011 def not_ilike(
1012 self, other: Any, escape: Optional[str] = None
1013 ) -> BinaryExpression[bool]: ...
1015 def notilike(
1016 self, other: Any, escape: Optional[str] = None
1017 ) -> BinaryExpression[bool]: ...
1019 def is_(self, other: Any) -> BinaryExpression[bool]: ...
1021 def is_not(self, other: Any) -> BinaryExpression[bool]: ...
1023 def isnot(self, other: Any) -> BinaryExpression[bool]: ...
1025 def startswith(
1026 self,
1027 other: Any,
1028 escape: Optional[str] = None,
1029 autoescape: bool = False,
1030 ) -> ColumnElement[bool]: ...
1032 def istartswith(
1033 self,
1034 other: Any,
1035 escape: Optional[str] = None,
1036 autoescape: bool = False,
1037 ) -> ColumnElement[bool]: ...
1039 def endswith(
1040 self,
1041 other: Any,
1042 escape: Optional[str] = None,
1043 autoescape: bool = False,
1044 ) -> ColumnElement[bool]: ...
1046 def iendswith(
1047 self,
1048 other: Any,
1049 escape: Optional[str] = None,
1050 autoescape: bool = False,
1051 ) -> ColumnElement[bool]: ...
1053 def contains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1055 def icontains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1057 def match(self, other: Any, **kwargs: Any) -> ColumnElement[bool]: ...
1059 def regexp_match(
1060 self, pattern: Any, flags: Optional[str] = None
1061 ) -> ColumnElement[bool]: ...
1063 def regexp_replace(
1064 self, pattern: Any, replacement: Any, flags: Optional[str] = None
1065 ) -> ColumnElement[str]: ...
1067 def desc(self) -> UnaryExpression[_T_co]: ...
1069 def asc(self) -> UnaryExpression[_T_co]: ...
1071 def nulls_first(self) -> UnaryExpression[_T_co]: ...
1073 def nullsfirst(self) -> UnaryExpression[_T_co]: ...
1075 def nulls_last(self) -> UnaryExpression[_T_co]: ...
1077 def nullslast(self) -> UnaryExpression[_T_co]: ...
1079 def collate(self, collation: str) -> CollationClause: ...
1081 def between(
1082 self, cleft: Any, cright: Any, symmetric: bool = False
1083 ) -> BinaryExpression[bool]: ...
1085 def distinct(self: _SQO[_T_co]) -> UnaryExpression[_T_co]: ...
1087 def any_(self) -> CollectionAggregate[Any]: ...
1089 def all_(self) -> CollectionAggregate[Any]: ...
1091 # numeric overloads. These need more tweaking
1092 # in particular they all need to have a variant for Optiona[_T]
1093 # because Optional only applies to the data side, not the expression
1094 # side
1096 @overload
1097 def __add__(
1098 self: _SQO[_NMT],
1099 other: Any,
1100 ) -> ColumnElement[_NMT]: ...
1102 @overload
1103 def __add__(
1104 self: _SQO[str],
1105 other: Any,
1106 ) -> ColumnElement[str]: ...
1108 @overload
1109 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1111 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1113 @overload
1114 def __radd__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1116 @overload
1117 def __radd__(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
1119 def __radd__(self, other: Any) -> ColumnElement[Any]: ...
1121 @overload
1122 def __sub__(
1123 self: _SQO[_NMT],
1124 other: Any,
1125 ) -> ColumnElement[_NMT]: ...
1127 @overload
1128 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1130 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1132 @overload
1133 def __rsub__(
1134 self: _SQO[_NMT],
1135 other: Any,
1136 ) -> ColumnElement[_NMT]: ...
1138 @overload
1139 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1141 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1143 @overload
1144 def __mul__(
1145 self: _SQO[_NMT],
1146 other: Any,
1147 ) -> ColumnElement[_NMT]: ...
1149 @overload
1150 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1152 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1154 @overload
1155 def __rmul__(
1156 self: _SQO[_NMT],
1157 other: Any,
1158 ) -> ColumnElement[_NMT]: ...
1160 @overload
1161 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1163 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1165 @overload
1166 def __mod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1168 @overload
1169 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1171 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1173 @overload
1174 def __rmod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1176 @overload
1177 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1179 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1181 @overload
1182 def __truediv__(
1183 self: _SQO[int], other: Any
1184 ) -> ColumnElement[_NUMERIC]: ...
1186 @overload
1187 def __truediv__(self: _SQO[_NT], other: Any) -> ColumnElement[_NT]: ...
1189 @overload
1190 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1192 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1194 @overload
1195 def __rtruediv__(
1196 self: _SQO[_NMT], other: Any
1197 ) -> ColumnElement[_NUMERIC]: ...
1199 @overload
1200 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1202 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1204 @overload
1205 def __floordiv__(
1206 self: _SQO[_NMT], other: Any
1207 ) -> ColumnElement[_NMT]: ...
1209 @overload
1210 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1212 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1214 @overload
1215 def __rfloordiv__(
1216 self: _SQO[_NMT], other: Any
1217 ) -> ColumnElement[_NMT]: ...
1219 @overload
1220 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1222 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1225class SQLColumnExpression(
1226 SQLCoreOperations[_T_co], roles.ExpressionElementRole[_T_co], TypingOnly
1227):
1228 """A type that may be used to indicate any SQL column element or object
1229 that acts in place of one.
1231 :class:`.SQLColumnExpression` is a base of
1232 :class:`.ColumnElement`, as well as within the bases of ORM elements
1233 such as :class:`.InstrumentedAttribute`, and may be used in :pep:`484`
1234 typing to indicate arguments or return values that should behave
1235 as column expressions.
1237 .. versionadded:: 2.0.0b4
1240 """
1242 __slots__ = ()
1245_SQO = SQLCoreOperations
1248class ColumnElement(
1249 roles.ColumnArgumentOrKeyRole,
1250 roles.StatementOptionRole,
1251 roles.WhereHavingRole,
1252 roles.BinaryElementRole[_T],
1253 roles.OrderByRole,
1254 roles.ColumnsClauseRole,
1255 roles.LimitOffsetRole,
1256 roles.DMLColumnRole,
1257 roles.DDLConstraintColumnRole,
1258 roles.DDLExpressionRole,
1259 SQLColumnExpression[_T],
1260 DQLDMLClauseElement,
1261):
1262 """Represent a column-oriented SQL expression suitable for usage in the
1263 "columns" clause, WHERE clause etc. of a statement.
1265 While the most familiar kind of :class:`_expression.ColumnElement` is the
1266 :class:`_schema.Column` object, :class:`_expression.ColumnElement`
1267 serves as the basis
1268 for any unit that may be present in a SQL expression, including
1269 the expressions themselves, SQL functions, bound parameters,
1270 literal expressions, keywords such as ``NULL``, etc.
1271 :class:`_expression.ColumnElement`
1272 is the ultimate base class for all such elements.
1274 A wide variety of SQLAlchemy Core functions work at the SQL expression
1275 level, and are intended to accept instances of
1276 :class:`_expression.ColumnElement` as
1277 arguments. These functions will typically document that they accept a
1278 "SQL expression" as an argument. What this means in terms of SQLAlchemy
1279 usually refers to an input which is either already in the form of a
1280 :class:`_expression.ColumnElement` object,
1281 or a value which can be **coerced** into
1282 one. The coercion rules followed by most, but not all, SQLAlchemy Core
1283 functions with regards to SQL expressions are as follows:
1285 * a literal Python value, such as a string, integer or floating
1286 point value, boolean, datetime, ``Decimal`` object, or virtually
1287 any other Python object, will be coerced into a "literal bound
1288 value". This generally means that a :func:`.bindparam` will be
1289 produced featuring the given value embedded into the construct; the
1290 resulting :class:`.BindParameter` object is an instance of
1291 :class:`_expression.ColumnElement`.
1292 The Python value will ultimately be sent
1293 to the DBAPI at execution time as a parameterized argument to the
1294 ``execute()`` or ``executemany()`` methods, after SQLAlchemy
1295 type-specific converters (e.g. those provided by any associated
1296 :class:`.TypeEngine` objects) are applied to the value.
1298 * any special object value, typically ORM-level constructs, which
1299 feature an accessor called ``__clause_element__()``. The Core
1300 expression system looks for this method when an object of otherwise
1301 unknown type is passed to a function that is looking to coerce the
1302 argument into a :class:`_expression.ColumnElement` and sometimes a
1303 :class:`_expression.SelectBase` expression.
1304 It is used within the ORM to
1305 convert from ORM-specific objects like mapped classes and
1306 mapped attributes into Core expression objects.
1308 * The Python ``None`` value is typically interpreted as ``NULL``,
1309 which in SQLAlchemy Core produces an instance of :func:`.null`.
1311 A :class:`_expression.ColumnElement` provides the ability to generate new
1312 :class:`_expression.ColumnElement`
1313 objects using Python expressions. This means that Python operators
1314 such as ``==``, ``!=`` and ``<`` are overloaded to mimic SQL operations,
1315 and allow the instantiation of further :class:`_expression.ColumnElement`
1316 instances
1317 which are composed from other, more fundamental
1318 :class:`_expression.ColumnElement`
1319 objects. For example, two :class:`.ColumnClause` objects can be added
1320 together with the addition operator ``+`` to produce
1321 a :class:`.BinaryExpression`.
1322 Both :class:`.ColumnClause` and :class:`.BinaryExpression` are subclasses
1323 of :class:`_expression.ColumnElement`:
1325 .. sourcecode:: pycon+sql
1327 >>> from sqlalchemy.sql import column
1328 >>> column("a") + column("b")
1329 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
1330 >>> print(column("a") + column("b"))
1331 {printsql}a + b
1333 .. seealso::
1335 :class:`_schema.Column`
1337 :func:`_expression.column`
1339 """
1341 __visit_name__ = "column_element"
1343 primary_key: bool = False
1344 _is_clone_of: Optional[ColumnElement[_T]]
1345 _is_column_element = True
1346 _insert_sentinel: bool = False
1347 _omit_from_statements = False
1348 _is_collection_aggregate = False
1350 foreign_keys: AbstractSet[ForeignKey] = frozenset()
1352 @util.memoized_property
1353 def _proxies(self) -> List[ColumnElement[Any]]:
1354 return []
1356 @util.non_memoized_property
1357 def _tq_label(self) -> Optional[str]:
1358 """The named label that can be used to target
1359 this column in a result set in a "table qualified" context.
1361 This label is almost always the label used when
1362 rendering <expr> AS <label> in a SELECT statement when using
1363 the LABEL_STYLE_TABLENAME_PLUS_COL label style, which is what the
1364 legacy ORM ``Query`` object uses as well.
1366 For a regular Column bound to a Table, this is typically the label
1367 <tablename>_<columnname>. For other constructs, different rules
1368 may apply, such as anonymized labels and others.
1370 .. versionchanged:: 1.4.21 renamed from ``._label``
1372 """
1373 return None
1375 key: Optional[str] = None
1376 """The 'key' that in some circumstances refers to this object in a
1377 Python namespace.
1379 This typically refers to the "key" of the column as present in the
1380 ``.c`` collection of a selectable, e.g. ``sometable.c["somekey"]`` would
1381 return a :class:`_schema.Column` with a ``.key`` of "somekey".
1383 """
1385 @HasMemoized.memoized_attribute
1386 def _tq_key_label(self) -> Optional[str]:
1387 """A label-based version of 'key' that in some circumstances refers
1388 to this object in a Python namespace.
1391 _tq_key_label comes into play when a select() statement is constructed
1392 with apply_labels(); in this case, all Column objects in the ``.c``
1393 collection are rendered as <tablename>_<columnname> in SQL; this is
1394 essentially the value of ._label. But to locate those columns in the
1395 ``.c`` collection, the name is along the lines of <tablename>_<key>;
1396 that's the typical value of .key_label.
1398 .. versionchanged:: 1.4.21 renamed from ``._key_label``
1400 """
1401 return self._proxy_key
1403 @property
1404 def _key_label(self) -> Optional[str]:
1405 """legacy; renamed to _tq_key_label"""
1406 return self._tq_key_label
1408 @property
1409 def _label(self) -> Optional[str]:
1410 """legacy; renamed to _tq_label"""
1411 return self._tq_label
1413 @property
1414 def _non_anon_label(self) -> Optional[str]:
1415 """the 'name' that naturally applies this element when rendered in
1416 SQL.
1418 Concretely, this is the "name" of a column or a label in a
1419 SELECT statement; ``<columnname>`` and ``<labelname>`` below:
1421 .. sourcecode:: sql
1423 SELECT <columnmame> FROM table
1425 SELECT column AS <labelname> FROM table
1427 Above, the two names noted will be what's present in the DBAPI
1428 ``cursor.description`` as the names.
1430 If this attribute returns ``None``, it means that the SQL element as
1431 written does not have a 100% fully predictable "name" that would appear
1432 in the ``cursor.description``. Examples include SQL functions, CAST
1433 functions, etc. While such things do return names in
1434 ``cursor.description``, they are only predictable on a
1435 database-specific basis; e.g. an expression like ``MAX(table.col)`` may
1436 appear as the string ``max`` on one database (like PostgreSQL) or may
1437 appear as the whole expression ``max(table.col)`` on SQLite.
1439 The default implementation looks for a ``.name`` attribute on the
1440 object, as has been the precedent established in SQLAlchemy for many
1441 years. An exception is made on the ``FunctionElement`` subclass
1442 so that the return value is always ``None``.
1444 .. versionadded:: 1.4.21
1448 """
1449 return getattr(self, "name", None)
1451 _render_label_in_columns_clause = True
1452 """A flag used by select._columns_plus_names that helps to determine
1453 we are actually going to render in terms of "SELECT <col> AS <label>".
1454 This flag can be returned as False for some Column objects that want
1455 to be rendered as simple "SELECT <col>"; typically columns that don't have
1456 any parent table and are named the same as what the label would be
1457 in any case.
1459 """
1461 _allow_label_resolve = True
1462 """A flag that can be flipped to prevent a column from being resolvable
1463 by string label name.
1465 The joined eager loader strategy in the ORM uses this, for example.
1467 """
1469 _is_implicitly_boolean = False
1471 _alt_names: Sequence[str] = ()
1473 if TYPE_CHECKING:
1475 def _ungroup(self) -> ColumnElement[_T]: ...
1477 @overload
1478 def self_group(self, against: None = None) -> ColumnElement[_T]: ...
1480 @overload
1481 def self_group(
1482 self, against: Optional[OperatorType] = None
1483 ) -> ColumnElement[Any]: ...
1485 def self_group(
1486 self, against: Optional[OperatorType] = None
1487 ) -> ColumnElement[Any]:
1488 if (
1489 against in (operators.and_, operators.or_, operators._asbool)
1490 and self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity
1491 ):
1492 return AsBoolean(self, operators.is_true, operators.is_false)
1493 elif against in (operators.any_op, operators.all_op):
1494 return Grouping(self)
1495 else:
1496 return self
1498 @overload
1499 def _negate(self: ColumnElement[bool]) -> ColumnElement[bool]: ...
1501 @overload
1502 def _negate(self: ColumnElement[_T]) -> ColumnElement[_T]: ...
1504 def _negate(self) -> ColumnElement[Any]:
1505 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
1506 return AsBoolean(self, operators.is_false, operators.is_true)
1507 else:
1508 grouped = self.self_group(against=operators.inv)
1509 assert isinstance(grouped, ColumnElement)
1510 return UnaryExpression(
1511 grouped,
1512 operator=operators.inv,
1513 )
1515 type: TypeEngine[_T]
1517 if not TYPE_CHECKING:
1519 @util.memoized_property
1520 def type(self) -> TypeEngine[_T]: # noqa: A001
1521 # used for delayed setup of
1522 # type_api
1523 return type_api.NULLTYPE
1525 @HasMemoized.memoized_attribute
1526 def comparator(self) -> TypeEngine.Comparator[_T]:
1527 try:
1528 comparator_factory = self.type.comparator_factory
1529 except AttributeError as err:
1530 raise TypeError(
1531 "Object %r associated with '.type' attribute "
1532 "is not a TypeEngine class or object" % self.type
1533 ) from err
1534 else:
1535 return comparator_factory(self)
1537 def __setstate__(self, state):
1538 self.__dict__.update(state)
1540 def __getattr__(self, key: str) -> Any:
1541 try:
1542 return getattr(self.comparator, key)
1543 except AttributeError as err:
1544 raise AttributeError(
1545 "Neither %r object nor %r object has an attribute %r"
1546 % (
1547 type(self).__name__,
1548 type(self.comparator).__name__,
1549 key,
1550 )
1551 ) from err
1553 def operate(
1554 self,
1555 op: operators.OperatorType,
1556 *other: Any,
1557 **kwargs: Any,
1558 ) -> ColumnElement[Any]:
1559 return op(self.comparator, *other, **kwargs) # type: ignore[no-any-return] # noqa: E501
1561 def reverse_operate(
1562 self, op: operators.OperatorType, other: Any, **kwargs: Any
1563 ) -> ColumnElement[Any]:
1564 return op(other, self.comparator, **kwargs) # type: ignore[no-any-return] # noqa: E501
1566 def _bind_param(
1567 self,
1568 operator: operators.OperatorType,
1569 obj: Any,
1570 type_: Optional[TypeEngine[_T]] = None,
1571 expanding: bool = False,
1572 ) -> BindParameter[_T]:
1573 return BindParameter(
1574 None,
1575 obj,
1576 _compared_to_operator=operator,
1577 type_=type_,
1578 _compared_to_type=self.type,
1579 unique=True,
1580 expanding=expanding,
1581 )
1583 @property
1584 def expression(self) -> ColumnElement[Any]:
1585 """Return a column expression.
1587 Part of the inspection interface; returns self.
1589 """
1590 return self
1592 @property
1593 def _select_iterable(self) -> _SelectIterable:
1594 return (self,)
1596 @util.memoized_property
1597 def base_columns(self) -> FrozenSet[ColumnElement[Any]]:
1598 return frozenset(c for c in self.proxy_set if not c._proxies)
1600 @util.memoized_property
1601 def proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1602 """set of all columns we are proxying
1604 as of 2.0 this is explicitly deannotated columns. previously it was
1605 effectively deannotated columns but wasn't enforced. annotated
1606 columns should basically not go into sets if at all possible because
1607 their hashing behavior is very non-performant.
1609 """
1610 return frozenset([self._deannotate()]).union(
1611 itertools.chain(*[c.proxy_set for c in self._proxies])
1612 )
1614 @util.memoized_property
1615 def _expanded_proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1616 return frozenset(_expand_cloned(self.proxy_set))
1618 def _uncached_proxy_list(self) -> List[ColumnElement[Any]]:
1619 """An 'uncached' version of proxy set.
1621 This list includes annotated columns which perform very poorly in
1622 set operations.
1624 """
1626 return [self] + list(
1627 itertools.chain(*[c._uncached_proxy_list() for c in self._proxies])
1628 )
1630 def shares_lineage(self, othercolumn: ColumnElement[Any]) -> bool:
1631 """Return True if the given :class:`_expression.ColumnElement`
1632 has a common ancestor to this :class:`_expression.ColumnElement`."""
1634 return bool(self.proxy_set.intersection(othercolumn.proxy_set))
1636 def _compare_name_for_result(self, other: ColumnElement[Any]) -> bool:
1637 """Return True if the given column element compares to this one
1638 when targeting within a result row."""
1640 return (
1641 hasattr(other, "name")
1642 and hasattr(self, "name")
1643 and other.name == self.name
1644 )
1646 @HasMemoized.memoized_attribute
1647 def _proxy_key(self) -> Optional[str]:
1648 if self._annotations and "proxy_key" in self._annotations:
1649 return cast(str, self._annotations["proxy_key"])
1651 name = self.key
1652 if not name:
1653 # there's a bit of a seeming contradiction which is that the
1654 # "_non_anon_label" of a column can in fact be an
1655 # "_anonymous_label"; this is when it's on a column that is
1656 # proxying for an anonymous expression in a subquery.
1657 name = self._non_anon_label
1659 if isinstance(name, _anonymous_label):
1660 return None
1661 else:
1662 return name
1664 @HasMemoized.memoized_attribute
1665 def _expression_label(self) -> Optional[str]:
1666 """a suggested label to use in the case that the column has no name,
1667 which should be used if possible as the explicit 'AS <label>'
1668 where this expression would normally have an anon label.
1670 this is essentially mostly what _proxy_key does except it returns
1671 None if the column has a normal name that can be used.
1673 """
1675 if getattr(self, "name", None) is not None:
1676 return None
1677 elif self._annotations and "proxy_key" in self._annotations:
1678 return cast(str, self._annotations["proxy_key"])
1679 else:
1680 return None
1682 def _make_proxy(
1683 self,
1684 selectable: FromClause,
1685 *,
1686 primary_key: ColumnSet,
1687 foreign_keys: Set[KeyedColumnElement[Any]],
1688 name: Optional[str] = None,
1689 key: Optional[str] = None,
1690 name_is_truncatable: bool = False,
1691 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
1692 **kw: Any,
1693 ) -> typing_Tuple[str, ColumnClause[_T]]:
1694 """Create a new :class:`_expression.ColumnElement` representing this
1695 :class:`_expression.ColumnElement` as it appears in the select list of
1696 a descending selectable.
1698 """
1699 if name is None:
1700 name = self._anon_name_label
1701 if key is None:
1702 key = self._proxy_key
1703 else:
1704 key = name
1706 assert key is not None
1708 co: ColumnClause[_T] = ColumnClause(
1709 (
1710 coercions.expect(roles.TruncatedLabelRole, name)
1711 if name_is_truncatable
1712 else name
1713 ),
1714 type_=getattr(self, "type", None),
1715 _selectable=selectable,
1716 )
1718 co._propagate_attrs = selectable._propagate_attrs
1719 if compound_select_cols:
1720 co._proxies = list(compound_select_cols)
1721 else:
1722 co._proxies = [self]
1723 if selectable._is_clone_of is not None:
1724 co._is_clone_of = selectable._is_clone_of.columns.get(key)
1725 return key, co
1727 def cast(self, type_: _TypeEngineArgument[_OPT]) -> Cast[_OPT]:
1728 """Produce a type cast, i.e. ``CAST(<expression> AS <type>)``.
1730 This is a shortcut to the :func:`_expression.cast` function.
1732 .. seealso::
1734 :ref:`tutorial_casts`
1736 :func:`_expression.cast`
1738 :func:`_expression.type_coerce`
1740 """
1741 return Cast(self, type_)
1743 def label(self, name: Optional[str]) -> Label[_T]:
1744 """Produce a column label, i.e. ``<columnname> AS <name>``.
1746 This is a shortcut to the :func:`_expression.label` function.
1748 If 'name' is ``None``, an anonymous label name will be generated.
1750 """
1751 return Label(name, self, self.type)
1753 def _anon_label(
1754 self, seed: Optional[str], add_hash: Optional[int] = None
1755 ) -> _anonymous_label:
1756 while self._is_clone_of is not None:
1757 self = self._is_clone_of
1759 # as of 1.4 anonymous label for ColumnElement uses hash(), not id(),
1760 # as the identifier, because a column and its annotated version are
1761 # the same thing in a SQL statement
1762 hash_value = hash(self)
1764 if add_hash:
1765 # this path is used for disambiguating anon labels that would
1766 # otherwise be the same name for the same element repeated.
1767 # an additional numeric value is factored in for each label.
1769 # shift hash(self) (which is id(self), typically 8 byte integer)
1770 # 16 bits leftward. fill extra add_hash on right
1771 assert add_hash < (2 << 15)
1772 assert seed
1773 hash_value = (hash_value << 16) | add_hash
1775 # extra underscore is added for labels with extra hash
1776 # values, to isolate the "deduped anon" namespace from the
1777 # regular namespace. eliminates chance of these
1778 # manufactured hash values overlapping with regular ones for some
1779 # undefined python interpreter
1780 seed = seed + "_"
1782 if isinstance(seed, _anonymous_label):
1783 # NOTE: the space after the hash is required
1784 return _anonymous_label(f"{seed}%({hash_value} )s")
1786 return _anonymous_label.safe_construct(hash_value, seed or "anon")
1788 @util.memoized_property
1789 def _anon_name_label(self) -> str:
1790 """Provides a constant 'anonymous label' for this ColumnElement.
1792 This is a label() expression which will be named at compile time.
1793 The same label() is returned each time ``anon_label`` is called so
1794 that expressions can reference ``anon_label`` multiple times,
1795 producing the same label name at compile time.
1797 The compiler uses this function automatically at compile time
1798 for expressions that are known to be 'unnamed' like binary
1799 expressions and function calls.
1801 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1802 public and is renamed to _anon_name_label. anon_name exists
1803 for backwards compat
1805 """
1806 name = getattr(self, "name", None)
1807 return self._anon_label(name)
1809 @util.memoized_property
1810 def _anon_key_label(self) -> _anonymous_label:
1811 """Provides a constant 'anonymous key label' for this ColumnElement.
1813 Compare to ``anon_label``, except that the "key" of the column,
1814 if available, is used to generate the label.
1816 This is used when a deduplicating key is placed into the columns
1817 collection of a selectable.
1819 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1820 public and is renamed to _anon_key_label. anon_key_label exists
1821 for backwards compat
1823 """
1824 return self._anon_label(self._proxy_key)
1826 @property
1827 @util.deprecated(
1828 "1.4",
1829 "The :attr:`_expression.ColumnElement.anon_label` attribute is now "
1830 "private, and the public accessor is deprecated.",
1831 )
1832 def anon_label(self) -> str:
1833 return self._anon_name_label
1835 @property
1836 @util.deprecated(
1837 "1.4",
1838 "The :attr:`_expression.ColumnElement.anon_key_label` attribute is "
1839 "now private, and the public accessor is deprecated.",
1840 )
1841 def anon_key_label(self) -> str:
1842 return self._anon_key_label
1844 def _dedupe_anon_label_idx(self, idx: int) -> str:
1845 """label to apply to a column that is anon labeled, but repeated
1846 in the SELECT, so that we have to make an "extra anon" label that
1847 disambiguates it from the previous appearance.
1849 these labels come out like "foo_bar_id__1" and have double underscores
1850 in them.
1852 """
1853 label = getattr(self, "name", None)
1855 # current convention is that if the element doesn't have a
1856 # ".name" (usually because it is not NamedColumn), we try to
1857 # use a "table qualified" form for the "dedupe anon" label,
1858 # based on the notion that a label like
1859 # "CAST(casttest.v1 AS DECIMAL) AS casttest_v1__1" looks better than
1860 # "CAST(casttest.v1 AS DECIMAL) AS anon__1"
1862 if label is None:
1863 return self._dedupe_anon_tq_label_idx(idx)
1864 else:
1865 return self._anon_label(label, add_hash=idx)
1867 @util.memoized_property
1868 def _anon_tq_label(self) -> _anonymous_label:
1869 return self._anon_label(getattr(self, "_tq_label", None))
1871 @util.memoized_property
1872 def _anon_tq_key_label(self) -> _anonymous_label:
1873 return self._anon_label(getattr(self, "_tq_key_label", None))
1875 def _dedupe_anon_tq_label_idx(self, idx: int) -> _anonymous_label:
1876 label = getattr(self, "_tq_label", None) or "anon"
1878 return self._anon_label(label, add_hash=idx)
1881class KeyedColumnElement(ColumnElement[_T]):
1882 """ColumnElement where ``.key`` is non-None."""
1884 _is_keyed_column_element = True
1886 key: str
1889class WrapsColumnExpression(ColumnElement[_T]):
1890 """Mixin that defines a :class:`_expression.ColumnElement`
1891 as a wrapper with special
1892 labeling behavior for an expression that already has a name.
1894 .. versionadded:: 1.4
1896 .. seealso::
1898 :ref:`change_4449`
1901 """
1903 @property
1904 def wrapped_column_expression(self) -> ColumnElement[_T]:
1905 raise NotImplementedError()
1907 @util.non_memoized_property
1908 def _tq_label(self) -> Optional[str]:
1909 wce = self.wrapped_column_expression
1910 if hasattr(wce, "_tq_label"):
1911 return wce._tq_label
1912 else:
1913 return None
1915 @property
1916 def _label(self) -> Optional[str]:
1917 return self._tq_label
1919 @property
1920 def _non_anon_label(self) -> Optional[str]:
1921 return None
1923 @util.non_memoized_property
1924 def _anon_name_label(self) -> str:
1925 wce = self.wrapped_column_expression
1927 # this logic tries to get the WrappedColumnExpression to render
1928 # with "<expr> AS <name>", where "<name>" is the natural name
1929 # within the expression itself. e.g. "CAST(table.foo) AS foo".
1930 if not wce._is_text_clause:
1931 nal = wce._non_anon_label
1932 if nal:
1933 return nal
1934 elif hasattr(wce, "_anon_name_label"):
1935 return wce._anon_name_label
1936 return super()._anon_name_label
1938 def _dedupe_anon_label_idx(self, idx: int) -> str:
1939 wce = self.wrapped_column_expression
1940 nal = wce._non_anon_label
1941 if nal:
1942 return self._anon_label(nal + "_")
1943 else:
1944 return self._dedupe_anon_tq_label_idx(idx)
1946 @property
1947 def _proxy_key(self):
1948 wce = self.wrapped_column_expression
1950 if not wce._is_text_clause:
1951 return wce._proxy_key
1952 return super()._proxy_key
1955class DMLTargetCopy(roles.InElementRole, KeyedColumnElement[_T]):
1956 """Refer to another column's VALUES or SET expression in an INSERT or
1957 UPDATE statement.
1959 See the public-facing :func:`_sql.from_dml_column` constructor for
1960 background.
1962 .. versionadded:: 2.1
1965 """
1967 def __init__(self, column: _DMLOnlyColumnArgument[_T]):
1968 self.column = coercions.expect(roles.ColumnArgumentRole, column)
1969 self.type = self.column.type
1971 __visit_name__ = "dmltargetcopy"
1973 _traverse_internals: _TraverseInternalsType = [
1974 ("column", InternalTraversal.dp_clauseelement),
1975 ]
1978class BindParameter(roles.InElementRole, KeyedColumnElement[_T]):
1979 r"""Represent a "bound expression".
1981 :class:`.BindParameter` is invoked explicitly using the
1982 :func:`.bindparam` function, as in::
1984 from sqlalchemy import bindparam
1986 stmt = select(users_table).where(
1987 users_table.c.name == bindparam("username")
1988 )
1990 Detailed discussion of how :class:`.BindParameter` is used is
1991 at :func:`.bindparam`.
1993 .. seealso::
1995 :func:`.bindparam`
1997 """
1999 __visit_name__ = "bindparam"
2001 _traverse_internals: _TraverseInternalsType = [
2002 ("key", InternalTraversal.dp_anon_name),
2003 ("type", InternalTraversal.dp_type),
2004 ("callable", InternalTraversal.dp_plain_dict),
2005 ("value", InternalTraversal.dp_plain_obj),
2006 ("literal_execute", InternalTraversal.dp_boolean),
2007 ]
2009 key: str
2010 _anon_map_key: Optional[str] = None
2011 type: TypeEngine[_T]
2012 value: Optional[_T]
2014 _is_crud = False
2015 _is_bind_parameter = True
2017 # bindparam implements its own _gen_cache_key() method however
2018 # we check subclasses for this flag, else no cache key is generated
2019 inherit_cache = True
2021 def __init__(
2022 self,
2023 key: Optional[str],
2024 value: Any = _NoArg.NO_ARG,
2025 type_: Optional[_TypeEngineArgument[_T]] = None,
2026 unique: bool = False,
2027 required: Union[bool, Literal[_NoArg.NO_ARG]] = _NoArg.NO_ARG,
2028 quote: Optional[bool] = None,
2029 callable_: Optional[Callable[[], Any]] = None,
2030 expanding: bool = False,
2031 isoutparam: bool = False,
2032 literal_execute: bool = False,
2033 _compared_to_operator: Optional[OperatorType] = None,
2034 _compared_to_type: Optional[TypeEngine[Any]] = None,
2035 _is_crud: bool = False,
2036 ):
2037 if required is _NoArg.NO_ARG:
2038 required = value is _NoArg.NO_ARG and callable_ is None
2039 if value is _NoArg.NO_ARG:
2040 value = None
2042 if quote is not None:
2043 key = quoted_name.construct(key, quote)
2045 if unique:
2046 self.key, self._anon_map_key = (
2047 _anonymous_label.safe_construct_with_key(
2048 id(self),
2049 (
2050 key
2051 if key is not None
2052 and not isinstance(key, _anonymous_label)
2053 else "param"
2054 ),
2055 sanitize_key=True,
2056 )
2057 )
2058 elif key:
2059 self.key = key
2060 else:
2061 self.key, self._anon_map_key = (
2062 _anonymous_label.safe_construct_with_key(id(self), "param")
2063 )
2065 # identifying key that won't change across
2066 # clones, used to identify the bind's logical
2067 # identity
2068 self._identifying_key = self.key
2070 # key that was passed in the first place, used to
2071 # generate new keys
2072 self._orig_key = key or "param"
2074 self.unique = unique
2075 self.value = value
2076 self.callable = callable_
2077 self.isoutparam = isoutparam
2078 self.required = required
2080 # indicate an "expanding" parameter; the compiler sets this
2081 # automatically in the compiler _render_in_expr_w_bindparam method
2082 # for an IN expression
2083 self.expanding = expanding
2085 # this is another hint to help w/ expanding and is typically
2086 # set in the compiler _render_in_expr_w_bindparam method for an
2087 # IN expression
2088 self.expand_op = None
2090 self.literal_execute = literal_execute
2091 if _is_crud:
2092 self._is_crud = True
2094 if type_ is None:
2095 if expanding:
2096 if value:
2097 check_value = value[0]
2098 else:
2099 check_value = type_api._NO_VALUE_IN_LIST
2100 else:
2101 check_value = value
2102 if _compared_to_type is not None:
2103 self.type = _compared_to_type.coerce_compared_value(
2104 _compared_to_operator, check_value
2105 )
2106 else:
2107 self.type = type_api._resolve_value_to_type(check_value)
2108 elif isinstance(type_, type):
2109 self.type = type_()
2110 elif is_tuple_type(type_):
2111 if value:
2112 if expanding:
2113 check_value = value[0]
2114 else:
2115 check_value = value
2116 cast("BindParameter[TupleAny]", self).type = (
2117 type_._resolve_values_to_types(check_value)
2118 )
2119 else:
2120 cast("BindParameter[TupleAny]", self).type = type_
2121 else:
2122 self.type = type_
2124 def _with_value(self, value, maintain_key=False, required=NO_ARG):
2125 """Return a copy of this :class:`.BindParameter` with the given value
2126 set.
2127 """
2128 cloned = self._clone(maintain_key=maintain_key)
2129 cloned.value = value
2130 cloned.callable = None
2131 cloned.required = required if required is not NO_ARG else self.required
2132 if cloned.type is type_api.NULLTYPE:
2133 cloned.type = type_api._resolve_value_to_type(value)
2134 return cloned
2136 @property
2137 def effective_value(self) -> Optional[_T]:
2138 """Return the value of this bound parameter,
2139 taking into account if the ``callable`` parameter
2140 was set.
2142 The ``callable`` value will be evaluated
2143 and returned if present, else ``value``.
2145 """
2146 if self.callable:
2147 # TODO: set up protocol for bind parameter callable
2148 return self.callable() # type: ignore
2149 else:
2150 return self.value
2152 def render_literal_execute(self) -> Self:
2153 """Produce a copy of this bound parameter that will enable the
2154 :paramref:`_sql.BindParameter.literal_execute` flag.
2156 The :paramref:`_sql.BindParameter.literal_execute` flag will
2157 have the effect of the parameter rendered in the compiled SQL
2158 string using ``[POSTCOMPILE]`` form, which is a special form that
2159 is converted to be a rendering of the literal value of the parameter
2160 at SQL execution time. The rationale is to support caching
2161 of SQL statement strings that can embed per-statement literal values,
2162 such as LIMIT and OFFSET parameters, in the final SQL string that
2163 is passed to the DBAPI. Dialects in particular may want to use
2164 this method within custom compilation schemes.
2166 .. versionadded:: 1.4.5
2168 .. seealso::
2170 :ref:`engine_thirdparty_caching`
2172 """
2173 c: Self = ClauseElement._clone(self)
2174 c.literal_execute = True
2175 return c
2177 def _negate_in_binary(self, negated_op, original_op):
2178 if self.expand_op is original_op:
2179 bind = self._clone()
2180 bind.expand_op = negated_op
2181 return bind
2182 else:
2183 return self
2185 def _with_binary_element_type(self, type_: TypeEngine[Any]) -> Self:
2186 c: Self = ClauseElement._clone(self)
2187 c.type = type_
2188 return c
2190 def _clone(self, maintain_key: bool = False, **kw: Any) -> Self:
2191 c: Self = ClauseElement._clone(self, **kw)
2192 # ensure all the BindParameter objects stay in cloned set.
2193 # in #7823, we changed "clone" so that a clone only keeps a reference
2194 # to the "original" element, since for column correspondence, that's
2195 # all we need. However, for BindParam, _cloned_set is used by
2196 # the "cache key bind match" lookup, which means if any of those
2197 # interim BindParameter objects became part of a cache key in the
2198 # cache, we need it. So here, make sure all clones keep carrying
2199 # forward.
2200 c._cloned_set.update(self._cloned_set)
2201 if not maintain_key and self.unique:
2202 c.key, c._anon_map_key = _anonymous_label.safe_construct_with_key(
2203 id(c), c._orig_key or "param", sanitize_key=True
2204 )
2205 return c
2207 def _gen_cache_key(self, anon_map, bindparams):
2208 _gen_cache_ok = self.__class__.__dict__.get("inherit_cache", False)
2210 if not _gen_cache_ok:
2211 if anon_map is not None:
2212 anon_map[NO_CACHE] = True
2213 return None
2215 id_, found = anon_map.get_anon(self)
2216 if found:
2217 return (id_, self.__class__)
2219 if bindparams is not None:
2220 bindparams.append(self)
2222 return (
2223 id_,
2224 self.__class__,
2225 self.type._static_cache_key,
2226 (
2227 anon_map[self._anon_map_key]
2228 if self._anon_map_key is not None
2229 else self.key
2230 ),
2231 self.literal_execute,
2232 )
2234 def _convert_to_unique(self):
2235 if not self.unique:
2236 self.unique = True
2237 self.key, self._anon_map_key = (
2238 _anonymous_label.safe_construct_with_key(
2239 id(self), self._orig_key or "param", sanitize_key=True
2240 )
2241 )
2243 def __getstate__(self):
2244 """execute a deferred value for serialization purposes."""
2246 d = self.__dict__.copy()
2247 v = self.value
2248 if self.callable:
2249 v = self.callable()
2250 d["callable"] = None
2251 d["value"] = v
2252 return d
2254 def __setstate__(self, state):
2255 if state.get("unique", False):
2256 anon_and_key = _anonymous_label.safe_construct_with_key(
2257 id(self), state.get("_orig_key", "param"), sanitize_key=True
2258 )
2259 state["key"], state["_anon_map_key"] = anon_and_key
2260 self.__dict__.update(state)
2262 def __repr__(self):
2263 return "%s(%r, %r, type_=%r)" % (
2264 self.__class__.__name__,
2265 self.key,
2266 self.value,
2267 self.type,
2268 )
2271class TypeClause(DQLDMLClauseElement):
2272 """Handle a type keyword in a SQL statement.
2274 Used by the ``Case`` statement.
2276 """
2278 __visit_name__ = "typeclause"
2280 _traverse_internals: _TraverseInternalsType = [
2281 ("type", InternalTraversal.dp_type)
2282 ]
2283 type: TypeEngine[Any]
2285 def __init__(self, type_: TypeEngine[Any]):
2286 self.type = type_
2289class TextClause(
2290 roles.DDLConstraintColumnRole,
2291 roles.DDLExpressionRole,
2292 roles.StatementOptionRole,
2293 roles.WhereHavingRole,
2294 roles.OrderByRole,
2295 roles.FromClauseRole,
2296 roles.SelectStatementRole,
2297 roles.InElementRole,
2298 Generative,
2299 Executable,
2300 DQLDMLClauseElement,
2301 roles.BinaryElementRole[Any],
2302 inspection.Inspectable["TextClause"],
2303):
2304 """Represent a literal SQL text fragment.
2306 E.g.::
2308 from sqlalchemy import text
2310 t = text("SELECT * FROM users")
2311 result = connection.execute(t)
2313 The :class:`_expression.TextClause` construct is produced using the
2314 :func:`_expression.text`
2315 function; see that function for full documentation.
2317 .. seealso::
2319 :func:`_expression.text`
2321 """
2323 __visit_name__ = "textclause"
2325 _traverse_internals: _TraverseInternalsType = [
2326 ("_bindparams", InternalTraversal.dp_string_clauseelement_dict),
2327 ("text", InternalTraversal.dp_string),
2328 ]
2330 _is_text_clause = True
2332 _is_textual = True
2334 _bind_params_regex = re.compile(r"(?<![:\w\x5c]):(\w+)(?!:)", re.UNICODE)
2335 _is_implicitly_boolean = False
2337 _render_label_in_columns_clause = False
2339 _omit_from_statements = False
2341 _is_collection_aggregate = False
2343 @property
2344 def _hide_froms(self) -> Iterable[FromClause]:
2345 return ()
2347 def __and__(self, other):
2348 # support use in select.where(), query.filter()
2349 return and_(self, other)
2351 @property
2352 def _select_iterable(self) -> _SelectIterable:
2353 return (self,)
2355 # help in those cases where text() is
2356 # interpreted in a column expression situation
2357 key: Optional[str] = None
2358 _label: Optional[str] = None
2360 _allow_label_resolve = False
2362 @property
2363 def _is_star(self): # type: ignore[override]
2364 return self.text == "*"
2366 def __init__(self, text: str):
2367 self._bindparams: Dict[str, BindParameter[Any]] = {}
2369 def repl(m):
2370 self._bindparams[m.group(1)] = BindParameter(m.group(1))
2371 return ":%s" % m.group(1)
2373 # scan the string and search for bind parameter names, add them
2374 # to the list of bindparams
2375 self.text = self._bind_params_regex.sub(repl, text)
2377 @_generative
2378 def bindparams(
2379 self,
2380 *binds: BindParameter[Any],
2381 **names_to_values: Any,
2382 ) -> Self:
2383 """Establish the values and/or types of bound parameters within
2384 this :class:`_expression.TextClause` construct.
2386 Given a text construct such as::
2388 from sqlalchemy import text
2390 stmt = text(
2391 "SELECT id, name FROM user WHERE name=:name AND timestamp=:timestamp"
2392 )
2394 the :meth:`_expression.TextClause.bindparams`
2395 method can be used to establish
2396 the initial value of ``:name`` and ``:timestamp``,
2397 using simple keyword arguments::
2399 stmt = stmt.bindparams(
2400 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2401 )
2403 Where above, new :class:`.BindParameter` objects
2404 will be generated with the names ``name`` and ``timestamp``, and
2405 values of ``jack`` and ``datetime.datetime(2012, 10, 8, 15, 12, 5)``,
2406 respectively. The types will be
2407 inferred from the values given, in this case :class:`.String` and
2408 :class:`.DateTime`.
2410 When specific typing behavior is needed, the positional ``*binds``
2411 argument can be used in which to specify :func:`.bindparam` constructs
2412 directly. These constructs must include at least the ``key``
2413 argument, then an optional value and type::
2415 from sqlalchemy import bindparam
2417 stmt = stmt.bindparams(
2418 bindparam("name", value="jack", type_=String),
2419 bindparam("timestamp", type_=DateTime),
2420 )
2422 Above, we specified the type of :class:`.DateTime` for the
2423 ``timestamp`` bind, and the type of :class:`.String` for the ``name``
2424 bind. In the case of ``name`` we also set the default value of
2425 ``"jack"``.
2427 Additional bound parameters can be supplied at statement execution
2428 time, e.g.::
2430 result = connection.execute(
2431 stmt, timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2432 )
2434 The :meth:`_expression.TextClause.bindparams`
2435 method can be called repeatedly,
2436 where it will re-use existing :class:`.BindParameter` objects to add
2437 new information. For example, we can call
2438 :meth:`_expression.TextClause.bindparams`
2439 first with typing information, and a
2440 second time with value information, and it will be combined::
2442 stmt = text(
2443 "SELECT id, name FROM user WHERE name=:name "
2444 "AND timestamp=:timestamp"
2445 )
2446 stmt = stmt.bindparams(
2447 bindparam("name", type_=String), bindparam("timestamp", type_=DateTime)
2448 )
2449 stmt = stmt.bindparams(
2450 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2451 )
2453 The :meth:`_expression.TextClause.bindparams`
2454 method also supports the concept of
2455 **unique** bound parameters. These are parameters that are
2456 "uniquified" on name at statement compilation time, so that multiple
2457 :func:`_expression.text`
2458 constructs may be combined together without the names
2459 conflicting. To use this feature, specify the
2460 :paramref:`.BindParameter.unique` flag on each :func:`.bindparam`
2461 object::
2463 stmt1 = text("select id from table where name=:name").bindparams(
2464 bindparam("name", value="name1", unique=True)
2465 )
2466 stmt2 = text("select id from table where name=:name").bindparams(
2467 bindparam("name", value="name2", unique=True)
2468 )
2470 union = union_all(stmt1.columns(column("id")), stmt2.columns(column("id")))
2472 The above statement will render as:
2474 .. sourcecode:: sql
2476 select id from table where name=:name_1
2477 UNION ALL select id from table where name=:name_2
2479 """ # noqa: E501
2480 self._bindparams = new_params = self._bindparams.copy()
2482 for bind in binds:
2483 try:
2484 # the regex used for text() currently will not match
2485 # a unique/anonymous key in any case, so use the _orig_key
2486 # so that a text() construct can support unique parameters
2487 existing = new_params[bind._orig_key]
2488 except KeyError as err:
2489 raise exc.ArgumentError(
2490 "This text() construct doesn't define a "
2491 "bound parameter named %r" % bind._orig_key
2492 ) from err
2493 else:
2494 new_params[existing._orig_key] = bind
2496 for key, value in names_to_values.items():
2497 try:
2498 existing = new_params[key]
2499 except KeyError as err:
2500 raise exc.ArgumentError(
2501 "This text() construct doesn't define a "
2502 "bound parameter named %r" % key
2503 ) from err
2504 else:
2505 new_params[key] = existing._with_value(value, required=False)
2506 return self
2508 @util.preload_module("sqlalchemy.sql.selectable")
2509 def columns(
2510 self,
2511 *cols: _ColumnExpressionArgument[Any],
2512 **types: _TypeEngineArgument[Any],
2513 ) -> TextualSelect:
2514 r"""Turn this :class:`_expression.TextClause` object into a
2515 :class:`_expression.TextualSelect`
2516 object that serves the same role as a SELECT
2517 statement.
2519 The :class:`_expression.TextualSelect` is part of the
2520 :class:`_expression.SelectBase`
2521 hierarchy and can be embedded into another statement by using the
2522 :meth:`_expression.TextualSelect.subquery` method to produce a
2523 :class:`.Subquery`
2524 object, which can then be SELECTed from.
2526 This function essentially bridges the gap between an entirely
2527 textual SELECT statement and the SQL expression language concept
2528 of a "selectable"::
2530 from sqlalchemy.sql import column, text
2532 stmt = text("SELECT id, name FROM some_table")
2533 stmt = stmt.columns(column("id"), column("name")).subquery("st")
2535 stmt = (
2536 select(mytable)
2537 .select_from(mytable.join(stmt, mytable.c.name == stmt.c.name))
2538 .where(stmt.c.id > 5)
2539 )
2541 Above, we pass a series of :func:`_expression.column` elements to the
2542 :meth:`_expression.TextClause.columns` method positionally. These
2543 :func:`_expression.column`
2544 elements now become first class elements upon the
2545 :attr:`_expression.TextualSelect.selected_columns` column collection,
2546 which then
2547 become part of the :attr:`.Subquery.c` collection after
2548 :meth:`_expression.TextualSelect.subquery` is invoked.
2550 The column expressions we pass to
2551 :meth:`_expression.TextClause.columns` may
2552 also be typed; when we do so, these :class:`.TypeEngine` objects become
2553 the effective return type of the column, so that SQLAlchemy's
2554 result-set-processing systems may be used on the return values.
2555 This is often needed for types such as date or boolean types, as well
2556 as for unicode processing on some dialect configurations::
2558 stmt = text("SELECT id, name, timestamp FROM some_table")
2559 stmt = stmt.columns(
2560 column("id", Integer),
2561 column("name", Unicode),
2562 column("timestamp", DateTime),
2563 )
2565 for id, name, timestamp in connection.execute(stmt):
2566 print(id, name, timestamp)
2568 As a shortcut to the above syntax, keyword arguments referring to
2569 types alone may be used, if only type conversion is needed::
2571 stmt = text("SELECT id, name, timestamp FROM some_table")
2572 stmt = stmt.columns(id=Integer, name=Unicode, timestamp=DateTime)
2574 for id, name, timestamp in connection.execute(stmt):
2575 print(id, name, timestamp)
2577 The positional form of :meth:`_expression.TextClause.columns`
2578 also provides the
2579 unique feature of **positional column targeting**, which is
2580 particularly useful when using the ORM with complex textual queries. If
2581 we specify the columns from our model to
2582 :meth:`_expression.TextClause.columns`,
2583 the result set will match to those columns positionally, meaning the
2584 name or origin of the column in the textual SQL doesn't matter::
2586 stmt = text(
2587 "SELECT users.id, addresses.id, users.id, "
2588 "users.name, addresses.email_address AS email "
2589 "FROM users JOIN addresses ON users.id=addresses.user_id "
2590 "WHERE users.id = 1"
2591 ).columns(
2592 User.id,
2593 Address.id,
2594 Address.user_id,
2595 User.name,
2596 Address.email_address,
2597 )
2599 query = (
2600 session.query(User)
2601 .from_statement(stmt)
2602 .options(contains_eager(User.addresses))
2603 )
2605 The :meth:`_expression.TextClause.columns` method provides a direct
2606 route to calling :meth:`_expression.FromClause.subquery` as well as
2607 :meth:`_expression.SelectBase.cte`
2608 against a textual SELECT statement::
2610 stmt = stmt.columns(id=Integer, name=String).cte("st")
2612 stmt = select(sometable).where(sometable.c.id == stmt.c.id)
2614 :param \*cols: A series of :class:`_expression.ColumnElement` objects,
2615 typically
2616 :class:`_schema.Column` objects from a :class:`_schema.Table`
2617 or ORM level
2618 column-mapped attributes, representing a set of columns that this
2619 textual string will SELECT from.
2621 :param \**types: A mapping of string names to :class:`.TypeEngine`
2622 type objects indicating the datatypes to use for names that are
2623 SELECTed from the textual string. Prefer to use the ``*cols``
2624 argument as it also indicates positional ordering.
2626 """
2627 selectable = util.preloaded.sql_selectable
2629 input_cols: List[NamedColumn[Any]] = [
2630 coercions.expect(roles.LabeledColumnExprRole, col) for col in cols
2631 ]
2633 positional_input_cols = [
2634 (
2635 ColumnClause(col.key, types.pop(col.key))
2636 if col.key in types
2637 else col
2638 )
2639 for col in input_cols
2640 ]
2641 keyed_input_cols: List[NamedColumn[Any]] = [
2642 ColumnClause(key, type_) for key, type_ in types.items()
2643 ]
2645 elem = selectable.TextualSelect.__new__(selectable.TextualSelect)
2646 elem._init(
2647 self,
2648 positional_input_cols + keyed_input_cols,
2649 positional=bool(positional_input_cols) and not keyed_input_cols,
2650 )
2651 return elem
2653 @property
2654 def type(self) -> TypeEngine[Any]:
2655 return type_api.NULLTYPE
2657 @property
2658 def comparator(self):
2659 # TODO: this seems wrong, it seems like we might not
2660 # be using this method.
2661 return self.type.comparator_factory(self) # type: ignore
2663 def self_group(
2664 self, against: Optional[OperatorType] = None
2665 ) -> Union[Self, Grouping[Any]]:
2666 if against is operators.in_op:
2667 return Grouping(self)
2668 else:
2669 return self
2672class Null(SingletonConstant, roles.ConstExprRole[None], ColumnElement[None]):
2673 """Represent the NULL keyword in a SQL statement.
2675 :class:`.Null` is accessed as a constant via the
2676 :func:`.null` function.
2678 """
2680 __visit_name__ = "null"
2682 _traverse_internals: _TraverseInternalsType = []
2683 _singleton: Null
2685 if not TYPE_CHECKING:
2687 @util.memoized_property
2688 def type(self) -> TypeEngine[_T]: # noqa: A001
2689 return type_api.NULLTYPE
2691 @classmethod
2692 def _instance(cls) -> Null:
2693 """Return a constant :class:`.Null` construct."""
2695 return Null._singleton
2698Null._create_singleton()
2701class False_(
2702 SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]
2703):
2704 """Represent the ``false`` keyword, or equivalent, in a SQL statement.
2706 :class:`.False_` is accessed as a constant via the
2707 :func:`.false` function.
2709 """
2711 __visit_name__ = "false"
2712 _traverse_internals: _TraverseInternalsType = []
2713 _singleton: False_
2715 if not TYPE_CHECKING:
2717 @util.memoized_property
2718 def type(self) -> TypeEngine[_T]: # noqa: A001
2719 return type_api.BOOLEANTYPE
2721 def _negate(self) -> True_:
2722 return True_._singleton
2724 @classmethod
2725 def _instance(cls) -> False_:
2726 return False_._singleton
2729False_._create_singleton()
2732class True_(SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]):
2733 """Represent the ``true`` keyword, or equivalent, in a SQL statement.
2735 :class:`.True_` is accessed as a constant via the
2736 :func:`.true` function.
2738 """
2740 __visit_name__ = "true"
2742 _traverse_internals: _TraverseInternalsType = []
2743 _singleton: True_
2745 if not TYPE_CHECKING:
2747 @util.memoized_property
2748 def type(self) -> TypeEngine[_T]: # noqa: A001
2749 return type_api.BOOLEANTYPE
2751 def _negate(self) -> False_:
2752 return False_._singleton
2754 @classmethod
2755 def _ifnone(
2756 cls, other: Optional[ColumnElement[Any]]
2757 ) -> ColumnElement[Any]:
2758 if other is None:
2759 return cls._instance()
2760 else:
2761 return other
2763 @classmethod
2764 def _instance(cls) -> True_:
2765 return True_._singleton
2768True_._create_singleton()
2771class ElementList(DQLDMLClauseElement):
2772 """Describe a list of clauses that will be space separated.
2774 This is a minimal version of :class:`.ClauseList` which is used by
2775 the :class:`.HasSyntaxExtension` class. It does not do any coercions
2776 so should be used internally only.
2778 .. versionadded:: 2.1
2780 """
2782 __visit_name__ = "element_list"
2784 _traverse_internals: _TraverseInternalsType = [
2785 ("clauses", InternalTraversal.dp_clauseelement_tuple),
2786 ]
2788 clauses: typing_Tuple[ClauseElement, ...]
2790 def __init__(self, clauses: Sequence[ClauseElement]):
2791 self.clauses = tuple(clauses)
2794class ClauseList(
2795 roles.InElementRole,
2796 roles.OrderByRole,
2797 roles.ColumnsClauseRole,
2798 roles.DMLColumnRole,
2799 DQLDMLClauseElement,
2800):
2801 """Describe a list of clauses, separated by an operator.
2803 By default, is comma-separated, such as a column listing.
2805 """
2807 __visit_name__ = "clauselist"
2809 # this is used only by the ORM in a legacy use case for
2810 # composite attributes
2811 _is_clause_list = True
2813 _traverse_internals: _TraverseInternalsType = [
2814 ("clauses", InternalTraversal.dp_clauseelement_list),
2815 ("operator", InternalTraversal.dp_operator),
2816 ]
2818 clauses: List[ColumnElement[Any]]
2820 def __init__(
2821 self,
2822 *clauses: _ColumnExpressionArgument[Any],
2823 operator: OperatorType = operators.comma_op,
2824 group: bool = True,
2825 group_contents: bool = True,
2826 _literal_as_text_role: Type[roles.SQLRole] = roles.WhereHavingRole,
2827 ):
2828 self.operator = operator
2829 self.group = group
2830 self.group_contents = group_contents
2831 clauses_iterator: Iterable[_ColumnExpressionArgument[Any]] = clauses
2832 text_converter_role: Type[roles.SQLRole] = _literal_as_text_role
2833 self._text_converter_role = text_converter_role
2835 if self.group_contents:
2836 self.clauses = [
2837 coercions.expect(
2838 text_converter_role, clause, apply_propagate_attrs=self
2839 ).self_group(against=self.operator)
2840 for clause in clauses_iterator
2841 ]
2842 else:
2843 self.clauses = [
2844 coercions.expect(
2845 text_converter_role, clause, apply_propagate_attrs=self
2846 )
2847 for clause in clauses_iterator
2848 ]
2849 self._is_implicitly_boolean = operators.is_boolean(self.operator)
2851 @classmethod
2852 def _construct_raw(
2853 cls,
2854 operator: OperatorType,
2855 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
2856 ) -> ClauseList:
2857 self = cls.__new__(cls)
2858 self.clauses = list(clauses) if clauses else []
2859 self.group = True
2860 self.operator = operator
2861 self.group_contents = True
2862 self._is_implicitly_boolean = False
2863 return self
2865 def __iter__(self) -> Iterator[ColumnElement[Any]]:
2866 return iter(self.clauses)
2868 def __len__(self) -> int:
2869 return len(self.clauses)
2871 @property
2872 def _select_iterable(self) -> _SelectIterable:
2873 return itertools.chain.from_iterable(
2874 [elem._select_iterable for elem in self.clauses]
2875 )
2877 def append(self, clause):
2878 if self.group_contents:
2879 self.clauses.append(
2880 coercions.expect(self._text_converter_role, clause).self_group(
2881 against=self.operator
2882 )
2883 )
2884 else:
2885 self.clauses.append(
2886 coercions.expect(self._text_converter_role, clause)
2887 )
2889 @util.ro_non_memoized_property
2890 def _from_objects(self) -> List[FromClause]:
2891 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
2893 def self_group(
2894 self, against: Optional[OperatorType] = None
2895 ) -> Union[Self, Grouping[Any]]:
2896 if self.group and operators.is_precedent(self.operator, against):
2897 return Grouping(self)
2898 else:
2899 return self
2902class OperatorExpression(ColumnElement[_T]):
2903 """base for expressions that contain an operator and operands
2905 .. versionadded:: 2.0
2907 """
2909 operator: OperatorType
2910 type: TypeEngine[_T]
2912 group: bool = True
2914 @property
2915 def is_comparison(self):
2916 return operators.is_comparison(self.operator)
2918 def self_group(
2919 self, against: Optional[OperatorType] = None
2920 ) -> Union[Self, Grouping[_T]]:
2921 if (
2922 self.group
2923 and operators.is_precedent(self.operator, against)
2924 or (
2925 # a negate against a non-boolean operator
2926 # doesn't make too much sense but we should
2927 # group for that
2928 against is operators.inv
2929 and not operators.is_boolean(self.operator)
2930 )
2931 ):
2932 return Grouping(self)
2933 else:
2934 return self
2936 @property
2937 def _flattened_operator_clauses(
2938 self,
2939 ) -> typing_Tuple[ColumnElement[Any], ...]:
2940 raise NotImplementedError()
2942 @classmethod
2943 def _construct_for_op(
2944 cls,
2945 left: ColumnElement[Any],
2946 right: ColumnElement[Any],
2947 op: OperatorType,
2948 *,
2949 type_: TypeEngine[_T],
2950 negate: Optional[OperatorType] = None,
2951 modifiers: Optional[Mapping[str, Any]] = None,
2952 ) -> OperatorExpression[_T]:
2953 if operators.is_associative(op):
2954 assert (
2955 negate is None
2956 ), f"negate not supported for associative operator {op}"
2958 multi = False
2959 if getattr(
2960 left, "operator", None
2961 ) is op and type_._compare_type_affinity(left.type):
2962 multi = True
2963 left_flattened = left._flattened_operator_clauses
2964 else:
2965 left_flattened = (left,)
2967 if getattr(
2968 right, "operator", None
2969 ) is op and type_._compare_type_affinity(right.type):
2970 multi = True
2971 right_flattened = right._flattened_operator_clauses
2972 else:
2973 right_flattened = (right,)
2975 if multi:
2976 return ExpressionClauseList._construct_for_list(
2977 op,
2978 type_,
2979 *(left_flattened + right_flattened),
2980 )
2982 if right._is_collection_aggregate:
2983 negate = None
2985 return BinaryExpression(
2986 left, right, op, type_=type_, negate=negate, modifiers=modifiers
2987 )
2990class ExpressionClauseList(OperatorExpression[_T]):
2991 """Describe a list of clauses, separated by an operator,
2992 in a column expression context.
2994 :class:`.ExpressionClauseList` differs from :class:`.ClauseList` in that
2995 it represents a column-oriented DQL expression only, not an open ended
2996 list of anything comma separated.
2998 .. versionadded:: 2.0
3000 """
3002 __visit_name__ = "expression_clauselist"
3004 _traverse_internals: _TraverseInternalsType = [
3005 ("clauses", InternalTraversal.dp_clauseelement_tuple),
3006 ("operator", InternalTraversal.dp_operator),
3007 ]
3009 clauses: typing_Tuple[ColumnElement[Any], ...]
3011 group: bool
3013 def __init__(
3014 self,
3015 operator: OperatorType,
3016 *clauses: _ColumnExpressionArgument[Any],
3017 type_: Optional[_TypeEngineArgument[_T]] = None,
3018 ):
3019 self.operator = operator
3021 self.clauses = tuple(
3022 coercions.expect(
3023 roles.ExpressionElementRole, clause, apply_propagate_attrs=self
3024 )
3025 for clause in clauses
3026 )
3027 self._is_implicitly_boolean = operators.is_boolean(self.operator)
3028 self.type = type_api.to_instance(type_) # type: ignore
3030 @property
3031 def _flattened_operator_clauses(
3032 self,
3033 ) -> typing_Tuple[ColumnElement[Any], ...]:
3034 return self.clauses
3036 def __iter__(self) -> Iterator[ColumnElement[Any]]:
3037 return iter(self.clauses)
3039 def __len__(self) -> int:
3040 return len(self.clauses)
3042 @property
3043 def _select_iterable(self) -> _SelectIterable:
3044 return (self,)
3046 @util.ro_non_memoized_property
3047 def _from_objects(self) -> List[FromClause]:
3048 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
3050 def _append_inplace(self, clause: ColumnElement[Any]) -> None:
3051 self.clauses += (clause,)
3053 @classmethod
3054 def _construct_for_list(
3055 cls,
3056 operator: OperatorType,
3057 type_: TypeEngine[_T],
3058 *clauses: ColumnElement[Any],
3059 group: bool = True,
3060 ) -> ExpressionClauseList[_T]:
3061 self = cls.__new__(cls)
3062 self.group = group
3063 if group:
3064 self.clauses = tuple(
3065 c.self_group(against=operator) for c in clauses
3066 )
3067 else:
3068 self.clauses = clauses
3069 self.operator = operator
3070 self.type = type_
3071 for c in clauses:
3072 if c._propagate_attrs:
3073 self._propagate_attrs = c._propagate_attrs
3074 break
3075 return self
3077 def _negate(self) -> Any:
3078 grouped = self.self_group(against=operators.inv)
3079 assert isinstance(grouped, ColumnElement)
3080 return UnaryExpression(grouped, operator=operators.inv)
3083class BooleanClauseList(ExpressionClauseList[bool]):
3084 __visit_name__ = "expression_clauselist"
3085 inherit_cache = True
3087 def __init__(self, *arg, **kw):
3088 raise NotImplementedError(
3089 "BooleanClauseList has a private constructor"
3090 )
3092 @classmethod
3093 def _process_clauses_for_boolean(
3094 cls,
3095 operator: OperatorType,
3096 continue_on: Any,
3097 skip_on: Any,
3098 clauses: Iterable[ColumnElement[Any]],
3099 ) -> typing_Tuple[int, List[ColumnElement[Any]]]:
3100 has_continue_on = None
3102 convert_clauses = []
3104 against = operators._asbool
3105 lcc = 0
3107 for clause in clauses:
3108 if clause is continue_on:
3109 # instance of continue_on, like and_(x, y, True, z), store it
3110 # if we didn't find one already, we will use it if there
3111 # are no other expressions here.
3112 has_continue_on = clause
3113 elif clause is skip_on:
3114 # instance of skip_on, e.g. and_(x, y, False, z), cancels
3115 # the rest out
3116 convert_clauses = [clause]
3117 lcc = 1
3118 break
3119 else:
3120 if not lcc:
3121 lcc = 1
3122 else:
3123 against = operator
3124 # technically this would be len(convert_clauses) + 1
3125 # however this only needs to indicate "greater than one"
3126 lcc = 2
3127 convert_clauses.append(clause)
3129 if not convert_clauses and has_continue_on is not None:
3130 convert_clauses = [has_continue_on]
3131 lcc = 1
3133 return lcc, [c.self_group(against=against) for c in convert_clauses]
3135 @classmethod
3136 def _construct(
3137 cls,
3138 operator: OperatorType,
3139 continue_on: Any,
3140 skip_on: Any,
3141 initial_clause: Any = _NoArg.NO_ARG,
3142 *clauses: Any,
3143 **kw: Any,
3144 ) -> ColumnElement[Any]:
3145 if initial_clause is _NoArg.NO_ARG:
3146 # no elements period. deprecated use case. return an empty
3147 # ClauseList construct that generates nothing unless it has
3148 # elements added to it.
3149 name = operator.__name__
3151 util.warn_deprecated(
3152 f"Invoking {name}() without arguments is deprecated, and "
3153 f"will be disallowed in a future release. For an empty "
3154 f"""{name}() construct, use '{name}({
3155 'true()' if continue_on is True_._singleton else 'false()'
3156 }, *args)' """
3157 f"""or '{name}({
3158 'True' if continue_on is True_._singleton else 'False'
3159 }, *args)'.""",
3160 version="1.4",
3161 )
3162 return cls._construct_raw(operator)
3164 lcc, convert_clauses = cls._process_clauses_for_boolean(
3165 operator,
3166 continue_on,
3167 skip_on,
3168 [
3169 coercions.expect(roles.WhereHavingRole, clause)
3170 for clause in util.coerce_generator_arg(
3171 (initial_clause,) + clauses
3172 )
3173 ],
3174 )
3176 if lcc > 1:
3177 # multiple elements. Return regular BooleanClauseList
3178 # which will link elements against the operator.
3180 flattened_clauses = itertools.chain.from_iterable(
3181 (
3182 (c for c in to_flat._flattened_operator_clauses)
3183 if getattr(to_flat, "operator", None) is operator
3184 else (to_flat,)
3185 )
3186 for to_flat in convert_clauses
3187 )
3189 return cls._construct_raw(operator, flattened_clauses) # type: ignore # noqa: E501
3190 else:
3191 assert lcc
3192 # just one element. return it as a single boolean element,
3193 # not a list and discard the operator.
3194 return convert_clauses[0]
3196 @classmethod
3197 def _construct_for_whereclause(
3198 cls, clauses: Iterable[ColumnElement[Any]]
3199 ) -> Optional[ColumnElement[bool]]:
3200 operator, continue_on, skip_on = (
3201 operators.and_,
3202 True_._singleton,
3203 False_._singleton,
3204 )
3206 lcc, convert_clauses = cls._process_clauses_for_boolean(
3207 operator,
3208 continue_on,
3209 skip_on,
3210 clauses, # these are assumed to be coerced already
3211 )
3213 if lcc > 1:
3214 # multiple elements. Return regular BooleanClauseList
3215 # which will link elements against the operator.
3216 return cls._construct_raw(operator, convert_clauses)
3217 elif lcc == 1:
3218 # just one element. return it as a single boolean element,
3219 # not a list and discard the operator.
3220 return convert_clauses[0]
3221 else:
3222 return None
3224 @classmethod
3225 def _construct_raw(
3226 cls,
3227 operator: OperatorType,
3228 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
3229 ) -> BooleanClauseList:
3230 self = cls.__new__(cls)
3231 self.clauses = tuple(clauses) if clauses else ()
3232 self.group = True
3233 self.operator = operator
3234 self.type = type_api.BOOLEANTYPE
3235 self._is_implicitly_boolean = True
3236 return self
3238 @classmethod
3239 def and_(
3240 cls,
3241 initial_clause: Union[
3242 Literal[True], _ColumnExpressionArgument[bool], _NoArg
3243 ] = _NoArg.NO_ARG,
3244 *clauses: _ColumnExpressionArgument[bool],
3245 ) -> ColumnElement[bool]:
3246 r"""Produce a conjunction of expressions joined by ``AND``.
3248 See :func:`_sql.and_` for full documentation.
3249 """
3250 return cls._construct(
3251 operators.and_,
3252 True_._singleton,
3253 False_._singleton,
3254 initial_clause,
3255 *clauses,
3256 )
3258 @classmethod
3259 def or_(
3260 cls,
3261 initial_clause: Union[
3262 Literal[False], _ColumnExpressionArgument[bool], _NoArg
3263 ] = _NoArg.NO_ARG,
3264 *clauses: _ColumnExpressionArgument[bool],
3265 ) -> ColumnElement[bool]:
3266 """Produce a conjunction of expressions joined by ``OR``.
3268 See :func:`_sql.or_` for full documentation.
3269 """
3270 return cls._construct(
3271 operators.or_,
3272 False_._singleton,
3273 True_._singleton,
3274 initial_clause,
3275 *clauses,
3276 )
3278 @property
3279 def _select_iterable(self) -> _SelectIterable:
3280 return (self,)
3282 def self_group(
3283 self, against: Optional[OperatorType] = None
3284 ) -> Union[Self, Grouping[bool]]:
3285 if not self.clauses:
3286 return self
3287 else:
3288 return super().self_group(against=against)
3291and_ = BooleanClauseList.and_
3292or_ = BooleanClauseList.or_
3295class Tuple(ClauseList, ColumnElement[TupleAny]):
3296 """Represent a SQL tuple."""
3298 __visit_name__ = "tuple"
3300 _traverse_internals: _TraverseInternalsType = (
3301 ClauseList._traverse_internals + []
3302 )
3304 type: TupleType
3306 @util.preload_module("sqlalchemy.sql.sqltypes")
3307 def __init__(
3308 self,
3309 *clauses: _ColumnExpressionArgument[Any],
3310 types: Optional[Sequence[_TypeEngineArgument[Any]]] = None,
3311 ):
3312 sqltypes = util.preloaded.sql_sqltypes
3314 if types is None:
3315 init_clauses: List[ColumnElement[Any]] = [
3316 coercions.expect(roles.ExpressionElementRole, c)
3317 for c in clauses
3318 ]
3319 else:
3320 if len(types) != len(clauses):
3321 raise exc.ArgumentError(
3322 "Wrong number of elements for %d-tuple: %r "
3323 % (len(types), clauses)
3324 )
3325 init_clauses = [
3326 coercions.expect(
3327 roles.ExpressionElementRole,
3328 c,
3329 type_=typ if not typ._isnull else None,
3330 )
3331 for typ, c in zip(types, clauses)
3332 ]
3334 self.type = sqltypes.TupleType(*[arg.type for arg in init_clauses])
3335 super().__init__(*init_clauses)
3337 @property
3338 def _select_iterable(self) -> _SelectIterable:
3339 return (self,)
3341 def _bind_param(self, operator, obj, type_=None, expanding=False):
3342 if expanding:
3343 return BindParameter(
3344 None,
3345 value=obj,
3346 _compared_to_operator=operator,
3347 unique=True,
3348 expanding=True,
3349 type_=type_,
3350 _compared_to_type=self.type,
3351 )
3352 else:
3353 return Tuple(
3354 *[
3355 BindParameter(
3356 None,
3357 o,
3358 _compared_to_operator=operator,
3359 _compared_to_type=compared_to_type,
3360 unique=True,
3361 type_=type_,
3362 )
3363 for o, compared_to_type in zip(obj, self.type.types)
3364 ]
3365 )
3367 def self_group(self, against: Optional[OperatorType] = None) -> Self:
3368 # Tuple is parenthesized by definition.
3369 return self
3372class Case(ColumnElement[_T]):
3373 """Represent a ``CASE`` expression.
3375 :class:`.Case` is produced using the :func:`.case` factory function,
3376 as in::
3378 from sqlalchemy import case
3380 stmt = select(users_table).where(
3381 case(
3382 (users_table.c.name == "wendy", "W"),
3383 (users_table.c.name == "jack", "J"),
3384 else_="E",
3385 )
3386 )
3388 Details on :class:`.Case` usage is at :func:`.case`.
3390 .. seealso::
3392 :func:`.case`
3394 """
3396 __visit_name__ = "case"
3398 _traverse_internals: _TraverseInternalsType = [
3399 ("value", InternalTraversal.dp_clauseelement),
3400 ("whens", InternalTraversal.dp_clauseelement_tuples),
3401 ("else_", InternalTraversal.dp_clauseelement),
3402 ]
3404 # for case(), the type is derived from the whens. so for the moment
3405 # users would have to cast() the case to get a specific type
3407 whens: List[typing_Tuple[ColumnElement[bool], ColumnElement[_T]]]
3408 else_: Optional[ColumnElement[_T]]
3409 value: Optional[ColumnElement[Any]]
3411 def __init__(
3412 self,
3413 *whens: Union[
3414 typing_Tuple[_ColumnExpressionArgument[bool], Any],
3415 Mapping[Any, Any],
3416 ],
3417 value: Optional[Any] = None,
3418 else_: Optional[Any] = None,
3419 ):
3420 new_whens: Iterable[Any] = coercions._expression_collection_was_a_list(
3421 "whens", "case", whens
3422 )
3423 try:
3424 new_whens = util.dictlike_iteritems(new_whens)
3425 except TypeError:
3426 pass
3428 self.whens = [
3429 (
3430 coercions.expect(
3431 roles.ExpressionElementRole,
3432 c,
3433 apply_propagate_attrs=self,
3434 ).self_group(),
3435 coercions.expect(roles.ExpressionElementRole, r),
3436 )
3437 for (c, r) in new_whens
3438 ]
3440 if value is None:
3441 self.value = None
3442 else:
3443 self.value = coercions.expect(roles.ExpressionElementRole, value)
3445 if else_ is not None:
3446 self.else_ = coercions.expect(roles.ExpressionElementRole, else_)
3447 else:
3448 self.else_ = None
3450 type_ = next(
3451 (
3452 then.type
3453 # Iterate `whens` in reverse to match previous behaviour
3454 # where type of final element took priority
3455 for *_, then in reversed(self.whens)
3456 if not then.type._isnull
3457 ),
3458 self.else_.type if self.else_ is not None else type_api.NULLTYPE,
3459 )
3460 self.type = cast(_T, type_)
3462 @util.ro_non_memoized_property
3463 def _from_objects(self) -> List[FromClause]:
3464 return list(
3465 itertools.chain(*[x._from_objects for x in self.get_children()])
3466 )
3469class Cast(WrapsColumnExpression[_T]):
3470 """Represent a ``CAST`` expression.
3472 :class:`.Cast` is produced using the :func:`.cast` factory function,
3473 as in::
3475 from sqlalchemy import cast, Numeric
3477 stmt = select(cast(product_table.c.unit_price, Numeric(10, 4)))
3479 Details on :class:`.Cast` usage is at :func:`.cast`.
3481 .. seealso::
3483 :ref:`tutorial_casts`
3485 :func:`.cast`
3487 :func:`.try_cast`
3489 :func:`.type_coerce` - an alternative to CAST that coerces the type
3490 on the Python side only, which is often sufficient to generate the
3491 correct SQL and data coercion.
3493 """
3495 __visit_name__ = "cast"
3497 _traverse_internals: _TraverseInternalsType = [
3498 ("clause", InternalTraversal.dp_clauseelement),
3499 ("type", InternalTraversal.dp_type),
3500 ]
3502 clause: ColumnElement[Any]
3503 type: TypeEngine[_T]
3504 typeclause: TypeClause
3506 def __init__(
3507 self,
3508 expression: _ColumnExpressionArgument[Any],
3509 type_: _TypeEngineArgument[_T],
3510 ):
3511 self.type = type_api.to_instance(type_)
3512 self.clause = coercions.expect(
3513 roles.ExpressionElementRole,
3514 expression,
3515 type_=self.type,
3516 apply_propagate_attrs=self,
3517 )
3518 self.typeclause = TypeClause(self.type)
3520 @util.ro_non_memoized_property
3521 def _from_objects(self) -> List[FromClause]:
3522 return self.clause._from_objects
3524 @property
3525 def wrapped_column_expression(self):
3526 return self.clause
3529class TryCast(Cast[_T]):
3530 """Represent a TRY_CAST expression.
3532 Details on :class:`.TryCast` usage is at :func:`.try_cast`.
3534 .. seealso::
3536 :func:`.try_cast`
3538 :ref:`tutorial_casts`
3539 """
3541 __visit_name__ = "try_cast"
3542 inherit_cache = True
3545class TypeCoerce(WrapsColumnExpression[_T]):
3546 """Represent a Python-side type-coercion wrapper.
3548 :class:`.TypeCoerce` supplies the :func:`_expression.type_coerce`
3549 function; see that function for usage details.
3551 .. seealso::
3553 :func:`_expression.type_coerce`
3555 :func:`.cast`
3557 """
3559 __visit_name__ = "type_coerce"
3561 _traverse_internals: _TraverseInternalsType = [
3562 ("clause", InternalTraversal.dp_clauseelement),
3563 ("type", InternalTraversal.dp_type),
3564 ]
3566 clause: ColumnElement[Any]
3567 type: TypeEngine[_T]
3569 def __init__(
3570 self,
3571 expression: _ColumnExpressionArgument[Any],
3572 type_: _TypeEngineArgument[_T],
3573 ):
3574 self.type = type_api.to_instance(type_)
3575 self.clause = coercions.expect(
3576 roles.ExpressionElementRole,
3577 expression,
3578 type_=self.type,
3579 apply_propagate_attrs=self,
3580 )
3582 @util.ro_non_memoized_property
3583 def _from_objects(self) -> List[FromClause]:
3584 return self.clause._from_objects
3586 @HasMemoized.memoized_attribute
3587 def typed_expression(self):
3588 if isinstance(self.clause, BindParameter):
3589 bp = self.clause._clone()
3590 bp.type = self.type
3591 return bp
3592 else:
3593 return self.clause
3595 @property
3596 def wrapped_column_expression(self):
3597 return self.clause
3599 def self_group(
3600 self, against: Optional[OperatorType] = None
3601 ) -> TypeCoerce[_T]:
3602 grouped = self.clause.self_group(against=against)
3603 if grouped is not self.clause:
3604 return TypeCoerce(grouped, self.type)
3605 else:
3606 return self
3609class Extract(ColumnElement[int]):
3610 """Represent a SQL EXTRACT clause, ``extract(field FROM expr)``."""
3612 __visit_name__ = "extract"
3614 _traverse_internals: _TraverseInternalsType = [
3615 ("expr", InternalTraversal.dp_clauseelement),
3616 ("field", InternalTraversal.dp_string),
3617 ]
3619 expr: ColumnElement[Any]
3620 field: str
3622 def __init__(self, field: str, expr: _ColumnExpressionArgument[Any]):
3623 self.type = type_api.INTEGERTYPE
3624 self.field = field
3625 self.expr = coercions.expect(roles.ExpressionElementRole, expr)
3627 @util.ro_non_memoized_property
3628 def _from_objects(self) -> List[FromClause]:
3629 return self.expr._from_objects
3632class _label_reference(ColumnElement[_T]):
3633 """Wrap a column expression as it appears in a 'reference' context.
3635 This expression is any that includes an _order_by_label_element,
3636 which is a Label, or a DESC / ASC construct wrapping a Label.
3638 The production of _label_reference() should occur when an expression
3639 is added to this context; this includes the ORDER BY or GROUP BY of a
3640 SELECT statement, as well as a few other places, such as the ORDER BY
3641 within an OVER clause.
3643 """
3645 __visit_name__ = "label_reference"
3647 _traverse_internals: _TraverseInternalsType = [
3648 ("element", InternalTraversal.dp_clauseelement)
3649 ]
3651 element: ColumnElement[_T]
3653 def __init__(self, element: ColumnElement[_T]):
3654 self.element = element
3655 self._propagate_attrs = element._propagate_attrs
3657 @util.ro_non_memoized_property
3658 def _from_objects(self) -> List[FromClause]:
3659 return []
3662class _textual_label_reference(ColumnElement[Any]):
3663 __visit_name__ = "textual_label_reference"
3665 _traverse_internals: _TraverseInternalsType = [
3666 ("element", InternalTraversal.dp_string)
3667 ]
3669 def __init__(self, element: str):
3670 self.element = element
3672 @util.memoized_property
3673 def _text_clause(self) -> TextClause:
3674 return TextClause(self.element)
3677class UnaryExpression(ColumnElement[_T]):
3678 """Define a 'unary' expression.
3680 A unary expression has a single column expression
3681 and an operator. The operator can be placed on the left
3682 (where it is called the 'operator') or right (where it is called the
3683 'modifier') of the column expression.
3685 :class:`.UnaryExpression` is the basis for several unary operators
3686 including those used by :func:`.desc`, :func:`.asc`, :func:`.distinct`,
3687 :func:`.nulls_first` and :func:`.nulls_last`.
3689 """
3691 __visit_name__ = "unary"
3693 _traverse_internals: _TraverseInternalsType = [
3694 ("element", InternalTraversal.dp_clauseelement),
3695 ("operator", InternalTraversal.dp_operator),
3696 ("modifier", InternalTraversal.dp_operator),
3697 ]
3699 element: ColumnElement[Any]
3700 operator: Optional[OperatorType]
3701 modifier: Optional[OperatorType]
3703 def __init__(
3704 self,
3705 element: ColumnElement[Any],
3706 *,
3707 operator: Optional[OperatorType] = None,
3708 modifier: Optional[OperatorType] = None,
3709 type_: Optional[_TypeEngineArgument[_T]] = None,
3710 wraps_column_expression: bool = False, # legacy, not used as of 2.0.42
3711 ):
3712 self.operator = operator
3713 self.modifier = modifier
3714 self._propagate_attrs = element._propagate_attrs
3715 self.element = element.self_group(
3716 against=self.operator or self.modifier
3717 )
3719 # if type is None, we get NULLTYPE, which is our _T. But I don't
3720 # know how to get the overloads to express that correctly
3721 self.type = type_api.to_instance(type_) # type: ignore
3723 def _wraps_unnamed_column(self):
3724 ungrouped = self.element._ungroup()
3725 return (
3726 not isinstance(ungrouped, NamedColumn)
3727 or ungrouped._non_anon_label is None
3728 )
3730 @classmethod
3731 def _create_nulls_first(
3732 cls,
3733 column: _ColumnExpressionArgument[_T],
3734 ) -> UnaryExpression[_T]:
3735 return UnaryExpression(
3736 coercions.expect(roles.ByOfRole, column),
3737 modifier=operators.nulls_first_op,
3738 )
3740 @classmethod
3741 def _create_nulls_last(
3742 cls,
3743 column: _ColumnExpressionArgument[_T],
3744 ) -> UnaryExpression[_T]:
3745 return UnaryExpression(
3746 coercions.expect(roles.ByOfRole, column),
3747 modifier=operators.nulls_last_op,
3748 )
3750 @classmethod
3751 def _create_desc(
3752 cls, column: _ColumnExpressionOrStrLabelArgument[_T]
3753 ) -> UnaryExpression[_T]:
3754 return UnaryExpression(
3755 coercions.expect(roles.ByOfRole, column),
3756 modifier=operators.desc_op,
3757 )
3759 @classmethod
3760 def _create_asc(
3761 cls,
3762 column: _ColumnExpressionOrStrLabelArgument[_T],
3763 ) -> UnaryExpression[_T]:
3764 return UnaryExpression(
3765 coercions.expect(roles.ByOfRole, column),
3766 modifier=operators.asc_op,
3767 )
3769 @classmethod
3770 def _create_distinct(
3771 cls,
3772 expr: _ColumnExpressionArgument[_T],
3773 ) -> UnaryExpression[_T]:
3774 col_expr: ColumnElement[_T] = coercions.expect(
3775 roles.ExpressionElementRole, expr
3776 )
3777 return UnaryExpression(
3778 col_expr,
3779 operator=operators.distinct_op,
3780 type_=col_expr.type,
3781 )
3783 @classmethod
3784 def _create_bitwise_not(
3785 cls,
3786 expr: _ColumnExpressionArgument[_T],
3787 ) -> UnaryExpression[_T]:
3788 col_expr: ColumnElement[_T] = coercions.expect(
3789 roles.ExpressionElementRole, expr
3790 )
3791 return UnaryExpression(
3792 col_expr,
3793 operator=operators.bitwise_not_op,
3794 type_=col_expr.type,
3795 )
3797 @property
3798 def _order_by_label_element(self) -> Optional[Label[Any]]:
3799 if operators.is_order_by_modifier(self.modifier):
3800 return self.element._order_by_label_element
3801 else:
3802 return None
3804 @util.ro_non_memoized_property
3805 def _from_objects(self) -> List[FromClause]:
3806 return self.element._from_objects
3808 def _negate(self) -> ColumnElement[Any]:
3809 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
3810 return UnaryExpression(
3811 self.self_group(against=operators.inv),
3812 operator=operators.inv,
3813 type_=type_api.BOOLEANTYPE,
3814 )
3815 else:
3816 return ColumnElement._negate(self)
3818 def self_group(
3819 self, against: Optional[OperatorType] = None
3820 ) -> Union[Self, Grouping[_T]]:
3821 if self.operator and operators.is_precedent(self.operator, against):
3822 return Grouping(self)
3823 else:
3824 return self
3827class CollectionAggregate(UnaryExpression[_T]):
3828 """Forms the basis for right-hand collection operator modifiers
3829 ANY and ALL.
3831 The ANY and ALL keywords are available in different ways on different
3832 backends. On PostgreSQL, they only work for an ARRAY type. On
3833 MySQL, they only work for subqueries.
3835 """
3837 inherit_cache = True
3838 _is_collection_aggregate = True
3840 @classmethod
3841 def _create_any(
3842 cls, expr: _ColumnExpressionArgument[_T]
3843 ) -> CollectionAggregate[bool]:
3844 col_expr: ColumnElement[_T] = coercions.expect(
3845 roles.ExpressionElementRole,
3846 expr,
3847 )
3848 col_expr = col_expr.self_group()
3849 return CollectionAggregate(
3850 col_expr,
3851 operator=operators.any_op,
3852 type_=type_api.BOOLEANTYPE,
3853 )
3855 @classmethod
3856 def _create_all(
3857 cls, expr: _ColumnExpressionArgument[_T]
3858 ) -> CollectionAggregate[bool]:
3859 col_expr: ColumnElement[_T] = coercions.expect(
3860 roles.ExpressionElementRole,
3861 expr,
3862 )
3863 col_expr = col_expr.self_group()
3864 return CollectionAggregate(
3865 col_expr,
3866 operator=operators.all_op,
3867 type_=type_api.BOOLEANTYPE,
3868 )
3870 # operate and reverse_operate are hardwired to
3871 # dispatch onto the type comparator directly, so that we can
3872 # ensure "reversed" behavior.
3873 def operate(
3874 self, op: OperatorType, *other: Any, **kwargs: Any
3875 ) -> ColumnElement[_T]:
3876 if not operators.is_comparison(op):
3877 raise exc.ArgumentError(
3878 "Only comparison operators may be used with ANY/ALL"
3879 )
3880 kwargs["reverse"] = True
3881 return self.comparator.operate(operators.mirror(op), *other, **kwargs)
3883 def reverse_operate(
3884 self, op: OperatorType, other: Any, **kwargs: Any
3885 ) -> ColumnElement[_T]:
3886 # comparison operators should never call reverse_operate
3887 assert not operators.is_comparison(op)
3888 raise exc.ArgumentError(
3889 "Only comparison operators may be used with ANY/ALL"
3890 )
3893class AsBoolean(WrapsColumnExpression[bool], UnaryExpression[bool]):
3894 inherit_cache = True
3896 def __init__(self, element, operator, negate):
3897 self.element = element
3898 self.type = type_api.BOOLEANTYPE
3899 self.operator = operator
3900 self.negate = negate
3901 self.modifier = None
3902 self._is_implicitly_boolean = element._is_implicitly_boolean
3904 @property
3905 def wrapped_column_expression(self):
3906 return self.element
3908 def self_group(self, against: Optional[OperatorType] = None) -> Self:
3909 return self
3911 def _negate(self):
3912 if isinstance(self.element, (True_, False_)):
3913 return self.element._negate()
3914 else:
3915 return AsBoolean(self.element, self.negate, self.operator)
3918class BinaryExpression(OperatorExpression[_T]):
3919 """Represent an expression that is ``LEFT <operator> RIGHT``.
3921 A :class:`.BinaryExpression` is generated automatically
3922 whenever two column expressions are used in a Python binary expression:
3924 .. sourcecode:: pycon+sql
3926 >>> from sqlalchemy.sql import column
3927 >>> column("a") + column("b")
3928 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
3929 >>> print(column("a") + column("b"))
3930 {printsql}a + b
3932 """
3934 __visit_name__ = "binary"
3936 _traverse_internals: _TraverseInternalsType = [
3937 ("left", InternalTraversal.dp_clauseelement),
3938 ("right", InternalTraversal.dp_clauseelement),
3939 ("operator", InternalTraversal.dp_operator),
3940 ("negate", InternalTraversal.dp_operator),
3941 ("modifiers", InternalTraversal.dp_plain_dict),
3942 (
3943 "type",
3944 InternalTraversal.dp_type,
3945 ),
3946 ]
3948 _cache_key_traversal = [
3949 ("left", InternalTraversal.dp_clauseelement),
3950 ("right", InternalTraversal.dp_clauseelement),
3951 ("operator", InternalTraversal.dp_operator),
3952 ("modifiers", InternalTraversal.dp_plain_dict),
3953 # "type" affects JSON CAST operators, so while redundant in most cases,
3954 # is needed for that one
3955 (
3956 "type",
3957 InternalTraversal.dp_type,
3958 ),
3959 ]
3961 _is_implicitly_boolean = True
3962 """Indicates that any database will know this is a boolean expression
3963 even if the database does not have an explicit boolean datatype.
3965 """
3967 left: ColumnElement[Any]
3968 right: ColumnElement[Any]
3969 modifiers: Mapping[str, Any]
3971 def __init__(
3972 self,
3973 left: ColumnElement[Any],
3974 right: ColumnElement[Any],
3975 operator: OperatorType,
3976 type_: Optional[_TypeEngineArgument[_T]] = None,
3977 negate: Optional[OperatorType] = None,
3978 modifiers: Optional[Mapping[str, Any]] = None,
3979 ):
3980 # allow compatibility with libraries that
3981 # refer to BinaryExpression directly and pass strings
3982 if isinstance(operator, str):
3983 operator = operators.custom_op(operator)
3984 self._orig = (left.__hash__(), right.__hash__())
3985 self._propagate_attrs = left._propagate_attrs or right._propagate_attrs
3986 self.left = left.self_group(against=operator)
3987 self.right = right.self_group(against=operator)
3988 self.operator = operator
3990 # if type is None, we get NULLTYPE, which is our _T. But I don't
3991 # know how to get the overloads to express that correctly
3992 self.type = type_api.to_instance(type_) # type: ignore
3994 self.negate = negate
3995 self._is_implicitly_boolean = operators.is_boolean(operator)
3997 if modifiers is None:
3998 self.modifiers = {}
3999 else:
4000 self.modifiers = modifiers
4002 @property
4003 def _flattened_operator_clauses(
4004 self,
4005 ) -> typing_Tuple[ColumnElement[Any], ...]:
4006 return (self.left, self.right)
4008 def __bool__(self):
4009 """Implement Python-side "bool" for BinaryExpression as a
4010 simple "identity" check for the left and right attributes,
4011 if the operator is "eq" or "ne". Otherwise the expression
4012 continues to not support "bool" like all other column expressions.
4014 The rationale here is so that ColumnElement objects can be hashable.
4015 What? Well, suppose you do this::
4017 c1, c2 = column("x"), column("y")
4018 s1 = set([c1, c2])
4020 We do that **a lot**, columns inside of sets is an extremely basic
4021 thing all over the ORM for example.
4023 So what happens if we do this? ::
4025 c1 in s1
4027 Hashing means it will normally use ``__hash__()`` of the object,
4028 but in case of hash collision, it's going to also do ``c1 == c1``
4029 and/or ``c1 == c2`` inside. Those operations need to return a
4030 True/False value. But because we override ``==`` and ``!=``, they're
4031 going to get a BinaryExpression. Hence we implement ``__bool__`` here
4032 so that these comparisons behave in this particular context mostly
4033 like regular object comparisons. Thankfully Python is OK with
4034 that! Otherwise we'd have to use special set classes for columns
4035 (which we used to do, decades ago).
4037 """
4038 if self.operator in (operators.eq, operators.ne):
4039 # this is using the eq/ne operator given int hash values,
4040 # rather than Operator, so that "bool" can be based on
4041 # identity
4042 return self.operator(*self._orig) # type: ignore
4043 else:
4044 raise TypeError("Boolean value of this clause is not defined")
4046 if typing.TYPE_CHECKING:
4048 def __invert__(
4049 self: BinaryExpression[_T],
4050 ) -> BinaryExpression[_T]: ...
4052 @util.ro_non_memoized_property
4053 def _from_objects(self) -> List[FromClause]:
4054 return self.left._from_objects + self.right._from_objects
4056 def _negate(self):
4057 if self.negate is not None:
4058 return BinaryExpression(
4059 self.left,
4060 self.right._negate_in_binary(self.negate, self.operator),
4061 self.negate,
4062 negate=self.operator,
4063 type_=self.type,
4064 modifiers=self.modifiers,
4065 )
4066 else:
4067 return self.self_group()._negate()
4070class Slice(ColumnElement[Any]):
4071 """Represent SQL for a Python array-slice object.
4073 This is not a specific SQL construct at this level, but
4074 may be interpreted by specific dialects, e.g. PostgreSQL.
4076 """
4078 __visit_name__ = "slice"
4080 _traverse_internals: _TraverseInternalsType = [
4081 ("start", InternalTraversal.dp_clauseelement),
4082 ("stop", InternalTraversal.dp_clauseelement),
4083 ("step", InternalTraversal.dp_clauseelement),
4084 ]
4086 def __init__(self, start, stop, step, _name=None):
4087 self.start = coercions.expect(
4088 roles.ExpressionElementRole,
4089 start,
4090 name=_name,
4091 type_=type_api.INTEGERTYPE,
4092 )
4093 self.stop = coercions.expect(
4094 roles.ExpressionElementRole,
4095 stop,
4096 name=_name,
4097 type_=type_api.INTEGERTYPE,
4098 )
4099 self.step = coercions.expect(
4100 roles.ExpressionElementRole,
4101 step,
4102 name=_name,
4103 type_=type_api.INTEGERTYPE,
4104 )
4105 self.type = type_api.NULLTYPE
4107 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4108 assert against is operator.getitem
4109 return self
4112class IndexExpression(BinaryExpression[Any]):
4113 """Represent the class of expressions that are like an "index"
4114 operation."""
4116 inherit_cache = True
4119class GroupedElement(DQLDMLClauseElement):
4120 """Represent any parenthesized expression"""
4122 __visit_name__ = "grouping"
4124 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4125 return self
4127 def _ungroup(self) -> ClauseElement:
4128 raise NotImplementedError()
4131class Grouping(GroupedElement, ColumnElement[_T]):
4132 """Represent a grouping within a column expression"""
4134 _traverse_internals: _TraverseInternalsType = [
4135 ("element", InternalTraversal.dp_clauseelement),
4136 ("type", InternalTraversal.dp_type),
4137 ]
4139 _cache_key_traversal = [
4140 ("element", InternalTraversal.dp_clauseelement),
4141 ]
4143 element: Union[TextClause, ClauseList, ColumnElement[_T]]
4145 def __init__(
4146 self, element: Union[TextClause, ClauseList, ColumnElement[_T]]
4147 ):
4148 self.element = element
4150 # nulltype assignment issue
4151 self.type = getattr(element, "type", type_api.NULLTYPE) # type: ignore
4152 self._propagate_attrs = element._propagate_attrs
4154 def _with_binary_element_type(self, type_):
4155 return self.__class__(self.element._with_binary_element_type(type_))
4157 def _ungroup(self) -> ColumnElement[_T]:
4158 assert isinstance(self.element, ColumnElement)
4159 return self.element._ungroup()
4161 @util.memoized_property
4162 def _is_implicitly_boolean(self):
4163 return self.element._is_implicitly_boolean
4165 @util.non_memoized_property
4166 def _tq_label(self) -> Optional[str]:
4167 return (
4168 getattr(self.element, "_tq_label", None) or self._anon_name_label
4169 )
4171 @util.non_memoized_property
4172 def _proxies(self) -> List[ColumnElement[Any]]:
4173 if isinstance(self.element, ColumnElement):
4174 return [self.element]
4175 else:
4176 return []
4178 @util.ro_non_memoized_property
4179 def _from_objects(self) -> List[FromClause]:
4180 return self.element._from_objects
4182 def __getattr__(self, attr):
4183 return getattr(self.element, attr)
4185 def __getstate__(self):
4186 return {"element": self.element, "type": self.type}
4188 def __setstate__(self, state):
4189 self.element = state["element"]
4190 self.type = state["type"]
4192 if TYPE_CHECKING:
4194 def self_group(
4195 self, against: Optional[OperatorType] = None
4196 ) -> Self: ...
4199class _OverrideBinds(Grouping[_T]):
4200 """used by cache_key->_apply_params_to_element to allow compilation /
4201 execution of a SQL element that's been cached, using an alternate set of
4202 bound parameter values.
4204 This is used by the ORM to swap new parameter values into expressions
4205 that are embedded into loader options like with_expression(),
4206 selectinload(). Previously, this task was accomplished using the
4207 .params() method which would perform a deep-copy instead. This deep
4208 copy proved to be too expensive for more complex expressions.
4210 See #11085
4212 """
4214 __visit_name__ = "override_binds"
4216 def __init__(
4217 self,
4218 element: ColumnElement[_T],
4219 bindparams: Sequence[BindParameter[Any]],
4220 replaces_params: Sequence[BindParameter[Any]],
4221 ):
4222 self.element = element
4223 self.translate = {
4224 k.key: v.value for k, v in zip(replaces_params, bindparams)
4225 }
4227 def _gen_cache_key(
4228 self, anon_map: anon_map, bindparams: List[BindParameter[Any]]
4229 ) -> Optional[typing_Tuple[Any, ...]]:
4230 """generate a cache key for the given element, substituting its bind
4231 values for the translation values present."""
4233 existing_bps: List[BindParameter[Any]] = []
4234 ck = self.element._gen_cache_key(anon_map, existing_bps)
4236 bindparams.extend(
4237 (
4238 bp._with_value(
4239 self.translate[bp.key], maintain_key=True, required=False
4240 )
4241 if bp.key in self.translate
4242 else bp
4243 )
4244 for bp in existing_bps
4245 )
4247 return ck
4250class Over(ColumnElement[_T]):
4251 """Represent an OVER clause.
4253 This is a special operator against a so-called
4254 "window" function, as well as any aggregate function,
4255 which produces results relative to the result set
4256 itself. Most modern SQL backends now support window functions.
4258 """
4260 __visit_name__ = "over"
4262 _traverse_internals: _TraverseInternalsType = [
4263 ("element", InternalTraversal.dp_clauseelement),
4264 ("order_by", InternalTraversal.dp_clauseelement),
4265 ("partition_by", InternalTraversal.dp_clauseelement),
4266 ("range_", InternalTraversal.dp_clauseelement),
4267 ("rows", InternalTraversal.dp_clauseelement),
4268 ("groups", InternalTraversal.dp_clauseelement),
4269 ]
4271 order_by: Optional[ClauseList] = None
4272 partition_by: Optional[ClauseList] = None
4274 element: ColumnElement[_T]
4275 """The underlying expression object to which this :class:`.Over`
4276 object refers."""
4278 range_: Optional[_FrameClause]
4279 rows: Optional[_FrameClause]
4280 groups: Optional[_FrameClause]
4282 def __init__(
4283 self,
4284 element: ColumnElement[_T],
4285 partition_by: Optional[_ByArgument] = None,
4286 order_by: Optional[_ByArgument] = None,
4287 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4288 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4289 groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4290 ):
4291 self.element = element
4292 if order_by is not None:
4293 self.order_by = ClauseList(
4294 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4295 )
4296 if partition_by is not None:
4297 self.partition_by = ClauseList(
4298 *util.to_list(partition_by),
4299 _literal_as_text_role=roles.ByOfRole,
4300 )
4302 if sum(bool(item) for item in (range_, rows, groups)) > 1:
4303 raise exc.ArgumentError(
4304 "only one of 'rows', 'range_', or 'groups' may be provided"
4305 )
4306 else:
4307 self.range_ = _FrameClause(range_) if range_ else None
4308 self.rows = _FrameClause(rows) if rows else None
4309 self.groups = _FrameClause(groups) if groups else None
4311 if not TYPE_CHECKING:
4313 @util.memoized_property
4314 def type(self) -> TypeEngine[_T]: # noqa: A001
4315 return self.element.type
4317 @util.ro_non_memoized_property
4318 def _from_objects(self) -> List[FromClause]:
4319 return list(
4320 itertools.chain(
4321 *[
4322 c._from_objects
4323 for c in (self.element, self.partition_by, self.order_by)
4324 if c is not None
4325 ]
4326 )
4327 )
4330class _FrameClauseType(Enum):
4331 RANGE_UNBOUNDED = 0
4332 RANGE_CURRENT = 1
4333 RANGE_PRECEDING = 2
4334 RANGE_FOLLOWING = 3
4337class _FrameClause(ClauseElement):
4338 """indicate the 'rows' or 'range' field of a window function, e.g. using
4339 :class:`.Over`.
4341 .. versionadded:: 2.1
4343 """
4345 __visit_name__ = "frame_clause"
4347 _traverse_internals: _TraverseInternalsType = [
4348 ("lower_integer_bind", InternalTraversal.dp_clauseelement),
4349 ("upper_integer_bind", InternalTraversal.dp_clauseelement),
4350 ("lower_type", InternalTraversal.dp_plain_obj),
4351 ("upper_type", InternalTraversal.dp_plain_obj),
4352 ]
4354 def __init__(
4355 self,
4356 range_: typing_Tuple[Optional[int], Optional[int]],
4357 ):
4358 try:
4359 r0, r1 = range_
4360 except (ValueError, TypeError) as ve:
4361 raise exc.ArgumentError("2-tuple expected for range/rows") from ve
4363 if r0 is None:
4364 self.lower_type = _FrameClauseType.RANGE_UNBOUNDED
4365 self.lower_integer_bind = None
4366 else:
4367 try:
4368 lower_integer = int(r0)
4369 except ValueError as err:
4370 raise exc.ArgumentError(
4371 "Integer or None expected for range value"
4372 ) from err
4373 else:
4374 if lower_integer == 0:
4375 self.lower_type = _FrameClauseType.RANGE_CURRENT
4376 self.lower_integer_bind = None
4377 elif lower_integer < 0:
4378 self.lower_type = _FrameClauseType.RANGE_PRECEDING
4379 self.lower_integer_bind = literal(
4380 abs(lower_integer), type_api.INTEGERTYPE
4381 )
4382 else:
4383 self.lower_type = _FrameClauseType.RANGE_FOLLOWING
4384 self.lower_integer_bind = literal(
4385 lower_integer, type_api.INTEGERTYPE
4386 )
4388 if r1 is None:
4389 self.upper_type = _FrameClauseType.RANGE_UNBOUNDED
4390 self.upper_integer_bind = None
4391 else:
4392 try:
4393 upper_integer = int(r1)
4394 except ValueError as err:
4395 raise exc.ArgumentError(
4396 "Integer or None expected for range value"
4397 ) from err
4398 else:
4399 if upper_integer == 0:
4400 self.upper_type = _FrameClauseType.RANGE_CURRENT
4401 self.upper_integer_bind = None
4402 elif upper_integer < 0:
4403 self.upper_type = _FrameClauseType.RANGE_PRECEDING
4404 self.upper_integer_bind = literal(
4405 abs(upper_integer), type_api.INTEGERTYPE
4406 )
4407 else:
4408 self.upper_type = _FrameClauseType.RANGE_FOLLOWING
4409 self.upper_integer_bind = literal(
4410 upper_integer, type_api.INTEGERTYPE
4411 )
4414class WithinGroup(ColumnElement[_T]):
4415 """Represent a WITHIN GROUP (ORDER BY) clause.
4417 This is a special operator against so-called
4418 "ordered set aggregate" and "hypothetical
4419 set aggregate" functions, including ``percentile_cont()``,
4420 ``rank()``, ``dense_rank()``, etc.
4422 It's supported only by certain database backends, such as PostgreSQL,
4423 Oracle Database and MS SQL Server.
4425 The :class:`.WithinGroup` construct extracts its type from the
4426 method :meth:`.FunctionElement.within_group_type`. If this returns
4427 ``None``, the function's ``.type`` is used.
4429 """
4431 __visit_name__ = "withingroup"
4433 _traverse_internals: _TraverseInternalsType = [
4434 ("element", InternalTraversal.dp_clauseelement),
4435 ("order_by", InternalTraversal.dp_clauseelement),
4436 ]
4438 order_by: Optional[ClauseList] = None
4440 def __init__(
4441 self,
4442 element: Union[FunctionElement[_T], FunctionFilter[_T]],
4443 *order_by: _ColumnExpressionArgument[Any],
4444 ):
4445 self.element = element
4446 if order_by is not None:
4447 self.order_by = ClauseList(
4448 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4449 )
4451 def __reduce__(self):
4452 return self.__class__, (self.element,) + (
4453 tuple(self.order_by) if self.order_by is not None else ()
4454 )
4456 def over(
4457 self,
4458 *,
4459 partition_by: Optional[_ByArgument] = None,
4460 order_by: Optional[_ByArgument] = None,
4461 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4462 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4463 groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4464 ) -> Over[_T]:
4465 """Produce an OVER clause against this :class:`.WithinGroup`
4466 construct.
4468 This function has the same signature as that of
4469 :meth:`.FunctionElement.over`.
4471 """
4472 return Over(
4473 self,
4474 partition_by=partition_by,
4475 order_by=order_by,
4476 range_=range_,
4477 rows=rows,
4478 groups=groups,
4479 )
4481 @overload
4482 def filter(self) -> Self: ...
4484 @overload
4485 def filter(
4486 self,
4487 __criterion0: _ColumnExpressionArgument[bool],
4488 *criterion: _ColumnExpressionArgument[bool],
4489 ) -> FunctionFilter[_T]: ...
4491 def filter(
4492 self, *criterion: _ColumnExpressionArgument[bool]
4493 ) -> Union[Self, FunctionFilter[_T]]:
4494 """Produce a FILTER clause against this function."""
4495 if not criterion:
4496 return self
4497 return FunctionFilter(self, *criterion)
4499 if not TYPE_CHECKING:
4501 @util.memoized_property
4502 def type(self) -> TypeEngine[_T]: # noqa: A001
4503 wgt = self.element.within_group_type(self)
4504 if wgt is not None:
4505 return wgt
4506 else:
4507 return self.element.type
4509 @util.ro_non_memoized_property
4510 def _from_objects(self) -> List[FromClause]:
4511 return list(
4512 itertools.chain(
4513 *[
4514 c._from_objects
4515 for c in (self.element, self.order_by)
4516 if c is not None
4517 ]
4518 )
4519 )
4522class FunctionFilter(Generative, ColumnElement[_T]):
4523 """Represent a function FILTER clause.
4525 This is a special operator against aggregate and window functions,
4526 which controls which rows are passed to it.
4527 It's supported only by certain database backends.
4529 Invocation of :class:`.FunctionFilter` is via
4530 :meth:`.FunctionElement.filter`::
4532 func.count(1).filter(True)
4534 .. seealso::
4536 :meth:`.FunctionElement.filter`
4538 """
4540 __visit_name__ = "funcfilter"
4542 _traverse_internals: _TraverseInternalsType = [
4543 ("func", InternalTraversal.dp_clauseelement),
4544 ("criterion", InternalTraversal.dp_clauseelement),
4545 ]
4547 criterion: Optional[ColumnElement[bool]] = None
4549 def __init__(
4550 self,
4551 func: Union[FunctionElement[_T], WithinGroup[_T]],
4552 *criterion: _ColumnExpressionArgument[bool],
4553 ):
4554 self.func = func
4555 self.filter.non_generative(self, *criterion) # type: ignore
4557 @_generative
4558 def filter(self, *criterion: _ColumnExpressionArgument[bool]) -> Self:
4559 """Produce an additional FILTER against the function.
4561 This method adds additional criteria to the initial criteria
4562 set up by :meth:`.FunctionElement.filter`.
4564 Multiple criteria are joined together at SQL render time
4565 via ``AND``.
4568 """
4570 for crit in list(criterion):
4571 crit = coercions.expect(roles.WhereHavingRole, crit)
4573 if self.criterion is not None:
4574 self.criterion = self.criterion & crit
4575 else:
4576 self.criterion = crit
4578 return self
4580 def over(
4581 self,
4582 partition_by: Optional[
4583 Union[
4584 Iterable[_ColumnExpressionArgument[Any]],
4585 _ColumnExpressionArgument[Any],
4586 ]
4587 ] = None,
4588 order_by: Optional[
4589 Union[
4590 Iterable[_ColumnExpressionArgument[Any]],
4591 _ColumnExpressionArgument[Any],
4592 ]
4593 ] = None,
4594 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4595 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4596 groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4597 ) -> Over[_T]:
4598 """Produce an OVER clause against this filtered function.
4600 Used against aggregate or so-called "window" functions,
4601 for database backends that support window functions.
4603 The expression::
4605 func.rank().filter(MyClass.y > 5).over(order_by="x")
4607 is shorthand for::
4609 from sqlalchemy import over, funcfilter
4611 over(funcfilter(func.rank(), MyClass.y > 5), order_by="x")
4613 See :func:`_expression.over` for a full description.
4615 """
4616 return Over(
4617 self,
4618 partition_by=partition_by,
4619 order_by=order_by,
4620 range_=range_,
4621 rows=rows,
4622 groups=groups,
4623 )
4625 def within_group(
4626 self, *order_by: _ColumnExpressionArgument[Any]
4627 ) -> WithinGroup[_T]:
4628 """Produce a WITHIN GROUP (ORDER BY expr) clause against
4629 this function.
4630 """
4631 return WithinGroup(self, *order_by)
4633 def within_group_type(
4634 self, within_group: WithinGroup[_T]
4635 ) -> Optional[TypeEngine[_T]]:
4636 return None
4638 def self_group(
4639 self, against: Optional[OperatorType] = None
4640 ) -> Union[Self, Grouping[_T]]:
4641 if operators.is_precedent(operators.filter_op, against):
4642 return Grouping(self)
4643 else:
4644 return self
4646 if not TYPE_CHECKING:
4648 @util.memoized_property
4649 def type(self) -> TypeEngine[_T]: # noqa: A001
4650 return self.func.type
4652 @util.ro_non_memoized_property
4653 def _from_objects(self) -> List[FromClause]:
4654 return list(
4655 itertools.chain(
4656 *[
4657 c._from_objects
4658 for c in (self.func, self.criterion)
4659 if c is not None
4660 ]
4661 )
4662 )
4665class NamedColumn(KeyedColumnElement[_T]):
4666 is_literal = False
4667 table: Optional[FromClause] = None
4668 name: str
4669 key: str
4671 def _compare_name_for_result(self, other):
4672 return (hasattr(other, "name") and self.name == other.name) or (
4673 hasattr(other, "_label") and self._label == other._label
4674 )
4676 @util.ro_memoized_property
4677 def description(self) -> str:
4678 return self.name
4680 @HasMemoized.memoized_attribute
4681 def _tq_key_label(self) -> Optional[str]:
4682 """table qualified label based on column key.
4684 for table-bound columns this is <tablename>_<column key/proxy key>;
4686 all other expressions it resolves to key/proxy key.
4688 """
4689 proxy_key = self._proxy_key
4690 if proxy_key and proxy_key != self.name:
4691 return self._gen_tq_label(proxy_key)
4692 else:
4693 return self._tq_label
4695 @HasMemoized.memoized_attribute
4696 def _tq_label(self) -> Optional[str]:
4697 """table qualified label based on column name.
4699 for table-bound columns this is <tablename>_<columnname>; all other
4700 expressions it resolves to .name.
4702 """
4703 return self._gen_tq_label(self.name)
4705 @HasMemoized.memoized_attribute
4706 def _render_label_in_columns_clause(self):
4707 return True
4709 @HasMemoized.memoized_attribute
4710 def _non_anon_label(self):
4711 return self.name
4713 def _gen_tq_label(
4714 self, name: str, dedupe_on_key: bool = True
4715 ) -> Optional[str]:
4716 return name
4718 def _bind_param(
4719 self,
4720 operator: OperatorType,
4721 obj: Any,
4722 type_: Optional[TypeEngine[_T]] = None,
4723 expanding: bool = False,
4724 ) -> BindParameter[_T]:
4725 return BindParameter(
4726 self.key,
4727 obj,
4728 _compared_to_operator=operator,
4729 _compared_to_type=self.type,
4730 type_=type_,
4731 unique=True,
4732 expanding=expanding,
4733 )
4735 def _make_proxy(
4736 self,
4737 selectable: FromClause,
4738 *,
4739 primary_key: ColumnSet,
4740 foreign_keys: Set[KeyedColumnElement[Any]],
4741 name: Optional[str] = None,
4742 key: Optional[str] = None,
4743 name_is_truncatable: bool = False,
4744 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
4745 disallow_is_literal: bool = False,
4746 **kw: Any,
4747 ) -> typing_Tuple[str, ColumnClause[_T]]:
4748 c = ColumnClause(
4749 (
4750 coercions.expect(roles.TruncatedLabelRole, name or self.name)
4751 if name_is_truncatable
4752 else (name or self.name)
4753 ),
4754 type_=self.type,
4755 _selectable=selectable,
4756 is_literal=False,
4757 )
4759 c._propagate_attrs = selectable._propagate_attrs
4760 if name is None:
4761 c.key = self.key
4762 if compound_select_cols:
4763 c._proxies = list(compound_select_cols)
4764 else:
4765 c._proxies = [self]
4767 if selectable._is_clone_of is not None:
4768 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
4769 return c.key, c
4772_PS = ParamSpec("_PS")
4775class Label(roles.LabeledColumnExprRole[_T], NamedColumn[_T]):
4776 """Represents a column label (AS).
4778 Represent a label, as typically applied to any column-level
4779 element using the ``AS`` sql keyword.
4781 """
4783 __visit_name__ = "label"
4785 _traverse_internals: _TraverseInternalsType = [
4786 ("name", InternalTraversal.dp_anon_name),
4787 ("type", InternalTraversal.dp_type),
4788 ("_element", InternalTraversal.dp_clauseelement),
4789 ]
4791 _cache_key_traversal = [
4792 ("name", InternalTraversal.dp_anon_name),
4793 ("_element", InternalTraversal.dp_clauseelement),
4794 ]
4796 _element: ColumnElement[_T]
4797 name: str
4799 def __init__(
4800 self,
4801 name: Optional[str],
4802 element: _ColumnExpressionArgument[_T],
4803 type_: Optional[_TypeEngineArgument[_T]] = None,
4804 ):
4805 orig_element = element
4806 element = coercions.expect(
4807 roles.ExpressionElementRole,
4808 element,
4809 apply_propagate_attrs=self,
4810 )
4811 while isinstance(element, Label):
4812 # TODO: this is only covered in test_text.py, but nothing
4813 # fails if it's removed. determine rationale
4814 element = element.element
4816 if name:
4817 self.name = name
4818 else:
4819 self.name = _anonymous_label.safe_construct(
4820 id(self), getattr(element, "name", "anon")
4821 )
4822 if isinstance(orig_element, Label):
4823 # TODO: no coverage for this block, again would be in
4824 # test_text.py where the resolve_label concept is important
4825 self._resolve_label = orig_element._label
4827 self.key = self._tq_label = self._tq_key_label = self.name
4828 self._element = element
4830 self.type = (
4831 type_api.to_instance(type_)
4832 if type_ is not None
4833 else self._element.type
4834 )
4836 self._proxies = [element]
4838 def __reduce__(self):
4839 return self.__class__, (self.name, self._element, self.type)
4841 @HasMemoized.memoized_attribute
4842 def _render_label_in_columns_clause(self):
4843 return True
4845 def _bind_param(self, operator, obj, type_=None, expanding=False):
4846 return BindParameter(
4847 None,
4848 obj,
4849 _compared_to_operator=operator,
4850 type_=type_,
4851 _compared_to_type=self.type,
4852 unique=True,
4853 expanding=expanding,
4854 )
4856 @util.memoized_property
4857 def _is_implicitly_boolean(self):
4858 return self.element._is_implicitly_boolean
4860 @HasMemoized.memoized_attribute
4861 def _allow_label_resolve(self):
4862 return self.element._allow_label_resolve
4864 @property
4865 def _order_by_label_element(self):
4866 return self
4868 def as_reference(self) -> _label_reference[_T]:
4869 """refer to this labeled expression in a clause such as GROUP BY,
4870 ORDER BY etc. as the label name itself, without expanding
4871 into the full expression.
4873 .. versionadded:: 2.1
4875 """
4876 return _label_reference(self)
4878 @HasMemoized.memoized_attribute
4879 def element(self) -> ColumnElement[_T]:
4880 return self._element.self_group(against=operators.as_)
4882 def self_group(self, against: Optional[OperatorType] = None) -> Label[_T]:
4883 return self._apply_to_inner(self._element.self_group, against=against)
4885 def _negate(self):
4886 return self._apply_to_inner(self._element._negate)
4888 def _apply_to_inner(
4889 self,
4890 fn: Callable[_PS, ColumnElement[_T]],
4891 *arg: _PS.args,
4892 **kw: _PS.kwargs,
4893 ) -> Label[_T]:
4894 sub_element = fn(*arg, **kw)
4895 if sub_element is not self._element:
4896 return Label(self.name, sub_element, type_=self.type)
4897 else:
4898 return self
4900 @property
4901 def primary_key(self): # type: ignore[override]
4902 return self.element.primary_key
4904 @property
4905 def foreign_keys(self): # type: ignore[override]
4906 return self.element.foreign_keys
4908 def _copy_internals(
4909 self,
4910 *,
4911 clone: _CloneCallableType = _clone,
4912 anonymize_labels: bool = False,
4913 **kw: Any,
4914 ) -> None:
4915 self._reset_memoizations()
4916 self._element = clone(self._element, **kw)
4917 if anonymize_labels:
4918 self.name = _anonymous_label.safe_construct(
4919 id(self), getattr(self.element, "name", "anon")
4920 )
4921 self.key = self._tq_label = self._tq_key_label = self.name
4923 @util.ro_non_memoized_property
4924 def _from_objects(self) -> List[FromClause]:
4925 return self.element._from_objects
4927 def _make_proxy(
4928 self,
4929 selectable: FromClause,
4930 *,
4931 primary_key: ColumnSet,
4932 foreign_keys: Set[KeyedColumnElement[Any]],
4933 name: Optional[str] = None,
4934 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
4935 **kw: Any,
4936 ) -> typing_Tuple[str, ColumnClause[_T]]:
4937 name = self.name if not name else name
4939 key, e = self.element._make_proxy(
4940 selectable,
4941 name=name,
4942 disallow_is_literal=True,
4943 name_is_truncatable=isinstance(name, _truncated_label),
4944 compound_select_cols=compound_select_cols,
4945 primary_key=primary_key,
4946 foreign_keys=foreign_keys,
4947 )
4949 # there was a note here to remove this assertion, which was here
4950 # to determine if we later could support a use case where
4951 # the key and name of a label are separate. But I don't know what
4952 # that case was. For now, this is an unexpected case that occurs
4953 # when a label name conflicts with other columns and select()
4954 # is attempting to disambiguate an explicit label, which is not what
4955 # the user would want. See issue #6090.
4956 if key != self.name and not isinstance(self.name, _anonymous_label):
4957 raise exc.InvalidRequestError(
4958 "Label name %s is being renamed to an anonymous label due "
4959 "to disambiguation "
4960 "which is not supported right now. Please use unique names "
4961 "for explicit labels." % (self.name)
4962 )
4964 e._propagate_attrs = selectable._propagate_attrs
4965 e._proxies.append(self)
4966 if self.type is not None:
4967 e.type = self.type
4969 return self.key, e
4972class ColumnClause(
4973 roles.DDLReferredColumnRole,
4974 roles.LabeledColumnExprRole[_T],
4975 roles.StrAsPlainColumnRole,
4976 Immutable,
4977 NamedColumn[_T],
4978):
4979 """Represents a column expression from any textual string.
4981 The :class:`.ColumnClause`, a lightweight analogue to the
4982 :class:`_schema.Column` class, is typically invoked using the
4983 :func:`_expression.column` function, as in::
4985 from sqlalchemy import column
4987 id, name = column("id"), column("name")
4988 stmt = select(id, name).select_from("user")
4990 The above statement would produce SQL like:
4992 .. sourcecode:: sql
4994 SELECT id, name FROM user
4996 :class:`.ColumnClause` is the immediate superclass of the schema-specific
4997 :class:`_schema.Column` object. While the :class:`_schema.Column`
4998 class has all the
4999 same capabilities as :class:`.ColumnClause`, the :class:`.ColumnClause`
5000 class is usable by itself in those cases where behavioral requirements
5001 are limited to simple SQL expression generation. The object has none of
5002 the associations with schema-level metadata or with execution-time
5003 behavior that :class:`_schema.Column` does,
5004 so in that sense is a "lightweight"
5005 version of :class:`_schema.Column`.
5007 Full details on :class:`.ColumnClause` usage is at
5008 :func:`_expression.column`.
5010 .. seealso::
5012 :func:`_expression.column`
5014 :class:`_schema.Column`
5016 """
5018 table: Optional[FromClause]
5019 is_literal: bool
5021 __visit_name__ = "column"
5023 _traverse_internals: _TraverseInternalsType = [
5024 ("name", InternalTraversal.dp_anon_name),
5025 ("type", InternalTraversal.dp_type),
5026 ("table", InternalTraversal.dp_clauseelement),
5027 ("is_literal", InternalTraversal.dp_boolean),
5028 ]
5030 onupdate: Optional[DefaultGenerator] = None
5031 default: Optional[DefaultGenerator] = None
5032 server_default: Optional[FetchedValue] = None
5033 server_onupdate: Optional[FetchedValue] = None
5035 _is_multiparam_column = False
5037 @property
5038 def _is_star(self): # type: ignore[override]
5039 return self.is_literal and self.name == "*"
5041 def __init__(
5042 self,
5043 text: str,
5044 type_: Optional[_TypeEngineArgument[_T]] = None,
5045 is_literal: bool = False,
5046 _selectable: Optional[FromClause] = None,
5047 ):
5048 self.key = self.name = text
5049 self.table = _selectable
5051 # if type is None, we get NULLTYPE, which is our _T. But I don't
5052 # know how to get the overloads to express that correctly
5053 self.type = type_api.to_instance(type_) # type: ignore
5055 self.is_literal = is_literal
5057 def get_children(self, *, column_tables=False, **kw):
5058 # override base get_children() to not return the Table
5059 # or selectable that is parent to this column. Traversals
5060 # expect the columns of tables and subqueries to be leaf nodes.
5061 return []
5063 @property
5064 def entity_namespace(self):
5065 if self.table is not None:
5066 return self.table.entity_namespace
5067 else:
5068 return super().entity_namespace
5070 def _clone(self, detect_subquery_cols=False, **kw):
5071 if (
5072 detect_subquery_cols
5073 and self.table is not None
5074 and self.table._is_subquery
5075 ):
5076 clone = kw.pop("clone")
5077 table = clone(self.table, **kw)
5078 new = table.c.corresponding_column(self)
5079 return new
5081 return super()._clone(**kw)
5083 @HasMemoized_ro_memoized_attribute
5084 def _from_objects(self) -> List[FromClause]:
5085 t = self.table
5086 if t is not None:
5087 return [t]
5088 else:
5089 return []
5091 @HasMemoized.memoized_attribute
5092 def _render_label_in_columns_clause(self):
5093 return self.table is not None
5095 @property
5096 def _ddl_label(self):
5097 return self._gen_tq_label(self.name, dedupe_on_key=False)
5099 def _compare_name_for_result(self, other):
5100 if (
5101 self.is_literal
5102 or self.table is None
5103 or self.table._is_textual
5104 or not hasattr(other, "proxy_set")
5105 or (
5106 isinstance(other, ColumnClause)
5107 and (
5108 other.is_literal
5109 or other.table is None
5110 or other.table._is_textual
5111 )
5112 )
5113 ):
5114 return (hasattr(other, "name") and self.name == other.name) or (
5115 hasattr(other, "_tq_label")
5116 and self._tq_label == other._tq_label
5117 )
5118 else:
5119 return other.proxy_set.intersection(self.proxy_set)
5121 def _gen_tq_label(
5122 self, name: str, dedupe_on_key: bool = True
5123 ) -> Optional[str]:
5124 """generate table-qualified label
5126 for a table-bound column this is <tablename>_<columnname>.
5128 used primarily for LABEL_STYLE_TABLENAME_PLUS_COL
5129 as well as the .columns collection on a Join object.
5131 """
5132 label: str
5133 t = self.table
5134 if self.is_literal:
5135 return None
5136 elif t is not None and is_named_from_clause(t):
5137 if has_schema_attr(t) and t.schema:
5138 label = (
5139 t.schema.replace(".", "_") + "_" + t.name + ("_" + name)
5140 )
5141 else:
5142 assert not TYPE_CHECKING or isinstance(t, NamedFromClause)
5143 label = t.name + ("_" + name)
5145 # propagate name quoting rules for labels.
5146 if is_quoted_name(name) and name.quote is not None:
5147 if is_quoted_name(label):
5148 label.quote = name.quote
5149 else:
5150 label = quoted_name(label, name.quote)
5151 elif is_quoted_name(t.name) and t.name.quote is not None:
5152 # can't get this situation to occur, so let's
5153 # assert false on it for now
5154 assert not isinstance(label, quoted_name)
5155 label = quoted_name(label, t.name.quote)
5157 if dedupe_on_key:
5158 # ensure the label name doesn't conflict with that of an
5159 # existing column. note that this implies that any Column
5160 # must **not** set up its _label before its parent table has
5161 # all of its other Column objects set up. There are several
5162 # tables in the test suite which will fail otherwise; example:
5163 # table "owner" has columns "name" and "owner_name". Therefore
5164 # column owner.name cannot use the label "owner_name", it has
5165 # to be "owner_name_1".
5166 if label in t.c:
5167 _label = label
5168 counter = 1
5169 while _label in t.c:
5170 _label = label + f"_{counter}"
5171 counter += 1
5172 label = _label
5174 return coercions.expect(roles.TruncatedLabelRole, label)
5176 else:
5177 return name
5179 def _make_proxy(
5180 self,
5181 selectable: FromClause,
5182 *,
5183 primary_key: ColumnSet,
5184 foreign_keys: Set[KeyedColumnElement[Any]],
5185 name: Optional[str] = None,
5186 key: Optional[str] = None,
5187 name_is_truncatable: bool = False,
5188 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5189 disallow_is_literal: bool = False,
5190 **kw: Any,
5191 ) -> typing_Tuple[str, ColumnClause[_T]]:
5192 # the "is_literal" flag normally should never be propagated; a proxied
5193 # column is always a SQL identifier and never the actual expression
5194 # being evaluated. however, there is a case where the "is_literal" flag
5195 # might be used to allow the given identifier to have a fixed quoting
5196 # pattern already, so maintain the flag for the proxy unless a
5197 # :class:`.Label` object is creating the proxy. See [ticket:4730].
5198 is_literal = (
5199 not disallow_is_literal
5200 and self.is_literal
5201 and (
5202 # note this does not accommodate for quoted_name differences
5203 # right now
5204 name is None
5205 or name == self.name
5206 )
5207 )
5208 c = self._constructor(
5209 (
5210 coercions.expect(roles.TruncatedLabelRole, name or self.name)
5211 if name_is_truncatable
5212 else (name or self.name)
5213 ),
5214 type_=self.type,
5215 _selectable=selectable,
5216 is_literal=is_literal,
5217 )
5218 c._propagate_attrs = selectable._propagate_attrs
5219 if name is None:
5220 c.key = self.key
5221 if compound_select_cols:
5222 c._proxies = list(compound_select_cols)
5223 else:
5224 c._proxies = [self]
5226 if selectable._is_clone_of is not None:
5227 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
5228 return c.key, c
5231class TableValuedColumn(NamedColumn[_T]):
5232 __visit_name__ = "table_valued_column"
5234 _traverse_internals: _TraverseInternalsType = [
5235 ("name", InternalTraversal.dp_anon_name),
5236 ("type", InternalTraversal.dp_type),
5237 ("scalar_alias", InternalTraversal.dp_clauseelement),
5238 ]
5240 def __init__(self, scalar_alias: NamedFromClause, type_: TypeEngine[_T]):
5241 self.scalar_alias = scalar_alias
5242 self.key = self.name = scalar_alias.name
5243 self.type = type_
5245 def _copy_internals(
5246 self, clone: _CloneCallableType = _clone, **kw: Any
5247 ) -> None:
5248 self.scalar_alias = clone(self.scalar_alias, **kw)
5249 self.key = self.name = self.scalar_alias.name
5251 @util.ro_non_memoized_property
5252 def _from_objects(self) -> List[FromClause]:
5253 return [self.scalar_alias]
5256class CollationClause(ColumnElement[str]):
5257 __visit_name__ = "collation"
5259 _traverse_internals: _TraverseInternalsType = [
5260 ("collation", InternalTraversal.dp_string)
5261 ]
5263 @classmethod
5264 @util.preload_module("sqlalchemy.sql.sqltypes")
5265 def _create_collation_expression(
5266 cls, expression: _ColumnExpressionArgument[str], collation: str
5267 ) -> BinaryExpression[str]:
5269 sqltypes = util.preloaded.sql_sqltypes
5271 expr = coercions.expect(roles.ExpressionElementRole[str], expression)
5273 if expr.type._type_affinity is sqltypes.String:
5274 collate_type = expr.type._with_collation(collation)
5275 else:
5276 collate_type = expr.type
5278 return BinaryExpression(
5279 expr,
5280 CollationClause(collation),
5281 operators.collate,
5282 type_=collate_type,
5283 )
5285 def __init__(self, collation):
5286 self.collation = collation
5289class _IdentifiedClause(Executable, ClauseElement):
5290 __visit_name__ = "identified"
5292 def __init__(self, ident):
5293 self.ident = ident
5296class SavepointClause(_IdentifiedClause):
5297 __visit_name__ = "savepoint"
5298 inherit_cache = False
5301class RollbackToSavepointClause(_IdentifiedClause):
5302 __visit_name__ = "rollback_to_savepoint"
5303 inherit_cache = False
5306class ReleaseSavepointClause(_IdentifiedClause):
5307 __visit_name__ = "release_savepoint"
5308 inherit_cache = False
5311class quoted_name(util.MemoizedSlots, str):
5312 """Represent a SQL identifier combined with quoting preferences.
5314 :class:`.quoted_name` is a Python unicode/str subclass which
5315 represents a particular identifier name along with a
5316 ``quote`` flag. This ``quote`` flag, when set to
5317 ``True`` or ``False``, overrides automatic quoting behavior
5318 for this identifier in order to either unconditionally quote
5319 or to not quote the name. If left at its default of ``None``,
5320 quoting behavior is applied to the identifier on a per-backend basis
5321 based on an examination of the token itself.
5323 A :class:`.quoted_name` object with ``quote=True`` is also
5324 prevented from being modified in the case of a so-called
5325 "name normalize" option. Certain database backends, such as
5326 Oracle Database, Firebird, and DB2 "normalize" case-insensitive names
5327 as uppercase. The SQLAlchemy dialects for these backends
5328 convert from SQLAlchemy's lower-case-means-insensitive convention
5329 to the upper-case-means-insensitive conventions of those backends.
5330 The ``quote=True`` flag here will prevent this conversion from occurring
5331 to support an identifier that's quoted as all lower case against
5332 such a backend.
5334 The :class:`.quoted_name` object is normally created automatically
5335 when specifying the name for key schema constructs such as
5336 :class:`_schema.Table`, :class:`_schema.Column`, and others.
5337 The class can also be
5338 passed explicitly as the name to any function that receives a name which
5339 can be quoted. Such as to use the :meth:`_engine.Engine.has_table`
5340 method with
5341 an unconditionally quoted name::
5343 from sqlalchemy import create_engine
5344 from sqlalchemy import inspect
5345 from sqlalchemy.sql import quoted_name
5347 engine = create_engine("oracle+oracledb://some_dsn")
5348 print(inspect(engine).has_table(quoted_name("some_table", True)))
5350 The above logic will run the "has table" logic against the Oracle Database
5351 backend, passing the name exactly as ``"some_table"`` without converting to
5352 upper case.
5354 """
5356 __slots__ = "quote", "lower", "upper"
5358 quote: Optional[bool]
5360 @overload
5361 @classmethod
5362 def construct(cls, value: str, quote: Optional[bool]) -> quoted_name: ...
5364 @overload
5365 @classmethod
5366 def construct(cls, value: None, quote: Optional[bool]) -> None: ...
5368 @classmethod
5369 def construct(
5370 cls, value: Optional[str], quote: Optional[bool]
5371 ) -> Optional[quoted_name]:
5372 if value is None:
5373 return None
5374 else:
5375 return quoted_name(value, quote)
5377 def __new__(cls, value: str, quote: Optional[bool]) -> quoted_name:
5378 assert (
5379 value is not None
5380 ), "use quoted_name.construct() for None passthrough"
5381 if isinstance(value, cls) and (quote is None or value.quote == quote):
5382 return value
5383 self = super().__new__(cls, value)
5385 self.quote = quote
5386 return self
5388 def __reduce__(self):
5389 return quoted_name, (str(self), self.quote)
5391 def _memoized_method_lower(self):
5392 if self.quote:
5393 return self
5394 else:
5395 return str(self).lower()
5397 def _memoized_method_upper(self):
5398 if self.quote:
5399 return self
5400 else:
5401 return str(self).upper()
5404def _find_columns(clause: ClauseElement) -> Set[ColumnClause[Any]]:
5405 """locate Column objects within the given expression."""
5407 cols: Set[ColumnClause[Any]] = set()
5408 traverse(clause, {}, {"column": cols.add})
5409 return cols
5412def _type_from_args(args: Sequence[ColumnElement[_T]]) -> TypeEngine[_T]:
5413 for a in args:
5414 if not a.type._isnull:
5415 return a.type
5416 else:
5417 return type_api.NULLTYPE # type: ignore
5420def _corresponding_column_or_error(fromclause, column, require_embedded=False):
5421 c = fromclause.corresponding_column(
5422 column, require_embedded=require_embedded
5423 )
5424 if c is None:
5425 raise exc.InvalidRequestError(
5426 "Given column '%s', attached to table '%s', "
5427 "failed to locate a corresponding column from table '%s'"
5428 % (column, getattr(column, "table", None), fromclause.description)
5429 )
5430 return c
5433class _memoized_property_but_not_nulltype(
5434 util.memoized_property["TypeEngine[_T]"]
5435):
5436 """memoized property, but dont memoize NullType"""
5438 def __get__(self, obj, cls):
5439 if obj is None:
5440 return self
5441 result = self.fget(obj)
5442 if not result._isnull:
5443 obj.__dict__[self.__name__] = result
5444 return result
5447class AnnotatedColumnElement(Annotated):
5448 _Annotated__element: ColumnElement[Any]
5450 def __init__(self, element, values):
5451 Annotated.__init__(self, element, values)
5452 for attr in (
5453 "comparator",
5454 "_proxy_key",
5455 "_tq_key_label",
5456 "_tq_label",
5457 "_non_anon_label",
5458 "type",
5459 ):
5460 self.__dict__.pop(attr, None)
5461 for attr in ("name", "key", "table"):
5462 if self.__dict__.get(attr, False) is None:
5463 self.__dict__.pop(attr)
5465 def _with_annotations(self, values):
5466 clone = super()._with_annotations(values)
5467 for attr in (
5468 "comparator",
5469 "_proxy_key",
5470 "_tq_key_label",
5471 "_tq_label",
5472 "_non_anon_label",
5473 ):
5474 clone.__dict__.pop(attr, None)
5475 return clone
5477 @util.memoized_property
5478 def name(self):
5479 """pull 'name' from parent, if not present"""
5480 return self._Annotated__element.name
5482 @_memoized_property_but_not_nulltype
5483 def type(self):
5484 """pull 'type' from parent and don't cache if null.
5486 type is routinely changed on existing columns within the
5487 mapped_column() initialization process, and "type" is also consulted
5488 during the creation of SQL expressions. Therefore it can change after
5489 it was already retrieved. At the same time we don't want annotated
5490 objects having overhead when expressions are produced, so continue
5491 to memoize, but only when we have a non-null type.
5493 """
5494 return self._Annotated__element.type
5496 @util.memoized_property
5497 def table(self):
5498 """pull 'table' from parent, if not present"""
5499 return self._Annotated__element.table
5501 @util.memoized_property
5502 def key(self):
5503 """pull 'key' from parent, if not present"""
5504 return self._Annotated__element.key
5506 @util.memoized_property
5507 def info(self) -> _InfoType:
5508 if TYPE_CHECKING:
5509 assert isinstance(self._Annotated__element, Column)
5510 return self._Annotated__element.info
5512 @util.memoized_property
5513 def _anon_name_label(self) -> str:
5514 return self._Annotated__element._anon_name_label
5517class _truncated_label(quoted_name):
5518 """A unicode subclass used to identify symbolic "
5519 "names that may require truncation."""
5521 __slots__ = ()
5523 def __new__(cls, value: str, quote: Optional[bool] = None) -> Any:
5524 quote = getattr(value, "quote", quote)
5525 # return super(_truncated_label, cls).__new__(cls, value, quote, True)
5526 return super().__new__(cls, value, quote)
5528 def __reduce__(self) -> Any:
5529 return self.__class__, (str(self), self.quote)
5531 def apply_map(self, map_: Mapping[str, Any]) -> str:
5532 return self
5535class conv(_truncated_label):
5536 """Mark a string indicating that a name has already been converted
5537 by a naming convention.
5539 This is a string subclass that indicates a name that should not be
5540 subject to any further naming conventions.
5542 E.g. when we create a :class:`.Constraint` using a naming convention
5543 as follows::
5545 m = MetaData(
5546 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5547 )
5548 t = Table(
5549 "t", m, Column("x", Integer), CheckConstraint("x > 5", name="x5")
5550 )
5552 The name of the above constraint will be rendered as ``"ck_t_x5"``.
5553 That is, the existing name ``x5`` is used in the naming convention as the
5554 ``constraint_name`` token.
5556 In some situations, such as in migration scripts, we may be rendering
5557 the above :class:`.CheckConstraint` with a name that's already been
5558 converted. In order to make sure the name isn't double-modified, the
5559 new name is applied using the :func:`_schema.conv` marker. We can
5560 use this explicitly as follows::
5563 m = MetaData(
5564 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5565 )
5566 t = Table(
5567 "t",
5568 m,
5569 Column("x", Integer),
5570 CheckConstraint("x > 5", name=conv("ck_t_x5")),
5571 )
5573 Where above, the :func:`_schema.conv` marker indicates that the constraint
5574 name here is final, and the name will render as ``"ck_t_x5"`` and not
5575 ``"ck_t_ck_t_x5"``
5577 .. seealso::
5579 :ref:`constraint_naming_conventions`
5581 """
5583 __slots__ = ()
5586# for backwards compatibility in case
5587# someone is re-implementing the
5588# _truncated_identifier() sequence in a custom
5589# compiler
5590_generated_label = _truncated_label
5591_anonymous_label_escape = re.compile(r"[%\(\) \$]+")
5594class _anonymous_label(_truncated_label):
5595 """A unicode subclass used to identify anonymously
5596 generated names."""
5598 __slots__ = ()
5600 @classmethod
5601 def safe_construct_with_key(
5602 cls, seed: int, body: str, sanitize_key: bool = False
5603 ) -> typing_Tuple[_anonymous_label, str]:
5604 # need to escape chars that interfere with format
5605 # strings in any case, issue #8724
5606 body = _anonymous_label_escape.sub("_", body)
5608 if sanitize_key:
5609 # sanitize_key is then an extra step used by BindParameter
5610 body = body.strip("_")
5612 key = f"{seed} {body.replace('%', '%%')}"
5613 label = _anonymous_label(f"%({key})s")
5614 return label, key
5616 @classmethod
5617 def safe_construct(
5618 cls, seed: int, body: str, sanitize_key: bool = False
5619 ) -> _anonymous_label:
5620 # need to escape chars that interfere with format
5621 # strings in any case, issue #8724
5622 body = _anonymous_label_escape.sub("_", body)
5624 if sanitize_key:
5625 # sanitize_key is then an extra step used by BindParameter
5626 body = body.strip("_")
5628 return _anonymous_label(f"%({seed} {body.replace('%', '%%')})s")
5630 def __add__(self, other: str) -> _anonymous_label:
5631 if "%" in other and not isinstance(other, _anonymous_label):
5632 other = str(other).replace("%", "%%")
5633 else:
5634 other = str(other)
5636 return _anonymous_label(
5637 quoted_name(
5638 str.__add__(self, other),
5639 self.quote,
5640 )
5641 )
5643 def __radd__(self, other: str) -> _anonymous_label:
5644 if "%" in other and not isinstance(other, _anonymous_label):
5645 other = str(other).replace("%", "%%")
5646 else:
5647 other = str(other)
5649 return _anonymous_label(
5650 quoted_name(
5651 str.__add__(other, self),
5652 self.quote,
5653 )
5654 )
5656 def apply_map(self, map_: Mapping[str, Any]) -> str:
5657 if self.quote is not None:
5658 # preserve quoting only if necessary
5659 return quoted_name(self % map_, self.quote)
5660 else:
5661 # else skip the constructor call
5662 return self % map_