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-2026 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 Literal
33from typing import Mapping
34from typing import Optional
35from typing import overload
36from typing import ParamSpec
37from typing import Sequence
38from typing import Set
39from typing import Tuple as typing_Tuple
40from typing import Type
41from typing import TYPE_CHECKING
42from typing import TypeVar
43from typing import Union
45from . import coercions
46from . import operators
47from . import roles
48from . import traversals
49from . import type_api
50from ._typing import has_schema_attr
51from ._typing import is_named_from_clause
52from ._typing import is_quoted_name
53from ._typing import is_tuple_type
54from .annotation import Annotated
55from .annotation import SupportsWrappingAnnotations
56from .base import _clone
57from .base import _expand_cloned
58from .base import _generative
59from .base import _NoArg
60from .base import Executable
61from .base import ExecutableStatement
62from .base import Generative
63from .base import HasMemoized
64from .base import Immutable
65from .base import NO_ARG
66from .base import SingletonConstant
67from .cache_key import MemoizedHasCacheKey
68from .cache_key import NO_CACHE
69from .coercions import _document_text_coercion # noqa
70from .operators import ColumnOperators
71from .operators import OperatorClass
72from .traversals import HasCopyInternals
73from .visitors import cloned_traverse
74from .visitors import ExternallyTraversible
75from .visitors import InternalTraversal
76from .visitors import traverse
77from .visitors import Visitable
78from .. import exc
79from .. import inspection
80from .. import util
81from ..util import deprecated
82from ..util import HasMemoized_ro_memoized_attribute
83from ..util import TypingOnly
84from ..util.compat import Template
85from ..util.typing import Self
86from ..util.typing import TupleAny
87from ..util.typing import Unpack
89if typing.TYPE_CHECKING:
90 from ._typing import _ByArgument
91 from ._typing import _ColumnExpressionArgument
92 from ._typing import _ColumnExpressionOrStrLabelArgument
93 from ._typing import _HasDialect
94 from ._typing import _InfoType
95 from ._typing import _OnlyColumnArgument
96 from ._typing import _PropagateAttrsType
97 from ._typing import _TypeEngineArgument
98 from .base import _EntityNamespace
99 from .base import ColumnSet
100 from .cache_key import _CacheKeyTraversalType
101 from .cache_key import CacheKey
102 from .compiler import Compiled
103 from .compiler import SQLCompiler
104 from .functions import FunctionElement
105 from .operators import OperatorType
106 from .schema import Column
107 from .schema import DefaultGenerator
108 from .schema import FetchedValue
109 from .schema import ForeignKey
110 from .selectable import _SelectIterable
111 from .selectable import FromClause
112 from .selectable import NamedFromClause
113 from .selectable import TextualSelect
114 from .sqltypes import TupleType
115 from .type_api import TypeEngine
116 from .visitors import _CloneCallableType
117 from .visitors import _TraverseInternalsType
118 from .visitors import anon_map
119 from ..engine import Connection
120 from ..engine import Dialect
121 from ..engine.interfaces import _CoreMultiExecuteParams
122 from ..engine.interfaces import _CoreSingleExecuteParams
123 from ..engine.interfaces import CacheStats
124 from ..engine.interfaces import CompiledCacheType
125 from ..engine.interfaces import CoreExecuteOptionsParameter
126 from ..engine.interfaces import SchemaTranslateMapType
127 from ..engine.result import Result
130_NUMERIC = Union[float, Decimal]
131_NUMBER = Union[float, int, Decimal]
133_T = TypeVar("_T", bound="Any")
134_T_co = TypeVar("_T_co", bound=Any, covariant=True)
135_OPT = TypeVar("_OPT", bound="Any")
136_NT = TypeVar("_NT", bound="_NUMERIC")
138_NMT = TypeVar("_NMT", bound="_NUMBER")
141@overload
142def literal(
143 value: Any,
144 type_: _TypeEngineArgument[_T],
145 literal_execute: bool = False,
146) -> BindParameter[_T]: ...
149@overload
150def literal(
151 value: _T,
152 type_: None = None,
153 literal_execute: bool = False,
154) -> BindParameter[_T]: ...
157@overload
158def literal(
159 value: Any,
160 type_: Optional[_TypeEngineArgument[Any]] = None,
161 literal_execute: bool = False,
162) -> BindParameter[Any]: ...
165def literal(
166 value: Any,
167 type_: Optional[_TypeEngineArgument[Any]] = None,
168 literal_execute: bool = False,
169) -> BindParameter[Any]:
170 r"""Return a literal clause, bound to a bind parameter.
172 Literal clauses are created automatically when non-
173 :class:`_expression.ClauseElement` objects (such as strings, ints, dates,
174 etc.) are
175 used in a comparison operation with a :class:`_expression.ColumnElement`
176 subclass,
177 such as a :class:`~sqlalchemy.schema.Column` object. Use this function
178 to force the generation of a literal clause, which will be created as a
179 :class:`BindParameter` with a bound value.
181 :param value: the value to be bound. Can be any Python object supported by
182 the underlying DB-API, or is translatable via the given type argument.
184 :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine` which will
185 provide bind-parameter translation for this literal.
187 :param literal_execute: optional bool, when True, the SQL engine will
188 attempt to render the bound value directly in the SQL statement at
189 execution time rather than providing as a parameter value.
191 .. versionadded:: 2.0
193 """
194 return coercions.expect(
195 roles.LiteralValueRole,
196 value,
197 type_=type_,
198 literal_execute=literal_execute,
199 )
202def literal_column(
203 text: str, type_: Optional[_TypeEngineArgument[_T]] = None
204) -> ColumnClause[_T]:
205 r"""Produce a :class:`.ColumnClause` object that has the
206 :paramref:`_expression.column.is_literal` flag set to True.
208 :func:`_expression.literal_column` is similar to
209 :func:`_expression.column`, except that
210 it is more often used as a "standalone" column expression that renders
211 exactly as stated; while :func:`_expression.column`
212 stores a string name that
213 will be assumed to be part of a table and may be quoted as such,
214 :func:`_expression.literal_column` can be that,
215 or any other arbitrary column-oriented
216 expression.
218 :param text: the text of the expression; can be any SQL expression.
219 Quoting rules will not be applied. To specify a column-name expression
220 which should be subject to quoting rules, use the :func:`column`
221 function.
223 :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine`
224 object which will
225 provide result-set translation and additional expression semantics for
226 this column. If left as ``None`` the type will be :class:`.NullType`.
228 .. seealso::
230 :func:`_expression.column`
232 :func:`_expression.text`
234 :ref:`tutorial_select_arbitrary_text`
236 """
237 return ColumnClause(text, type_=type_, is_literal=True)
240class CompilerElement(Visitable):
241 """base class for SQL elements that can be compiled to produce a
242 SQL string.
244 .. versionadded:: 2.0
246 """
248 __slots__ = ()
249 __visit_name__ = "compiler_element"
251 supports_execution = False
253 stringify_dialect = "default"
255 @util.preload_module("sqlalchemy.engine.default")
256 @util.preload_module("sqlalchemy.engine.url")
257 def compile(
258 self,
259 bind: Optional[_HasDialect] = None,
260 dialect: Optional[Dialect] = None,
261 **kw: Any,
262 ) -> Compiled:
263 """Compile this SQL expression.
265 The return value is a :class:`~.Compiled` object.
266 Calling ``str()`` or ``unicode()`` on the returned value will yield a
267 string representation of the result. The
268 :class:`~.Compiled` object also can return a
269 dictionary of bind parameter names and values
270 using the ``params`` accessor.
272 :param bind: An :class:`.Connection` or :class:`.Engine` which
273 can provide a :class:`.Dialect` in order to generate a
274 :class:`.Compiled` object. If the ``bind`` and
275 ``dialect`` parameters are both omitted, a default SQL compiler
276 is used.
278 :param column_keys: Used for INSERT and UPDATE statements, a list of
279 column names which should be present in the VALUES clause of the
280 compiled statement. If ``None``, all columns from the target table
281 object are rendered.
283 :param dialect: A :class:`.Dialect` instance which can generate
284 a :class:`.Compiled` object. This argument takes precedence over
285 the ``bind`` argument.
287 :param compile_kwargs: optional dictionary of additional parameters
288 that will be passed through to the compiler within all "visit"
289 methods. This allows any custom flag to be passed through to
290 a custom compilation construct, for example. It is also used
291 for the case of passing the ``literal_binds`` flag through::
293 from sqlalchemy.sql import table, column, select
295 t = table("t", column("x"))
297 s = select(t).where(t.c.x == 5)
299 print(s.compile(compile_kwargs={"literal_binds": True}))
301 .. seealso::
303 :ref:`faq_sql_expression_string`
305 """
307 if dialect is None:
308 if bind:
309 dialect = bind.dialect
310 elif self.stringify_dialect == "default":
311 dialect = self._default_dialect()
312 else:
313 url = util.preloaded.engine_url
314 dialect = url.URL.create(
315 self.stringify_dialect
316 ).get_dialect()()
318 return self._compiler(dialect, **kw)
320 def _default_dialect(self):
321 default = util.preloaded.engine_default
322 return default.StrCompileDialect()
324 def _compiler(self, dialect: Dialect, **kw: Any) -> Compiled:
325 """Return a compiler appropriate for this ClauseElement, given a
326 Dialect."""
328 if TYPE_CHECKING:
329 assert isinstance(self, ClauseElement)
330 return dialect.statement_compiler(dialect, self, **kw)
332 def __str__(self) -> str:
333 return str(self.compile())
336@inspection._self_inspects
337class ClauseElement(
338 SupportsWrappingAnnotations,
339 MemoizedHasCacheKey,
340 HasCopyInternals,
341 ExternallyTraversible,
342 CompilerElement,
343):
344 """Base class for elements of a programmatically constructed SQL
345 expression.
347 """
349 __visit_name__ = "clause"
351 if TYPE_CHECKING:
353 @util.memoized_property
354 def _propagate_attrs(self) -> _PropagateAttrsType:
355 """like annotations, however these propagate outwards liberally
356 as SQL constructs are built, and are set up at construction time.
358 """
359 ...
361 else:
362 _propagate_attrs = util.EMPTY_DICT
364 @util.ro_memoized_property
365 def description(self) -> Optional[str]:
366 return None
368 _is_clone_of: Optional[Self] = None
370 is_clause_element = True
371 is_selectable = False
372 is_dml = False
373 _is_column_element = False
374 _is_keyed_column_element = False
375 _is_table = False
376 _gen_static_annotations_cache_key = False
377 _is_textual = False
378 _is_from_clause = False
379 _is_returns_rows = False
380 _is_text_clause = False
381 _is_from_container = False
382 _is_select_container = False
383 _is_select_base = False
384 _is_select_statement = False
385 _is_bind_parameter = False
386 _is_clause_list = False
387 _is_lambda_element = False
388 _is_singleton_constant = False
389 _is_immutable = False
390 _is_star = False
392 @property
393 def _order_by_label_element(self) -> Optional[Label[Any]]:
394 return None
396 _cache_key_traversal: _CacheKeyTraversalType = None
398 negation_clause: ColumnElement[bool]
400 if typing.TYPE_CHECKING:
402 def get_children(
403 self, *, omit_attrs: typing_Tuple[str, ...] = ..., **kw: Any
404 ) -> Iterable[ClauseElement]: ...
406 @util.ro_non_memoized_property
407 def _from_objects(self) -> List[FromClause]:
408 return []
410 def _set_propagate_attrs(self, values: Mapping[str, Any]) -> Self:
411 # usually, self._propagate_attrs is empty here. one case where it's
412 # not is a subquery against ORM select, that is then pulled as a
413 # property of an aliased class. should all be good
415 # assert not self._propagate_attrs
417 self._propagate_attrs = util.immutabledict(values)
418 return self
420 def _default_compiler(self) -> SQLCompiler:
421 dialect = self._default_dialect()
422 return dialect.statement_compiler(dialect, self) # type: ignore[no-any-return] # noqa: E501
424 def _clone(self, **kw: Any) -> Self:
425 """Create a shallow copy of this ClauseElement.
427 This method may be used by a generative API. Its also used as
428 part of the "deep" copy afforded by a traversal that combines
429 the _copy_internals() method.
431 """
433 skip = self._memoized_keys
434 c = self.__class__.__new__(self.__class__)
436 if skip:
437 # ensure this iteration remains atomic
438 c.__dict__ = {
439 k: v for k, v in self.__dict__.copy().items() if k not in skip
440 }
441 else:
442 c.__dict__ = self.__dict__.copy()
444 # this is a marker that helps to "equate" clauses to each other
445 # when a Select returns its list of FROM clauses. the cloning
446 # process leaves around a lot of remnants of the previous clause
447 # typically in the form of column expressions still attached to the
448 # old table.
449 cc = self._is_clone_of
450 c._is_clone_of = cc if cc is not None else self
451 return c
453 def _negate_in_binary(self, negated_op, original_op):
454 """a hook to allow the right side of a binary expression to respond
455 to a negation of the binary expression.
457 Used for the special case of expanding bind parameter with IN.
459 """
460 return self
462 def _with_binary_element_type(self, type_):
463 """in the context of binary expression, convert the type of this
464 object to the one given.
466 applies only to :class:`_expression.ColumnElement` classes.
468 """
469 return self
471 @property
472 def _constructor(self): # type: ignore[override]
473 """return the 'constructor' for this ClauseElement.
475 This is for the purposes for creating a new object of
476 this type. Usually, its just the element's __class__.
477 However, the "Annotated" version of the object overrides
478 to return the class of its proxied element.
480 """
481 return self.__class__
483 @HasMemoized.memoized_attribute
484 def _cloned_set(self):
485 """Return the set consisting all cloned ancestors of this
486 ClauseElement.
488 Includes this ClauseElement. This accessor tends to be used for
489 FromClause objects to identify 'equivalent' FROM clauses, regardless
490 of transformative operations.
492 """
493 s = util.column_set()
494 f: Optional[ClauseElement] = self
496 # note this creates a cycle, asserted in test_memusage. however,
497 # turning this into a plain @property adds tends of thousands of method
498 # calls to Core / ORM performance tests, so the small overhead
499 # introduced by the relatively small amount of short term cycles
500 # produced here is preferable
501 while f is not None:
502 s.add(f)
503 f = f._is_clone_of
504 return s
506 def _de_clone(self):
507 while self._is_clone_of is not None:
508 self = self._is_clone_of
509 return self
511 @util.ro_non_memoized_property
512 def entity_namespace(self) -> _EntityNamespace:
513 raise AttributeError(
514 "This SQL expression has no entity namespace "
515 "with which to filter from."
516 )
518 def __getstate__(self):
519 d = self.__dict__.copy()
520 d.pop("_is_clone_of", None)
521 d.pop("_generate_cache_key", None)
522 return d
524 def _execute_on_connection(
525 self,
526 connection: Connection,
527 distilled_params: _CoreMultiExecuteParams,
528 execution_options: CoreExecuteOptionsParameter,
529 ) -> Result[Unpack[TupleAny]]:
530 if self.supports_execution:
531 if TYPE_CHECKING:
532 assert isinstance(self, Executable)
533 return connection._execute_clauseelement(
534 self, distilled_params, execution_options
535 )
536 else:
537 raise exc.ObjectNotExecutableError(self)
539 def _execute_on_scalar(
540 self,
541 connection: Connection,
542 distilled_params: _CoreMultiExecuteParams,
543 execution_options: CoreExecuteOptionsParameter,
544 ) -> Any:
545 """an additional hook for subclasses to provide a different
546 implementation for connection.scalar() vs. connection.execute().
548 .. versionadded:: 2.0
550 """
551 return self._execute_on_connection(
552 connection, distilled_params, execution_options
553 ).scalar()
555 def _get_embedded_bindparams(self) -> Sequence[BindParameter[Any]]:
556 """Return the list of :class:`.BindParameter` objects embedded in the
557 object.
559 This accomplishes the same purpose as ``visitors.traverse()`` or
560 similar would provide, however by making use of the cache key
561 it takes advantage of memoization of the key to result in fewer
562 net method calls, assuming the statement is also going to be
563 executed.
565 """
567 key = self._generate_cache_key()
568 if key is None:
569 bindparams: List[BindParameter[Any]] = []
571 traverse(self, {}, {"bindparam": bindparams.append})
572 return bindparams
574 else:
575 return key.bindparams
577 def unique_params(
578 self,
579 __optionaldict: Optional[Dict[str, Any]] = None,
580 /,
581 **kwargs: Any,
582 ) -> Self:
583 """Return a copy with :func:`_expression.bindparam` elements
584 replaced.
586 Same functionality as :meth:`_expression.ClauseElement.params`,
587 except adds `unique=True`
588 to affected bind parameters so that multiple statements can be
589 used.
591 """
592 return self._replace_params(True, __optionaldict, kwargs)
594 def params(
595 self,
596 __optionaldict: Optional[Mapping[str, Any]] = None,
597 /,
598 **kwargs: Any,
599 ) -> Self:
600 """Return a copy with :func:`_expression.bindparam` elements
601 replaced.
603 Returns a copy of this ClauseElement with
604 :func:`_expression.bindparam`
605 elements replaced with values taken from the given dictionary::
607 >>> clause = column("x") + bindparam("foo")
608 >>> print(clause.compile().params)
609 {'foo':None}
610 >>> print(clause.params({"foo": 7}).compile().params)
611 {'foo':7}
613 """
614 return self._replace_params(False, __optionaldict, kwargs)
616 @deprecated(
617 "2.1",
618 "The params() and unique_params() methods on non-statement "
619 "ClauseElement objects are deprecated; params() is now limited to "
620 "statement level objects such as select(), insert(), union(), etc. ",
621 )
622 def _replace_params(
623 self,
624 unique: bool,
625 optionaldict: Optional[Mapping[str, Any]],
626 kwargs: Dict[str, Any],
627 ) -> Self:
628 if optionaldict:
629 kwargs.update(optionaldict)
631 def visit_bindparam(bind: BindParameter[Any]) -> None:
632 if bind.key in kwargs:
633 bind.value = kwargs[bind.key]
634 bind.required = False
635 if unique:
636 bind._convert_to_unique()
638 return cloned_traverse(
639 self,
640 {"maintain_key": True, "detect_subquery_cols": True},
641 {"bindparam": visit_bindparam},
642 )
644 def compare(self, other: ClauseElement, **kw: Any) -> bool:
645 r"""Compare this :class:`_expression.ClauseElement` to
646 the given :class:`_expression.ClauseElement`.
648 Subclasses should override the default behavior, which is a
649 straight identity comparison.
651 \**kw are arguments consumed by subclass ``compare()`` methods and
652 may be used to modify the criteria for comparison
653 (see :class:`_expression.ColumnElement`).
655 """
656 return traversals.compare(self, other, **kw)
658 def self_group(
659 self, against: Optional[OperatorType] = None
660 ) -> ClauseElement:
661 """Apply a 'grouping' to this :class:`_expression.ClauseElement`.
663 This method is overridden by subclasses to return a "grouping"
664 construct, i.e. parenthesis. In particular it's used by "binary"
665 expressions to provide a grouping around themselves when placed into a
666 larger expression, as well as by :func:`_expression.select`
667 constructs when placed into the FROM clause of another
668 :func:`_expression.select`. (Note that subqueries should be
669 normally created using the :meth:`_expression.Select.alias` method,
670 as many
671 platforms require nested SELECT statements to be named).
673 As expressions are composed together, the application of
674 :meth:`self_group` is automatic - end-user code should never
675 need to use this method directly. Note that SQLAlchemy's
676 clause constructs take operator precedence into account -
677 so parenthesis might not be needed, for example, in
678 an expression like ``x OR (y AND z)`` - AND takes precedence
679 over OR.
681 The base :meth:`self_group` method of
682 :class:`_expression.ClauseElement`
683 just returns self.
684 """
685 return self
687 def _ungroup(self) -> ClauseElement:
688 """Return this :class:`_expression.ClauseElement`
689 without any groupings.
690 """
692 return self
694 def _compile_w_cache(
695 self,
696 dialect: Dialect,
697 *,
698 compiled_cache: Optional[CompiledCacheType],
699 column_keys: List[str],
700 for_executemany: bool = False,
701 schema_translate_map: Optional[SchemaTranslateMapType] = None,
702 **kw: Any,
703 ) -> tuple[
704 Compiled,
705 Sequence[BindParameter[Any]] | None,
706 _CoreSingleExecuteParams | None,
707 CacheStats,
708 ]:
709 elem_cache_key: Optional[CacheKey]
711 if compiled_cache is not None and dialect._supports_statement_cache:
712 elem_cache_key = self._generate_cache_key()
713 else:
714 elem_cache_key = None
716 extracted_params: Optional[Sequence[BindParameter[Any]]]
717 if elem_cache_key is not None:
718 if TYPE_CHECKING:
719 assert compiled_cache is not None
721 cache_key, extracted_params, param_dict = elem_cache_key
722 key = (
723 dialect,
724 cache_key,
725 tuple(column_keys),
726 bool(schema_translate_map),
727 for_executemany,
728 )
729 compiled_sql = compiled_cache.get(key)
731 if compiled_sql is None:
732 cache_hit = dialect.CACHE_MISS
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 )
741 # ensure that params of the current statement are not
742 # left in the cache
743 assert not compiled_sql._collect_params # type: ignore[attr-defined] # noqa: E501
744 compiled_cache[key] = compiled_sql
745 else:
746 cache_hit = dialect.CACHE_HIT
747 else:
748 param_dict = None
749 extracted_params = None
750 compiled_sql = self._compiler(
751 dialect,
752 cache_key=None,
753 column_keys=column_keys,
754 for_executemany=for_executemany,
755 schema_translate_map=schema_translate_map,
756 **kw,
757 )
758 # here instead the params need to be extracted, since we don't
759 # have them otherwise
760 assert compiled_sql._collect_params # type: ignore[attr-defined] # noqa: E501
762 if not dialect._supports_statement_cache:
763 cache_hit = dialect.NO_DIALECT_SUPPORT
764 elif compiled_cache is None:
765 cache_hit = dialect.CACHING_DISABLED
766 else:
767 cache_hit = dialect.NO_CACHE_KEY
769 return compiled_sql, extracted_params, param_dict, cache_hit
771 def __invert__(self):
772 # undocumented element currently used by the ORM for
773 # relationship.contains()
774 if hasattr(self, "negation_clause"):
775 return self.negation_clause
776 else:
777 return self._negate()
779 def _negate(self) -> ClauseElement:
780 # TODO: this code is uncovered and in all likelihood is not included
781 # in any codepath. So this should raise NotImplementedError in 2.1
782 grouped = self.self_group(against=operators.inv)
783 assert isinstance(grouped, ColumnElement)
784 return UnaryExpression(grouped, operator=operators.inv)
786 def __bool__(self):
787 raise TypeError("Boolean value of this clause is not defined")
789 def __repr__(self):
790 friendly = self.description
791 if friendly is None:
792 return object.__repr__(self)
793 else:
794 return "<%s.%s at 0x%x; %s>" % (
795 self.__module__,
796 self.__class__.__name__,
797 id(self),
798 friendly,
799 )
802class DQLDMLClauseElement(ClauseElement):
803 """represents a :class:`.ClauseElement` that compiles to a DQL or DML
804 expression, not DDL.
806 .. versionadded:: 2.0
808 """
810 if typing.TYPE_CHECKING:
812 def _compiler(self, dialect: Dialect, **kw: Any) -> SQLCompiler:
813 """Return a compiler appropriate for this ClauseElement, given a
814 Dialect."""
815 ...
817 def compile( # noqa: A001
818 self,
819 bind: Optional[_HasDialect] = None,
820 dialect: Optional[Dialect] = None,
821 **kw: Any,
822 ) -> SQLCompiler: ...
825class CompilerColumnElement(
826 roles.DMLColumnRole,
827 roles.DDLConstraintColumnRole,
828 roles.ColumnsClauseRole,
829 CompilerElement,
830):
831 """A compiler-only column element used for ad-hoc string compilations.
833 .. versionadded:: 2.0
835 """
837 __slots__ = ()
839 _propagate_attrs = util.EMPTY_DICT
840 _is_collection_aggregate = False
841 _is_implicitly_boolean = False
843 def _with_binary_element_type(self, type_):
844 raise NotImplementedError()
846 def _gen_cache_key(self, anon_map, bindparams):
847 raise NotImplementedError()
849 @property
850 def _from_objects(self) -> List[FromClause]:
851 raise NotImplementedError()
854# SQLCoreOperations should be suiting the ExpressionElementRole
855# and ColumnsClauseRole. however the MRO issues become too elaborate
856# at the moment.
857class SQLCoreOperations(Generic[_T_co], ColumnOperators, TypingOnly):
858 __slots__ = ()
860 # annotations for comparison methods
861 # these are from operators->Operators / ColumnOperators,
862 # redefined with the specific types returned by ColumnElement hierarchies
863 if typing.TYPE_CHECKING:
865 @util.non_memoized_property
866 def _propagate_attrs(self) -> _PropagateAttrsType: ...
868 def operate(
869 self, op: OperatorType, *other: Any, **kwargs: Any
870 ) -> ColumnElement[Any]: ...
872 def reverse_operate(
873 self, op: OperatorType, other: Any, **kwargs: Any
874 ) -> ColumnElement[Any]: ...
876 @overload
877 def op(
878 self,
879 opstring: str,
880 precedence: int = ...,
881 is_comparison: bool = ...,
882 *,
883 return_type: _TypeEngineArgument[_OPT],
884 python_impl: Optional[Callable[..., Any]] = None,
885 operator_class: OperatorClass = ...,
886 visit_name: Optional[str] = ...,
887 ) -> Callable[[Any], BinaryExpression[_OPT]]: ...
889 @overload
890 def op(
891 self,
892 opstring: str,
893 precedence: int = ...,
894 is_comparison: bool = ...,
895 return_type: Optional[_TypeEngineArgument[Any]] = ...,
896 python_impl: Optional[Callable[..., Any]] = ...,
897 operator_class: OperatorClass = ...,
898 visit_name: Optional[str] = ...,
899 ) -> Callable[[Any], BinaryExpression[Any]]: ...
901 def op(
902 self,
903 opstring: str,
904 precedence: int = 0,
905 is_comparison: bool = False,
906 return_type: Optional[_TypeEngineArgument[Any]] = None,
907 python_impl: Optional[Callable[..., Any]] = None,
908 operator_class: OperatorClass = OperatorClass.BASE,
909 visit_name: Optional[str] = None,
910 ) -> Callable[[Any], BinaryExpression[Any]]: ...
912 def bool_op(
913 self,
914 opstring: str,
915 precedence: int = 0,
916 python_impl: Optional[Callable[..., Any]] = None,
917 ) -> Callable[[Any], BinaryExpression[bool]]: ...
919 def __and__(self, other: Any) -> BooleanClauseList: ...
921 def __or__(self, other: Any) -> BooleanClauseList: ...
923 def __invert__(self) -> ColumnElement[_T_co]: ...
925 def __lt__(self, other: Any) -> ColumnElement[bool]: ...
927 def __le__(self, other: Any) -> ColumnElement[bool]: ...
929 # declare also that this class has an hash method otherwise
930 # it may be assumed to be None by type checkers since the
931 # object defines __eq__ and python sets it to None in that case:
932 # https://docs.python.org/3/reference/datamodel.html#object.__hash__
933 def __hash__(self) -> int: ...
935 def __eq__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
936 ...
938 def __ne__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
939 ...
941 def is_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
943 def is_not_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
945 def __gt__(self, other: Any) -> ColumnElement[bool]: ...
947 def __ge__(self, other: Any) -> ColumnElement[bool]: ...
949 def __neg__(self) -> UnaryExpression[_T_co]: ...
951 def __contains__(self, other: Any) -> ColumnElement[bool]: ...
953 def __getitem__(self, index: Any) -> ColumnElement[Any]: ...
955 @overload
956 def __lshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
958 @overload
959 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
961 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
963 @overload
964 def __rlshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
966 @overload
967 def __rlshift__(self, other: Any) -> ColumnElement[Any]: ...
969 def __rlshift__(self, other: Any) -> ColumnElement[Any]: ...
971 @overload
972 def __rshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
974 @overload
975 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
977 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
979 @overload
980 def __rrshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
982 @overload
983 def __rrshift__(self, other: Any) -> ColumnElement[Any]: ...
985 def __rrshift__(self, other: Any) -> ColumnElement[Any]: ...
987 def __matmul__(self, other: Any) -> ColumnElement[Any]: ...
989 def __rmatmul__(self, other: Any) -> ColumnElement[Any]: ...
991 @overload
992 def concat(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
994 @overload
995 def concat(self, other: Any) -> ColumnElement[Any]: ...
997 def concat(self, other: Any) -> ColumnElement[Any]: ...
999 def like(
1000 self, other: Any, escape: Optional[str] = None
1001 ) -> BinaryExpression[bool]: ...
1003 def ilike(
1004 self, other: Any, escape: Optional[str] = None
1005 ) -> BinaryExpression[bool]: ...
1007 def bitwise_xor(self, other: Any) -> BinaryExpression[Any]: ...
1009 def bitwise_or(self, other: Any) -> BinaryExpression[Any]: ...
1011 def bitwise_and(self, other: Any) -> BinaryExpression[Any]: ...
1013 def bitwise_not(self) -> UnaryExpression[_T_co]: ...
1015 def bitwise_lshift(self, other: Any) -> BinaryExpression[Any]: ...
1017 def bitwise_rshift(self, other: Any) -> BinaryExpression[Any]: ...
1019 def in_(
1020 self,
1021 other: Union[
1022 Iterable[Any], BindParameter[Any], roles.InElementRole
1023 ],
1024 ) -> BinaryExpression[bool]: ...
1026 def not_in(
1027 self,
1028 other: Union[
1029 Iterable[Any], BindParameter[Any], roles.InElementRole
1030 ],
1031 ) -> BinaryExpression[bool]: ...
1033 def notin_(
1034 self,
1035 other: Union[
1036 Iterable[Any], BindParameter[Any], roles.InElementRole
1037 ],
1038 ) -> BinaryExpression[bool]: ...
1040 def not_like(
1041 self, other: Any, escape: Optional[str] = None
1042 ) -> BinaryExpression[bool]: ...
1044 def notlike(
1045 self, other: Any, escape: Optional[str] = None
1046 ) -> BinaryExpression[bool]: ...
1048 def not_ilike(
1049 self, other: Any, escape: Optional[str] = None
1050 ) -> BinaryExpression[bool]: ...
1052 def notilike(
1053 self, other: Any, escape: Optional[str] = None
1054 ) -> BinaryExpression[bool]: ...
1056 def is_(self, other: Any) -> BinaryExpression[bool]: ...
1058 def is_not(self, other: Any) -> BinaryExpression[bool]: ...
1060 def isnot(self, other: Any) -> BinaryExpression[bool]: ...
1062 def startswith(
1063 self,
1064 other: Any,
1065 escape: Optional[str] = None,
1066 autoescape: bool = False,
1067 ) -> ColumnElement[bool]: ...
1069 def istartswith(
1070 self,
1071 other: Any,
1072 escape: Optional[str] = None,
1073 autoescape: bool = False,
1074 ) -> ColumnElement[bool]: ...
1076 def endswith(
1077 self,
1078 other: Any,
1079 escape: Optional[str] = None,
1080 autoescape: bool = False,
1081 ) -> ColumnElement[bool]: ...
1083 def iendswith(
1084 self,
1085 other: Any,
1086 escape: Optional[str] = None,
1087 autoescape: bool = False,
1088 ) -> ColumnElement[bool]: ...
1090 def contains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1092 def icontains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1094 def match(self, other: Any, **kwargs: Any) -> ColumnElement[bool]: ...
1096 def regexp_match(
1097 self, pattern: Any, flags: Optional[str] = None
1098 ) -> ColumnElement[bool]: ...
1100 def regexp_replace(
1101 self, pattern: Any, replacement: Any, flags: Optional[str] = None
1102 ) -> ColumnElement[str]: ...
1104 def desc(self) -> UnaryExpression[_T_co]: ...
1106 def asc(self) -> UnaryExpression[_T_co]: ...
1108 def nulls_first(self) -> UnaryExpression[_T_co]: ...
1110 def nullsfirst(self) -> UnaryExpression[_T_co]: ...
1112 def nulls_last(self) -> UnaryExpression[_T_co]: ...
1114 def nullslast(self) -> UnaryExpression[_T_co]: ...
1116 def collate(
1117 self, collation: str, collation_schema: Optional[str] = None
1118 ) -> CollationClause: ...
1120 def between(
1121 self, cleft: Any, cright: Any, symmetric: bool = False
1122 ) -> BinaryExpression[bool]: ...
1124 def distinct(self: _SQO[_T_co]) -> UnaryExpression[_T_co]: ...
1126 def any_(self) -> CollectionAggregate[Any]: ...
1128 def all_(self) -> CollectionAggregate[Any]: ...
1130 # numeric overloads. These need more tweaking
1131 # in particular they all need to have a variant for Optional[_T]
1132 # because Optional only applies to the data side, not the expression
1133 # side
1135 @overload
1136 def __add__(
1137 self: _SQO[_NMT],
1138 other: Any,
1139 ) -> ColumnElement[_NMT]: ...
1141 @overload
1142 def __add__(
1143 self: _SQO[str],
1144 other: Any,
1145 ) -> ColumnElement[str]: ...
1147 @overload
1148 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1150 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1152 @overload
1153 def __radd__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1155 @overload
1156 def __radd__(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
1158 def __radd__(self, other: Any) -> ColumnElement[Any]: ...
1160 @overload
1161 def __sub__(
1162 self: _SQO[_NMT],
1163 other: Any,
1164 ) -> ColumnElement[_NMT]: ...
1166 @overload
1167 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1169 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1171 @overload
1172 def __rsub__(
1173 self: _SQO[_NMT],
1174 other: Any,
1175 ) -> ColumnElement[_NMT]: ...
1177 @overload
1178 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1180 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1182 @overload
1183 def __mul__(
1184 self: _SQO[_NMT],
1185 other: Any,
1186 ) -> ColumnElement[_NMT]: ...
1188 @overload
1189 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1191 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1193 @overload
1194 def __rmul__(
1195 self: _SQO[_NMT],
1196 other: Any,
1197 ) -> ColumnElement[_NMT]: ...
1199 @overload
1200 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1202 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1204 @overload
1205 def __mod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1207 @overload
1208 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1210 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1212 @overload
1213 def __rmod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1215 @overload
1216 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1218 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1220 @overload
1221 def __truediv__(
1222 self: _SQO[int], other: Any
1223 ) -> ColumnElement[_NUMERIC]: ...
1225 @overload
1226 def __truediv__(self: _SQO[_NT], other: Any) -> ColumnElement[_NT]: ...
1228 @overload
1229 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1231 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1233 @overload
1234 def __rtruediv__(
1235 self: _SQO[_NMT], other: Any
1236 ) -> ColumnElement[_NUMERIC]: ...
1238 @overload
1239 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1241 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1243 @overload
1244 def __floordiv__(
1245 self: _SQO[_NMT], other: Any
1246 ) -> ColumnElement[_NMT]: ...
1248 @overload
1249 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1251 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1253 @overload
1254 def __rfloordiv__(
1255 self: _SQO[_NMT], other: Any
1256 ) -> ColumnElement[_NMT]: ...
1258 @overload
1259 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1261 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1264class SQLColumnExpression(
1265 SQLCoreOperations[_T_co], roles.ExpressionElementRole[_T_co], TypingOnly
1266):
1267 """A type that may be used to indicate any SQL column element or object
1268 that acts in place of one.
1270 :class:`.SQLColumnExpression` is a base of
1271 :class:`.ColumnElement`, as well as within the bases of ORM elements
1272 such as :class:`.InstrumentedAttribute`, and may be used in :pep:`484`
1273 typing to indicate arguments or return values that should behave
1274 as column expressions.
1276 .. versionadded:: 2.0.0b4
1279 """
1281 __slots__ = ()
1284_SQO = SQLCoreOperations
1287class ColumnElement(
1288 roles.ColumnArgumentOrKeyRole,
1289 roles.StatementOptionRole,
1290 roles.WhereHavingRole,
1291 roles.BinaryElementRole[_T],
1292 roles.OrderByRole,
1293 roles.ColumnsClauseRole,
1294 roles.LimitOffsetRole,
1295 roles.DMLColumnRole,
1296 roles.DDLConstraintColumnRole,
1297 roles.DDLExpressionRole,
1298 SQLColumnExpression[_T],
1299 DQLDMLClauseElement,
1300):
1301 """Represent a column-oriented SQL expression suitable for usage in the
1302 "columns" clause, WHERE clause etc. of a statement.
1304 While the most familiar kind of :class:`_expression.ColumnElement` is the
1305 :class:`_schema.Column` object, :class:`_expression.ColumnElement`
1306 serves as the basis
1307 for any unit that may be present in a SQL expression, including
1308 the expressions themselves, SQL functions, bound parameters,
1309 literal expressions, keywords such as ``NULL``, etc.
1310 :class:`_expression.ColumnElement`
1311 is the ultimate base class for all such elements.
1313 A wide variety of SQLAlchemy Core functions work at the SQL expression
1314 level, and are intended to accept instances of
1315 :class:`_expression.ColumnElement` as
1316 arguments. These functions will typically document that they accept a
1317 "SQL expression" as an argument. What this means in terms of SQLAlchemy
1318 usually refers to an input which is either already in the form of a
1319 :class:`_expression.ColumnElement` object,
1320 or a value which can be **coerced** into
1321 one. The coercion rules followed by most, but not all, SQLAlchemy Core
1322 functions with regards to SQL expressions are as follows:
1324 * a literal Python value, such as a string, integer or floating
1325 point value, boolean, datetime, ``Decimal`` object, or virtually
1326 any other Python object, will be coerced into a "literal bound
1327 value". This generally means that a :func:`.bindparam` will be
1328 produced featuring the given value embedded into the construct; the
1329 resulting :class:`.BindParameter` object is an instance of
1330 :class:`_expression.ColumnElement`.
1331 The Python value will ultimately be sent
1332 to the DBAPI at execution time as a parameterized argument to the
1333 ``execute()`` or ``executemany()`` methods, after SQLAlchemy
1334 type-specific converters (e.g. those provided by any associated
1335 :class:`.TypeEngine` objects) are applied to the value.
1337 * any special object value, typically ORM-level constructs, which
1338 feature an accessor called ``__clause_element__()``. The Core
1339 expression system looks for this method when an object of otherwise
1340 unknown type is passed to a function that is looking to coerce the
1341 argument into a :class:`_expression.ColumnElement` and sometimes a
1342 :class:`_expression.SelectBase` expression.
1343 It is used within the ORM to
1344 convert from ORM-specific objects like mapped classes and
1345 mapped attributes into Core expression objects.
1347 * The Python ``None`` value is typically interpreted as ``NULL``,
1348 which in SQLAlchemy Core produces an instance of :func:`.null`.
1350 A :class:`_expression.ColumnElement` provides the ability to generate new
1351 :class:`_expression.ColumnElement`
1352 objects using Python expressions. This means that Python operators
1353 such as ``==``, ``!=`` and ``<`` are overloaded to mimic SQL operations,
1354 and allow the instantiation of further :class:`_expression.ColumnElement`
1355 instances
1356 which are composed from other, more fundamental
1357 :class:`_expression.ColumnElement`
1358 objects. For example, two :class:`.ColumnClause` objects can be added
1359 together with the addition operator ``+`` to produce
1360 a :class:`.BinaryExpression`.
1361 Both :class:`.ColumnClause` and :class:`.BinaryExpression` are subclasses
1362 of :class:`_expression.ColumnElement`:
1364 .. sourcecode:: pycon+sql
1366 >>> from sqlalchemy.sql import column
1367 >>> column("a") + column("b")
1368 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
1369 >>> print(column("a") + column("b"))
1370 {printsql}a + b
1372 .. seealso::
1374 :class:`_schema.Column`
1376 :func:`_expression.column`
1378 """
1380 __visit_name__ = "column_element"
1382 primary_key: bool = False
1383 _is_clone_of: Optional[ColumnElement[_T]]
1384 _is_column_element = True
1385 _insert_sentinel: bool = False
1386 _omit_from_statements = False
1387 _is_collection_aggregate = False
1389 foreign_keys: AbstractSet[ForeignKey] = frozenset()
1391 @util.memoized_property
1392 def _proxies(self) -> List[ColumnElement[Any]]:
1393 return []
1395 @util.non_memoized_property
1396 def _tq_label(self) -> Optional[str]:
1397 """The named label that can be used to target
1398 this column in a result set in a "table qualified" context.
1400 This label is almost always the label used when
1401 rendering <expr> AS <label> in a SELECT statement when using
1402 the LABEL_STYLE_TABLENAME_PLUS_COL label style, which is what the
1403 legacy ORM ``Query`` object uses as well.
1405 For a regular Column bound to a Table, this is typically the label
1406 <tablename>_<columnname>. For other constructs, different rules
1407 may apply, such as anonymized labels and others.
1409 .. versionchanged:: 1.4.21 renamed from ``._label``
1411 """
1412 return None
1414 key: Optional[str] = None
1415 """The 'key' that in some circumstances refers to this object in a
1416 Python namespace.
1418 This typically refers to the "key" of the column as present in the
1419 ``.c`` collection of a selectable, e.g. ``sometable.c["somekey"]`` would
1420 return a :class:`_schema.Column` with a ``.key`` of "somekey".
1422 """
1424 @HasMemoized.memoized_attribute
1425 def _tq_key_label(self) -> Optional[str]:
1426 """A label-based version of 'key' that in some circumstances refers
1427 to this object in a Python namespace.
1430 _tq_key_label comes into play when a select() statement is constructed
1431 with apply_labels(); in this case, all Column objects in the ``.c``
1432 collection are rendered as <tablename>_<columnname> in SQL; this is
1433 essentially the value of ._label. But to locate those columns in the
1434 ``.c`` collection, the name is along the lines of <tablename>_<key>;
1435 that's the typical value of .key_label.
1437 .. versionchanged:: 1.4.21 renamed from ``._key_label``
1439 """
1440 return self._proxy_key
1442 @property
1443 def _key_label(self) -> Optional[str]:
1444 """legacy; renamed to _tq_key_label"""
1445 return self._tq_key_label
1447 @property
1448 def _label(self) -> Optional[str]:
1449 """legacy; renamed to _tq_label"""
1450 return self._tq_label
1452 @property
1453 def _non_anon_label(self) -> Optional[str]:
1454 """the 'name' that naturally applies this element when rendered in
1455 SQL.
1457 Concretely, this is the "name" of a column or a label in a
1458 SELECT statement; ``<columnname>`` and ``<labelname>`` below:
1460 .. sourcecode:: sql
1462 SELECT <columnmame> FROM table
1464 SELECT column AS <labelname> FROM table
1466 Above, the two names noted will be what's present in the DBAPI
1467 ``cursor.description`` as the names.
1469 If this attribute returns ``None``, it means that the SQL element as
1470 written does not have a 100% fully predictable "name" that would appear
1471 in the ``cursor.description``. Examples include SQL functions, CAST
1472 functions, etc. While such things do return names in
1473 ``cursor.description``, they are only predictable on a
1474 database-specific basis; e.g. an expression like ``MAX(table.col)`` may
1475 appear as the string ``max`` on one database (like PostgreSQL) or may
1476 appear as the whole expression ``max(table.col)`` on SQLite.
1478 The default implementation looks for a ``.name`` attribute on the
1479 object, as has been the precedent established in SQLAlchemy for many
1480 years. An exception is made on the ``FunctionElement`` subclass
1481 so that the return value is always ``None``.
1483 .. versionadded:: 1.4.21
1487 """
1488 return getattr(self, "name", None)
1490 _render_label_in_columns_clause = True
1491 """A flag used by select._columns_plus_names that helps to determine
1492 we are actually going to render in terms of "SELECT <col> AS <label>".
1493 This flag can be returned as False for some Column objects that want
1494 to be rendered as simple "SELECT <col>"; typically columns that don't have
1495 any parent table and are named the same as what the label would be
1496 in any case.
1498 """
1500 _allow_label_resolve = True
1501 """A flag that can be flipped to prevent a column from being resolvable
1502 by string label name.
1504 The joined eager loader strategy in the ORM uses this, for example.
1506 """
1508 _is_implicitly_boolean = False
1510 _alt_names: Sequence[str] = ()
1512 if TYPE_CHECKING:
1514 def _ungroup(self) -> ColumnElement[_T]: ...
1516 @overload
1517 def self_group(self, against: None = None) -> ColumnElement[_T]: ...
1519 @overload
1520 def self_group(
1521 self, against: Optional[OperatorType] = None
1522 ) -> ColumnElement[Any]: ...
1524 def self_group(
1525 self, against: Optional[OperatorType] = None
1526 ) -> ColumnElement[Any]:
1527 if (
1528 against in (operators.and_, operators.or_, operators._asbool)
1529 and self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity
1530 ):
1531 return AsBoolean(self, operators.is_true, operators.is_false)
1532 elif against in (operators.any_op, operators.all_op):
1533 return Grouping(self)
1534 else:
1535 return self
1537 @overload
1538 def _negate(self: ColumnElement[bool]) -> ColumnElement[bool]: ...
1540 @overload
1541 def _negate(self: ColumnElement[_T]) -> ColumnElement[_T]: ...
1543 def _negate(self) -> ColumnElement[Any]:
1544 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
1545 return AsBoolean(self, operators.is_false, operators.is_true)
1546 else:
1547 grouped = self.self_group(against=operators.inv)
1548 assert isinstance(grouped, ColumnElement)
1549 return UnaryExpression(
1550 grouped,
1551 operator=operators.inv,
1552 )
1554 type: TypeEngine[_T]
1556 if not TYPE_CHECKING:
1558 @util.memoized_property
1559 def type(self) -> TypeEngine[_T]: # noqa: A001
1560 # used for delayed setup of
1561 # type_api
1562 return type_api.NULLTYPE
1564 @HasMemoized.memoized_attribute
1565 def comparator(self) -> TypeEngine.Comparator[_T]:
1566 try:
1567 comparator_factory = self.type.comparator_factory
1568 except AttributeError as err:
1569 raise TypeError(
1570 "Object %r associated with '.type' attribute "
1571 "is not a TypeEngine class or object" % self.type
1572 ) from err
1573 else:
1574 return comparator_factory(self)
1576 def __setstate__(self, state):
1577 self.__dict__.update(state)
1579 def __getattr__(self, key: str) -> Any:
1580 try:
1581 return getattr(self.comparator, key)
1582 except AttributeError as err:
1583 raise AttributeError(
1584 "Neither %r object nor %r object has an attribute %r"
1585 % (
1586 type(self).__name__,
1587 type(self.comparator).__name__,
1588 key,
1589 )
1590 ) from err
1592 def operate(
1593 self,
1594 op: operators.OperatorType,
1595 *other: Any,
1596 **kwargs: Any,
1597 ) -> ColumnElement[Any]:
1598 return op(self.comparator, *other, **kwargs) # type: ignore[no-any-return] # noqa: E501
1600 def reverse_operate(
1601 self, op: operators.OperatorType, other: Any, **kwargs: Any
1602 ) -> ColumnElement[Any]:
1603 return op(other, self.comparator, **kwargs) # type: ignore[no-any-return] # noqa: E501
1605 def _bind_param(
1606 self,
1607 operator: operators.OperatorType,
1608 obj: Any,
1609 type_: Optional[TypeEngine[_T]] = None,
1610 expanding: bool = False,
1611 ) -> BindParameter[_T]:
1612 return BindParameter(
1613 None,
1614 obj,
1615 _compared_to_operator=operator,
1616 type_=type_,
1617 _compared_to_type=self.type,
1618 unique=True,
1619 expanding=expanding,
1620 )
1622 @property
1623 def expression(self) -> ColumnElement[Any]:
1624 """Return a column expression.
1626 Part of the inspection interface; returns self.
1628 """
1629 return self
1631 @property
1632 def _select_iterable(self) -> _SelectIterable:
1633 return (self,)
1635 @util.memoized_property
1636 def base_columns(self) -> FrozenSet[ColumnElement[Any]]:
1637 return frozenset(c for c in self.proxy_set if not c._proxies)
1639 @util.memoized_property
1640 def proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1641 """set of all columns we are proxying
1643 as of 2.0 this is explicitly deannotated columns. previously it was
1644 effectively deannotated columns but wasn't enforced. annotated
1645 columns should basically not go into sets if at all possible because
1646 their hashing behavior is very non-performant.
1648 """
1649 return frozenset([self._deannotate()]).union(
1650 itertools.chain(*[c.proxy_set for c in self._proxies])
1651 )
1653 @util.memoized_property
1654 def _expanded_proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1655 return frozenset(_expand_cloned(self.proxy_set))
1657 def _uncached_proxy_list(self) -> List[ColumnElement[Any]]:
1658 """An 'uncached' version of proxy set.
1660 This list includes annotated columns which perform very poorly in
1661 set operations.
1663 """
1665 return [self] + list(
1666 itertools.chain(*[c._uncached_proxy_list() for c in self._proxies])
1667 )
1669 def shares_lineage(self, othercolumn: ColumnElement[Any]) -> bool:
1670 """Return True if the given :class:`_expression.ColumnElement`
1671 has a common ancestor to this :class:`_expression.ColumnElement`."""
1673 return bool(self.proxy_set.intersection(othercolumn.proxy_set))
1675 def _compare_name_for_result(self, other: ColumnElement[Any]) -> bool:
1676 """Return True if the given column element compares to this one
1677 when targeting within a result row."""
1679 return (
1680 hasattr(other, "name")
1681 and hasattr(self, "name")
1682 and other.name == self.name
1683 )
1685 @HasMemoized.memoized_attribute
1686 def _proxy_key(self) -> Optional[str]:
1687 if self._annotations and "proxy_key" in self._annotations:
1688 return cast(str, self._annotations["proxy_key"])
1690 name = self.key
1691 if not name:
1692 # there's a bit of a seeming contradiction which is that the
1693 # "_non_anon_label" of a column can in fact be an
1694 # "_anonymous_label"; this is when it's on a column that is
1695 # proxying for an anonymous expression in a subquery.
1696 name = self._non_anon_label
1698 if isinstance(name, _anonymous_label):
1699 return None
1700 else:
1701 return name
1703 @HasMemoized.memoized_attribute
1704 def _expression_label(self) -> Optional[str]:
1705 """a suggested label to use in the case that the column has no name,
1706 which should be used if possible as the explicit 'AS <label>'
1707 where this expression would normally have an anon label.
1709 this is essentially mostly what _proxy_key does except it returns
1710 None if the column has a normal name that can be used.
1712 """
1714 if getattr(self, "name", None) is not None:
1715 return None
1716 elif self._annotations and "proxy_key" in self._annotations:
1717 return cast(str, self._annotations["proxy_key"])
1718 else:
1719 return None
1721 def _make_proxy(
1722 self,
1723 selectable: FromClause,
1724 *,
1725 primary_key: ColumnSet,
1726 foreign_keys: Set[KeyedColumnElement[Any]],
1727 name: Optional[str] = None,
1728 key: Optional[str] = None,
1729 name_is_truncatable: bool = False,
1730 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
1731 **kw: Any,
1732 ) -> typing_Tuple[str, ColumnClause[_T]]:
1733 """Create a new :class:`_expression.ColumnElement` representing this
1734 :class:`_expression.ColumnElement` as it appears in the select list of
1735 a descending selectable.
1737 """
1738 if name is None:
1739 name = self._anon_name_label
1740 if key is None:
1741 key = self._proxy_key
1742 else:
1743 key = name
1745 assert key is not None
1747 co: ColumnClause[_T] = ColumnClause(
1748 (
1749 coercions.expect(roles.TruncatedLabelRole, name)
1750 if name_is_truncatable
1751 else name
1752 ),
1753 type_=getattr(self, "type", None),
1754 _selectable=selectable,
1755 )
1757 co._propagate_attrs = selectable._propagate_attrs
1758 if compound_select_cols:
1759 co._proxies = list(compound_select_cols)
1760 else:
1761 co._proxies = [self]
1762 if selectable._is_clone_of is not None:
1763 co._is_clone_of = selectable._is_clone_of.columns.get(key)
1764 return key, co
1766 def cast(self, type_: _TypeEngineArgument[_OPT]) -> Cast[_OPT]:
1767 """Produce a type cast, i.e. ``CAST(<expression> AS <type>)``.
1769 This is a shortcut to the :func:`_expression.cast` function.
1771 .. seealso::
1773 :ref:`tutorial_casts`
1775 :func:`_expression.cast`
1777 :func:`_expression.type_coerce`
1779 """
1780 return Cast(self, type_)
1782 def label(self, name: Optional[str]) -> Label[_T]:
1783 """Produce a column label, i.e. ``<columnname> AS <name>``.
1785 This is a shortcut to the :func:`_expression.label` function.
1787 If 'name' is ``None``, an anonymous label name will be generated.
1789 """
1790 return Label(name, self, self.type)
1792 def _anon_label(
1793 self, seed: Optional[str], add_hash: Optional[int] = None
1794 ) -> _anonymous_label:
1795 while self._is_clone_of is not None:
1796 self = self._is_clone_of
1798 # as of 1.4 anonymous label for ColumnElement uses hash(), not id(),
1799 # as the identifier, because a column and its annotated version are
1800 # the same thing in a SQL statement
1801 hash_value = hash(self)
1803 if add_hash:
1804 # this path is used for disambiguating anon labels that would
1805 # otherwise be the same name for the same element repeated.
1806 # an additional numeric value is factored in for each label.
1808 # shift hash(self) (which is id(self), typically 8 byte integer)
1809 # 16 bits leftward. fill extra add_hash on right
1810 assert add_hash < (2 << 15)
1811 assert seed
1812 hash_value = (hash_value << 16) | add_hash
1814 # extra underscore is added for labels with extra hash
1815 # values, to isolate the "deduped anon" namespace from the
1816 # regular namespace. eliminates chance of these
1817 # manufactured hash values overlapping with regular ones for some
1818 # undefined python interpreter
1819 seed = seed + "_"
1821 if isinstance(seed, _anonymous_label):
1822 # NOTE: the space after the hash is required
1823 return _anonymous_label(f"{seed}%({hash_value} )s")
1825 return _anonymous_label.safe_construct(hash_value, seed or "anon")
1827 @util.memoized_property
1828 def _anon_name_label(self) -> str:
1829 """Provides a constant 'anonymous label' for this ColumnElement.
1831 This is a label() expression which will be named at compile time.
1832 The same label() is returned each time ``anon_label`` is called so
1833 that expressions can reference ``anon_label`` multiple times,
1834 producing the same label name at compile time.
1836 The compiler uses this function automatically at compile time
1837 for expressions that are known to be 'unnamed' like binary
1838 expressions and function calls.
1840 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1841 public and is renamed to _anon_name_label. anon_name exists
1842 for backwards compat
1844 """
1845 name = getattr(self, "name", None)
1846 return self._anon_label(name)
1848 @util.memoized_property
1849 def _anon_key_label(self) -> _anonymous_label:
1850 """Provides a constant 'anonymous key label' for this ColumnElement.
1852 Compare to ``anon_label``, except that the "key" of the column,
1853 if available, is used to generate the label.
1855 This is used when a deduplicating key is placed into the columns
1856 collection of a selectable.
1858 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1859 public and is renamed to _anon_key_label. anon_key_label exists
1860 for backwards compat
1862 """
1863 return self._anon_label(self._proxy_key)
1865 @property
1866 @util.deprecated(
1867 "1.4",
1868 "The :attr:`_expression.ColumnElement.anon_label` attribute is now "
1869 "private, and the public accessor is deprecated.",
1870 )
1871 def anon_label(self) -> str:
1872 return self._anon_name_label
1874 @property
1875 @util.deprecated(
1876 "1.4",
1877 "The :attr:`_expression.ColumnElement.anon_key_label` attribute is "
1878 "now private, and the public accessor is deprecated.",
1879 )
1880 def anon_key_label(self) -> str:
1881 return self._anon_key_label
1883 def _dedupe_anon_label_idx(self, idx: int) -> str:
1884 """label to apply to a column that is anon labeled, but repeated
1885 in the SELECT, so that we have to make an "extra anon" label that
1886 disambiguates it from the previous appearance.
1888 these labels come out like "foo_bar_id__1" and have double underscores
1889 in them.
1891 """
1892 label = getattr(self, "name", None)
1894 # current convention is that if the element doesn't have a
1895 # ".name" (usually because it is not NamedColumn), we try to
1896 # use a "table qualified" form for the "dedupe anon" label,
1897 # based on the notion that a label like
1898 # "CAST(casttest.v1 AS DECIMAL) AS casttest_v1__1" looks better than
1899 # "CAST(casttest.v1 AS DECIMAL) AS anon__1"
1901 if label is None:
1902 return self._dedupe_anon_tq_label_idx(idx)
1903 else:
1904 return self._anon_label(label, add_hash=idx)
1906 @util.memoized_property
1907 def _anon_tq_label(self) -> _anonymous_label:
1908 return self._anon_label(getattr(self, "_tq_label", None))
1910 @util.memoized_property
1911 def _anon_tq_key_label(self) -> _anonymous_label:
1912 return self._anon_label(getattr(self, "_tq_key_label", None))
1914 def _dedupe_anon_tq_label_idx(self, idx: int) -> _anonymous_label:
1915 label = getattr(self, "_tq_label", None) or "anon"
1917 return self._anon_label(label, add_hash=idx)
1920class KeyedColumnElement(ColumnElement[_T]):
1921 """ColumnElement where ``.key`` is non-None."""
1923 _is_keyed_column_element = True
1925 key: str
1928class WrapsColumnExpression(ColumnElement[_T]):
1929 """Mixin that defines a :class:`_expression.ColumnElement`
1930 as a wrapper with special
1931 labeling behavior for an expression that already has a name.
1933 .. versionadded:: 1.4
1935 .. seealso::
1937 :ref:`change_4449`
1940 """
1942 @property
1943 def wrapped_column_expression(self) -> ColumnElement[_T]:
1944 raise NotImplementedError()
1946 @util.non_memoized_property
1947 def _tq_label(self) -> Optional[str]:
1948 wce = self.wrapped_column_expression
1949 if hasattr(wce, "_tq_label"):
1950 return wce._tq_label
1951 else:
1952 return None
1954 @property
1955 def _label(self) -> Optional[str]:
1956 return self._tq_label
1958 @property
1959 def _non_anon_label(self) -> Optional[str]:
1960 return None
1962 @util.non_memoized_property
1963 def _anon_name_label(self) -> str:
1964 wce = self.wrapped_column_expression
1966 # this logic tries to get the WrappedColumnExpression to render
1967 # with "<expr> AS <name>", where "<name>" is the natural name
1968 # within the expression itself. e.g. "CAST(table.foo) AS foo".
1969 if not wce._is_text_clause:
1970 nal = wce._non_anon_label
1971 if nal:
1972 return nal
1973 elif hasattr(wce, "_anon_name_label"):
1974 return wce._anon_name_label
1975 return super()._anon_name_label
1977 def _dedupe_anon_label_idx(self, idx: int) -> str:
1978 wce = self.wrapped_column_expression
1979 nal = wce._non_anon_label
1980 if nal:
1981 return self._anon_label(nal + "_")
1982 else:
1983 return self._dedupe_anon_tq_label_idx(idx)
1985 @property
1986 def _proxy_key(self):
1987 wce = self.wrapped_column_expression
1989 if not wce._is_text_clause:
1990 return wce._proxy_key
1991 return super()._proxy_key
1994class DMLTargetCopy(roles.InElementRole, KeyedColumnElement[_T]):
1995 """Refer to another column's VALUES or SET expression in an INSERT or
1996 UPDATE statement.
1998 See the public-facing :func:`_sql.from_dml_column` constructor for
1999 background.
2001 .. versionadded:: 2.1
2004 """
2006 def __init__(self, column: _OnlyColumnArgument[_T]):
2007 self.column = coercions.expect(roles.ColumnArgumentRole, column)
2008 self.type = self.column.type
2010 __visit_name__ = "dmltargetcopy"
2012 _traverse_internals: _TraverseInternalsType = [
2013 ("column", InternalTraversal.dp_clauseelement),
2014 ]
2017class BindParameter(roles.InElementRole, KeyedColumnElement[_T]):
2018 r"""Represent a "bound expression".
2020 :class:`.BindParameter` is invoked explicitly using the
2021 :func:`.bindparam` function, as in::
2023 from sqlalchemy import bindparam
2025 stmt = select(users_table).where(
2026 users_table.c.name == bindparam("username")
2027 )
2029 Detailed discussion of how :class:`.BindParameter` is used is
2030 at :func:`.bindparam`.
2032 .. seealso::
2034 :func:`.bindparam`
2036 """
2038 __visit_name__ = "bindparam"
2040 _traverse_internals: _TraverseInternalsType = [
2041 ("key", InternalTraversal.dp_anon_name),
2042 ("type", InternalTraversal.dp_type),
2043 ("callable", InternalTraversal.dp_plain_dict),
2044 ("value", InternalTraversal.dp_plain_obj),
2045 ("literal_execute", InternalTraversal.dp_boolean),
2046 ]
2048 key: str
2049 _anon_map_key: Optional[str] = None
2050 type: TypeEngine[_T]
2051 value: Optional[_T]
2053 _is_crud = False
2054 _is_bind_parameter = True
2056 # bindparam implements its own _gen_cache_key() method however
2057 # we check subclasses for this flag, else no cache key is generated
2058 inherit_cache = True
2060 def __init__(
2061 self,
2062 key: Optional[str],
2063 value: Any = _NoArg.NO_ARG,
2064 type_: Optional[_TypeEngineArgument[_T]] = None,
2065 unique: bool = False,
2066 required: Union[bool, Literal[_NoArg.NO_ARG]] = _NoArg.NO_ARG,
2067 quote: Optional[bool] = None,
2068 callable_: Optional[Callable[[], Any]] = None,
2069 expanding: bool = False,
2070 isoutparam: bool = False,
2071 literal_execute: bool = False,
2072 _compared_to_operator: Optional[OperatorType] = None,
2073 _compared_to_type: Optional[TypeEngine[Any]] = None,
2074 _is_crud: bool = False,
2075 ):
2076 if required is _NoArg.NO_ARG:
2077 required = value is _NoArg.NO_ARG and callable_ is None
2078 if value is _NoArg.NO_ARG:
2079 value = None
2081 if quote is not None:
2082 key = quoted_name.construct(key, quote)
2084 if unique:
2085 self.key, self._anon_map_key = (
2086 _anonymous_label.safe_construct_with_key(
2087 id(self),
2088 (
2089 key
2090 if key is not None
2091 and not isinstance(key, _anonymous_label)
2092 else "param"
2093 ),
2094 sanitize_key=True,
2095 )
2096 )
2097 elif key:
2098 self.key = key
2099 else:
2100 self.key, self._anon_map_key = (
2101 _anonymous_label.safe_construct_with_key(id(self), "param")
2102 )
2104 # identifying key that won't change across
2105 # clones, used to identify the bind's logical
2106 # identity
2107 self._identifying_key = self.key
2109 # key that was passed in the first place, used to
2110 # generate new keys
2111 self._orig_key = key or "param"
2113 self.unique = unique
2114 self.value = value
2115 self.callable = callable_
2116 self.isoutparam = isoutparam
2117 self.required = required
2119 # indicate an "expanding" parameter; the compiler sets this
2120 # automatically in the compiler _render_in_expr_w_bindparam method
2121 # for an IN expression
2122 self.expanding = expanding
2124 # this is another hint to help w/ expanding and is typically
2125 # set in the compiler _render_in_expr_w_bindparam method for an
2126 # IN expression
2127 self.expand_op = None
2129 self.literal_execute = literal_execute
2130 if _is_crud:
2131 self._is_crud = True
2133 if type_ is None:
2134 if expanding:
2135 if value:
2136 check_value = value[0]
2137 else:
2138 check_value = type_api._NO_VALUE_IN_LIST
2139 else:
2140 check_value = value
2141 if _compared_to_type is not None:
2142 self.type = _compared_to_type.coerce_compared_value(
2143 _compared_to_operator, check_value
2144 )
2145 else:
2146 self.type = type_api._resolve_value_to_type(check_value)
2147 elif isinstance(type_, type):
2148 self.type = type_()
2149 elif is_tuple_type(type_):
2150 if value:
2151 if expanding:
2152 check_value = value[0]
2153 else:
2154 check_value = value
2155 cast("BindParameter[TupleAny]", self).type = (
2156 type_._resolve_values_to_types(check_value)
2157 )
2158 else:
2159 cast("BindParameter[TupleAny]", self).type = type_
2160 else:
2161 self.type = type_
2163 def _with_value(self, value, maintain_key=False, required=NO_ARG):
2164 """Return a copy of this :class:`.BindParameter` with the given value
2165 set.
2166 """
2167 cloned = self._clone(maintain_key=maintain_key)
2168 cloned.value = value
2169 cloned.callable = None
2170 cloned.required = required if required is not NO_ARG else self.required
2171 if cloned.type is type_api.NULLTYPE:
2172 cloned.type = type_api._resolve_value_to_type(value)
2173 return cloned
2175 @property
2176 def effective_value(self) -> Optional[_T]:
2177 """Return the value of this bound parameter,
2178 taking into account if the ``callable`` parameter
2179 was set.
2181 The ``callable`` value will be evaluated
2182 and returned if present, else ``value``.
2184 """
2185 if self.callable:
2186 # TODO: set up protocol for bind parameter callable
2187 return self.callable() # type: ignore[no-any-return]
2188 else:
2189 return self.value
2191 def render_literal_execute(self) -> Self:
2192 """Produce a copy of this bound parameter that will enable the
2193 :paramref:`_sql.BindParameter.literal_execute` flag.
2195 The :paramref:`_sql.BindParameter.literal_execute` flag will
2196 have the effect of the parameter rendered in the compiled SQL
2197 string using ``[POSTCOMPILE]`` form, which is a special form that
2198 is converted to be a rendering of the literal value of the parameter
2199 at SQL execution time. The rationale is to support caching
2200 of SQL statement strings that can embed per-statement literal values,
2201 such as LIMIT and OFFSET parameters, in the final SQL string that
2202 is passed to the DBAPI. Dialects in particular may want to use
2203 this method within custom compilation schemes.
2205 .. versionadded:: 1.4.5
2207 .. seealso::
2209 :ref:`engine_thirdparty_caching`
2211 """
2212 c: Self = ClauseElement._clone(self)
2213 c.literal_execute = True
2214 return c
2216 def _negate_in_binary(self, negated_op, original_op):
2217 if self.expand_op is original_op:
2218 bind = self._clone()
2219 bind.expand_op = negated_op
2220 return bind
2221 else:
2222 return self
2224 def _with_binary_element_type(self, type_: TypeEngine[Any]) -> Self:
2225 c: Self = ClauseElement._clone(self)
2226 c.type = type_
2227 return c
2229 def _clone(self, maintain_key: bool = False, **kw: Any) -> Self:
2230 c: Self = ClauseElement._clone(self, **kw)
2231 # ensure all the BindParameter objects stay in cloned set.
2232 # in #7823, we changed "clone" so that a clone only keeps a reference
2233 # to the "original" element, since for column correspondence, that's
2234 # all we need. However, for BindParam, _cloned_set is used by
2235 # the "cache key bind match" lookup, which means if any of those
2236 # interim BindParameter objects became part of a cache key in the
2237 # cache, we need it. So here, make sure all clones keep carrying
2238 # forward.
2239 c._cloned_set.update(self._cloned_set)
2240 if not maintain_key and self.unique:
2241 c.key, c._anon_map_key = _anonymous_label.safe_construct_with_key(
2242 id(c), c._orig_key or "param", sanitize_key=True
2243 )
2244 return c
2246 def _gen_cache_key(self, anon_map, bindparams):
2247 _gen_cache_ok = self.__class__.__dict__.get("inherit_cache", False)
2249 if not _gen_cache_ok:
2250 if anon_map is not None:
2251 anon_map[NO_CACHE] = True
2252 return None
2254 id_, found = anon_map.get_anon(self)
2255 if found:
2256 return (id_, self.__class__)
2258 if bindparams is not None:
2259 bindparams.append(self)
2261 return (
2262 id_,
2263 self.__class__,
2264 self.type._static_cache_key,
2265 (
2266 anon_map[self._anon_map_key]
2267 if self._anon_map_key is not None
2268 else self.key
2269 ),
2270 self.literal_execute,
2271 )
2273 def _convert_to_unique(self):
2274 if not self.unique:
2275 self.unique = True
2276 self.key, self._anon_map_key = (
2277 _anonymous_label.safe_construct_with_key(
2278 id(self), self._orig_key or "param", sanitize_key=True
2279 )
2280 )
2282 def __getstate__(self):
2283 """execute a deferred value for serialization purposes."""
2285 d = self.__dict__.copy()
2286 v = self.value
2287 if self.callable:
2288 v = self.callable()
2289 d["callable"] = None
2290 d["value"] = v
2291 return d
2293 def __setstate__(self, state):
2294 if state.get("unique", False):
2295 anon_and_key = _anonymous_label.safe_construct_with_key(
2296 id(self), state.get("_orig_key", "param"), sanitize_key=True
2297 )
2298 state["key"], state["_anon_map_key"] = anon_and_key
2299 self.__dict__.update(state)
2301 def __repr__(self):
2302 return "%s(%r, %r, type_=%r)" % (
2303 self.__class__.__name__,
2304 self.key,
2305 self.value,
2306 self.type,
2307 )
2310class TypeClause(DQLDMLClauseElement):
2311 """Handle a type keyword in a SQL statement.
2313 Used by the ``Case`` statement.
2315 """
2317 __visit_name__ = "typeclause"
2319 _traverse_internals: _TraverseInternalsType = [
2320 ("type", InternalTraversal.dp_type)
2321 ]
2322 type: TypeEngine[Any]
2324 def __init__(self, type_: TypeEngine[Any]):
2325 self.type = type_
2328class AbstractTextClause(
2329 roles.DDLConstraintColumnRole,
2330 roles.DDLExpressionRole,
2331 roles.StatementOptionRole,
2332 roles.WhereHavingRole,
2333 roles.OrderByRole,
2334 roles.FromClauseRole,
2335 roles.SelectStatementRole,
2336 roles.InElementRole,
2337 Generative,
2338 ExecutableStatement,
2339 DQLDMLClauseElement,
2340 roles.BinaryElementRole[Any],
2341):
2342 """Base class for textual SQL constructs like TextClause and TString."""
2344 __visit_name__: str
2346 _is_text_clause = True
2347 _is_textual = True
2348 _is_implicitly_boolean = False
2349 _render_label_in_columns_clause = False
2350 _omit_from_statements = False
2351 _is_collection_aggregate = False
2353 @property
2354 def _hide_froms(self) -> Iterable[FromClause]:
2355 return ()
2357 def __and__(self, other):
2358 # support use in select.where(), query.filter()
2359 return and_(self, other)
2361 @property
2362 def _select_iterable(self) -> _SelectIterable:
2363 return (self,)
2365 # help in those cases where text/tstring() is
2366 # interpreted in a column expression situation
2367 key: Optional[str] = None
2368 _label: Optional[str] = None
2370 _allow_label_resolve = False
2372 @property
2373 def type(self) -> TypeEngine[Any]:
2374 return type_api.NULLTYPE
2376 @property
2377 def comparator(self):
2378 return self.type.comparator_factory(self) # type: ignore[arg-type]
2380 def self_group(
2381 self, against: Optional[OperatorType] = None
2382 ) -> Union[Self, Grouping[Any]]:
2383 if against is operators.in_op:
2384 return Grouping(self)
2385 else:
2386 return self
2388 def bindparams(
2389 self,
2390 *binds: BindParameter[Any],
2391 **names_to_values: Any,
2392 ) -> Self:
2393 """Establish the values and/or types of bound parameters within
2394 this :class:`_expression.AbstractTextClause` construct.
2396 This is implemented only for :class:`.TextClause` will raise
2397 ``NotImplementedError`` for :class:`.TString`.
2399 """
2400 raise NotImplementedError()
2402 @util.preload_module("sqlalchemy.sql.selectable")
2403 def columns(
2404 self,
2405 *cols: _OnlyColumnArgument[Any],
2406 **types: _TypeEngineArgument[Any],
2407 ) -> TextualSelect:
2408 r"""Turn this :class:`_expression.AbstractTextClause` object into a
2409 :class:`_expression.TextualSelect`
2410 object that serves the same role as a SELECT
2411 statement.
2413 The :class:`_expression.TextualSelect` is part of the
2414 :class:`_expression.SelectBase`
2415 hierarchy and can be embedded into another statement by using the
2416 :meth:`_expression.TextualSelect.subquery` method to produce a
2417 :class:`.Subquery`
2418 object, which can then be SELECTed from.
2420 This function essentially bridges the gap between an entirely
2421 textual SELECT statement and the SQL expression language concept
2422 of a "selectable"::
2424 from sqlalchemy.sql import column, text
2426 stmt = text("SELECT id, name FROM some_table")
2427 stmt = stmt.columns(column("id"), column("name")).subquery("st")
2429 stmt = (
2430 select(mytable)
2431 .select_from(mytable.join(stmt, mytable.c.name == stmt.c.name))
2432 .where(stmt.c.id > 5)
2433 )
2435 Above, we pass a series of :func:`_expression.column` elements to the
2436 :meth:`_expression.AbstractTextClause.columns` method positionally.
2437 These :func:`_expression.column` elements now become first class
2438 elements upon the :attr:`_expression.TextualSelect.selected_columns`
2439 column collection, which then become part of the :attr:`.Subquery.c`
2440 collection after :meth:`_expression.TextualSelect.subquery` is invoked.
2442 The column expressions we pass to
2443 :meth:`_expression.AbstractTextClause.columns` may also be typed; when
2444 we do so, these :class:`.TypeEngine` objects become the effective
2445 return type of the column, so that SQLAlchemy's result-set-processing
2446 systems may be used on the return values. This is often needed for
2447 types such as date or boolean types, as well as for unicode processing
2448 on some dialect configurations::
2450 stmt = text("SELECT id, name, timestamp FROM some_table")
2451 stmt = stmt.columns(
2452 column("id", Integer),
2453 column("name", Unicode),
2454 column("timestamp", DateTime),
2455 )
2457 for id, name, timestamp in connection.execute(stmt):
2458 print(id, name, timestamp)
2460 As a shortcut to the above syntax, keyword arguments referring to
2461 types alone may be used, if only type conversion is needed::
2463 stmt = text("SELECT id, name, timestamp FROM some_table")
2464 stmt = stmt.columns(id=Integer, name=Unicode, timestamp=DateTime)
2466 for id, name, timestamp in connection.execute(stmt):
2467 print(id, name, timestamp)
2469 The positional form of :meth:`_expression.AbstractTextClause.columns`
2470 also provides the unique feature of **positional column targeting**,
2471 which is particularly useful when using the ORM with complex textual
2472 queries. If we specify the columns from our model to
2473 :meth:`_expression.AbstractTextClause.columns`, the result set will
2474 match to those columns positionally, meaning the name or origin of the
2475 column in the textual SQL doesn't matter::
2477 stmt = text(
2478 "SELECT users.id, addresses.id, users.id, "
2479 "users.name, addresses.email_address AS email "
2480 "FROM users JOIN addresses ON users.id=addresses.user_id "
2481 "WHERE users.id = 1"
2482 ).columns(
2483 User.id,
2484 Address.id,
2485 Address.user_id,
2486 User.name,
2487 Address.email_address,
2488 )
2490 query = (
2491 session.query(User)
2492 .from_statement(stmt)
2493 .options(contains_eager(User.addresses))
2494 )
2496 The :meth:`_expression.AbstractTextClause.columns` method provides a
2497 direct route to calling :meth:`_expression.FromClause.subquery` as well
2498 as :meth:`_expression.SelectBase.cte` against a textual SELECT
2499 statement::
2501 stmt = stmt.columns(id=Integer, name=String).cte("st")
2503 stmt = select(sometable).where(sometable.c.id == stmt.c.id)
2505 :param \*cols: A series of :class:`_expression.ColumnElement` objects,
2506 typically
2507 :class:`_schema.Column` objects from a :class:`_schema.Table`
2508 or ORM level
2509 column-mapped attributes, representing a set of columns that this
2510 textual string will SELECT from.
2512 :param \**types: A mapping of string names to :class:`.TypeEngine`
2513 type objects indicating the datatypes to use for names that are
2514 SELECTed from the textual string. Prefer to use the ``*cols``
2515 argument as it also indicates positional ordering.
2517 """
2518 selectable = util.preloaded.sql_selectable
2520 input_cols: List[NamedColumn[Any]] = [
2521 coercions.expect(roles.LabeledColumnExprRole, col) for col in cols
2522 ]
2524 positional_input_cols = [
2525 (
2526 ColumnClause(col.key, types.pop(col.key))
2527 if col.key in types
2528 else col
2529 )
2530 for col in input_cols
2531 ]
2532 keyed_input_cols: List[NamedColumn[Any]] = [
2533 ColumnClause(key, type_) for key, type_ in types.items()
2534 ]
2536 elem = selectable.TextualSelect.__new__(selectable.TextualSelect)
2537 elem._init(
2538 self,
2539 positional_input_cols + keyed_input_cols,
2540 positional=bool(positional_input_cols) and not keyed_input_cols,
2541 )
2542 return elem
2545class TextClause(AbstractTextClause, inspection.Inspectable["TextClause"]):
2546 """Represent a literal SQL text fragment.
2548 E.g.::
2550 from sqlalchemy import text
2552 t = text("SELECT * FROM users")
2553 result = connection.execute(t)
2555 The :class:`_expression.TextClause` construct is produced using the
2556 :func:`_expression.text`
2557 function; see that function for full documentation.
2559 .. seealso::
2561 :func:`_expression.text`
2563 """
2565 __visit_name__ = "textclause"
2567 _traverse_internals: _TraverseInternalsType = [
2568 ("_bindparams", InternalTraversal.dp_string_clauseelement_dict),
2569 ("text", InternalTraversal.dp_string),
2570 ] + ExecutableStatement._executable_traverse_internals
2572 _bind_params_regex = re.compile(r"(?<![:\w\x5c]):(\w+)(?!:)", re.UNICODE)
2574 @property
2575 def _is_star(self) -> bool: # type: ignore[override]
2576 return self.text == "*"
2578 def __init__(self, text: str):
2579 self._bindparams: Dict[str, BindParameter[Any]] = {}
2581 def repl(m):
2582 self._bindparams[m.group(1)] = BindParameter(m.group(1))
2583 return ":%s" % m.group(1)
2585 # scan the string and search for bind parameter names, add them
2586 # to the list of bindparams
2587 self.text = self._bind_params_regex.sub(repl, text)
2589 @_generative
2590 def bindparams(
2591 self,
2592 *binds: BindParameter[Any],
2593 **names_to_values: Any,
2594 ) -> Self:
2595 """Establish the values and/or types of bound parameters within
2596 this :class:`_expression.TextClause` construct.
2598 Given a text construct such as::
2600 from sqlalchemy import text
2602 stmt = text(
2603 "SELECT id, name FROM user WHERE name=:name AND timestamp=:timestamp"
2604 )
2606 the :meth:`_expression.TextClause.bindparams`
2607 method can be used to establish
2608 the initial value of ``:name`` and ``:timestamp``,
2609 using simple keyword arguments::
2611 stmt = stmt.bindparams(
2612 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2613 )
2615 Where above, new :class:`.BindParameter` objects
2616 will be generated with the names ``name`` and ``timestamp``, and
2617 values of ``jack`` and ``datetime.datetime(2012, 10, 8, 15, 12, 5)``,
2618 respectively. The types will be
2619 inferred from the values given, in this case :class:`.String` and
2620 :class:`.DateTime`.
2622 When specific typing behavior is needed, the positional ``*binds``
2623 argument can be used in which to specify :func:`.bindparam` constructs
2624 directly. These constructs must include at least the ``key``
2625 argument, then an optional value and type::
2627 from sqlalchemy import bindparam
2629 stmt = stmt.bindparams(
2630 bindparam("name", value="jack", type_=String),
2631 bindparam("timestamp", type_=DateTime),
2632 )
2634 Above, we specified the type of :class:`.DateTime` for the
2635 ``timestamp`` bind, and the type of :class:`.String` for the ``name``
2636 bind. In the case of ``name`` we also set the default value of
2637 ``"jack"``.
2639 Additional bound parameters can be supplied at statement execution
2640 time, e.g.::
2642 result = connection.execute(
2643 stmt, timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2644 )
2646 The :meth:`_expression.TextClause.bindparams`
2647 method can be called repeatedly,
2648 where it will reuse existing :class:`.BindParameter` objects to add
2649 new information. For example, we can call
2650 :meth:`_expression.TextClause.bindparams`
2651 first with typing information, and a
2652 second time with value information, and it will be combined::
2654 stmt = text(
2655 "SELECT id, name FROM user WHERE name=:name "
2656 "AND timestamp=:timestamp"
2657 )
2658 stmt = stmt.bindparams(
2659 bindparam("name", type_=String), bindparam("timestamp", type_=DateTime)
2660 )
2661 stmt = stmt.bindparams(
2662 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2663 )
2665 The :meth:`_expression.TextClause.bindparams`
2666 method also supports the concept of
2667 **unique** bound parameters. These are parameters that are
2668 "uniquified" on name at statement compilation time, so that multiple
2669 :func:`_expression.text`
2670 constructs may be combined together without the names
2671 conflicting. To use this feature, specify the
2672 :paramref:`.BindParameter.unique` flag on each :func:`.bindparam`
2673 object::
2675 stmt1 = text("select id from table where name=:name").bindparams(
2676 bindparam("name", value="name1", unique=True)
2677 )
2678 stmt2 = text("select id from table where name=:name").bindparams(
2679 bindparam("name", value="name2", unique=True)
2680 )
2682 union = union_all(stmt1.columns(column("id")), stmt2.columns(column("id")))
2684 The above statement will render as:
2686 .. sourcecode:: sql
2688 select id from table where name=:name_1
2689 UNION ALL select id from table where name=:name_2
2691 """ # noqa: E501
2692 self._bindparams = new_params = self._bindparams.copy()
2694 for bind in binds:
2695 try:
2696 # the regex used for text() currently will not match
2697 # a unique/anonymous key in any case, so use the _orig_key
2698 # so that a text() construct can support unique parameters
2699 existing = new_params[bind._orig_key]
2700 except KeyError as err:
2701 raise exc.ArgumentError(
2702 "This text() construct doesn't define a "
2703 "bound parameter named %r" % bind._orig_key
2704 ) from err
2705 else:
2706 new_params[existing._orig_key] = bind
2708 for key, value in names_to_values.items():
2709 try:
2710 existing = new_params[key]
2711 except KeyError as err:
2712 raise exc.ArgumentError(
2713 "This text() construct doesn't define a "
2714 "bound parameter named %r" % key
2715 ) from err
2716 else:
2717 new_params[key] = existing._with_value(value, required=False)
2718 return self
2720 @property
2721 def type(self) -> TypeEngine[Any]:
2722 return type_api.NULLTYPE
2724 @property
2725 def comparator(self):
2726 # TODO: this seems wrong, it seems like we might not
2727 # be using this method.
2728 return self.type.comparator_factory(self) # type: ignore[arg-type]
2730 def self_group(
2731 self, against: Optional[OperatorType] = None
2732 ) -> Union[Self, Grouping[Any]]:
2733 if against is operators.in_op:
2734 return Grouping(self)
2735 else:
2736 return self
2739class TString(AbstractTextClause, inspection.Inspectable["TString"]):
2740 """Represent a SQL template string using Python 3.14+ t-strings.
2742 E.g.::
2744 from sqlalchemy import tstring, column
2746 a = 5
2747 b = 10
2748 stmt = tstring(t"select {a}, {b}")
2749 result = connection.execute(stmt)
2751 The :class:`_expression.TString` construct is produced using the
2752 :func:`_expression.tstring` function; see that function for full
2753 documentation.
2755 .. versionadded:: 2.1
2757 .. seealso::
2759 :func:`_expression.tstring`
2761 """
2763 __visit_name__ = "tstring"
2765 _traverse_internals: _TraverseInternalsType = [
2766 ("parts", InternalTraversal.dp_clauseelement_list)
2767 ] + ExecutableStatement._executable_traverse_internals
2769 @property
2770 def _is_star(self) -> bool: # type: ignore[override]
2771 return (
2772 len(self.parts) == 1
2773 and isinstance(self.parts[0], TextClause)
2774 and self.parts[0]._is_star
2775 )
2777 def __init__(self, template: Template):
2778 """Construct a :class:`_expression.TString` from a Python 3.14+
2779 template string.
2781 :param template: a Python 3.14+ template string (t-string) that
2782 contains SQL fragments and Python expressions to be interpolated.
2784 """
2785 self.parts: List[ClauseElement] = []
2787 if not isinstance(template, Template):
2788 raise exc.ArgumentError("pep-750 Tstring (e.g. t'...') expected")
2790 for part in template:
2791 if isinstance(part, str):
2792 self.parts.append(TextClause(part))
2793 else:
2794 assert hasattr(part, "value")
2795 self.parts.append(
2796 coercions.expect(roles.TStringElementRole, part.value)
2797 )
2799 def bindparams(
2800 self,
2801 *binds: BindParameter[Any],
2802 **names_to_values: Any,
2803 ) -> Self:
2804 """Not supported for TString constructs.
2806 TString constructs do not support .bindparams(). Bind parameters
2807 are automatically created from interpolated values.
2809 """
2810 raise NotImplementedError(
2811 "TString constructs do not support .bindparams(). "
2812 "Bind parameters are automatically created "
2813 "from interpolated values."
2814 )
2817class Null(SingletonConstant, roles.ConstExprRole[None], ColumnElement[None]):
2818 """Represent the NULL keyword in a SQL statement.
2820 :class:`.Null` is accessed as a constant via the
2821 :func:`.null` function.
2823 """
2825 __visit_name__ = "null"
2827 _traverse_internals: _TraverseInternalsType = []
2828 _singleton: Null
2830 if not TYPE_CHECKING:
2832 @util.memoized_property
2833 def type(self) -> TypeEngine[_T]: # noqa: A001
2834 return type_api.NULLTYPE
2836 @classmethod
2837 def _instance(cls) -> Null:
2838 """Return a constant :class:`.Null` construct."""
2840 return Null._singleton
2843Null._create_singleton()
2846class False_(
2847 SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]
2848):
2849 """Represent the ``false`` keyword, or equivalent, in a SQL statement.
2851 :class:`.False_` is accessed as a constant via the
2852 :func:`.false` function.
2854 """
2856 __visit_name__ = "false"
2857 _traverse_internals: _TraverseInternalsType = []
2858 _singleton: False_
2860 if not TYPE_CHECKING:
2862 @util.memoized_property
2863 def type(self) -> TypeEngine[_T]: # noqa: A001
2864 return type_api.BOOLEANTYPE
2866 def _negate(self) -> True_:
2867 return True_._singleton
2869 @classmethod
2870 def _instance(cls) -> False_:
2871 return False_._singleton
2874False_._create_singleton()
2877class True_(SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]):
2878 """Represent the ``true`` keyword, or equivalent, in a SQL statement.
2880 :class:`.True_` is accessed as a constant via the
2881 :func:`.true` function.
2883 """
2885 __visit_name__ = "true"
2887 _traverse_internals: _TraverseInternalsType = []
2888 _singleton: True_
2890 if not TYPE_CHECKING:
2892 @util.memoized_property
2893 def type(self) -> TypeEngine[_T]: # noqa: A001
2894 return type_api.BOOLEANTYPE
2896 def _negate(self) -> False_:
2897 return False_._singleton
2899 @classmethod
2900 def _ifnone(
2901 cls, other: Optional[ColumnElement[Any]]
2902 ) -> ColumnElement[Any]:
2903 if other is None:
2904 return cls._instance()
2905 else:
2906 return other
2908 @classmethod
2909 def _instance(cls) -> True_:
2910 return True_._singleton
2913True_._create_singleton()
2916class ElementList(DQLDMLClauseElement):
2917 """Describe a list of clauses that will be space separated.
2919 This is a minimal version of :class:`.ClauseList` which is used by
2920 the :class:`.HasSyntaxExtension` class. It does not do any coercions
2921 so should be used internally only.
2923 .. versionadded:: 2.1
2925 """
2927 __visit_name__ = "element_list"
2929 _traverse_internals: _TraverseInternalsType = [
2930 ("clauses", InternalTraversal.dp_clauseelement_tuple),
2931 ]
2933 clauses: typing_Tuple[ClauseElement, ...]
2935 def __init__(self, clauses: Sequence[ClauseElement]):
2936 self.clauses = tuple(clauses)
2939class OrderByList(
2940 roles.OrderByRole,
2941 operators.OrderingOperators,
2942 DQLDMLClauseElement,
2943):
2944 """Describe a list of clauses that will be comma separated to nest
2945 within an ORDER BY.
2947 .. versionadded:: 2.1
2949 """
2951 __visit_name__ = "order_by_list"
2953 _traverse_internals: _TraverseInternalsType = [
2954 ("clauses", InternalTraversal.dp_clauseelement_tuple),
2955 ]
2957 clauses: List[ColumnElement[Any]]
2959 def __init__(
2960 self,
2961 clauses: Iterable[Union[OrderByList, _ColumnExpressionArgument[Any]]],
2962 ):
2963 text_converter_role: Type[roles.SQLRole] = roles.ByOfRole
2964 self._text_converter_role = text_converter_role
2966 self.clauses = [
2967 coercions.expect(
2968 text_converter_role, clause, apply_propagate_attrs=self
2969 )
2970 for clause in clauses
2971 ]
2973 def __iter__(self) -> Iterator[ColumnElement[Any]]:
2974 return iter(self.clauses)
2976 def __len__(self) -> int:
2977 return len(self.clauses)
2979 @property
2980 def _select_iterable(self) -> _SelectIterable:
2981 return itertools.chain.from_iterable(
2982 [elem._select_iterable for elem in self.clauses]
2983 )
2985 @util.ro_non_memoized_property
2986 def _from_objects(self) -> List[FromClause]:
2987 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
2989 def self_group(
2990 self, against: Optional[OperatorType] = None
2991 ) -> Union[Self, Grouping[Any]]:
2992 return self
2994 def desc(self) -> OrderByList:
2995 return OrderByList([e.desc() for e in self.clauses])
2997 def asc(self) -> OrderByList:
2998 return OrderByList([e.asc() for e in self.clauses])
3000 def nulls_first(self) -> OrderByList:
3001 return OrderByList([e.nulls_first() for e in self.clauses])
3003 def nulls_last(self) -> OrderByList:
3004 return OrderByList([e.nulls_last() for e in self.clauses])
3007class ClauseList(
3008 roles.InElementRole,
3009 roles.OrderByRole,
3010 roles.ColumnsClauseRole,
3011 roles.DMLColumnRole,
3012 DQLDMLClauseElement,
3013):
3014 """Describe a list of clauses, separated by an operator.
3016 By default, is comma-separated, such as a column listing.
3018 """
3020 __visit_name__ = "clauselist"
3022 # Used by ORM context.py to identify ClauseList objects in legacy
3023 # composite attribute queries (see test_query_cols_legacy test)
3024 _is_clause_list = True
3026 _traverse_internals: _TraverseInternalsType = [
3027 ("clauses", InternalTraversal.dp_clauseelement_list),
3028 ("operator", InternalTraversal.dp_operator),
3029 ]
3031 clauses: List[ColumnElement[Any]]
3033 def __init__(
3034 self,
3035 *clauses: _ColumnExpressionArgument[Any],
3036 operator: OperatorType = operators.comma_op,
3037 group: bool = True,
3038 group_contents: bool = True,
3039 _literal_as_text_role: Type[roles.SQLRole] = roles.WhereHavingRole,
3040 ):
3041 self.operator = operator
3042 self.group = group
3043 self.group_contents = group_contents
3044 clauses_iterator: Iterable[_ColumnExpressionArgument[Any]] = clauses
3045 text_converter_role: Type[roles.SQLRole] = _literal_as_text_role
3046 self._text_converter_role = text_converter_role
3048 if self.group_contents:
3049 self.clauses = [
3050 coercions.expect(
3051 text_converter_role, clause, apply_propagate_attrs=self
3052 ).self_group(against=self.operator)
3053 for clause in clauses_iterator
3054 ]
3055 else:
3056 self.clauses = [
3057 coercions.expect(
3058 text_converter_role, clause, apply_propagate_attrs=self
3059 )
3060 for clause in clauses_iterator
3061 ]
3062 self._is_implicitly_boolean = operators.is_boolean(self.operator)
3064 @classmethod
3065 def _construct_raw(
3066 cls,
3067 operator: OperatorType,
3068 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
3069 ) -> ClauseList:
3070 self = cls.__new__(cls)
3071 self.clauses = list(clauses) if clauses else []
3072 self.group = True
3073 self.operator = operator
3074 self.group_contents = True
3075 self._is_implicitly_boolean = False
3076 return self
3078 def __iter__(self) -> Iterator[ColumnElement[Any]]:
3079 return iter(self.clauses)
3081 def __len__(self) -> int:
3082 return len(self.clauses)
3084 @property
3085 def _select_iterable(self) -> _SelectIterable:
3086 return itertools.chain.from_iterable(
3087 [elem._select_iterable for elem in self.clauses]
3088 )
3090 def append(self, clause):
3091 if self.group_contents:
3092 self.clauses.append(
3093 coercions.expect(self._text_converter_role, clause).self_group(
3094 against=self.operator
3095 )
3096 )
3097 else:
3098 self.clauses.append(
3099 coercions.expect(self._text_converter_role, clause)
3100 )
3102 @util.ro_non_memoized_property
3103 def _from_objects(self) -> List[FromClause]:
3104 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
3106 def self_group(
3107 self, against: Optional[OperatorType] = None
3108 ) -> Union[Self, Grouping[Any]]:
3109 if self.group and operators.is_precedent(self.operator, against):
3110 return Grouping(self)
3111 else:
3112 return self
3115class OperatorExpression(ColumnElement[_T]):
3116 """base for expressions that contain an operator and operands
3118 .. versionadded:: 2.0
3120 """
3122 operator: OperatorType
3123 type: TypeEngine[_T]
3125 group: bool = True
3127 @property
3128 def is_comparison(self):
3129 return operators.is_comparison(self.operator)
3131 def self_group(
3132 self, against: Optional[OperatorType] = None
3133 ) -> Union[Self, Grouping[_T]]:
3134 if (
3135 self.group
3136 and operators.is_precedent(self.operator, against)
3137 or (
3138 # a negate against a non-boolean operator
3139 # doesn't make too much sense but we should
3140 # group for that
3141 against is operators.inv
3142 and not operators.is_boolean(self.operator)
3143 )
3144 ):
3145 return Grouping(self)
3146 else:
3147 return self
3149 @property
3150 def _flattened_operator_clauses(
3151 self,
3152 ) -> typing_Tuple[ColumnElement[Any], ...]:
3153 raise NotImplementedError()
3155 @classmethod
3156 def _construct_for_op(
3157 cls,
3158 left: ColumnElement[Any],
3159 right: ColumnElement[Any],
3160 op: OperatorType,
3161 *,
3162 type_: TypeEngine[_T],
3163 negate: Optional[OperatorType] = None,
3164 modifiers: Optional[Mapping[str, Any]] = None,
3165 ) -> OperatorExpression[_T]:
3166 if operators.is_associative(op):
3167 assert (
3168 negate is None
3169 ), f"negate not supported for associative operator {op}"
3171 multi = False
3172 if getattr(
3173 left, "operator", None
3174 ) is op and type_._compare_type_affinity(left.type):
3175 multi = True
3176 left_flattened = left._flattened_operator_clauses
3177 else:
3178 left_flattened = (left,)
3180 if getattr(
3181 right, "operator", None
3182 ) is op and type_._compare_type_affinity(right.type):
3183 multi = True
3184 right_flattened = right._flattened_operator_clauses
3185 else:
3186 right_flattened = (right,)
3188 if multi:
3189 return ExpressionClauseList._construct_for_list(
3190 op,
3191 type_,
3192 *(left_flattened + right_flattened),
3193 )
3195 if left._is_collection_aggregate or right._is_collection_aggregate:
3196 negate = None
3198 return BinaryExpression(
3199 left, right, op, type_=type_, negate=negate, modifiers=modifiers
3200 )
3203class ExpressionClauseList(OperatorExpression[_T]):
3204 """Describe a list of clauses, separated by an operator,
3205 in a column expression context.
3207 :class:`.ExpressionClauseList` differs from :class:`.ClauseList` in that
3208 it represents a column-oriented DQL expression only, not an open ended
3209 list of anything comma separated.
3211 .. versionadded:: 2.0
3213 """
3215 __visit_name__ = "expression_clauselist"
3217 _traverse_internals: _TraverseInternalsType = [
3218 ("clauses", InternalTraversal.dp_clauseelement_tuple),
3219 ("operator", InternalTraversal.dp_operator),
3220 ]
3222 clauses: typing_Tuple[ColumnElement[Any], ...]
3224 group: bool
3226 def __init__(
3227 self,
3228 operator: OperatorType,
3229 *clauses: _ColumnExpressionArgument[Any],
3230 type_: Optional[_TypeEngineArgument[_T]] = None,
3231 ):
3232 self.operator = operator
3234 self.clauses = tuple(
3235 coercions.expect(
3236 roles.ExpressionElementRole, clause, apply_propagate_attrs=self
3237 )
3238 for clause in clauses
3239 )
3240 self._is_implicitly_boolean = operators.is_boolean(self.operator)
3241 self.type = type_api.to_instance(type_) # type: ignore[assignment]
3243 @property
3244 def _flattened_operator_clauses(
3245 self,
3246 ) -> typing_Tuple[ColumnElement[Any], ...]:
3247 return self.clauses
3249 def __iter__(self) -> Iterator[ColumnElement[Any]]:
3250 return iter(self.clauses)
3252 def __len__(self) -> int:
3253 return len(self.clauses)
3255 @property
3256 def _select_iterable(self) -> _SelectIterable:
3257 return (self,)
3259 @util.ro_non_memoized_property
3260 def _from_objects(self) -> List[FromClause]:
3261 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
3263 def _append_inplace(self, clause: ColumnElement[Any]) -> None:
3264 self.clauses += (clause,)
3266 @classmethod
3267 def _construct_for_list(
3268 cls,
3269 operator: OperatorType,
3270 type_: TypeEngine[_T],
3271 *clauses: ColumnElement[Any],
3272 group: bool = True,
3273 ) -> ExpressionClauseList[_T]:
3274 self = cls.__new__(cls)
3275 self.group = group
3276 if group:
3277 self.clauses = tuple(
3278 c.self_group(against=operator) for c in clauses
3279 )
3280 else:
3281 self.clauses = clauses
3282 self.operator = operator
3283 self.type = type_
3284 for c in clauses:
3285 if c._propagate_attrs:
3286 self._propagate_attrs = c._propagate_attrs
3287 break
3288 return self
3290 def _negate(self) -> Any:
3291 grouped = self.self_group(against=operators.inv)
3292 assert isinstance(grouped, ColumnElement)
3293 return UnaryExpression(grouped, operator=operators.inv)
3296class BooleanClauseList(ExpressionClauseList[bool]):
3297 __visit_name__ = "expression_clauselist"
3298 inherit_cache = True
3300 def __init__(self, *arg, **kw):
3301 raise NotImplementedError(
3302 "BooleanClauseList has a private constructor"
3303 )
3305 @classmethod
3306 def _process_clauses_for_boolean(
3307 cls,
3308 operator: OperatorType,
3309 continue_on: Any,
3310 skip_on: Any,
3311 clauses: Iterable[ColumnElement[Any]],
3312 ) -> typing_Tuple[int, List[ColumnElement[Any]]]:
3313 has_continue_on = None
3315 convert_clauses = []
3317 against = operators._asbool
3318 lcc = 0
3320 for clause in clauses:
3321 if clause is continue_on:
3322 # instance of continue_on, like and_(x, y, True, z), store it
3323 # if we didn't find one already, we will use it if there
3324 # are no other expressions here.
3325 has_continue_on = clause
3326 elif clause is skip_on:
3327 # instance of skip_on, e.g. and_(x, y, False, z), cancels
3328 # the rest out
3329 convert_clauses = [clause]
3330 lcc = 1
3331 break
3332 else:
3333 if not lcc:
3334 lcc = 1
3335 else:
3336 against = operator
3337 # technically this would be len(convert_clauses) + 1
3338 # however this only needs to indicate "greater than one"
3339 lcc = 2
3340 convert_clauses.append(clause)
3342 if not convert_clauses and has_continue_on is not None:
3343 convert_clauses = [has_continue_on]
3344 lcc = 1
3346 return lcc, [c.self_group(against=against) for c in convert_clauses]
3348 @classmethod
3349 def _construct(
3350 cls,
3351 operator: OperatorType,
3352 continue_on: Any,
3353 skip_on: Any,
3354 initial_clause: Any = _NoArg.NO_ARG,
3355 *clauses: Any,
3356 **kw: Any,
3357 ) -> ColumnElement[Any]:
3358 if initial_clause is _NoArg.NO_ARG:
3359 # no elements period. deprecated use case. return an empty
3360 # ClauseList construct that generates nothing unless it has
3361 # elements added to it.
3362 name = operator.__name__
3364 util.warn_deprecated(
3365 f"Invoking {name}() without arguments is deprecated, and "
3366 f"will be disallowed in a future release. For an empty "
3367 f"""{name}() construct, use '{name}({
3368 'true()' if continue_on is True_._singleton else 'false()'
3369 }, *args)' """
3370 f"""or '{name}({
3371 'True' if continue_on is True_._singleton else 'False'
3372 }, *args)'.""",
3373 version="1.4",
3374 )
3375 return cls._construct_raw(operator)
3377 lcc, convert_clauses = cls._process_clauses_for_boolean(
3378 operator,
3379 continue_on,
3380 skip_on,
3381 [
3382 coercions.expect(roles.WhereHavingRole, clause)
3383 for clause in util.coerce_generator_arg(
3384 (initial_clause,) + clauses
3385 )
3386 ],
3387 )
3389 if lcc > 1:
3390 # multiple elements. Return regular BooleanClauseList
3391 # which will link elements against the operator.
3393 flattened_clauses = itertools.chain.from_iterable(
3394 (
3395 (c for c in to_flat._flattened_operator_clauses)
3396 if getattr(to_flat, "operator", None) is operator
3397 else (to_flat,)
3398 )
3399 for to_flat in convert_clauses
3400 )
3402 return cls._construct_raw(operator, flattened_clauses) # type: ignore[arg-type] # noqa: E501
3403 else:
3404 assert lcc
3405 # just one element. return it as a single boolean element,
3406 # not a list and discard the operator.
3407 return convert_clauses[0]
3409 @classmethod
3410 def _construct_for_whereclause(
3411 cls, clauses: Iterable[ColumnElement[Any]]
3412 ) -> Optional[ColumnElement[bool]]:
3413 operator, continue_on, skip_on = (
3414 operators.and_,
3415 True_._singleton,
3416 False_._singleton,
3417 )
3419 lcc, convert_clauses = cls._process_clauses_for_boolean(
3420 operator,
3421 continue_on,
3422 skip_on,
3423 clauses, # these are assumed to be coerced already
3424 )
3426 if lcc > 1:
3427 # multiple elements. Return regular BooleanClauseList
3428 # which will link elements against the operator.
3429 return cls._construct_raw(operator, convert_clauses)
3430 elif lcc == 1:
3431 # just one element. return it as a single boolean element,
3432 # not a list and discard the operator.
3433 return convert_clauses[0]
3434 else:
3435 return None
3437 @classmethod
3438 def _construct_raw(
3439 cls,
3440 operator: OperatorType,
3441 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
3442 ) -> BooleanClauseList:
3443 self = cls.__new__(cls)
3444 self.clauses = tuple(clauses) if clauses else ()
3445 self.group = True
3446 self.operator = operator
3447 self.type = type_api.BOOLEANTYPE
3448 self._is_implicitly_boolean = True
3449 return self
3451 @classmethod
3452 def and_(
3453 cls,
3454 initial_clause: Union[
3455 Literal[True], _ColumnExpressionArgument[bool], _NoArg
3456 ] = _NoArg.NO_ARG,
3457 *clauses: _ColumnExpressionArgument[bool],
3458 ) -> ColumnElement[bool]:
3459 r"""Produce a conjunction of expressions joined by ``AND``.
3461 See :func:`_sql.and_` for full documentation.
3462 """
3463 return cls._construct(
3464 operators.and_,
3465 True_._singleton,
3466 False_._singleton,
3467 initial_clause,
3468 *clauses,
3469 )
3471 @classmethod
3472 def or_(
3473 cls,
3474 initial_clause: Union[
3475 Literal[False], _ColumnExpressionArgument[bool], _NoArg
3476 ] = _NoArg.NO_ARG,
3477 *clauses: _ColumnExpressionArgument[bool],
3478 ) -> ColumnElement[bool]:
3479 """Produce a conjunction of expressions joined by ``OR``.
3481 See :func:`_sql.or_` for full documentation.
3482 """
3483 return cls._construct(
3484 operators.or_,
3485 False_._singleton,
3486 True_._singleton,
3487 initial_clause,
3488 *clauses,
3489 )
3491 @property
3492 def _select_iterable(self) -> _SelectIterable:
3493 return (self,)
3495 def self_group(
3496 self, against: Optional[OperatorType] = None
3497 ) -> Union[Self, Grouping[bool]]:
3498 if not self.clauses:
3499 return self
3500 else:
3501 return super().self_group(against=against)
3504and_ = BooleanClauseList.and_
3505or_ = BooleanClauseList.or_
3508class Tuple(ClauseList, ColumnElement[TupleAny]):
3509 """Represent a SQL tuple."""
3511 __visit_name__ = "tuple"
3513 _traverse_internals: _TraverseInternalsType = (
3514 ClauseList._traverse_internals + []
3515 )
3517 type: TupleType
3519 @util.preload_module("sqlalchemy.sql.sqltypes")
3520 def __init__(
3521 self,
3522 *clauses: _ColumnExpressionArgument[Any],
3523 types: Optional[Sequence[_TypeEngineArgument[Any]]] = None,
3524 ):
3525 sqltypes = util.preloaded.sql_sqltypes
3527 if types is None:
3528 init_clauses: List[ColumnElement[Any]] = [
3529 coercions.expect(roles.ExpressionElementRole, c)
3530 for c in clauses
3531 ]
3532 else:
3533 if len(types) != len(clauses):
3534 raise exc.ArgumentError(
3535 "Wrong number of elements for %d-tuple: %r "
3536 % (len(types), clauses)
3537 )
3538 init_clauses = [
3539 coercions.expect(
3540 roles.ExpressionElementRole,
3541 c,
3542 type_=typ if not typ._isnull else None,
3543 )
3544 for typ, c in zip(types, clauses)
3545 ]
3547 self.type = sqltypes.TupleType(*[arg.type for arg in init_clauses])
3548 super().__init__(*init_clauses)
3550 @property
3551 def _select_iterable(self) -> _SelectIterable:
3552 return (self,)
3554 def _bind_param(self, operator, obj, type_=None, expanding=False):
3555 if expanding:
3556 return BindParameter(
3557 None,
3558 value=obj,
3559 _compared_to_operator=operator,
3560 unique=True,
3561 expanding=True,
3562 type_=type_,
3563 _compared_to_type=self.type,
3564 )
3565 else:
3566 return Tuple(
3567 *[
3568 BindParameter(
3569 None,
3570 o,
3571 _compared_to_operator=operator,
3572 _compared_to_type=compared_to_type,
3573 unique=True,
3574 type_=type_,
3575 )
3576 for o, compared_to_type in zip(obj, self.type.types)
3577 ]
3578 )
3580 def self_group(self, against: Optional[OperatorType] = None) -> Self:
3581 # Tuple is parenthesized by definition.
3582 return self
3585class Case(ColumnElement[_T]):
3586 """Represent a ``CASE`` expression.
3588 :class:`.Case` is produced using the :func:`.case` factory function,
3589 as in::
3591 from sqlalchemy import case
3593 stmt = select(users_table).where(
3594 case(
3595 (users_table.c.name == "wendy", "W"),
3596 (users_table.c.name == "jack", "J"),
3597 else_="E",
3598 )
3599 )
3601 Details on :class:`.Case` usage is at :func:`.case`.
3603 .. seealso::
3605 :func:`.case`
3607 """
3609 __visit_name__ = "case"
3611 _traverse_internals: _TraverseInternalsType = [
3612 ("value", InternalTraversal.dp_clauseelement),
3613 ("whens", InternalTraversal.dp_clauseelement_tuples),
3614 ("else_", InternalTraversal.dp_clauseelement),
3615 ]
3617 # for case(), the type is derived from the whens. so for the moment
3618 # users would have to cast() the case to get a specific type
3620 whens: List[typing_Tuple[ColumnElement[bool], ColumnElement[_T]]]
3621 else_: Optional[ColumnElement[_T]]
3622 value: Optional[ColumnElement[Any]]
3624 def __init__(
3625 self,
3626 *whens: Union[
3627 typing_Tuple[_ColumnExpressionArgument[bool], Any],
3628 Mapping[Any, Any],
3629 ],
3630 value: Optional[Any] = None,
3631 else_: Optional[Any] = None,
3632 ):
3633 new_whens: Iterable[Any] = coercions._expression_collection_was_a_list(
3634 "whens", "case", whens
3635 )
3636 try:
3637 new_whens = util.dictlike_iteritems(new_whens)
3638 except TypeError:
3639 pass
3641 self.whens = [
3642 (
3643 coercions.expect(
3644 roles.ExpressionElementRole,
3645 c,
3646 apply_propagate_attrs=self,
3647 ).self_group(),
3648 coercions.expect(roles.ExpressionElementRole, r),
3649 )
3650 for (c, r) in new_whens
3651 ]
3653 if value is None:
3654 self.value = None
3655 else:
3656 self.value = coercions.expect(roles.ExpressionElementRole, value)
3658 if else_ is not None:
3659 self.else_ = coercions.expect(roles.ExpressionElementRole, else_)
3660 else:
3661 self.else_ = None
3663 type_ = next(
3664 (
3665 then.type
3666 # Iterate `whens` in reverse to match previous behaviour
3667 # where type of final element took priority
3668 for *_, then in reversed(self.whens)
3669 if not then.type._isnull
3670 ),
3671 self.else_.type if self.else_ is not None else type_api.NULLTYPE,
3672 )
3673 self.type = cast(_T, type_)
3675 @util.ro_non_memoized_property
3676 def _from_objects(self) -> List[FromClause]:
3677 return list(
3678 itertools.chain(*[x._from_objects for x in self.get_children()])
3679 )
3682class Cast(WrapsColumnExpression[_T]):
3683 """Represent a ``CAST`` expression.
3685 :class:`.Cast` is produced using the :func:`.cast` factory function,
3686 as in::
3688 from sqlalchemy import cast, Numeric
3690 stmt = select(cast(product_table.c.unit_price, Numeric(10, 4)))
3692 Details on :class:`.Cast` usage is at :func:`.cast`.
3694 .. seealso::
3696 :ref:`tutorial_casts`
3698 :func:`.cast`
3700 :func:`.try_cast`
3702 :func:`.type_coerce` - an alternative to CAST that coerces the type
3703 on the Python side only, which is often sufficient to generate the
3704 correct SQL and data coercion.
3706 """
3708 __visit_name__ = "cast"
3710 _traverse_internals: _TraverseInternalsType = [
3711 ("clause", InternalTraversal.dp_clauseelement),
3712 ("type", InternalTraversal.dp_type),
3713 ]
3715 clause: ColumnElement[Any]
3716 type: TypeEngine[_T]
3717 typeclause: TypeClause
3719 def __init__(
3720 self,
3721 expression: _ColumnExpressionArgument[Any],
3722 type_: _TypeEngineArgument[_T],
3723 ):
3724 self.type = type_api.to_instance(type_)
3725 self.clause = coercions.expect(
3726 roles.ExpressionElementRole,
3727 expression,
3728 type_=self.type,
3729 apply_propagate_attrs=self,
3730 )
3731 self.typeclause = TypeClause(self.type)
3733 @util.ro_non_memoized_property
3734 def _from_objects(self) -> List[FromClause]:
3735 return self.clause._from_objects
3737 @property
3738 def wrapped_column_expression(self):
3739 return self.clause
3742class TryCast(Cast[_T]):
3743 """Represent a TRY_CAST expression.
3745 Details on :class:`.TryCast` usage is at :func:`.try_cast`.
3747 .. seealso::
3749 :func:`.try_cast`
3751 :ref:`tutorial_casts`
3752 """
3754 __visit_name__ = "try_cast"
3755 inherit_cache = True
3758class TypeCoerce(WrapsColumnExpression[_T]):
3759 """Represent a Python-side type-coercion wrapper.
3761 :class:`.TypeCoerce` supplies the :func:`_expression.type_coerce`
3762 function; see that function for usage details.
3764 .. seealso::
3766 :func:`_expression.type_coerce`
3768 :func:`.cast`
3770 """
3772 __visit_name__ = "type_coerce"
3774 _traverse_internals: _TraverseInternalsType = [
3775 ("clause", InternalTraversal.dp_clauseelement),
3776 ("type", InternalTraversal.dp_type),
3777 ]
3779 clause: ColumnElement[Any]
3780 type: TypeEngine[_T]
3782 def __init__(
3783 self,
3784 expression: _ColumnExpressionArgument[Any],
3785 type_: _TypeEngineArgument[_T],
3786 ):
3787 self.type = type_api.to_instance(type_)
3788 self.clause = coercions.expect(
3789 roles.ExpressionElementRole,
3790 expression,
3791 type_=self.type,
3792 apply_propagate_attrs=self,
3793 )
3795 @util.ro_non_memoized_property
3796 def _from_objects(self) -> List[FromClause]:
3797 return self.clause._from_objects
3799 @HasMemoized.memoized_attribute
3800 def typed_expression(self):
3801 if isinstance(self.clause, BindParameter):
3802 bp = self.clause._clone()
3803 bp.type = self.type
3804 return bp
3805 else:
3806 return self.clause
3808 @property
3809 def wrapped_column_expression(self):
3810 return self.clause
3812 def self_group(
3813 self, against: Optional[OperatorType] = None
3814 ) -> TypeCoerce[_T]:
3815 grouped = self.clause.self_group(against=against)
3816 if grouped is not self.clause:
3817 return TypeCoerce(grouped, self.type)
3818 else:
3819 return self
3822class Extract(ColumnElement[int]):
3823 """Represent a SQL EXTRACT clause, ``extract(field FROM expr)``."""
3825 __visit_name__ = "extract"
3827 _traverse_internals: _TraverseInternalsType = [
3828 ("expr", InternalTraversal.dp_clauseelement),
3829 ("field", InternalTraversal.dp_string),
3830 ]
3832 expr: ColumnElement[Any]
3833 field: str
3835 def __init__(self, field: str, expr: _ColumnExpressionArgument[Any]):
3836 self.type = type_api.INTEGERTYPE
3837 self.field = field
3838 self.expr = coercions.expect(roles.ExpressionElementRole, expr)
3840 @util.ro_non_memoized_property
3841 def _from_objects(self) -> List[FromClause]:
3842 return self.expr._from_objects
3845class _label_reference(ColumnElement[_T]):
3846 """Wrap a column expression as it appears in a 'reference' context.
3848 This expression is any that includes an _order_by_label_element,
3849 which is a Label, or a DESC / ASC construct wrapping a Label.
3851 The production of _label_reference() should occur when an expression
3852 is added to this context; this includes the ORDER BY or GROUP BY of a
3853 SELECT statement, as well as a few other places, such as the ORDER BY
3854 within an OVER clause.
3856 """
3858 __visit_name__ = "label_reference"
3860 _traverse_internals: _TraverseInternalsType = [
3861 ("element", InternalTraversal.dp_clauseelement)
3862 ]
3864 element: ColumnElement[_T]
3866 def __init__(self, element: ColumnElement[_T]):
3867 self.element = element
3868 self._propagate_attrs = element._propagate_attrs
3870 @util.ro_non_memoized_property
3871 def _from_objects(self) -> List[FromClause]:
3872 return []
3875class _textual_label_reference(ColumnElement[Any]):
3876 __visit_name__ = "textual_label_reference"
3878 _traverse_internals: _TraverseInternalsType = [
3879 ("element", InternalTraversal.dp_string)
3880 ]
3882 def __init__(self, element: str):
3883 self.element = element
3885 @util.memoized_property
3886 def _text_clause(self) -> TextClause:
3887 return TextClause(self.element)
3890class UnaryExpression(ColumnElement[_T]):
3891 """Define a 'unary' expression.
3893 A unary expression has a single column expression
3894 and an operator. The operator can be placed on the left
3895 (where it is called the 'operator') or right (where it is called the
3896 'modifier') of the column expression.
3898 :class:`.UnaryExpression` is the basis for several unary operators
3899 including those used by :func:`.desc`, :func:`.asc`, :func:`.distinct`,
3900 :func:`.nulls_first` and :func:`.nulls_last`.
3902 """
3904 __visit_name__ = "unary"
3906 _traverse_internals: _TraverseInternalsType = [
3907 ("element", InternalTraversal.dp_clauseelement),
3908 ("operator", InternalTraversal.dp_operator),
3909 ("modifier", InternalTraversal.dp_operator),
3910 ]
3912 element: ColumnElement[Any]
3913 operator: Optional[OperatorType]
3914 modifier: Optional[OperatorType]
3916 def __init__(
3917 self,
3918 element: ColumnElement[Any],
3919 *,
3920 operator: Optional[OperatorType] = None,
3921 modifier: Optional[OperatorType] = None,
3922 type_: Optional[_TypeEngineArgument[_T]] = None,
3923 wraps_column_expression: bool = False, # legacy, not used as of 2.0.42
3924 ):
3925 self.operator = operator
3926 self.modifier = modifier
3927 self._propagate_attrs = element._propagate_attrs
3928 self.element = element.self_group(
3929 against=self.operator or self.modifier
3930 )
3932 # if type is None, we get NULLTYPE, which is our _T. But I don't
3933 # know how to get the overloads to express that correctly
3934 self.type = type_api.to_instance(type_) # type: ignore[assignment]
3936 def _wraps_unnamed_column(self):
3937 ungrouped = self.element._ungroup()
3938 return (
3939 not isinstance(ungrouped, NamedColumn)
3940 or ungrouped._non_anon_label is None
3941 )
3943 @classmethod
3944 def _create_nulls_first(
3945 cls,
3946 column: _ColumnExpressionArgument[_T],
3947 ) -> UnaryExpression[_T]:
3948 return UnaryExpression(
3949 coercions.expect(roles.ByOfRole, column),
3950 modifier=operators.nulls_first_op,
3951 )
3953 @classmethod
3954 def _create_nulls_last(
3955 cls,
3956 column: _ColumnExpressionArgument[_T],
3957 ) -> UnaryExpression[_T]:
3958 return UnaryExpression(
3959 coercions.expect(roles.ByOfRole, column),
3960 modifier=operators.nulls_last_op,
3961 )
3963 @classmethod
3964 def _create_desc(
3965 cls, column: _ColumnExpressionOrStrLabelArgument[_T]
3966 ) -> UnaryExpression[_T]:
3968 return UnaryExpression(
3969 coercions.expect(roles.ByOfRole, column),
3970 modifier=operators.desc_op,
3971 )
3973 @classmethod
3974 def _create_asc(
3975 cls,
3976 column: _ColumnExpressionOrStrLabelArgument[_T],
3977 ) -> UnaryExpression[_T]:
3978 return UnaryExpression(
3979 coercions.expect(roles.ByOfRole, column),
3980 modifier=operators.asc_op,
3981 )
3983 @classmethod
3984 def _create_distinct(
3985 cls,
3986 expr: _ColumnExpressionArgument[_T],
3987 ) -> UnaryExpression[_T]:
3988 col_expr: ColumnElement[_T] = coercions.expect(
3989 roles.ExpressionElementRole, expr
3990 )
3991 return UnaryExpression(
3992 col_expr,
3993 operator=operators.distinct_op,
3994 type_=col_expr.type,
3995 )
3997 @classmethod
3998 def _create_bitwise_not(
3999 cls,
4000 expr: _ColumnExpressionArgument[_T],
4001 ) -> UnaryExpression[_T]:
4002 col_expr: ColumnElement[_T] = coercions.expect(
4003 roles.ExpressionElementRole, expr
4004 )
4005 return UnaryExpression(
4006 col_expr,
4007 operator=operators.bitwise_not_op,
4008 type_=col_expr.type,
4009 )
4011 @property
4012 def _order_by_label_element(self) -> Optional[Label[Any]]:
4013 if operators.is_order_by_modifier(self.modifier):
4014 return self.element._order_by_label_element
4015 else:
4016 return None
4018 @util.ro_non_memoized_property
4019 def _from_objects(self) -> List[FromClause]:
4020 return self.element._from_objects
4022 def _negate(self) -> ColumnElement[Any]:
4023 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
4024 return UnaryExpression(
4025 self.self_group(against=operators.inv),
4026 operator=operators.inv,
4027 type_=type_api.BOOLEANTYPE,
4028 )
4029 else:
4030 return ColumnElement._negate(self)
4032 def self_group(
4033 self, against: Optional[OperatorType] = None
4034 ) -> Union[Self, Grouping[_T]]:
4035 if self.operator and operators.is_precedent(self.operator, against):
4036 return Grouping(self)
4037 else:
4038 return self
4041class CollectionAggregate(UnaryExpression[_T]):
4042 """Forms the basis for right-hand collection operator modifiers
4043 ANY and ALL.
4045 The ANY and ALL keywords are available in different ways on different
4046 backends. On PostgreSQL, they only work for an ARRAY type. On
4047 MySQL, they only work for subqueries.
4049 """
4051 inherit_cache = True
4052 _is_collection_aggregate = True
4054 @classmethod
4055 def _create_any(
4056 cls, expr: _ColumnExpressionArgument[_T]
4057 ) -> CollectionAggregate[bool]:
4058 """create CollectionAggregate for the legacy
4059 ARRAY.Comparator.any() method"""
4060 col_expr: ColumnElement[_T] = coercions.expect(
4061 roles.ExpressionElementRole,
4062 expr,
4063 )
4064 col_expr = col_expr.self_group()
4065 return CollectionAggregate(
4066 col_expr,
4067 operator=operators.any_op,
4068 type_=type_api.BOOLEANTYPE,
4069 )
4071 @classmethod
4072 def _create_all(
4073 cls, expr: _ColumnExpressionArgument[_T]
4074 ) -> CollectionAggregate[bool]:
4075 """create CollectionAggregate for the legacy
4076 ARRAY.Comparator.all() method"""
4077 col_expr: ColumnElement[_T] = coercions.expect(
4078 roles.ExpressionElementRole,
4079 expr,
4080 )
4081 col_expr = col_expr.self_group()
4082 return CollectionAggregate(
4083 col_expr,
4084 operator=operators.all_op,
4085 type_=type_api.BOOLEANTYPE,
4086 )
4088 @util.preload_module("sqlalchemy.sql.sqltypes")
4089 def _bind_param(
4090 self,
4091 operator: operators.OperatorType,
4092 obj: Any,
4093 type_: Optional[TypeEngine[_T]] = None,
4094 expanding: bool = False,
4095 ) -> BindParameter[_T]:
4096 """For new style any_(), all_(), ensure compared literal value
4097 receives appropriate bound parameter type."""
4099 # a CollectionAggregate is specific to ARRAY or int
4100 # only. So for ARRAY case, make sure we use correct element type
4101 sqltypes = util.preloaded.sql_sqltypes
4102 if self.element.type._type_affinity is sqltypes.ARRAY:
4103 compared_to_type = cast(
4104 sqltypes.ARRAY[Any], self.element.type
4105 ).item_type
4106 else:
4107 compared_to_type = self.element.type
4109 return BindParameter(
4110 None,
4111 obj,
4112 _compared_to_operator=operator,
4113 type_=type_,
4114 _compared_to_type=compared_to_type,
4115 unique=True,
4116 expanding=expanding,
4117 )
4119 # operate and reverse_operate are hardwired to
4120 # dispatch onto the type comparator directly, so that we can
4121 # ensure "reversed" behavior.
4122 def operate(
4123 self, op: OperatorType, *other: Any, **kwargs: Any
4124 ) -> ColumnElement[_T]:
4125 if not operators.is_comparison(op):
4126 raise exc.ArgumentError(
4127 "Only comparison operators may be used with ANY/ALL"
4128 )
4129 kwargs["reverse"] = True
4130 return self.comparator.operate(operators.mirror(op), *other, **kwargs)
4132 def reverse_operate(
4133 self, op: OperatorType, other: Any, **kwargs: Any
4134 ) -> ColumnElement[_T]:
4135 # comparison operators should never call reverse_operate
4136 assert not operators.is_comparison(op)
4137 raise exc.ArgumentError(
4138 "Only comparison operators may be used with ANY/ALL"
4139 )
4142class AsBoolean(WrapsColumnExpression[bool], UnaryExpression[bool]):
4143 inherit_cache = True
4145 def __init__(self, element, operator, negate):
4146 self.element = element
4147 self.type = type_api.BOOLEANTYPE
4148 self.operator = operator
4149 self.negate = negate
4150 self.modifier = None
4151 self._is_implicitly_boolean = element._is_implicitly_boolean
4153 @property
4154 def wrapped_column_expression(self):
4155 return self.element
4157 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4158 return self
4160 def _negate(self):
4161 if isinstance(self.element, (True_, False_)):
4162 return self.element._negate()
4163 else:
4164 return AsBoolean(self.element, self.negate, self.operator)
4167class BinaryExpression(OperatorExpression[_T]):
4168 """Represent an expression that is ``LEFT <operator> RIGHT``.
4170 A :class:`.BinaryExpression` is generated automatically
4171 whenever two column expressions are used in a Python binary expression:
4173 .. sourcecode:: pycon+sql
4175 >>> from sqlalchemy.sql import column
4176 >>> column("a") + column("b")
4177 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
4178 >>> print(column("a") + column("b"))
4179 {printsql}a + b
4181 """
4183 __visit_name__ = "binary"
4185 _traverse_internals: _TraverseInternalsType = [
4186 ("left", InternalTraversal.dp_clauseelement),
4187 ("right", InternalTraversal.dp_clauseelement),
4188 ("operator", InternalTraversal.dp_operator),
4189 ("negate", InternalTraversal.dp_operator),
4190 ("modifiers", InternalTraversal.dp_plain_dict),
4191 (
4192 "type",
4193 InternalTraversal.dp_type,
4194 ),
4195 ]
4197 _cache_key_traversal = [
4198 ("left", InternalTraversal.dp_clauseelement),
4199 ("right", InternalTraversal.dp_clauseelement),
4200 ("operator", InternalTraversal.dp_operator),
4201 ("modifiers", InternalTraversal.dp_plain_dict),
4202 # "type" affects JSON CAST operators, so while redundant in most cases,
4203 # is needed for that one
4204 (
4205 "type",
4206 InternalTraversal.dp_type,
4207 ),
4208 ]
4210 _is_implicitly_boolean = True
4211 """Indicates that any database will know this is a boolean expression
4212 even if the database does not have an explicit boolean datatype.
4214 """
4216 left: ColumnElement[Any]
4217 right: ColumnElement[Any]
4218 modifiers: Mapping[str, Any]
4220 def __init__(
4221 self,
4222 left: ColumnElement[Any],
4223 right: ColumnElement[Any],
4224 operator: OperatorType,
4225 type_: Optional[_TypeEngineArgument[_T]] = None,
4226 negate: Optional[OperatorType] = None,
4227 modifiers: Optional[Mapping[str, Any]] = None,
4228 ):
4229 # allow compatibility with libraries that
4230 # refer to BinaryExpression directly and pass strings
4231 if isinstance(operator, str):
4232 operator = operators.custom_op(operator)
4233 self._orig = (left.__hash__(), right.__hash__())
4234 self._propagate_attrs = left._propagate_attrs or right._propagate_attrs
4235 self.left = left.self_group(against=operator)
4236 self.right = right.self_group(against=operator)
4237 self.operator = operator
4239 # if type is None, we get NULLTYPE, which is our _T. But I don't
4240 # know how to get the overloads to express that correctly
4241 self.type = type_api.to_instance(type_) # type: ignore[assignment]
4243 self.negate = negate
4244 self._is_implicitly_boolean = operators.is_boolean(operator)
4246 if modifiers is None:
4247 self.modifiers = {}
4248 else:
4249 self.modifiers = modifiers
4251 @property
4252 def _flattened_operator_clauses(
4253 self,
4254 ) -> typing_Tuple[ColumnElement[Any], ...]:
4255 return (self.left, self.right)
4257 def __bool__(self):
4258 """Implement Python-side "bool" for BinaryExpression as a
4259 simple "identity" check for the left and right attributes,
4260 if the operator is "eq" or "ne". Otherwise the expression
4261 continues to not support "bool" like all other column expressions.
4263 The rationale here is so that ColumnElement objects can be hashable.
4264 What? Well, suppose you do this::
4266 c1, c2 = column("x"), column("y")
4267 s1 = set([c1, c2])
4269 We do that **a lot**, columns inside of sets is an extremely basic
4270 thing all over the ORM for example.
4272 So what happens if we do this? ::
4274 c1 in s1
4276 Hashing means it will normally use ``__hash__()`` of the object,
4277 but in case of hash collision, it's going to also do ``c1 == c1``
4278 and/or ``c1 == c2`` inside. Those operations need to return a
4279 True/False value. But because we override ``==`` and ``!=``, they're
4280 going to get a BinaryExpression. Hence we implement ``__bool__`` here
4281 so that these comparisons behave in this particular context mostly
4282 like regular object comparisons. Thankfully Python is OK with
4283 that! Otherwise we'd have to use special set classes for columns
4284 (which we used to do, decades ago).
4286 """
4287 if self.operator in (operators.eq, operators.ne):
4288 # this is using the eq/ne operator given int hash values,
4289 # rather than Operator, so that "bool" can be based on
4290 # identity
4291 return self.operator(*self._orig) # type: ignore[call-overload]
4292 else:
4293 raise TypeError("Boolean value of this clause is not defined")
4295 if typing.TYPE_CHECKING:
4297 def __invert__(
4298 self: BinaryExpression[_T],
4299 ) -> BinaryExpression[_T]: ...
4301 @util.ro_non_memoized_property
4302 def _from_objects(self) -> List[FromClause]:
4303 return self.left._from_objects + self.right._from_objects
4305 def _negate(self):
4306 if self.negate is not None:
4307 return BinaryExpression(
4308 self.left,
4309 self.right._negate_in_binary(self.negate, self.operator),
4310 self.negate,
4311 negate=self.operator,
4312 type_=self.type,
4313 modifiers=self.modifiers,
4314 )
4315 else:
4316 return self.self_group()._negate()
4319class Slice(ColumnElement[Any]):
4320 """Represent SQL for a Python array-slice object.
4322 This is not a specific SQL construct at this level, but
4323 may be interpreted by specific dialects, e.g. PostgreSQL.
4325 """
4327 __visit_name__ = "slice"
4329 _traverse_internals: _TraverseInternalsType = [
4330 ("start", InternalTraversal.dp_clauseelement),
4331 ("stop", InternalTraversal.dp_clauseelement),
4332 ("step", InternalTraversal.dp_clauseelement),
4333 ]
4335 def __init__(self, start, stop, step, _name=None):
4336 self.start = coercions.expect(
4337 roles.ExpressionElementRole,
4338 start,
4339 name=_name,
4340 type_=type_api.INTEGERTYPE,
4341 )
4342 self.stop = coercions.expect(
4343 roles.ExpressionElementRole,
4344 stop,
4345 name=_name,
4346 type_=type_api.INTEGERTYPE,
4347 )
4348 self.step = coercions.expect(
4349 roles.ExpressionElementRole,
4350 step,
4351 name=_name,
4352 type_=type_api.INTEGERTYPE,
4353 )
4354 self.type = type_api.NULLTYPE
4356 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4357 assert against is operator.getitem
4358 return self
4361class IndexExpression(BinaryExpression[Any]):
4362 """Represent the class of expressions that are like an "index"
4363 operation."""
4365 inherit_cache = True
4368class GroupedElement(DQLDMLClauseElement):
4369 """Represent any parenthesized expression"""
4371 __visit_name__ = "grouping"
4373 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4374 return self
4376 def _ungroup(self) -> ClauseElement:
4377 raise NotImplementedError()
4380class Grouping(GroupedElement, ColumnElement[_T]):
4381 """Represent a grouping within a column expression"""
4383 _traverse_internals: _TraverseInternalsType = [
4384 ("element", InternalTraversal.dp_clauseelement),
4385 ("type", InternalTraversal.dp_type),
4386 ]
4388 _cache_key_traversal = [
4389 ("element", InternalTraversal.dp_clauseelement),
4390 ]
4392 element: Union[
4393 AbstractTextClause,
4394 ClauseList,
4395 ColumnElement[_T],
4396 CompilerColumnElement,
4397 ]
4399 def __init__(
4400 self,
4401 element: Union[
4402 AbstractTextClause,
4403 ClauseList,
4404 ColumnElement[_T],
4405 CompilerColumnElement,
4406 ],
4407 ):
4408 self.element = element
4410 # nulltype assignment issue
4411 self.type = getattr(element, "type", type_api.NULLTYPE) # type: ignore[arg-type] # noqa: E501
4412 self._propagate_attrs = element._propagate_attrs
4414 def _with_binary_element_type(self, type_):
4415 return self.__class__(self.element._with_binary_element_type(type_))
4417 def _ungroup(self) -> ColumnElement[_T]:
4418 assert isinstance(self.element, ColumnElement)
4419 return self.element._ungroup()
4421 @util.memoized_property
4422 def _is_implicitly_boolean(self):
4423 return self.element._is_implicitly_boolean
4425 @util.non_memoized_property
4426 def _tq_label(self) -> Optional[str]:
4427 return (
4428 getattr(self.element, "_tq_label", None) or self._anon_name_label
4429 )
4431 @util.non_memoized_property
4432 def _proxies(self) -> List[ColumnElement[Any]]:
4433 if isinstance(self.element, ColumnElement):
4434 return [self.element]
4435 else:
4436 return []
4438 @util.ro_non_memoized_property
4439 def _from_objects(self) -> List[FromClause]:
4440 return self.element._from_objects
4442 def __getattr__(self, attr):
4443 return getattr(self.element, attr)
4445 def __getstate__(self):
4446 return {"element": self.element, "type": self.type}
4448 def __setstate__(self, state):
4449 self.element = state["element"]
4450 self.type = state["type"]
4452 if TYPE_CHECKING:
4454 def self_group(
4455 self, against: Optional[OperatorType] = None
4456 ) -> Self: ...
4459class _OverrideBinds(Grouping[_T]):
4460 """used by cache_key->_apply_params_to_element to allow compilation /
4461 execution of a SQL element that's been cached, using an alternate set of
4462 bound parameter values.
4464 This is used by the ORM to swap new parameter values into expressions
4465 that are embedded into loader options like with_expression(),
4466 selectinload(). Previously, this task was accomplished using the
4467 .params() method which would perform a deep-copy instead. This deep
4468 copy proved to be too expensive for more complex expressions.
4470 See #11085
4472 """
4474 __visit_name__ = "override_binds"
4476 def __init__(
4477 self,
4478 element: ColumnElement[_T],
4479 bindparams: Sequence[BindParameter[Any]],
4480 replaces_params: Sequence[BindParameter[Any]],
4481 ):
4482 self.element = element
4483 self.translate = {
4484 k.key: v.value for k, v in zip(replaces_params, bindparams)
4485 }
4487 def _gen_cache_key(
4488 self, anon_map: anon_map, bindparams: List[BindParameter[Any]]
4489 ) -> Optional[typing_Tuple[Any, ...]]:
4490 """generate a cache key for the given element, substituting its bind
4491 values for the translation values present."""
4493 existing_bps: List[BindParameter[Any]] = []
4494 ck = self.element._gen_cache_key(anon_map, existing_bps)
4496 bindparams.extend(
4497 (
4498 bp._with_value(
4499 self.translate[bp.key], maintain_key=True, required=False
4500 )
4501 if bp.key in self.translate
4502 else bp
4503 )
4504 for bp in existing_bps
4505 )
4507 return ck
4510_FrameIntTuple = tuple[int | None, int | None]
4513class Over(ColumnElement[_T]):
4514 """Represent an OVER clause.
4516 This is a special operator against a so-called
4517 "window" function, as well as any aggregate function,
4518 which produces results relative to the result set
4519 itself. Most modern SQL backends now support window functions.
4521 """
4523 __visit_name__ = "over"
4525 _traverse_internals: _TraverseInternalsType = [
4526 ("element", InternalTraversal.dp_clauseelement),
4527 ("order_by", InternalTraversal.dp_clauseelement),
4528 ("partition_by", InternalTraversal.dp_clauseelement),
4529 ("range_", InternalTraversal.dp_clauseelement),
4530 ("rows", InternalTraversal.dp_clauseelement),
4531 ("groups", InternalTraversal.dp_clauseelement),
4532 ("exclude", InternalTraversal.dp_string),
4533 ]
4535 order_by: Optional[ClauseList] = None
4536 partition_by: Optional[ClauseList] = None
4538 element: ColumnElement[_T]
4539 """The underlying expression object to which this :class:`.Over`
4540 object refers."""
4542 range_: FrameClause | None
4543 rows: FrameClause | None
4544 groups: FrameClause | None
4545 exclude: str | None
4547 def __init__(
4548 self,
4549 element: ColumnElement[_T],
4550 partition_by: Optional[_ByArgument] = None,
4551 order_by: Optional[_ByArgument] = None,
4552 range_: _FrameIntTuple | FrameClause | None = None,
4553 rows: _FrameIntTuple | FrameClause | None = None,
4554 groups: _FrameIntTuple | FrameClause | None = None,
4555 exclude: str | None = None,
4556 ):
4557 self.element = element
4558 if order_by is not None:
4559 self.order_by = ClauseList(
4560 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4561 )
4562 if partition_by is not None:
4563 self.partition_by = ClauseList(
4564 *util.to_list(partition_by),
4565 _literal_as_text_role=roles.ByOfRole,
4566 )
4568 if sum(item is not None for item in (range_, rows, groups)) > 1:
4569 raise exc.ArgumentError(
4570 "only one of 'rows', 'range_', or 'groups' may be provided"
4571 )
4572 else:
4573 self.range_ = FrameClause._parse(range_, coerce_int=False)
4574 self.rows = FrameClause._parse(rows, coerce_int=True)
4575 self.groups = FrameClause._parse(groups, coerce_int=True)
4577 self.exclude = exclude
4579 if exclude is not None and (
4580 range_ is None and rows is None and groups is None
4581 ):
4582 raise exc.ArgumentError(
4583 "'exclude' requires that one of 'rows', "
4584 "'range_', or 'groups' is also specified"
4585 )
4587 if not TYPE_CHECKING:
4589 @util.memoized_property
4590 def type(self) -> TypeEngine[_T]: # noqa: A001
4591 return self.element.type
4593 @util.ro_non_memoized_property
4594 def _from_objects(self) -> List[FromClause]:
4595 return list(
4596 itertools.chain(
4597 *[
4598 c._from_objects
4599 for c in (self.element, self.partition_by, self.order_by)
4600 if c is not None
4601 ]
4602 )
4603 )
4606class FrameClauseType(Enum):
4607 """Frame clause type enum for FrameClause lower_type and upper_type.
4609 .. versionadded:: 2.1
4611 """
4613 UNBOUNDED = 0
4614 """Produces an "UNBOUNDED PRECEDING" or "UNBOUNDED FOLLOWING" frame
4615 clause depending on the position.
4616 Requires a ``None`` value for the corresponding bound value.
4617 """
4618 CURRENT = 1
4619 """Produces a "CURRENT ROW" frame clause.
4620 Requires a ``None`` value for the corresponding bound value.
4621 """
4622 PRECEDING = 2
4623 """Produces a "PRECEDING" frame clause."""
4624 FOLLOWING = 3
4625 """Produces a "FOLLOWING" frame clause."""
4628_require_none = (
4629 FrameClauseType.CURRENT,
4630 FrameClauseType.UNBOUNDED,
4631)
4634class FrameClause(ClauseElement):
4635 """Indicate the 'rows' 'range' or 'group' field of a window function,
4636 e.g. using :class:`.Over`.
4638 .. versionadded:: 2.1
4640 """
4642 __visit_name__ = "frame_clause"
4644 _traverse_internals: _TraverseInternalsType = [
4645 ("lower_bind", InternalTraversal.dp_clauseelement),
4646 ("upper_bind", InternalTraversal.dp_clauseelement),
4647 ("lower_type", InternalTraversal.dp_plain_obj),
4648 ("upper_type", InternalTraversal.dp_plain_obj),
4649 ]
4651 def __init__(
4652 self,
4653 start: Any,
4654 end: Any,
4655 start_frame_type: FrameClauseType,
4656 end_frame_type: FrameClauseType,
4657 _validate: bool = True,
4658 ) -> None:
4659 """Creates a new FrameClause specifying the bounds of a window frame.
4661 :param start: The start value.
4662 :param end: The end value.
4663 :param start_frame_type: The :class:`FrameClauseType` for the
4664 start value.
4665 :param end_frame_type: The :class:`FrameClauseType` for the end value.
4666 """
4667 self.lower_bind = self._as_literal(start)
4668 self.upper_bind = self._as_literal(end)
4669 self.lower_type = FrameClauseType(start_frame_type)
4670 self.upper_type = FrameClauseType(end_frame_type)
4671 if _validate:
4672 if (
4673 self.lower_type in _require_none
4674 and self.lower_bind is not None
4675 ):
4676 raise exc.ArgumentError(
4677 "Cannot specify a value for start with frame type "
4678 f"{self.lower_type.name}"
4679 )
4680 if (
4681 self.upper_type in _require_none
4682 and self.upper_bind is not None
4683 ):
4684 raise exc.ArgumentError(
4685 "Cannot specify a value for end with frame type "
4686 f"{self.upper_type.name}"
4687 )
4689 @classmethod
4690 def _as_literal(cls, value: Any) -> BindParameter[Any] | None:
4691 if value is None:
4692 return None
4693 elif isinstance(value, int):
4694 return literal(value, type_api.INTEGERTYPE)
4695 elif isinstance(value, BindParameter):
4696 return value
4697 else:
4698 return literal(value) # let the default type resolution occur
4700 @classmethod
4701 def _handle_int(
4702 cls, value: Any | None, coerce_int: bool
4703 ) -> tuple[int | None, FrameClauseType]:
4704 if value is None:
4705 return None, FrameClauseType.UNBOUNDED
4707 if coerce_int:
4708 try:
4709 integer = int(value)
4710 except ValueError as err:
4711 raise exc.ArgumentError(
4712 "Integer or None expected for values in rows/groups frame"
4713 ) from err
4714 elif not isinstance(value, int):
4715 raise exc.ArgumentError(
4716 "When using a tuple to specify a range only integer or none "
4717 "values are allowed in the range frame. To specify a "
4718 "different type use the FrameClause directly."
4719 )
4720 else:
4721 integer = value
4722 if integer == 0:
4723 return None, FrameClauseType.CURRENT
4724 elif integer < 0:
4725 return abs(integer), FrameClauseType.PRECEDING
4726 else:
4727 return integer, FrameClauseType.FOLLOWING
4729 @classmethod
4730 def _parse(
4731 cls,
4732 range_: _FrameIntTuple | FrameClause | None,
4733 coerce_int: bool,
4734 ) -> FrameClause | None:
4735 if range_ is None or isinstance(range_, FrameClause):
4736 return range_
4738 try:
4739 r0, r1 = range_
4740 except (ValueError, TypeError) as ve:
4741 raise exc.ArgumentError(
4742 "2-tuple expected for range/rows/groups"
4743 ) from ve
4745 l_b, l_t = cls._handle_int(r0, coerce_int)
4746 u_b, u_t = cls._handle_int(r1, coerce_int)
4748 return FrameClause(
4749 start=l_b,
4750 end=u_b,
4751 start_frame_type=l_t,
4752 end_frame_type=u_t,
4753 _validate=False,
4754 )
4757class AggregateOrderBy(WrapsColumnExpression[_T]):
4758 """Represent an aggregate ORDER BY expression.
4760 This is a special operator against aggregate functions such as
4761 ``array_agg()``, ``json_arrayagg()`` ``string_agg()``, etc. that provides
4762 for an ORDER BY expression, using a syntax that's compatible with
4763 the backend.
4765 :class:`.AggregateOrderBy` is a generalized version of the
4766 :class:`.WithinGroup` construct, the latter of which always provides a
4767 "WITHIN GROUP (ORDER BY ...)" expression. :class:`.AggregateOrderBy` will
4768 also compile to "WITHIN GROUP (ORDER BY ...)" on backends such as Oracle
4769 and SQL Server that don't have another style of aggregate function
4770 ordering.
4772 .. versionadded:: 2.1
4775 """
4777 __visit_name__ = "aggregateorderby"
4779 _traverse_internals: _TraverseInternalsType = [
4780 ("element", InternalTraversal.dp_clauseelement),
4781 ("order_by", InternalTraversal.dp_clauseelement),
4782 ]
4784 order_by: ClauseList
4786 def __init__(
4787 self,
4788 element: Union[FunctionElement[_T], FunctionFilter[_T]],
4789 *order_by: _ColumnExpressionArgument[Any],
4790 ):
4791 self.element = element
4792 if not order_by:
4793 raise TypeError("at least one ORDER BY element is required")
4794 self.order_by = ClauseList(
4795 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4796 )
4798 if not TYPE_CHECKING:
4800 @util.memoized_property
4801 def type(self) -> TypeEngine[_T]: # noqa: A001
4802 return self.element.type
4804 @property
4805 def wrapped_column_expression(self) -> ColumnElement[_T]:
4806 return self.element
4808 def __reduce__(self):
4809 return self.__class__, (self.element,) + (
4810 tuple(self.order_by) if self.order_by is not None else ()
4811 )
4813 def over(
4814 self,
4815 *,
4816 partition_by: _ByArgument | None = None,
4817 order_by: _ByArgument | None = None,
4818 rows: _FrameIntTuple | FrameClause | None = None,
4819 range_: _FrameIntTuple | FrameClause | None = None,
4820 groups: _FrameIntTuple | FrameClause | None = None,
4821 exclude: str | None = None,
4822 ) -> Over[_T]:
4823 """Produce an OVER clause against this :class:`.WithinGroup`
4824 construct.
4826 This function has the same signature as that of
4827 :meth:`.FunctionElement.over`.
4829 """
4830 return Over(
4831 self,
4832 partition_by=partition_by,
4833 order_by=order_by,
4834 range_=range_,
4835 rows=rows,
4836 groups=groups,
4837 exclude=exclude,
4838 )
4840 @overload
4841 def filter(self) -> Self: ...
4843 @overload
4844 def filter(
4845 self,
4846 __criterion0: _ColumnExpressionArgument[bool],
4847 *criterion: _ColumnExpressionArgument[bool],
4848 ) -> FunctionFilter[_T]: ...
4850 def filter(
4851 self, *criterion: _ColumnExpressionArgument[bool]
4852 ) -> Union[Self, FunctionFilter[_T]]:
4853 """Produce a FILTER clause against this function."""
4854 if not criterion:
4855 return self
4856 return FunctionFilter(self, *criterion)
4858 @util.ro_non_memoized_property
4859 def _from_objects(self) -> List[FromClause]:
4860 return list(
4861 itertools.chain(
4862 *[
4863 c._from_objects
4864 for c in (self.element, self.order_by)
4865 if c is not None
4866 ]
4867 )
4868 )
4871class WithinGroup(AggregateOrderBy[_T]):
4872 """Represent a WITHIN GROUP (ORDER BY) clause.
4874 This is a special operator against so-called
4875 "ordered set aggregate" and "hypothetical
4876 set aggregate" functions, including ``percentile_cont()``,
4877 ``rank()``, ``dense_rank()``, etc.
4879 It's supported only by certain database backends, such as PostgreSQL,
4880 Oracle Database and MS SQL Server.
4882 The :class:`.WithinGroup` construct extracts its type from the
4883 method :meth:`.FunctionElement.within_group_type`. If this returns
4884 ``None``, the function's ``.type`` is used.
4886 """
4888 __visit_name__ = "withingroup"
4889 inherit_cache = True
4891 if not TYPE_CHECKING:
4893 @util.memoized_property
4894 def type(self) -> TypeEngine[_T]: # noqa: A001
4895 wgt = self.element.within_group_type(self)
4896 if wgt is not None:
4897 return wgt
4898 else:
4899 return self.element.type
4902class FunctionFilter(Generative, ColumnElement[_T]):
4903 """Represent a function FILTER clause.
4905 This is a special operator against aggregate and window functions,
4906 which controls which rows are passed to it.
4907 It's supported only by certain database backends.
4909 Invocation of :class:`.FunctionFilter` is via
4910 :meth:`.FunctionElement.filter`::
4912 func.count(1).filter(True)
4914 .. seealso::
4916 :meth:`.FunctionElement.filter`
4918 """
4920 __visit_name__ = "funcfilter"
4922 _traverse_internals: _TraverseInternalsType = [
4923 ("func", InternalTraversal.dp_clauseelement),
4924 ("criterion", InternalTraversal.dp_clauseelement),
4925 ]
4927 criterion: Optional[ColumnElement[bool]] = None
4929 def __init__(
4930 self,
4931 func: Union[FunctionElement[_T], AggregateOrderBy[_T]],
4932 *criterion: _ColumnExpressionArgument[bool],
4933 ):
4934 self.func = func
4935 self.filter.non_generative(self, *criterion) # type: ignore[attr-defined] # noqa: E501
4937 @_generative
4938 def filter(self, *criterion: _ColumnExpressionArgument[bool]) -> Self:
4939 """Produce an additional FILTER against the function.
4941 This method adds additional criteria to the initial criteria
4942 set up by :meth:`.FunctionElement.filter`.
4944 Multiple criteria are joined together at SQL render time
4945 via ``AND``.
4948 """
4950 for crit in list(criterion):
4951 crit = coercions.expect(roles.WhereHavingRole, crit)
4953 if self.criterion is not None:
4954 self.criterion = self.criterion & crit
4955 else:
4956 self.criterion = crit
4958 return self
4960 def over(
4961 self,
4962 partition_by: _ByArgument | None = None,
4963 order_by: _ByArgument | None = None,
4964 range_: _FrameIntTuple | FrameClause | None = None,
4965 rows: _FrameIntTuple | FrameClause | None = None,
4966 groups: _FrameIntTuple | FrameClause | None = None,
4967 exclude: str | None = None,
4968 ) -> Over[_T]:
4969 """Produce an OVER clause against this filtered function.
4971 Used against aggregate or so-called "window" functions,
4972 for database backends that support window functions.
4974 The expression::
4976 func.rank().filter(MyClass.y > 5).over(order_by="x")
4978 is shorthand for::
4980 from sqlalchemy import over, funcfilter
4982 over(funcfilter(func.rank(), MyClass.y > 5), order_by="x")
4984 See :func:`_expression.over` for a full description.
4986 """
4987 return Over(
4988 self,
4989 partition_by=partition_by,
4990 order_by=order_by,
4991 range_=range_,
4992 rows=rows,
4993 groups=groups,
4994 exclude=exclude,
4995 )
4997 def within_group(
4998 self, *order_by: _ColumnExpressionArgument[Any]
4999 ) -> WithinGroup[_T]:
5000 """Produce a WITHIN GROUP (ORDER BY expr) clause against
5001 this function.
5002 """
5003 return WithinGroup(self, *order_by)
5005 def within_group_type(
5006 self, within_group: WithinGroup[_T]
5007 ) -> Optional[TypeEngine[_T]]:
5008 return None
5010 def self_group(
5011 self, against: Optional[OperatorType] = None
5012 ) -> Union[Self, Grouping[_T]]:
5013 if operators.is_precedent(operators.filter_op, against):
5014 return Grouping(self)
5015 else:
5016 return self
5018 if not TYPE_CHECKING:
5020 @util.memoized_property
5021 def type(self) -> TypeEngine[_T]: # noqa: A001
5022 return self.func.type
5024 @util.ro_non_memoized_property
5025 def _from_objects(self) -> List[FromClause]:
5026 return list(
5027 itertools.chain(
5028 *[
5029 c._from_objects
5030 for c in (self.func, self.criterion)
5031 if c is not None
5032 ]
5033 )
5034 )
5037class NamedColumn(KeyedColumnElement[_T]):
5038 is_literal = False
5039 table: Optional[FromClause] = None
5040 name: str
5041 key: str
5043 def _compare_name_for_result(self, other):
5044 return (hasattr(other, "name") and self.name == other.name) or (
5045 hasattr(other, "_label") and self._label == other._label
5046 )
5048 @util.ro_memoized_property
5049 def description(self) -> str:
5050 return self.name
5052 @HasMemoized.memoized_attribute
5053 def _tq_key_label(self) -> Optional[str]:
5054 """table qualified label based on column key.
5056 for table-bound columns this is <tablename>_<column key/proxy key>;
5058 all other expressions it resolves to key/proxy key.
5060 """
5061 proxy_key = self._proxy_key
5062 if proxy_key and proxy_key != self.name:
5063 return self._gen_tq_label(proxy_key)
5064 else:
5065 return self._tq_label
5067 @HasMemoized.memoized_attribute
5068 def _tq_label(self) -> Optional[str]:
5069 """table qualified label based on column name.
5071 for table-bound columns this is <tablename>_<columnname>; all other
5072 expressions it resolves to .name.
5074 """
5075 return self._gen_tq_label(self.name)
5077 @HasMemoized.memoized_attribute
5078 def _render_label_in_columns_clause(self):
5079 return True
5081 @HasMemoized.memoized_attribute
5082 def _non_anon_label(self):
5083 return self.name
5085 def _gen_tq_label(
5086 self, name: str, dedupe_on_key: bool = True
5087 ) -> Optional[str]:
5088 return name
5090 def _bind_param(
5091 self,
5092 operator: OperatorType,
5093 obj: Any,
5094 type_: Optional[TypeEngine[_T]] = None,
5095 expanding: bool = False,
5096 ) -> BindParameter[_T]:
5097 return BindParameter(
5098 self.key,
5099 obj,
5100 _compared_to_operator=operator,
5101 _compared_to_type=self.type,
5102 type_=type_,
5103 unique=True,
5104 expanding=expanding,
5105 )
5107 def _make_proxy(
5108 self,
5109 selectable: FromClause,
5110 *,
5111 primary_key: ColumnSet,
5112 foreign_keys: Set[KeyedColumnElement[Any]],
5113 name: Optional[str] = None,
5114 key: Optional[str] = None,
5115 name_is_truncatable: bool = False,
5116 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5117 disallow_is_literal: bool = False,
5118 **kw: Any,
5119 ) -> typing_Tuple[str, ColumnClause[_T]]:
5120 c = ColumnClause(
5121 (
5122 coercions.expect(roles.TruncatedLabelRole, name or self.name)
5123 if name_is_truncatable
5124 else (name or self.name)
5125 ),
5126 type_=self.type,
5127 _selectable=selectable,
5128 is_literal=False,
5129 )
5131 c._propagate_attrs = selectable._propagate_attrs
5132 if name is None:
5133 c.key = self.key
5134 if compound_select_cols:
5135 c._proxies = list(compound_select_cols)
5136 else:
5137 c._proxies = [self]
5139 if selectable._is_clone_of is not None:
5140 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
5141 return c.key, c
5144_PS = ParamSpec("_PS")
5147class Label(roles.LabeledColumnExprRole[_T], NamedColumn[_T]):
5148 """Represents a column label (AS).
5150 Represent a label, as typically applied to any column-level
5151 element using the ``AS`` sql keyword.
5153 """
5155 __visit_name__ = "label"
5157 _traverse_internals: _TraverseInternalsType = [
5158 ("name", InternalTraversal.dp_anon_name),
5159 ("type", InternalTraversal.dp_type),
5160 ("_element", InternalTraversal.dp_clauseelement),
5161 ]
5163 _cache_key_traversal = [
5164 ("name", InternalTraversal.dp_anon_name),
5165 ("_element", InternalTraversal.dp_clauseelement),
5166 ]
5168 _element: ColumnElement[_T]
5169 name: str
5171 def __init__(
5172 self,
5173 name: Optional[str],
5174 element: _ColumnExpressionArgument[_T],
5175 type_: Optional[_TypeEngineArgument[_T]] = None,
5176 ):
5177 orig_element = element
5178 element = coercions.expect(
5179 roles.ExpressionElementRole,
5180 element,
5181 apply_propagate_attrs=self,
5182 )
5183 while isinstance(element, Label):
5184 # TODO: this is only covered in test_text.py, but nothing
5185 # fails if it's removed. determine rationale
5186 element = element.element
5188 if name:
5189 self.name = name
5190 else:
5191 self.name = _anonymous_label.safe_construct(
5192 id(self), getattr(element, "name", "anon")
5193 )
5194 if isinstance(orig_element, Label):
5195 # TODO: no coverage for this block, again would be in
5196 # test_text.py where the resolve_label concept is important
5197 self._resolve_label = orig_element._label
5199 self.key = self._tq_label = self._tq_key_label = self.name
5200 self._element = element
5202 self.type = (
5203 type_api.to_instance(type_)
5204 if type_ is not None
5205 else self._element.type
5206 )
5208 self._proxies = [element]
5210 def __reduce__(self):
5211 return self.__class__, (self.name, self._element, self.type)
5213 @HasMemoized.memoized_attribute
5214 def _render_label_in_columns_clause(self):
5215 return True
5217 def _bind_param(self, operator, obj, type_=None, expanding=False):
5218 return BindParameter(
5219 None,
5220 obj,
5221 _compared_to_operator=operator,
5222 type_=type_,
5223 _compared_to_type=self.type,
5224 unique=True,
5225 expanding=expanding,
5226 )
5228 @util.memoized_property
5229 def _is_implicitly_boolean(self):
5230 return self.element._is_implicitly_boolean
5232 @HasMemoized.memoized_attribute
5233 def _allow_label_resolve(self):
5234 return self.element._allow_label_resolve
5236 @property
5237 def _order_by_label_element(self):
5238 return self
5240 def as_reference(self) -> _label_reference[_T]:
5241 """refer to this labeled expression in a clause such as GROUP BY,
5242 ORDER BY etc. as the label name itself, without expanding
5243 into the full expression.
5245 .. versionadded:: 2.1
5247 """
5248 return _label_reference(self)
5250 @HasMemoized.memoized_attribute
5251 def element(self) -> ColumnElement[_T]:
5252 return self._element.self_group(against=operators.as_)
5254 def self_group(self, against: Optional[OperatorType] = None) -> Label[_T]:
5255 return self._apply_to_inner(self._element.self_group, against=against)
5257 def _negate(self):
5258 return self._apply_to_inner(self._element._negate)
5260 def _apply_to_inner(
5261 self,
5262 fn: Callable[_PS, ColumnElement[_T]],
5263 *arg: _PS.args,
5264 **kw: _PS.kwargs,
5265 ) -> Label[_T]:
5266 sub_element = fn(*arg, **kw)
5267 if sub_element is not self._element:
5268 return Label(self.name, sub_element, type_=self.type)
5269 else:
5270 return self
5272 @property
5273 def primary_key(self): # type: ignore[override]
5274 return self.element.primary_key
5276 @property
5277 def foreign_keys(self): # type: ignore[override]
5278 return self.element.foreign_keys
5280 def _copy_internals(
5281 self,
5282 *,
5283 clone: _CloneCallableType = _clone,
5284 anonymize_labels: bool = False,
5285 **kw: Any,
5286 ) -> None:
5287 self._reset_memoizations()
5288 self._element = clone(self._element, **kw)
5289 if anonymize_labels:
5290 self.name = _anonymous_label.safe_construct(
5291 id(self), getattr(self.element, "name", "anon")
5292 )
5293 self.key = self._tq_label = self._tq_key_label = self.name
5295 @util.ro_non_memoized_property
5296 def _from_objects(self) -> List[FromClause]:
5297 return self.element._from_objects
5299 def _make_proxy(
5300 self,
5301 selectable: FromClause,
5302 *,
5303 primary_key: ColumnSet,
5304 foreign_keys: Set[KeyedColumnElement[Any]],
5305 name: Optional[str] = None,
5306 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5307 **kw: Any,
5308 ) -> typing_Tuple[str, ColumnClause[_T]]:
5309 name = self.name if not name else name
5311 key, e = self.element._make_proxy(
5312 selectable,
5313 name=name,
5314 disallow_is_literal=True,
5315 name_is_truncatable=isinstance(name, _truncated_label),
5316 compound_select_cols=compound_select_cols,
5317 primary_key=primary_key,
5318 foreign_keys=foreign_keys,
5319 )
5321 # there was a note here to remove this assertion, which was here
5322 # to determine if we later could support a use case where
5323 # the key and name of a label are separate. But I don't know what
5324 # that case was. For now, this is an unexpected case that occurs
5325 # when a label name conflicts with other columns and select()
5326 # is attempting to disambiguate an explicit label, which is not what
5327 # the user would want. See issue #6090.
5328 if key != self.name and not isinstance(self.name, _anonymous_label):
5329 raise exc.InvalidRequestError(
5330 "Label name %s is being renamed to an anonymous label due "
5331 "to disambiguation "
5332 "which is not supported right now. Please use unique names "
5333 "for explicit labels." % (self.name)
5334 )
5336 e._propagate_attrs = selectable._propagate_attrs
5337 e._proxies.append(self)
5338 if self.type is not None:
5339 e.type = self.type
5341 return self.key, e
5344class ColumnClause(
5345 roles.DDLReferredColumnRole,
5346 roles.LabeledColumnExprRole[_T],
5347 roles.StrAsPlainColumnRole,
5348 Immutable,
5349 NamedColumn[_T],
5350):
5351 """Represents a column expression from any textual string.
5353 The :class:`.ColumnClause`, a lightweight analogue to the
5354 :class:`_schema.Column` class, is typically invoked using the
5355 :func:`_expression.column` function, as in::
5357 from sqlalchemy import column
5359 id, name = column("id"), column("name")
5360 stmt = select(id, name).select_from("user")
5362 The above statement would produce SQL like:
5364 .. sourcecode:: sql
5366 SELECT id, name FROM user
5368 :class:`.ColumnClause` is the immediate superclass of the schema-specific
5369 :class:`_schema.Column` object. While the :class:`_schema.Column`
5370 class has all the
5371 same capabilities as :class:`.ColumnClause`, the :class:`.ColumnClause`
5372 class is usable by itself in those cases where behavioral requirements
5373 are limited to simple SQL expression generation. The object has none of
5374 the associations with schema-level metadata or with execution-time
5375 behavior that :class:`_schema.Column` does,
5376 so in that sense is a "lightweight"
5377 version of :class:`_schema.Column`.
5379 Full details on :class:`.ColumnClause` usage is at
5380 :func:`_expression.column`.
5382 .. seealso::
5384 :func:`_expression.column`
5386 :class:`_schema.Column`
5388 """
5390 table: Optional[FromClause]
5391 is_literal: bool
5393 __visit_name__ = "column"
5395 _traverse_internals: _TraverseInternalsType = [
5396 ("name", InternalTraversal.dp_anon_name),
5397 ("type", InternalTraversal.dp_type),
5398 ("table", InternalTraversal.dp_clauseelement),
5399 ("is_literal", InternalTraversal.dp_boolean),
5400 ]
5402 onupdate: Optional[DefaultGenerator] = None
5403 default: Optional[DefaultGenerator] = None
5404 server_default: Optional[FetchedValue] = None
5405 server_onupdate: Optional[FetchedValue] = None
5407 _is_multiparam_column = False
5409 @property
5410 def _is_star(self): # type: ignore[override]
5411 return self.is_literal and self.name == "*"
5413 def __init__(
5414 self,
5415 text: str,
5416 type_: Optional[_TypeEngineArgument[_T]] = None,
5417 is_literal: bool = False,
5418 _selectable: Optional[FromClause] = None,
5419 ):
5420 self.key = self.name = text
5421 self.table = _selectable
5423 # if type is None, we get NULLTYPE, which is our _T. But I don't
5424 # know how to get the overloads to express that correctly
5425 self.type = type_api.to_instance(type_) # type: ignore[assignment]
5427 self.is_literal = is_literal
5429 def get_children(self, *, column_tables=False, **kw):
5430 # override base get_children() to not return the Table
5431 # or selectable that is parent to this column. Traversals
5432 # expect the columns of tables and subqueries to be leaf nodes.
5433 return []
5435 @property
5436 def entity_namespace(self):
5437 if self.table is not None:
5438 return self.table.entity_namespace
5439 else:
5440 return super().entity_namespace
5442 def _clone(self, detect_subquery_cols=False, **kw):
5443 if (
5444 detect_subquery_cols
5445 and self.table is not None
5446 and self.table._is_subquery
5447 ):
5448 clone = kw.pop("clone")
5449 table = clone(self.table, **kw)
5450 new = table.c.corresponding_column(self)
5451 return new
5453 return super()._clone(**kw)
5455 @HasMemoized_ro_memoized_attribute
5456 def _from_objects(self) -> List[FromClause]:
5457 t = self.table
5458 if t is not None:
5459 return [t]
5460 else:
5461 return []
5463 @HasMemoized.memoized_attribute
5464 def _render_label_in_columns_clause(self):
5465 return self.table is not None
5467 @property
5468 def _ddl_label(self):
5469 return self._gen_tq_label(self.name, dedupe_on_key=False)
5471 def _compare_name_for_result(self, other):
5472 if (
5473 self.is_literal
5474 or self.table is None
5475 or self.table._is_textual
5476 or not hasattr(other, "proxy_set")
5477 or (
5478 isinstance(other, ColumnClause)
5479 and (
5480 other.is_literal
5481 or other.table is None
5482 or other.table._is_textual
5483 )
5484 )
5485 ):
5486 return (hasattr(other, "name") and self.name == other.name) or (
5487 hasattr(other, "_tq_label")
5488 and self._tq_label == other._tq_label
5489 )
5490 else:
5491 return other.proxy_set.intersection(self.proxy_set)
5493 def _gen_tq_label(
5494 self, name: str, dedupe_on_key: bool = True
5495 ) -> Optional[str]:
5496 """generate table-qualified label
5498 for a table-bound column this is <tablename>_<columnname>.
5500 used primarily for LABEL_STYLE_TABLENAME_PLUS_COL
5501 as well as the .columns collection on a Join object.
5503 """
5504 label: str
5505 t = self.table
5506 if self.is_literal:
5507 return None
5508 elif t is not None and is_named_from_clause(t):
5509 if has_schema_attr(t) and t.schema:
5510 label = (
5511 t.schema.replace(".", "_") + "_" + t.name + ("_" + name)
5512 )
5513 else:
5514 assert not TYPE_CHECKING or isinstance(t, NamedFromClause)
5515 label = t.name + ("_" + name)
5517 # propagate name quoting rules for labels.
5518 if is_quoted_name(name) and name.quote is not None:
5519 if is_quoted_name(label):
5520 label.quote = name.quote
5521 else:
5522 label = quoted_name(label, name.quote)
5523 elif is_quoted_name(t.name) and t.name.quote is not None:
5524 # can't get this situation to occur, so let's
5525 # assert false on it for now
5526 assert not isinstance(label, quoted_name)
5527 label = quoted_name(label, t.name.quote)
5529 if dedupe_on_key:
5530 # ensure the label name doesn't conflict with that of an
5531 # existing column. note that this implies that any Column
5532 # must **not** set up its _label before its parent table has
5533 # all of its other Column objects set up. There are several
5534 # tables in the test suite which will fail otherwise; example:
5535 # table "owner" has columns "name" and "owner_name". Therefore
5536 # column owner.name cannot use the label "owner_name", it has
5537 # to be "owner_name_1".
5538 if label in t.c:
5539 _label = label
5540 counter = 1
5541 while _label in t.c:
5542 _label = label + f"_{counter}"
5543 counter += 1
5544 label = _label
5546 return coercions.expect(roles.TruncatedLabelRole, label)
5548 else:
5549 return name
5551 def _make_proxy(
5552 self,
5553 selectable: FromClause,
5554 *,
5555 primary_key: ColumnSet,
5556 foreign_keys: Set[KeyedColumnElement[Any]],
5557 name: Optional[str] = None,
5558 key: Optional[str] = None,
5559 name_is_truncatable: bool = False,
5560 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5561 disallow_is_literal: bool = False,
5562 **kw: Any,
5563 ) -> typing_Tuple[str, ColumnClause[_T]]:
5564 # the "is_literal" flag normally should never be propagated; a proxied
5565 # column is always a SQL identifier and never the actual expression
5566 # being evaluated. however, there is a case where the "is_literal" flag
5567 # might be used to allow the given identifier to have a fixed quoting
5568 # pattern already, so maintain the flag for the proxy unless a
5569 # :class:`.Label` object is creating the proxy. See [ticket:4730].
5570 is_literal = (
5571 not disallow_is_literal
5572 and self.is_literal
5573 and (
5574 # note this does not accommodate for quoted_name differences
5575 # right now
5576 name is None
5577 or name == self.name
5578 )
5579 )
5580 c = self._constructor(
5581 (
5582 coercions.expect(roles.TruncatedLabelRole, name or self.name)
5583 if name_is_truncatable
5584 else (name or self.name)
5585 ),
5586 type_=self.type,
5587 _selectable=selectable,
5588 is_literal=is_literal,
5589 )
5590 c._propagate_attrs = selectable._propagate_attrs
5591 if name is None:
5592 c.key = self.key
5593 if compound_select_cols:
5594 c._proxies = list(compound_select_cols)
5595 else:
5596 c._proxies = [self]
5598 if selectable._is_clone_of is not None:
5599 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
5600 return c.key, c
5603class TableValuedColumn(NamedColumn[_T]):
5604 __visit_name__ = "table_valued_column"
5606 _traverse_internals: _TraverseInternalsType = [
5607 ("name", InternalTraversal.dp_anon_name),
5608 ("type", InternalTraversal.dp_type),
5609 ("scalar_alias", InternalTraversal.dp_clauseelement),
5610 ]
5612 def __init__(self, scalar_alias: NamedFromClause, type_: TypeEngine[_T]):
5613 self.scalar_alias = scalar_alias
5614 self.key = self.name = scalar_alias.name
5615 self.type = type_
5617 def _copy_internals(
5618 self, clone: _CloneCallableType = _clone, **kw: Any
5619 ) -> None:
5620 self.scalar_alias = clone(self.scalar_alias, **kw)
5621 self.key = self.name = self.scalar_alias.name
5623 @util.ro_non_memoized_property
5624 def _from_objects(self) -> List[FromClause]:
5625 return [self.scalar_alias]
5628class CollationClause(ColumnElement[str]):
5629 __visit_name__ = "collation"
5631 _traverse_internals: _TraverseInternalsType = [
5632 ("collation", InternalTraversal.dp_string),
5633 ("collation_schema", InternalTraversal.dp_string),
5634 ]
5636 @classmethod
5637 @util.preload_module("sqlalchemy.sql.sqltypes")
5638 def _create_collation_expression(
5639 cls,
5640 expression: _ColumnExpressionArgument[str],
5641 collation: str,
5642 collation_schema: Optional[str] = None,
5643 ) -> BinaryExpression[str]:
5645 sqltypes = util.preloaded.sql_sqltypes
5647 expr = coercions.expect(roles.ExpressionElementRole[str], expression)
5649 if expr.type._type_affinity is sqltypes.String:
5650 collate_type = expr.type._with_collation(
5651 collation, collation_schema
5652 )
5653 else:
5654 collate_type = expr.type
5656 return BinaryExpression(
5657 expr,
5658 CollationClause(collation, collation_schema),
5659 operators.collate,
5660 type_=collate_type,
5661 )
5663 def __init__(self, collation, collation_schema=None):
5664 self.collation = collation
5665 self.collation_schema = collation_schema
5668class _IdentifiedClause(Executable, ClauseElement):
5669 __visit_name__ = "identified"
5671 def __init__(self, ident):
5672 self.ident = ident
5675class SavepointClause(_IdentifiedClause):
5676 __visit_name__ = "savepoint"
5677 inherit_cache = False
5680class RollbackToSavepointClause(_IdentifiedClause):
5681 __visit_name__ = "rollback_to_savepoint"
5682 inherit_cache = False
5685class ReleaseSavepointClause(_IdentifiedClause):
5686 __visit_name__ = "release_savepoint"
5687 inherit_cache = False
5690class quoted_name(util.MemoizedSlots, str):
5691 """Represent a SQL identifier combined with quoting preferences.
5693 :class:`.quoted_name` is a Python unicode/str subclass which
5694 represents a particular identifier name along with a
5695 ``quote`` flag. This ``quote`` flag, when set to
5696 ``True`` or ``False``, overrides automatic quoting behavior
5697 for this identifier in order to either unconditionally quote
5698 or to not quote the name. If left at its default of ``None``,
5699 quoting behavior is applied to the identifier on a per-backend basis
5700 based on an examination of the token itself.
5702 A :class:`.quoted_name` object with ``quote=True`` is also
5703 prevented from being modified in the case of a so-called
5704 "name normalize" option. Certain database backends, such as
5705 Oracle Database, Firebird, and DB2 "normalize" case-insensitive names
5706 as uppercase. The SQLAlchemy dialects for these backends
5707 convert from SQLAlchemy's lower-case-means-insensitive convention
5708 to the upper-case-means-insensitive conventions of those backends.
5709 The ``quote=True`` flag here will prevent this conversion from occurring
5710 to support an identifier that's quoted as all lower case against
5711 such a backend.
5713 The :class:`.quoted_name` object is normally created automatically
5714 when specifying the name for key schema constructs such as
5715 :class:`_schema.Table`, :class:`_schema.Column`, and others.
5716 The class can also be
5717 passed explicitly as the name to any function that receives a name which
5718 can be quoted, such as :meth:`.Inspector.has_table` with an
5719 unconditionally quoted name::
5721 from sqlalchemy import create_engine
5722 from sqlalchemy import inspect
5723 from sqlalchemy.sql import quoted_name
5725 engine = create_engine("oracle+oracledb://some_dsn")
5726 print(inspect(engine).has_table(quoted_name("some_table", True)))
5728 The above logic will run the "has table" logic against the Oracle Database
5729 backend, passing the name exactly as ``"some_table"`` without converting to
5730 upper case.
5732 A :class:`.quoted_name` object with ``quote=False`` may be passed to APIs
5733 that apply automatic quoting in order to keep the given name unquoted,
5734 such as when a PostgreSQL ``INHERITS`` option refers to a schema-qualified
5735 table name like ``my_schema.some_table``.
5737 """
5739 __slots__ = "quote", "lower", "upper"
5741 quote: Optional[bool]
5743 @overload
5744 @classmethod
5745 def construct(cls, value: str, quote: Optional[bool]) -> quoted_name: ...
5747 @overload
5748 @classmethod
5749 def construct(cls, value: None, quote: Optional[bool]) -> None: ...
5751 @classmethod
5752 def construct(
5753 cls, value: Optional[str], quote: Optional[bool]
5754 ) -> Optional[quoted_name]:
5755 if value is None:
5756 return None
5757 else:
5758 return quoted_name(value, quote)
5760 def __new__(cls, value: str, quote: Optional[bool]) -> Self:
5761 assert (
5762 value is not None
5763 ), "use quoted_name.construct() for None passthrough"
5764 if isinstance(value, cls) and (quote is None or value.quote == quote):
5765 return value
5766 self = super().__new__(cls, value)
5768 self.quote = quote
5769 return self
5771 def __reduce__(self):
5772 return quoted_name, (str(self), self.quote)
5774 def _memoized_method_lower(self):
5775 if self.quote:
5776 return self
5777 else:
5778 return str(self).lower()
5780 def _memoized_method_upper(self):
5781 if self.quote:
5782 return self
5783 else:
5784 return str(self).upper()
5787def _find_columns(clause: ClauseElement) -> Set[ColumnClause[Any]]:
5788 """locate Column objects within the given expression."""
5790 cols: Set[ColumnClause[Any]] = set()
5791 traverse(clause, {}, {"column": cols.add})
5792 return cols
5795def _type_from_args(args: Sequence[ColumnElement[_T]]) -> TypeEngine[_T]:
5796 for a in args:
5797 if not a.type._isnull:
5798 return a.type
5799 else:
5800 return type_api.NULLTYPE # type: ignore[return-value]
5803def _corresponding_column_or_error(fromclause, column, require_embedded=False):
5804 c = fromclause.corresponding_column(
5805 column, require_embedded=require_embedded
5806 )
5807 if c is None:
5808 raise exc.InvalidRequestError(
5809 "Given column '%s', attached to table '%s', "
5810 "failed to locate a corresponding column from table '%s'"
5811 % (column, getattr(column, "table", None), fromclause.description)
5812 )
5813 return c
5816class _memoized_property_but_not_nulltype(
5817 util.memoized_property["TypeEngine[_T]"]
5818):
5819 """memoized property, but dont memoize NullType"""
5821 def __get__(self, obj, cls):
5822 if obj is None:
5823 return self
5824 result = self.fget(obj)
5825 if not result._isnull:
5826 obj.__dict__[self.__name__] = result
5827 return result
5830class AnnotatedColumnElement(Annotated):
5831 _Annotated__element: ColumnElement[Any]
5833 def __init__(self, element, values):
5834 Annotated.__init__(self, element, values)
5835 for attr in (
5836 "comparator",
5837 "_proxy_key",
5838 "_tq_key_label",
5839 "_tq_label",
5840 "_non_anon_label",
5841 "type",
5842 ):
5843 self.__dict__.pop(attr, None)
5844 for attr in ("name", "key", "table"):
5845 if self.__dict__.get(attr, False) is None:
5846 self.__dict__.pop(attr)
5848 def _with_annotations(self, values):
5849 clone = super()._with_annotations(values)
5850 for attr in (
5851 "comparator",
5852 "_proxy_key",
5853 "_tq_key_label",
5854 "_tq_label",
5855 "_non_anon_label",
5856 ):
5857 clone.__dict__.pop(attr, None)
5858 return clone
5860 @util.memoized_property
5861 def name(self):
5862 """pull 'name' from parent, if not present"""
5863 return self._Annotated__element.name
5865 @_memoized_property_but_not_nulltype
5866 def type(self):
5867 """pull 'type' from parent and don't cache if null.
5869 type is routinely changed on existing columns within the
5870 mapped_column() initialization process, and "type" is also consulted
5871 during the creation of SQL expressions. Therefore it can change after
5872 it was already retrieved. At the same time we don't want annotated
5873 objects having overhead when expressions are produced, so continue
5874 to memoize, but only when we have a non-null type.
5876 """
5877 return self._Annotated__element.type
5879 @util.memoized_property
5880 def table(self):
5881 """pull 'table' from parent, if not present"""
5882 return self._Annotated__element.table
5884 @util.memoized_property
5885 def key(self):
5886 """pull 'key' from parent, if not present"""
5887 return self._Annotated__element.key
5889 @util.memoized_property
5890 def info(self) -> _InfoType:
5891 if TYPE_CHECKING:
5892 assert isinstance(self._Annotated__element, Column)
5893 return self._Annotated__element.info
5895 @util.memoized_property
5896 def _anon_name_label(self) -> str:
5897 return self._Annotated__element._anon_name_label
5900class _truncated_label(quoted_name):
5901 """A unicode subclass used to identify symbolic "
5902 "names that may require truncation."""
5904 __slots__ = ()
5906 def __new__(cls, value: str, quote: Optional[bool] = None) -> Self:
5907 quote = getattr(value, "quote", quote)
5908 # return super(_truncated_label, cls).__new__(cls, value, quote, True)
5909 return super().__new__(cls, value, quote)
5911 def __reduce__(self) -> Any:
5912 return self.__class__, (str(self), self.quote)
5914 def apply_map(self, map_: Mapping[str, Any]) -> str:
5915 return self
5918class conv(_truncated_label):
5919 """Mark a string indicating that a name has already been converted
5920 by a naming convention.
5922 This is a string subclass that indicates a name that should not be
5923 subject to any further naming conventions.
5925 E.g. when we create a :class:`.Constraint` using a naming convention
5926 as follows::
5928 m = MetaData(
5929 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5930 )
5931 t = Table(
5932 "t", m, Column("x", Integer), CheckConstraint("x > 5", name="x5")
5933 )
5935 The name of the above constraint will be rendered as ``"ck_t_x5"``.
5936 That is, the existing name ``x5`` is used in the naming convention as the
5937 ``constraint_name`` token.
5939 In some situations, such as in migration scripts, we may be rendering
5940 the above :class:`.CheckConstraint` with a name that's already been
5941 converted. In order to make sure the name isn't double-modified, the
5942 new name is applied using the :func:`_schema.conv` marker. We can
5943 use this explicitly as follows::
5946 m = MetaData(
5947 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5948 )
5949 t = Table(
5950 "t",
5951 m,
5952 Column("x", Integer),
5953 CheckConstraint("x > 5", name=conv("ck_t_x5")),
5954 )
5956 Where above, the :func:`_schema.conv` marker indicates that the constraint
5957 name here is final, and the name will render as ``"ck_t_x5"`` and not
5958 ``"ck_t_ck_t_x5"``
5960 .. seealso::
5962 :ref:`constraint_naming_conventions`
5964 """
5966 __slots__ = ()
5969# for backwards compatibility in case
5970# someone is re-implementing the
5971# _truncated_identifier() sequence in a custom
5972# compiler
5973_generated_label = _truncated_label
5974_anonymous_label_escape = re.compile(r"[%\(\) \$]+")
5977class _anonymous_label(_truncated_label):
5978 """A unicode subclass used to identify anonymously
5979 generated names."""
5981 __slots__ = ()
5983 @classmethod
5984 def safe_construct_with_key(
5985 cls, seed: int | str, body: str, sanitize_key: bool = False
5986 ) -> typing_Tuple[_anonymous_label, str]:
5987 # need to escape chars that interfere with format
5988 # strings in any case, issue #8724
5989 body = _anonymous_label_escape.sub("_", body)
5991 if sanitize_key:
5992 # sanitize_key is then an extra step used by BindParameter
5993 body = body.strip("_")
5995 key = f"{seed} {body.replace('%', '%%')}"
5996 label = _anonymous_label(f"%({key})s")
5997 return label, key
5999 @classmethod
6000 def safe_construct(
6001 cls, seed: int | str, body: str, sanitize_key: bool = False
6002 ) -> _anonymous_label:
6003 # need to escape chars that interfere with format
6004 # strings in any case, issue #8724
6005 body = _anonymous_label_escape.sub("_", body)
6007 if sanitize_key:
6008 # sanitize_key is then an extra step used by BindParameter
6009 body = body.strip("_")
6011 return _anonymous_label(f"%({seed} {body.replace('%', '%%')})s")
6013 def __add__(self, other: str) -> _anonymous_label:
6014 if "%" in other and not isinstance(other, _anonymous_label):
6015 other = str(other).replace("%", "%%")
6016 else:
6017 other = str(other)
6019 return _anonymous_label(
6020 quoted_name(
6021 str.__add__(self, other),
6022 self.quote,
6023 )
6024 )
6026 def __radd__(self, other: str) -> _anonymous_label:
6027 if "%" in other and not isinstance(other, _anonymous_label):
6028 other = str(other).replace("%", "%%")
6029 else:
6030 other = str(other)
6032 return _anonymous_label(
6033 quoted_name(
6034 str.__add__(other, self),
6035 self.quote,
6036 )
6037 )
6039 def apply_map(self, map_: Mapping[str, Any]) -> str:
6040 if self.quote is not None:
6041 # preserve quoting only if necessary
6042 return quoted_name(self % map_, self.quote)
6043 else:
6044 # else skip the constructor call
6045 return self % map_