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 # also used by the compiler to render the text; the two must find
2573 # the same names, else a name is rendered for which no
2574 # BindParameter exists
2575 _bind_params_regex = re.compile(
2576 r"(?<![:\w\x5c]):(\w+)(?![:\w])", re.UNICODE
2577 )
2579 @property
2580 def _is_star(self) -> bool: # type: ignore[override]
2581 return self.text == "*"
2583 def __init__(self, text: str):
2584 self._bindparams: Dict[str, BindParameter[Any]] = {}
2586 def repl(m):
2587 self._bindparams[m.group(1)] = BindParameter(m.group(1))
2588 return ":%s" % m.group(1)
2590 # scan the string and search for bind parameter names, add them
2591 # to the list of bindparams
2592 self.text = self._bind_params_regex.sub(repl, text)
2594 @_generative
2595 def bindparams(
2596 self,
2597 *binds: BindParameter[Any],
2598 **names_to_values: Any,
2599 ) -> Self:
2600 """Establish the values and/or types of bound parameters within
2601 this :class:`_expression.TextClause` construct.
2603 Given a text construct such as::
2605 from sqlalchemy import text
2607 stmt = text(
2608 "SELECT id, name FROM user WHERE name=:name AND timestamp=:timestamp"
2609 )
2611 the :meth:`_expression.TextClause.bindparams`
2612 method can be used to establish
2613 the initial value of ``:name`` and ``:timestamp``,
2614 using simple keyword arguments::
2616 stmt = stmt.bindparams(
2617 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2618 )
2620 Where above, new :class:`.BindParameter` objects
2621 will be generated with the names ``name`` and ``timestamp``, and
2622 values of ``jack`` and ``datetime.datetime(2012, 10, 8, 15, 12, 5)``,
2623 respectively. The types will be
2624 inferred from the values given, in this case :class:`.String` and
2625 :class:`.DateTime`.
2627 When specific typing behavior is needed, the positional ``*binds``
2628 argument can be used in which to specify :func:`.bindparam` constructs
2629 directly. These constructs must include at least the ``key``
2630 argument, then an optional value and type::
2632 from sqlalchemy import bindparam
2634 stmt = stmt.bindparams(
2635 bindparam("name", value="jack", type_=String),
2636 bindparam("timestamp", type_=DateTime),
2637 )
2639 Above, we specified the type of :class:`.DateTime` for the
2640 ``timestamp`` bind, and the type of :class:`.String` for the ``name``
2641 bind. In the case of ``name`` we also set the default value of
2642 ``"jack"``.
2644 Additional bound parameters can be supplied at statement execution
2645 time, e.g.::
2647 result = connection.execute(
2648 stmt, timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2649 )
2651 The :meth:`_expression.TextClause.bindparams`
2652 method can be called repeatedly,
2653 where it will reuse existing :class:`.BindParameter` objects to add
2654 new information. For example, we can call
2655 :meth:`_expression.TextClause.bindparams`
2656 first with typing information, and a
2657 second time with value information, and it will be combined::
2659 stmt = text(
2660 "SELECT id, name FROM user WHERE name=:name "
2661 "AND timestamp=:timestamp"
2662 )
2663 stmt = stmt.bindparams(
2664 bindparam("name", type_=String), bindparam("timestamp", type_=DateTime)
2665 )
2666 stmt = stmt.bindparams(
2667 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2668 )
2670 The :meth:`_expression.TextClause.bindparams`
2671 method also supports the concept of
2672 **unique** bound parameters. These are parameters that are
2673 "uniquified" on name at statement compilation time, so that multiple
2674 :func:`_expression.text`
2675 constructs may be combined together without the names
2676 conflicting. To use this feature, specify the
2677 :paramref:`.BindParameter.unique` flag on each :func:`.bindparam`
2678 object::
2680 stmt1 = text("select id from table where name=:name").bindparams(
2681 bindparam("name", value="name1", unique=True)
2682 )
2683 stmt2 = text("select id from table where name=:name").bindparams(
2684 bindparam("name", value="name2", unique=True)
2685 )
2687 union = union_all(stmt1.columns(column("id")), stmt2.columns(column("id")))
2689 The above statement will render as:
2691 .. sourcecode:: sql
2693 select id from table where name=:name_1
2694 UNION ALL select id from table where name=:name_2
2696 """ # noqa: E501
2697 self._bindparams = new_params = self._bindparams.copy()
2699 for bind in binds:
2700 try:
2701 # the regex used for text() currently will not match
2702 # a unique/anonymous key in any case, so use the _orig_key
2703 # so that a text() construct can support unique parameters
2704 existing = new_params[bind._orig_key]
2705 except KeyError as err:
2706 raise exc.ArgumentError(
2707 "This text() construct doesn't define a "
2708 "bound parameter named %r" % bind._orig_key
2709 ) from err
2710 else:
2711 new_params[existing._orig_key] = bind
2713 for key, value in names_to_values.items():
2714 try:
2715 existing = new_params[key]
2716 except KeyError as err:
2717 raise exc.ArgumentError(
2718 "This text() construct doesn't define a "
2719 "bound parameter named %r" % key
2720 ) from err
2721 else:
2722 new_params[key] = existing._with_value(value, required=False)
2723 return self
2725 @property
2726 def type(self) -> TypeEngine[Any]:
2727 return type_api.NULLTYPE
2729 @property
2730 def comparator(self):
2731 # TODO: this seems wrong, it seems like we might not
2732 # be using this method.
2733 return self.type.comparator_factory(self) # type: ignore[arg-type]
2735 def self_group(
2736 self, against: Optional[OperatorType] = None
2737 ) -> Union[Self, Grouping[Any]]:
2738 if against is operators.in_op:
2739 return Grouping(self)
2740 else:
2741 return self
2744class _TStringText(ClauseElement):
2745 """Represent a literal string portion of a :class:`.TString`.
2747 Unlike :class:`.TextClause`, the text is not scanned for bound parameter
2748 names and has no escaping rules; it is rendered as is.
2750 .. versionadded:: 2.1.2
2752 """
2754 __visit_name__ = "tstring_text"
2756 _traverse_internals: _TraverseInternalsType = [
2757 ("text", InternalTraversal.dp_string)
2758 ]
2760 def __init__(self, text: str):
2761 self.text = text
2763 @property
2764 def _is_star(self) -> bool: # type: ignore[override]
2765 return self.text == "*"
2768class TString(AbstractTextClause, inspection.Inspectable["TString"]):
2769 """Represent a SQL template string using Python 3.14+ t-strings.
2771 E.g.::
2773 from sqlalchemy import tstring, column
2775 a = 5
2776 b = 10
2777 stmt = tstring(t"select {a}, {b}")
2778 result = connection.execute(stmt)
2780 The :class:`_expression.TString` construct is produced using the
2781 :func:`_expression.tstring` function; see that function for full
2782 documentation.
2784 .. versionadded:: 2.1
2786 .. seealso::
2788 :func:`_expression.tstring`
2790 """
2792 __visit_name__ = "tstring"
2794 _traverse_internals: _TraverseInternalsType = [
2795 ("parts", InternalTraversal.dp_clauseelement_list)
2796 ] + ExecutableStatement._executable_traverse_internals
2798 @property
2799 def _is_star(self) -> bool: # type: ignore[override]
2800 return (
2801 len(self.parts) == 1
2802 and isinstance(self.parts[0], _TStringText)
2803 and self.parts[0]._is_star
2804 )
2806 def __init__(self, template: Template):
2807 """Construct a :class:`_expression.TString` from a Python 3.14+
2808 template string.
2810 :param template: a Python 3.14+ template string (t-string) that
2811 contains SQL fragments and Python expressions to be interpolated.
2813 """
2814 self.parts: List[ClauseElement] = []
2816 if not isinstance(template, Template):
2817 raise exc.ArgumentError("pep-750 Tstring (e.g. t'...') expected")
2819 for part in template:
2820 if isinstance(part, str):
2821 self.parts.append(_TStringText(part))
2822 else:
2823 assert hasattr(part, "value")
2824 self.parts.append(
2825 coercions.expect(roles.TStringElementRole, part.value)
2826 )
2828 def bindparams(
2829 self,
2830 *binds: BindParameter[Any],
2831 **names_to_values: Any,
2832 ) -> Self:
2833 """Not supported for TString constructs.
2835 TString constructs do not support .bindparams(). Bind parameters
2836 are automatically created from interpolated values.
2838 """
2839 raise NotImplementedError(
2840 "TString constructs do not support .bindparams(). "
2841 "Bind parameters are automatically created "
2842 "from interpolated values."
2843 )
2846class Null(SingletonConstant, roles.ConstExprRole[None], ColumnElement[None]):
2847 """Represent the NULL keyword in a SQL statement.
2849 :class:`.Null` is accessed as a constant via the
2850 :func:`.null` function.
2852 """
2854 __visit_name__ = "null"
2856 _traverse_internals: _TraverseInternalsType = []
2857 _singleton: Null
2859 if not TYPE_CHECKING:
2861 @util.memoized_property
2862 def type(self) -> TypeEngine[_T]: # noqa: A001
2863 return type_api.NULLTYPE
2865 @classmethod
2866 def _instance(cls) -> Null:
2867 """Return a constant :class:`.Null` construct."""
2869 return Null._singleton
2872Null._create_singleton()
2875class False_(
2876 SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]
2877):
2878 """Represent the ``false`` keyword, or equivalent, in a SQL statement.
2880 :class:`.False_` is accessed as a constant via the
2881 :func:`.false` function.
2883 """
2885 __visit_name__ = "false"
2886 _traverse_internals: _TraverseInternalsType = []
2887 _singleton: False_
2889 if not TYPE_CHECKING:
2891 @util.memoized_property
2892 def type(self) -> TypeEngine[_T]: # noqa: A001
2893 return type_api.BOOLEANTYPE
2895 def _negate(self) -> True_:
2896 return True_._singleton
2898 @classmethod
2899 def _instance(cls) -> False_:
2900 return False_._singleton
2903False_._create_singleton()
2906class True_(SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]):
2907 """Represent the ``true`` keyword, or equivalent, in a SQL statement.
2909 :class:`.True_` is accessed as a constant via the
2910 :func:`.true` function.
2912 """
2914 __visit_name__ = "true"
2916 _traverse_internals: _TraverseInternalsType = []
2917 _singleton: True_
2919 if not TYPE_CHECKING:
2921 @util.memoized_property
2922 def type(self) -> TypeEngine[_T]: # noqa: A001
2923 return type_api.BOOLEANTYPE
2925 def _negate(self) -> False_:
2926 return False_._singleton
2928 @classmethod
2929 def _ifnone(
2930 cls, other: Optional[ColumnElement[Any]]
2931 ) -> ColumnElement[Any]:
2932 if other is None:
2933 return cls._instance()
2934 else:
2935 return other
2937 @classmethod
2938 def _instance(cls) -> True_:
2939 return True_._singleton
2942True_._create_singleton()
2945class ElementList(DQLDMLClauseElement):
2946 """Describe a list of clauses that will be space separated.
2948 This is a minimal version of :class:`.ClauseList` which is used by
2949 the :class:`.HasSyntaxExtension` class. It does not do any coercions
2950 so should be used internally only.
2952 .. versionadded:: 2.1
2954 """
2956 __visit_name__ = "element_list"
2958 _traverse_internals: _TraverseInternalsType = [
2959 ("clauses", InternalTraversal.dp_clauseelement_tuple),
2960 ]
2962 clauses: typing_Tuple[ClauseElement, ...]
2964 def __init__(self, clauses: Sequence[ClauseElement]):
2965 self.clauses = tuple(clauses)
2968class OrderByList(
2969 roles.OrderByRole,
2970 operators.OrderingOperators,
2971 DQLDMLClauseElement,
2972):
2973 """Describe a list of clauses that will be comma separated to nest
2974 within an ORDER BY.
2976 .. versionadded:: 2.1
2978 """
2980 __visit_name__ = "order_by_list"
2982 _traverse_internals: _TraverseInternalsType = [
2983 ("clauses", InternalTraversal.dp_clauseelement_tuple),
2984 ]
2986 clauses: List[ColumnElement[Any]]
2988 def __init__(
2989 self,
2990 clauses: Iterable[Union[OrderByList, _ColumnExpressionArgument[Any]]],
2991 ):
2992 text_converter_role: Type[roles.SQLRole] = roles.ByOfRole
2993 self._text_converter_role = text_converter_role
2995 self.clauses = [
2996 coercions.expect(
2997 text_converter_role, clause, apply_propagate_attrs=self
2998 )
2999 for clause in clauses
3000 ]
3002 def __iter__(self) -> Iterator[ColumnElement[Any]]:
3003 return iter(self.clauses)
3005 def __len__(self) -> int:
3006 return len(self.clauses)
3008 @property
3009 def _select_iterable(self) -> _SelectIterable:
3010 return itertools.chain.from_iterable(
3011 [elem._select_iterable for elem in self.clauses]
3012 )
3014 @util.ro_non_memoized_property
3015 def _from_objects(self) -> List[FromClause]:
3016 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
3018 def self_group(
3019 self, against: Optional[OperatorType] = None
3020 ) -> Union[Self, Grouping[Any]]:
3021 return self
3023 def desc(self) -> OrderByList:
3024 return OrderByList([e.desc() for e in self.clauses])
3026 def asc(self) -> OrderByList:
3027 return OrderByList([e.asc() for e in self.clauses])
3029 def nulls_first(self) -> OrderByList:
3030 return OrderByList([e.nulls_first() for e in self.clauses])
3032 def nulls_last(self) -> OrderByList:
3033 return OrderByList([e.nulls_last() for e in self.clauses])
3036class ClauseList(
3037 roles.InElementRole,
3038 roles.OrderByRole,
3039 roles.ColumnsClauseRole,
3040 roles.DMLColumnRole,
3041 DQLDMLClauseElement,
3042):
3043 """Describe a list of clauses, separated by an operator.
3045 By default, is comma-separated, such as a column listing.
3047 """
3049 __visit_name__ = "clauselist"
3051 # Used by ORM context.py to identify ClauseList objects in legacy
3052 # composite attribute queries (see test_query_cols_legacy test)
3053 _is_clause_list = True
3055 _traverse_internals: _TraverseInternalsType = [
3056 ("clauses", InternalTraversal.dp_clauseelement_list),
3057 ("operator", InternalTraversal.dp_operator),
3058 ]
3060 clauses: List[ColumnElement[Any]]
3062 def __init__(
3063 self,
3064 *clauses: _ColumnExpressionArgument[Any],
3065 operator: OperatorType = operators.comma_op,
3066 group: bool = True,
3067 group_contents: bool = True,
3068 _literal_as_text_role: Type[roles.SQLRole] = roles.WhereHavingRole,
3069 ):
3070 self.operator = operator
3071 self.group = group
3072 self.group_contents = group_contents
3073 clauses_iterator: Iterable[_ColumnExpressionArgument[Any]] = clauses
3074 text_converter_role: Type[roles.SQLRole] = _literal_as_text_role
3075 self._text_converter_role = text_converter_role
3077 if self.group_contents:
3078 self.clauses = [
3079 coercions.expect(
3080 text_converter_role, clause, apply_propagate_attrs=self
3081 ).self_group(against=self.operator)
3082 for clause in clauses_iterator
3083 ]
3084 else:
3085 self.clauses = [
3086 coercions.expect(
3087 text_converter_role, clause, apply_propagate_attrs=self
3088 )
3089 for clause in clauses_iterator
3090 ]
3091 self._is_implicitly_boolean = operators.is_boolean(self.operator)
3093 @classmethod
3094 def _construct_raw(
3095 cls,
3096 operator: OperatorType,
3097 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
3098 ) -> ClauseList:
3099 self = cls.__new__(cls)
3100 self.clauses = list(clauses) if clauses else []
3101 self.group = True
3102 self.operator = operator
3103 self.group_contents = True
3104 self._is_implicitly_boolean = False
3105 return self
3107 def __iter__(self) -> Iterator[ColumnElement[Any]]:
3108 return iter(self.clauses)
3110 def __len__(self) -> int:
3111 return len(self.clauses)
3113 @property
3114 def _select_iterable(self) -> _SelectIterable:
3115 return itertools.chain.from_iterable(
3116 [elem._select_iterable for elem in self.clauses]
3117 )
3119 def append(self, clause):
3120 if self.group_contents:
3121 self.clauses.append(
3122 coercions.expect(self._text_converter_role, clause).self_group(
3123 against=self.operator
3124 )
3125 )
3126 else:
3127 self.clauses.append(
3128 coercions.expect(self._text_converter_role, clause)
3129 )
3131 @util.ro_non_memoized_property
3132 def _from_objects(self) -> List[FromClause]:
3133 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
3135 def self_group(
3136 self, against: Optional[OperatorType] = None
3137 ) -> Union[Self, Grouping[Any]]:
3138 if self.group and operators.is_precedent(self.operator, against):
3139 return Grouping(self)
3140 else:
3141 return self
3144class OperatorExpression(ColumnElement[_T]):
3145 """base for expressions that contain an operator and operands
3147 .. versionadded:: 2.0
3149 """
3151 operator: OperatorType
3152 type: TypeEngine[_T]
3154 group: bool = True
3156 @property
3157 def is_comparison(self):
3158 return operators.is_comparison(self.operator)
3160 def self_group(
3161 self, against: Optional[OperatorType] = None
3162 ) -> Union[Self, Grouping[_T]]:
3163 if (
3164 self.group
3165 and operators.is_precedent(self.operator, against)
3166 or (
3167 # a negate against a non-boolean operator
3168 # doesn't make too much sense but we should
3169 # group for that
3170 against is operators.inv
3171 and not operators.is_boolean(self.operator)
3172 )
3173 ):
3174 return Grouping(self)
3175 else:
3176 return self
3178 @property
3179 def _flattened_operator_clauses(
3180 self,
3181 ) -> typing_Tuple[ColumnElement[Any], ...]:
3182 raise NotImplementedError()
3184 @classmethod
3185 def _construct_for_op(
3186 cls,
3187 left: ColumnElement[Any],
3188 right: ColumnElement[Any],
3189 op: OperatorType,
3190 *,
3191 type_: TypeEngine[_T],
3192 negate: Optional[OperatorType] = None,
3193 modifiers: Optional[Mapping[str, Any]] = None,
3194 ) -> OperatorExpression[_T]:
3195 if operators.is_associative(op):
3196 assert (
3197 negate is None
3198 ), f"negate not supported for associative operator {op}"
3200 multi = False
3201 if getattr(
3202 left, "operator", None
3203 ) is op and type_._compare_type_affinity(left.type):
3204 multi = True
3205 left_flattened = left._flattened_operator_clauses
3206 else:
3207 left_flattened = (left,)
3209 if getattr(
3210 right, "operator", None
3211 ) is op and type_._compare_type_affinity(right.type):
3212 multi = True
3213 right_flattened = right._flattened_operator_clauses
3214 else:
3215 right_flattened = (right,)
3217 if multi:
3218 return ExpressionClauseList._construct_for_list(
3219 op,
3220 type_,
3221 *(left_flattened + right_flattened),
3222 )
3224 if left._is_collection_aggregate or right._is_collection_aggregate:
3225 negate = None
3227 return BinaryExpression(
3228 left, right, op, type_=type_, negate=negate, modifiers=modifiers
3229 )
3232class ExpressionClauseList(OperatorExpression[_T]):
3233 """Describe a list of clauses, separated by an operator,
3234 in a column expression context.
3236 :class:`.ExpressionClauseList` differs from :class:`.ClauseList` in that
3237 it represents a column-oriented DQL expression only, not an open ended
3238 list of anything comma separated.
3240 .. versionadded:: 2.0
3242 """
3244 __visit_name__ = "expression_clauselist"
3246 _traverse_internals: _TraverseInternalsType = [
3247 ("clauses", InternalTraversal.dp_clauseelement_tuple),
3248 ("operator", InternalTraversal.dp_operator),
3249 ]
3251 clauses: typing_Tuple[ColumnElement[Any], ...]
3253 group: bool
3255 def __init__(
3256 self,
3257 operator: OperatorType,
3258 *clauses: _ColumnExpressionArgument[Any],
3259 type_: Optional[_TypeEngineArgument[_T]] = None,
3260 ):
3261 self.operator = operator
3263 self.clauses = tuple(
3264 coercions.expect(
3265 roles.ExpressionElementRole, clause, apply_propagate_attrs=self
3266 )
3267 for clause in clauses
3268 )
3269 self._is_implicitly_boolean = operators.is_boolean(self.operator)
3270 self.type = type_api.to_instance(type_) # type: ignore[assignment]
3272 @property
3273 def _flattened_operator_clauses(
3274 self,
3275 ) -> typing_Tuple[ColumnElement[Any], ...]:
3276 return self.clauses
3278 def __iter__(self) -> Iterator[ColumnElement[Any]]:
3279 return iter(self.clauses)
3281 def __len__(self) -> int:
3282 return len(self.clauses)
3284 @property
3285 def _select_iterable(self) -> _SelectIterable:
3286 return (self,)
3288 @util.ro_non_memoized_property
3289 def _from_objects(self) -> List[FromClause]:
3290 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
3292 def _append_inplace(self, clause: ColumnElement[Any]) -> None:
3293 self.clauses += (clause,)
3295 @classmethod
3296 def _construct_for_list(
3297 cls,
3298 operator: OperatorType,
3299 type_: TypeEngine[_T],
3300 *clauses: ColumnElement[Any],
3301 group: bool = True,
3302 ) -> ExpressionClauseList[_T]:
3303 self = cls.__new__(cls)
3304 self.group = group
3305 if group:
3306 self.clauses = tuple(
3307 c.self_group(against=operator) for c in clauses
3308 )
3309 else:
3310 self.clauses = clauses
3311 self.operator = operator
3312 self.type = type_
3313 for c in clauses:
3314 if c._propagate_attrs:
3315 self._propagate_attrs = c._propagate_attrs
3316 break
3317 return self
3319 def _negate(self) -> Any:
3320 grouped = self.self_group(against=operators.inv)
3321 assert isinstance(grouped, ColumnElement)
3322 return UnaryExpression(grouped, operator=operators.inv)
3325class BooleanClauseList(ExpressionClauseList[bool]):
3326 __visit_name__ = "expression_clauselist"
3327 inherit_cache = True
3329 def __init__(self, *arg, **kw):
3330 raise NotImplementedError(
3331 "BooleanClauseList has a private constructor"
3332 )
3334 @classmethod
3335 def _process_clauses_for_boolean(
3336 cls,
3337 operator: OperatorType,
3338 continue_on: Any,
3339 skip_on: Any,
3340 clauses: Iterable[ColumnElement[Any]],
3341 ) -> typing_Tuple[int, List[ColumnElement[Any]]]:
3342 has_continue_on = None
3344 convert_clauses = []
3346 against = operators._asbool
3347 lcc = 0
3349 for clause in clauses:
3350 if clause is continue_on:
3351 # instance of continue_on, like and_(x, y, True, z), store it
3352 # if we didn't find one already, we will use it if there
3353 # are no other expressions here.
3354 has_continue_on = clause
3355 elif clause is skip_on:
3356 # instance of skip_on, e.g. and_(x, y, False, z), cancels
3357 # the rest out
3358 convert_clauses = [clause]
3359 lcc = 1
3360 break
3361 else:
3362 if not lcc:
3363 lcc = 1
3364 else:
3365 against = operator
3366 # technically this would be len(convert_clauses) + 1
3367 # however this only needs to indicate "greater than one"
3368 lcc = 2
3369 convert_clauses.append(clause)
3371 if not convert_clauses and has_continue_on is not None:
3372 convert_clauses = [has_continue_on]
3373 lcc = 1
3375 return lcc, [c.self_group(against=against) for c in convert_clauses]
3377 @classmethod
3378 def _construct(
3379 cls,
3380 operator: OperatorType,
3381 continue_on: Any,
3382 skip_on: Any,
3383 initial_clause: Any = _NoArg.NO_ARG,
3384 *clauses: Any,
3385 **kw: Any,
3386 ) -> ColumnElement[Any]:
3387 if initial_clause is _NoArg.NO_ARG:
3388 # no elements period. deprecated use case. return an empty
3389 # ClauseList construct that generates nothing unless it has
3390 # elements added to it.
3391 name = operator.__name__
3393 util.warn_deprecated(
3394 f"Invoking {name}() without arguments is deprecated, and "
3395 f"will be disallowed in a future release. For an empty "
3396 f"""{name}() construct, use '{name}({
3397 'true()' if continue_on is True_._singleton else 'false()'
3398 }, *args)' """
3399 f"""or '{name}({
3400 'True' if continue_on is True_._singleton else 'False'
3401 }, *args)'.""",
3402 version="1.4",
3403 )
3404 return cls._construct_raw(operator)
3406 lcc, convert_clauses = cls._process_clauses_for_boolean(
3407 operator,
3408 continue_on,
3409 skip_on,
3410 [
3411 coercions.expect(roles.WhereHavingRole, clause)
3412 for clause in util.coerce_generator_arg(
3413 (initial_clause,) + clauses
3414 )
3415 ],
3416 )
3418 if lcc > 1:
3419 # multiple elements. Return regular BooleanClauseList
3420 # which will link elements against the operator.
3422 flattened_clauses = itertools.chain.from_iterable(
3423 (
3424 (c for c in to_flat._flattened_operator_clauses)
3425 if getattr(to_flat, "operator", None) is operator
3426 else (to_flat,)
3427 )
3428 for to_flat in convert_clauses
3429 )
3431 return cls._construct_raw(operator, flattened_clauses) # type: ignore[arg-type] # noqa: E501
3432 else:
3433 assert lcc
3434 # just one element. return it as a single boolean element,
3435 # not a list and discard the operator.
3436 return convert_clauses[0]
3438 @classmethod
3439 def _construct_for_whereclause(
3440 cls, clauses: Iterable[ColumnElement[Any]]
3441 ) -> Optional[ColumnElement[bool]]:
3442 operator, continue_on, skip_on = (
3443 operators.and_,
3444 True_._singleton,
3445 False_._singleton,
3446 )
3448 lcc, convert_clauses = cls._process_clauses_for_boolean(
3449 operator,
3450 continue_on,
3451 skip_on,
3452 clauses, # these are assumed to be coerced already
3453 )
3455 if lcc > 1:
3456 # multiple elements. Return regular BooleanClauseList
3457 # which will link elements against the operator.
3458 return cls._construct_raw(operator, convert_clauses)
3459 elif lcc == 1:
3460 # just one element. return it as a single boolean element,
3461 # not a list and discard the operator.
3462 return convert_clauses[0]
3463 else:
3464 return None
3466 @classmethod
3467 def _construct_raw(
3468 cls,
3469 operator: OperatorType,
3470 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
3471 ) -> BooleanClauseList:
3472 self = cls.__new__(cls)
3473 self.clauses = tuple(clauses) if clauses else ()
3474 self.group = True
3475 self.operator = operator
3476 self.type = type_api.BOOLEANTYPE
3477 self._is_implicitly_boolean = True
3478 return self
3480 @classmethod
3481 def and_(
3482 cls,
3483 initial_clause: Union[
3484 Literal[True], _ColumnExpressionArgument[bool], _NoArg
3485 ] = _NoArg.NO_ARG,
3486 *clauses: _ColumnExpressionArgument[bool],
3487 ) -> ColumnElement[bool]:
3488 r"""Produce a conjunction of expressions joined by ``AND``.
3490 See :func:`_sql.and_` for full documentation.
3491 """
3492 return cls._construct(
3493 operators.and_,
3494 True_._singleton,
3495 False_._singleton,
3496 initial_clause,
3497 *clauses,
3498 )
3500 @classmethod
3501 def or_(
3502 cls,
3503 initial_clause: Union[
3504 Literal[False], _ColumnExpressionArgument[bool], _NoArg
3505 ] = _NoArg.NO_ARG,
3506 *clauses: _ColumnExpressionArgument[bool],
3507 ) -> ColumnElement[bool]:
3508 """Produce a conjunction of expressions joined by ``OR``.
3510 See :func:`_sql.or_` for full documentation.
3511 """
3512 return cls._construct(
3513 operators.or_,
3514 False_._singleton,
3515 True_._singleton,
3516 initial_clause,
3517 *clauses,
3518 )
3520 @property
3521 def _select_iterable(self) -> _SelectIterable:
3522 return (self,)
3524 def self_group(
3525 self, against: Optional[OperatorType] = None
3526 ) -> Union[Self, Grouping[bool]]:
3527 if not self.clauses:
3528 return self
3529 else:
3530 return super().self_group(against=against)
3533and_ = BooleanClauseList.and_
3534or_ = BooleanClauseList.or_
3537class Tuple(ClauseList, ColumnElement[TupleAny]):
3538 """Represent a SQL tuple."""
3540 __visit_name__ = "tuple"
3542 _traverse_internals: _TraverseInternalsType = (
3543 ClauseList._traverse_internals + []
3544 )
3546 type: TupleType
3548 @util.preload_module("sqlalchemy.sql.sqltypes")
3549 def __init__(
3550 self,
3551 *clauses: _ColumnExpressionArgument[Any],
3552 types: Optional[Sequence[_TypeEngineArgument[Any]]] = None,
3553 ):
3554 sqltypes = util.preloaded.sql_sqltypes
3556 if types is None:
3557 init_clauses: List[ColumnElement[Any]] = [
3558 coercions.expect(roles.ExpressionElementRole, c)
3559 for c in clauses
3560 ]
3561 else:
3562 if len(types) != len(clauses):
3563 raise exc.ArgumentError(
3564 "Wrong number of elements for %d-tuple: %r "
3565 % (len(types), clauses)
3566 )
3567 init_clauses = [
3568 coercions.expect(
3569 roles.ExpressionElementRole,
3570 c,
3571 type_=typ if not typ._isnull else None,
3572 )
3573 for typ, c in zip(types, clauses)
3574 ]
3576 self.type = sqltypes.TupleType(*[arg.type for arg in init_clauses])
3577 super().__init__(*init_clauses)
3579 @property
3580 def _select_iterable(self) -> _SelectIterable:
3581 return (self,)
3583 def _bind_param(self, operator, obj, type_=None, expanding=False):
3584 if expanding:
3585 return BindParameter(
3586 None,
3587 value=obj,
3588 _compared_to_operator=operator,
3589 unique=True,
3590 expanding=True,
3591 type_=type_,
3592 _compared_to_type=self.type,
3593 )
3594 else:
3595 return Tuple(
3596 *[
3597 BindParameter(
3598 None,
3599 o,
3600 _compared_to_operator=operator,
3601 _compared_to_type=compared_to_type,
3602 unique=True,
3603 type_=type_,
3604 )
3605 for o, compared_to_type in zip(obj, self.type.types)
3606 ]
3607 )
3609 def self_group(self, against: Optional[OperatorType] = None) -> Self:
3610 # Tuple is parenthesized by definition.
3611 return self
3614class Case(ColumnElement[_T]):
3615 """Represent a ``CASE`` expression.
3617 :class:`.Case` is produced using the :func:`.case` factory function,
3618 as in::
3620 from sqlalchemy import case
3622 stmt = select(users_table).where(
3623 case(
3624 (users_table.c.name == "wendy", "W"),
3625 (users_table.c.name == "jack", "J"),
3626 else_="E",
3627 )
3628 )
3630 Details on :class:`.Case` usage is at :func:`.case`.
3632 .. seealso::
3634 :func:`.case`
3636 """
3638 __visit_name__ = "case"
3640 _traverse_internals: _TraverseInternalsType = [
3641 ("value", InternalTraversal.dp_clauseelement),
3642 ("whens", InternalTraversal.dp_clauseelement_tuples),
3643 ("else_", InternalTraversal.dp_clauseelement),
3644 ]
3646 # for case(), the type is derived from the whens. so for the moment
3647 # users would have to cast() the case to get a specific type
3649 whens: List[typing_Tuple[ColumnElement[bool], ColumnElement[_T]]]
3650 else_: Optional[ColumnElement[_T]]
3651 value: Optional[ColumnElement[Any]]
3653 def __init__(
3654 self,
3655 *whens: Union[
3656 typing_Tuple[_ColumnExpressionArgument[bool], Any],
3657 Mapping[Any, Any],
3658 ],
3659 value: Optional[Any] = None,
3660 else_: Optional[Any] = None,
3661 ):
3662 new_whens: Iterable[Any] = coercions._expression_collection_was_a_list(
3663 "whens", "case", whens
3664 )
3665 try:
3666 new_whens = util.dictlike_iteritems(new_whens)
3667 except TypeError:
3668 pass
3670 self.whens = [
3671 (
3672 coercions.expect(
3673 roles.ExpressionElementRole,
3674 c,
3675 apply_propagate_attrs=self,
3676 ).self_group(),
3677 coercions.expect(roles.ExpressionElementRole, r),
3678 )
3679 for (c, r) in new_whens
3680 ]
3682 if value is None:
3683 self.value = None
3684 else:
3685 self.value = coercions.expect(roles.ExpressionElementRole, value)
3687 if else_ is not None:
3688 self.else_ = coercions.expect(roles.ExpressionElementRole, else_)
3689 else:
3690 self.else_ = None
3692 type_ = next(
3693 (
3694 then.type
3695 # Iterate `whens` in reverse to match previous behaviour
3696 # where type of final element took priority
3697 for *_, then in reversed(self.whens)
3698 if not then.type._isnull
3699 ),
3700 self.else_.type if self.else_ is not None else type_api.NULLTYPE,
3701 )
3702 self.type = cast(_T, type_)
3704 @util.ro_non_memoized_property
3705 def _from_objects(self) -> List[FromClause]:
3706 return list(
3707 itertools.chain(*[x._from_objects for x in self.get_children()])
3708 )
3711class Cast(WrapsColumnExpression[_T]):
3712 """Represent a ``CAST`` expression.
3714 :class:`.Cast` is produced using the :func:`.cast` factory function,
3715 as in::
3717 from sqlalchemy import cast, Numeric
3719 stmt = select(cast(product_table.c.unit_price, Numeric(10, 4)))
3721 Details on :class:`.Cast` usage is at :func:`.cast`.
3723 .. seealso::
3725 :ref:`tutorial_casts`
3727 :func:`.cast`
3729 :func:`.try_cast`
3731 :func:`.type_coerce` - an alternative to CAST that coerces the type
3732 on the Python side only, which is often sufficient to generate the
3733 correct SQL and data coercion.
3735 """
3737 __visit_name__ = "cast"
3739 _traverse_internals: _TraverseInternalsType = [
3740 ("clause", InternalTraversal.dp_clauseelement),
3741 ("type", InternalTraversal.dp_type),
3742 ]
3744 clause: ColumnElement[Any]
3745 type: TypeEngine[_T]
3746 typeclause: TypeClause
3748 def __init__(
3749 self,
3750 expression: _ColumnExpressionArgument[Any],
3751 type_: _TypeEngineArgument[_T],
3752 ):
3753 self.type = type_api.to_instance(type_)
3754 self.clause = coercions.expect(
3755 roles.ExpressionElementRole,
3756 expression,
3757 type_=self.type,
3758 apply_propagate_attrs=self,
3759 )
3760 self.typeclause = TypeClause(self.type)
3762 @util.ro_non_memoized_property
3763 def _from_objects(self) -> List[FromClause]:
3764 return self.clause._from_objects
3766 @property
3767 def wrapped_column_expression(self):
3768 return self.clause
3771class TryCast(Cast[_T]):
3772 """Represent a TRY_CAST expression.
3774 Details on :class:`.TryCast` usage is at :func:`.try_cast`.
3776 .. seealso::
3778 :func:`.try_cast`
3780 :ref:`tutorial_casts`
3781 """
3783 __visit_name__ = "try_cast"
3784 inherit_cache = True
3787class TypeCoerce(WrapsColumnExpression[_T]):
3788 """Represent a Python-side type-coercion wrapper.
3790 :class:`.TypeCoerce` supplies the :func:`_expression.type_coerce`
3791 function; see that function for usage details.
3793 .. seealso::
3795 :func:`_expression.type_coerce`
3797 :func:`.cast`
3799 """
3801 __visit_name__ = "type_coerce"
3803 _traverse_internals: _TraverseInternalsType = [
3804 ("clause", InternalTraversal.dp_clauseelement),
3805 ("type", InternalTraversal.dp_type),
3806 ]
3808 clause: ColumnElement[Any]
3809 type: TypeEngine[_T]
3811 def __init__(
3812 self,
3813 expression: _ColumnExpressionArgument[Any],
3814 type_: _TypeEngineArgument[_T],
3815 ):
3816 self.type = type_api.to_instance(type_)
3817 self.clause = coercions.expect(
3818 roles.ExpressionElementRole,
3819 expression,
3820 type_=self.type,
3821 apply_propagate_attrs=self,
3822 )
3824 @util.ro_non_memoized_property
3825 def _from_objects(self) -> List[FromClause]:
3826 return self.clause._from_objects
3828 @HasMemoized.memoized_attribute
3829 def typed_expression(self):
3830 if isinstance(self.clause, BindParameter):
3831 bp = self.clause._clone()
3832 bp.type = self.type
3833 return bp
3834 else:
3835 return self.clause
3837 @property
3838 def wrapped_column_expression(self):
3839 return self.clause
3841 def self_group(
3842 self, against: Optional[OperatorType] = None
3843 ) -> TypeCoerce[_T]:
3844 grouped = self.clause.self_group(against=against)
3845 if grouped is not self.clause:
3846 return TypeCoerce(grouped, self.type)
3847 else:
3848 return self
3851class Extract(ColumnElement[int]):
3852 """Represent a SQL EXTRACT clause, ``extract(field FROM expr)``."""
3854 __visit_name__ = "extract"
3856 _traverse_internals: _TraverseInternalsType = [
3857 ("expr", InternalTraversal.dp_clauseelement),
3858 ("field", InternalTraversal.dp_string),
3859 ]
3861 expr: ColumnElement[Any]
3862 field: str
3864 def __init__(self, field: str, expr: _ColumnExpressionArgument[Any]):
3865 self.type = type_api.INTEGERTYPE
3866 self.field = field
3867 self.expr = coercions.expect(roles.ExpressionElementRole, expr)
3869 @util.ro_non_memoized_property
3870 def _from_objects(self) -> List[FromClause]:
3871 return self.expr._from_objects
3874class _label_reference(ColumnElement[_T]):
3875 """Wrap a column expression as it appears in a 'reference' context.
3877 This expression is any that includes an _order_by_label_element,
3878 which is a Label, or a DESC / ASC construct wrapping a Label.
3880 The production of _label_reference() should occur when an expression
3881 is added to this context; this includes the ORDER BY or GROUP BY of a
3882 SELECT statement, as well as a few other places, such as the ORDER BY
3883 within an OVER clause.
3885 """
3887 __visit_name__ = "label_reference"
3889 _traverse_internals: _TraverseInternalsType = [
3890 ("element", InternalTraversal.dp_clauseelement)
3891 ]
3893 element: ColumnElement[_T]
3895 def __init__(self, element: ColumnElement[_T]):
3896 self.element = element
3897 self._propagate_attrs = element._propagate_attrs
3899 @util.ro_non_memoized_property
3900 def _from_objects(self) -> List[FromClause]:
3901 return []
3904class _textual_label_reference(ColumnElement[Any]):
3905 __visit_name__ = "textual_label_reference"
3907 _traverse_internals: _TraverseInternalsType = [
3908 ("element", InternalTraversal.dp_string)
3909 ]
3911 def __init__(self, element: str):
3912 self.element = element
3914 @util.memoized_property
3915 def _text_clause(self) -> TextClause:
3916 return TextClause(self.element)
3919class UnaryExpression(ColumnElement[_T]):
3920 """Define a 'unary' expression.
3922 A unary expression has a single column expression
3923 and an operator. The operator can be placed on the left
3924 (where it is called the 'operator') or right (where it is called the
3925 'modifier') of the column expression.
3927 :class:`.UnaryExpression` is the basis for several unary operators
3928 including those used by :func:`.desc`, :func:`.asc`, :func:`.distinct`,
3929 :func:`.nulls_first` and :func:`.nulls_last`.
3931 """
3933 __visit_name__ = "unary"
3935 _traverse_internals: _TraverseInternalsType = [
3936 ("element", InternalTraversal.dp_clauseelement),
3937 ("operator", InternalTraversal.dp_operator),
3938 ("modifier", InternalTraversal.dp_operator),
3939 ]
3941 element: ColumnElement[Any]
3942 operator: Optional[OperatorType]
3943 modifier: Optional[OperatorType]
3945 def __init__(
3946 self,
3947 element: ColumnElement[Any],
3948 *,
3949 operator: Optional[OperatorType] = None,
3950 modifier: Optional[OperatorType] = None,
3951 type_: Optional[_TypeEngineArgument[_T]] = None,
3952 wraps_column_expression: bool = False, # legacy, not used as of 2.0.42
3953 ):
3954 self.operator = operator
3955 self.modifier = modifier
3956 self._propagate_attrs = element._propagate_attrs
3957 self.element = element.self_group(
3958 against=self.operator or self.modifier
3959 )
3961 # if type is None, we get NULLTYPE, which is our _T. But I don't
3962 # know how to get the overloads to express that correctly
3963 self.type = type_api.to_instance(type_) # type: ignore[assignment]
3965 def _wraps_unnamed_column(self):
3966 ungrouped = self.element._ungroup()
3967 return (
3968 not isinstance(ungrouped, NamedColumn)
3969 or ungrouped._non_anon_label is None
3970 )
3972 @classmethod
3973 def _create_nulls_first(
3974 cls,
3975 column: _ColumnExpressionArgument[_T],
3976 ) -> UnaryExpression[_T]:
3977 return UnaryExpression(
3978 coercions.expect(roles.ByOfRole, column),
3979 modifier=operators.nulls_first_op,
3980 )
3982 @classmethod
3983 def _create_nulls_last(
3984 cls,
3985 column: _ColumnExpressionArgument[_T],
3986 ) -> UnaryExpression[_T]:
3987 return UnaryExpression(
3988 coercions.expect(roles.ByOfRole, column),
3989 modifier=operators.nulls_last_op,
3990 )
3992 @classmethod
3993 def _create_desc(
3994 cls, column: _ColumnExpressionOrStrLabelArgument[_T]
3995 ) -> UnaryExpression[_T]:
3997 return UnaryExpression(
3998 coercions.expect(roles.ByOfRole, column),
3999 modifier=operators.desc_op,
4000 )
4002 @classmethod
4003 def _create_asc(
4004 cls,
4005 column: _ColumnExpressionOrStrLabelArgument[_T],
4006 ) -> UnaryExpression[_T]:
4007 return UnaryExpression(
4008 coercions.expect(roles.ByOfRole, column),
4009 modifier=operators.asc_op,
4010 )
4012 @classmethod
4013 def _create_distinct(
4014 cls,
4015 expr: _ColumnExpressionArgument[_T],
4016 ) -> UnaryExpression[_T]:
4017 col_expr: ColumnElement[_T] = coercions.expect(
4018 roles.ExpressionElementRole, expr
4019 )
4020 return UnaryExpression(
4021 col_expr,
4022 operator=operators.distinct_op,
4023 type_=col_expr.type,
4024 )
4026 @classmethod
4027 def _create_bitwise_not(
4028 cls,
4029 expr: _ColumnExpressionArgument[_T],
4030 ) -> UnaryExpression[_T]:
4031 col_expr: ColumnElement[_T] = coercions.expect(
4032 roles.ExpressionElementRole, expr
4033 )
4034 return UnaryExpression(
4035 col_expr,
4036 operator=operators.bitwise_not_op,
4037 type_=col_expr.type,
4038 )
4040 @property
4041 def _order_by_label_element(self) -> Optional[Label[Any]]:
4042 if operators.is_order_by_modifier(self.modifier):
4043 return self.element._order_by_label_element
4044 else:
4045 return None
4047 @util.ro_non_memoized_property
4048 def _from_objects(self) -> List[FromClause]:
4049 return self.element._from_objects
4051 def _negate(self) -> ColumnElement[Any]:
4052 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
4053 return UnaryExpression(
4054 self.self_group(against=operators.inv),
4055 operator=operators.inv,
4056 type_=type_api.BOOLEANTYPE,
4057 )
4058 else:
4059 return ColumnElement._negate(self)
4061 def self_group(
4062 self, against: Optional[OperatorType] = None
4063 ) -> Union[Self, Grouping[_T]]:
4064 if self.operator and operators.is_precedent(self.operator, against):
4065 return Grouping(self)
4066 else:
4067 return self
4070class CollectionAggregate(UnaryExpression[_T]):
4071 """Forms the basis for right-hand collection operator modifiers
4072 ANY and ALL.
4074 The ANY and ALL keywords are available in different ways on different
4075 backends. On PostgreSQL, they only work for an ARRAY type. On
4076 MySQL, they only work for subqueries.
4078 """
4080 inherit_cache = True
4081 _is_collection_aggregate = True
4083 @classmethod
4084 def _create_any(
4085 cls, expr: _ColumnExpressionArgument[_T]
4086 ) -> CollectionAggregate[bool]:
4087 """create CollectionAggregate for the legacy
4088 ARRAY.Comparator.any() method"""
4089 col_expr: ColumnElement[_T] = coercions.expect(
4090 roles.ExpressionElementRole,
4091 expr,
4092 )
4093 col_expr = col_expr.self_group()
4094 return CollectionAggregate(
4095 col_expr,
4096 operator=operators.any_op,
4097 type_=type_api.BOOLEANTYPE,
4098 )
4100 @classmethod
4101 def _create_all(
4102 cls, expr: _ColumnExpressionArgument[_T]
4103 ) -> CollectionAggregate[bool]:
4104 """create CollectionAggregate for the legacy
4105 ARRAY.Comparator.all() method"""
4106 col_expr: ColumnElement[_T] = coercions.expect(
4107 roles.ExpressionElementRole,
4108 expr,
4109 )
4110 col_expr = col_expr.self_group()
4111 return CollectionAggregate(
4112 col_expr,
4113 operator=operators.all_op,
4114 type_=type_api.BOOLEANTYPE,
4115 )
4117 @util.preload_module("sqlalchemy.sql.sqltypes")
4118 def _bind_param(
4119 self,
4120 operator: operators.OperatorType,
4121 obj: Any,
4122 type_: Optional[TypeEngine[_T]] = None,
4123 expanding: bool = False,
4124 ) -> BindParameter[_T]:
4125 """For new style any_(), all_(), ensure compared literal value
4126 receives appropriate bound parameter type."""
4128 # a CollectionAggregate is specific to ARRAY or int
4129 # only. So for ARRAY case, make sure we use correct element type
4130 sqltypes = util.preloaded.sql_sqltypes
4131 if self.element.type._type_affinity is sqltypes.ARRAY:
4132 compared_to_type = cast(
4133 sqltypes.ARRAY[Any], self.element.type
4134 ).item_type
4135 else:
4136 compared_to_type = self.element.type
4138 return BindParameter(
4139 None,
4140 obj,
4141 _compared_to_operator=operator,
4142 type_=type_,
4143 _compared_to_type=compared_to_type,
4144 unique=True,
4145 expanding=expanding,
4146 )
4148 # operate and reverse_operate are hardwired to
4149 # dispatch onto the type comparator directly, so that we can
4150 # ensure "reversed" behavior.
4151 def operate(
4152 self, op: OperatorType, *other: Any, **kwargs: Any
4153 ) -> ColumnElement[_T]:
4154 if not operators.is_comparison(op):
4155 raise exc.ArgumentError(
4156 "Only comparison operators may be used with ANY/ALL"
4157 )
4158 kwargs["reverse"] = True
4159 return self.comparator.operate(operators.mirror(op), *other, **kwargs)
4161 def reverse_operate(
4162 self, op: OperatorType, other: Any, **kwargs: Any
4163 ) -> ColumnElement[_T]:
4164 # comparison operators should never call reverse_operate
4165 assert not operators.is_comparison(op)
4166 raise exc.ArgumentError(
4167 "Only comparison operators may be used with ANY/ALL"
4168 )
4171class AsBoolean(WrapsColumnExpression[bool], UnaryExpression[bool]):
4172 inherit_cache = True
4174 def __init__(self, element, operator, negate):
4175 self.element = element
4176 self.type = type_api.BOOLEANTYPE
4177 self.operator = operator
4178 self.negate = negate
4179 self.modifier = None
4180 self._is_implicitly_boolean = element._is_implicitly_boolean
4182 @property
4183 def wrapped_column_expression(self):
4184 return self.element
4186 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4187 return self
4189 def _negate(self):
4190 if isinstance(self.element, (True_, False_)):
4191 return self.element._negate()
4192 else:
4193 return AsBoolean(self.element, self.negate, self.operator)
4196class BinaryExpression(OperatorExpression[_T]):
4197 """Represent an expression that is ``LEFT <operator> RIGHT``.
4199 A :class:`.BinaryExpression` is generated automatically
4200 whenever two column expressions are used in a Python binary expression:
4202 .. sourcecode:: pycon+sql
4204 >>> from sqlalchemy.sql import column
4205 >>> column("a") + column("b")
4206 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
4207 >>> print(column("a") + column("b"))
4208 {printsql}a + b
4210 """
4212 __visit_name__ = "binary"
4214 _traverse_internals: _TraverseInternalsType = [
4215 ("left", InternalTraversal.dp_clauseelement),
4216 ("right", InternalTraversal.dp_clauseelement),
4217 ("operator", InternalTraversal.dp_operator),
4218 ("negate", InternalTraversal.dp_operator),
4219 ("modifiers", InternalTraversal.dp_plain_dict),
4220 (
4221 "type",
4222 InternalTraversal.dp_type,
4223 ),
4224 ]
4226 _cache_key_traversal = [
4227 ("left", InternalTraversal.dp_clauseelement),
4228 ("right", InternalTraversal.dp_clauseelement),
4229 ("operator", InternalTraversal.dp_operator),
4230 ("modifiers", InternalTraversal.dp_plain_dict),
4231 # "type" affects JSON CAST operators, so while redundant in most cases,
4232 # is needed for that one
4233 (
4234 "type",
4235 InternalTraversal.dp_type,
4236 ),
4237 ]
4239 _is_implicitly_boolean = True
4240 """Indicates that any database will know this is a boolean expression
4241 even if the database does not have an explicit boolean datatype.
4243 """
4245 left: ColumnElement[Any]
4246 right: ColumnElement[Any]
4247 modifiers: Mapping[str, Any]
4249 def __init__(
4250 self,
4251 left: ColumnElement[Any],
4252 right: ColumnElement[Any],
4253 operator: OperatorType,
4254 type_: Optional[_TypeEngineArgument[_T]] = None,
4255 negate: Optional[OperatorType] = None,
4256 modifiers: Optional[Mapping[str, Any]] = None,
4257 ):
4258 # allow compatibility with libraries that
4259 # refer to BinaryExpression directly and pass strings
4260 if isinstance(operator, str):
4261 operator = operators.custom_op(operator)
4262 self._orig = (left.__hash__(), right.__hash__())
4263 self._propagate_attrs = left._propagate_attrs or right._propagate_attrs
4264 self.left = left.self_group(against=operator)
4265 self.right = right.self_group(against=operator)
4266 self.operator = operator
4268 # if type is None, we get NULLTYPE, which is our _T. But I don't
4269 # know how to get the overloads to express that correctly
4270 self.type = type_api.to_instance(type_) # type: ignore[assignment]
4272 self.negate = negate
4273 self._is_implicitly_boolean = operators.is_boolean(operator)
4275 if modifiers is None:
4276 self.modifiers = {}
4277 else:
4278 self.modifiers = modifiers
4280 @property
4281 def _flattened_operator_clauses(
4282 self,
4283 ) -> typing_Tuple[ColumnElement[Any], ...]:
4284 return (self.left, self.right)
4286 def __bool__(self):
4287 """Implement Python-side "bool" for BinaryExpression as a
4288 simple "identity" check for the left and right attributes,
4289 if the operator is "eq" or "ne". Otherwise the expression
4290 continues to not support "bool" like all other column expressions.
4292 The rationale here is so that ColumnElement objects can be hashable.
4293 What? Well, suppose you do this::
4295 c1, c2 = column("x"), column("y")
4296 s1 = set([c1, c2])
4298 We do that **a lot**, columns inside of sets is an extremely basic
4299 thing all over the ORM for example.
4301 So what happens if we do this? ::
4303 c1 in s1
4305 Hashing means it will normally use ``__hash__()`` of the object,
4306 but in case of hash collision, it's going to also do ``c1 == c1``
4307 and/or ``c1 == c2`` inside. Those operations need to return a
4308 True/False value. But because we override ``==`` and ``!=``, they're
4309 going to get a BinaryExpression. Hence we implement ``__bool__`` here
4310 so that these comparisons behave in this particular context mostly
4311 like regular object comparisons. Thankfully Python is OK with
4312 that! Otherwise we'd have to use special set classes for columns
4313 (which we used to do, decades ago).
4315 """
4316 if self.operator in (operators.eq, operators.ne):
4317 # this is using the eq/ne operator given int hash values,
4318 # rather than Operator, so that "bool" can be based on
4319 # identity
4320 return self.operator(*self._orig) # type: ignore[call-overload]
4321 else:
4322 raise TypeError("Boolean value of this clause is not defined")
4324 if typing.TYPE_CHECKING:
4326 def __invert__(
4327 self: BinaryExpression[_T],
4328 ) -> BinaryExpression[_T]: ...
4330 @util.ro_non_memoized_property
4331 def _from_objects(self) -> List[FromClause]:
4332 return self.left._from_objects + self.right._from_objects
4334 def _negate(self):
4335 if self.negate is not None:
4336 return BinaryExpression(
4337 self.left,
4338 self.right._negate_in_binary(self.negate, self.operator),
4339 self.negate,
4340 negate=self.operator,
4341 type_=self.type,
4342 modifiers=self.modifiers,
4343 )
4344 else:
4345 return self.self_group()._negate()
4348class Slice(ColumnElement[Any]):
4349 """Represent SQL for a Python array-slice object.
4351 This is not a specific SQL construct at this level, but
4352 may be interpreted by specific dialects, e.g. PostgreSQL.
4354 """
4356 __visit_name__ = "slice"
4358 _traverse_internals: _TraverseInternalsType = [
4359 ("start", InternalTraversal.dp_clauseelement),
4360 ("stop", InternalTraversal.dp_clauseelement),
4361 ("step", InternalTraversal.dp_clauseelement),
4362 ]
4364 def __init__(self, start, stop, step, _name=None):
4365 self.start = coercions.expect(
4366 roles.ExpressionElementRole,
4367 start,
4368 name=_name,
4369 type_=type_api.INTEGERTYPE,
4370 )
4371 self.stop = coercions.expect(
4372 roles.ExpressionElementRole,
4373 stop,
4374 name=_name,
4375 type_=type_api.INTEGERTYPE,
4376 )
4377 self.step = coercions.expect(
4378 roles.ExpressionElementRole,
4379 step,
4380 name=_name,
4381 type_=type_api.INTEGERTYPE,
4382 )
4383 self.type = type_api.NULLTYPE
4385 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4386 assert against is operator.getitem
4387 return self
4390class IndexExpression(BinaryExpression[Any]):
4391 """Represent the class of expressions that are like an "index"
4392 operation."""
4394 inherit_cache = True
4397class GroupedElement(DQLDMLClauseElement):
4398 """Represent any parenthesized expression"""
4400 __visit_name__ = "grouping"
4402 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4403 return self
4405 def _ungroup(self) -> ClauseElement:
4406 raise NotImplementedError()
4409class Grouping(GroupedElement, ColumnElement[_T]):
4410 """Represent a grouping within a column expression"""
4412 _traverse_internals: _TraverseInternalsType = [
4413 ("element", InternalTraversal.dp_clauseelement),
4414 ("type", InternalTraversal.dp_type),
4415 ]
4417 _cache_key_traversal = [
4418 ("element", InternalTraversal.dp_clauseelement),
4419 ]
4421 element: Union[
4422 AbstractTextClause,
4423 ClauseList,
4424 ColumnElement[_T],
4425 CompilerColumnElement,
4426 ]
4428 def __init__(
4429 self,
4430 element: Union[
4431 AbstractTextClause,
4432 ClauseList,
4433 ColumnElement[_T],
4434 CompilerColumnElement,
4435 ],
4436 ):
4437 self.element = element
4439 # nulltype assignment issue
4440 self.type = getattr(element, "type", type_api.NULLTYPE) # type: ignore[arg-type] # noqa: E501
4441 self._propagate_attrs = element._propagate_attrs
4443 def _with_binary_element_type(self, type_):
4444 return self.__class__(self.element._with_binary_element_type(type_))
4446 def _ungroup(self) -> ColumnElement[_T]:
4447 assert isinstance(self.element, ColumnElement)
4448 return self.element._ungroup()
4450 @util.memoized_property
4451 def _is_implicitly_boolean(self):
4452 return self.element._is_implicitly_boolean
4454 @util.non_memoized_property
4455 def _tq_label(self) -> Optional[str]:
4456 return (
4457 getattr(self.element, "_tq_label", None) or self._anon_name_label
4458 )
4460 @util.non_memoized_property
4461 def _proxies(self) -> List[ColumnElement[Any]]:
4462 if isinstance(self.element, ColumnElement):
4463 return [self.element]
4464 else:
4465 return []
4467 @util.ro_non_memoized_property
4468 def _from_objects(self) -> List[FromClause]:
4469 return self.element._from_objects
4471 def __getattr__(self, attr):
4472 return getattr(self.element, attr)
4474 def __getstate__(self):
4475 return {"element": self.element, "type": self.type}
4477 def __setstate__(self, state):
4478 self.element = state["element"]
4479 self.type = state["type"]
4481 if TYPE_CHECKING:
4483 def self_group(
4484 self, against: Optional[OperatorType] = None
4485 ) -> Self: ...
4488class _OverrideBinds(Grouping[_T]):
4489 """used by cache_key->_apply_params_to_element to allow compilation /
4490 execution of a SQL element that's been cached, using an alternate set of
4491 bound parameter values.
4493 This is used by the ORM to swap new parameter values into expressions
4494 that are embedded into loader options like with_expression(),
4495 selectinload(). Previously, this task was accomplished using the
4496 .params() method which would perform a deep-copy instead. This deep
4497 copy proved to be too expensive for more complex expressions.
4499 See #11085
4501 """
4503 __visit_name__ = "override_binds"
4505 def __init__(
4506 self,
4507 element: ColumnElement[_T],
4508 bindparams: Sequence[BindParameter[Any]],
4509 replaces_params: Sequence[BindParameter[Any]],
4510 ):
4511 self.element = element
4513 # as with Grouping, take on the type of the element we wrap;
4514 # otherwise the construct reports NULLTYPE and the result set
4515 # for a column that's compiled from here has no result processor
4516 self.type = element.type
4517 self.translate = {
4518 k.key: v.value for k, v in zip(replaces_params, bindparams)
4519 }
4521 def _gen_cache_key(
4522 self, anon_map: anon_map, bindparams: List[BindParameter[Any]]
4523 ) -> Optional[typing_Tuple[Any, ...]]:
4524 """generate a cache key for the given element, substituting its bind
4525 values for the translation values present."""
4527 existing_bps: List[BindParameter[Any]] = []
4528 ck = self.element._gen_cache_key(anon_map, existing_bps)
4530 bindparams.extend(
4531 (
4532 bp._with_value(
4533 self.translate[bp.key], maintain_key=True, required=False
4534 )
4535 if bp.key in self.translate
4536 else bp
4537 )
4538 for bp in existing_bps
4539 )
4541 # ck derives from _gen_cache_key, a compiled function in
4542 # _cache_key_cy that mypy sees as untyped
4543 return ck # type: ignore[no-any-return]
4546_FrameIntTuple = tuple[int | None, int | None]
4549class Over(ColumnElement[_T]):
4550 """Represent an OVER clause.
4552 This is a special operator against a so-called
4553 "window" function, as well as any aggregate function,
4554 which produces results relative to the result set
4555 itself. Most modern SQL backends now support window functions.
4557 """
4559 __visit_name__ = "over"
4561 _traverse_internals: _TraverseInternalsType = [
4562 ("element", InternalTraversal.dp_clauseelement),
4563 ("order_by", InternalTraversal.dp_clauseelement),
4564 ("partition_by", InternalTraversal.dp_clauseelement),
4565 ("range_", InternalTraversal.dp_clauseelement),
4566 ("rows", InternalTraversal.dp_clauseelement),
4567 ("groups", InternalTraversal.dp_clauseelement),
4568 ("exclude", InternalTraversal.dp_string),
4569 ]
4571 order_by: Optional[ClauseList] = None
4572 partition_by: Optional[ClauseList] = None
4574 element: ColumnElement[_T]
4575 """The underlying expression object to which this :class:`.Over`
4576 object refers."""
4578 range_: FrameClause | None
4579 rows: FrameClause | None
4580 groups: FrameClause | None
4581 exclude: str | None
4583 def __init__(
4584 self,
4585 element: ColumnElement[_T],
4586 partition_by: Optional[_ByArgument] = None,
4587 order_by: Optional[_ByArgument] = None,
4588 range_: _FrameIntTuple | FrameClause | None = None,
4589 rows: _FrameIntTuple | FrameClause | None = None,
4590 groups: _FrameIntTuple | FrameClause | None = None,
4591 exclude: str | None = None,
4592 ):
4593 self.element = element
4594 if order_by is not None:
4595 self.order_by = ClauseList(
4596 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4597 )
4598 if partition_by is not None:
4599 self.partition_by = ClauseList(
4600 *util.to_list(partition_by),
4601 _literal_as_text_role=roles.ByOfRole,
4602 )
4604 if sum(item is not None for item in (range_, rows, groups)) > 1:
4605 raise exc.ArgumentError(
4606 "only one of 'rows', 'range_', or 'groups' may be provided"
4607 )
4608 else:
4609 self.range_ = FrameClause._parse(range_, coerce_int=False)
4610 self.rows = FrameClause._parse(rows, coerce_int=True)
4611 self.groups = FrameClause._parse(groups, coerce_int=True)
4613 self.exclude = exclude
4615 if exclude is not None and (
4616 range_ is None and rows is None and groups is None
4617 ):
4618 raise exc.ArgumentError(
4619 "'exclude' requires that one of 'rows', "
4620 "'range_', or 'groups' is also specified"
4621 )
4623 if not TYPE_CHECKING:
4625 @util.memoized_property
4626 def type(self) -> TypeEngine[_T]: # noqa: A001
4627 return self.element.type
4629 @util.ro_non_memoized_property
4630 def _from_objects(self) -> List[FromClause]:
4631 return list(
4632 itertools.chain(
4633 *[
4634 c._from_objects
4635 for c in (self.element, self.partition_by, self.order_by)
4636 if c is not None
4637 ]
4638 )
4639 )
4642class FrameClauseType(Enum):
4643 """Frame clause type enum for FrameClause lower_type and upper_type.
4645 .. versionadded:: 2.1
4647 """
4649 UNBOUNDED = 0
4650 """Produces an "UNBOUNDED PRECEDING" or "UNBOUNDED FOLLOWING" frame
4651 clause depending on the position.
4652 Requires a ``None`` value for the corresponding bound value.
4653 """
4654 CURRENT = 1
4655 """Produces a "CURRENT ROW" frame clause.
4656 Requires a ``None`` value for the corresponding bound value.
4657 """
4658 PRECEDING = 2
4659 """Produces a "PRECEDING" frame clause."""
4660 FOLLOWING = 3
4661 """Produces a "FOLLOWING" frame clause."""
4664_require_none = (
4665 FrameClauseType.CURRENT,
4666 FrameClauseType.UNBOUNDED,
4667)
4670class FrameClause(ClauseElement):
4671 """Indicate the 'rows' 'range' or 'group' field of a window function,
4672 e.g. using :class:`.Over`.
4674 .. versionadded:: 2.1
4676 """
4678 __visit_name__ = "frame_clause"
4680 _traverse_internals: _TraverseInternalsType = [
4681 ("lower_bind", InternalTraversal.dp_clauseelement),
4682 ("upper_bind", InternalTraversal.dp_clauseelement),
4683 ("lower_type", InternalTraversal.dp_plain_obj),
4684 ("upper_type", InternalTraversal.dp_plain_obj),
4685 ]
4687 def __init__(
4688 self,
4689 start: Any,
4690 end: Any,
4691 start_frame_type: FrameClauseType,
4692 end_frame_type: FrameClauseType,
4693 _validate: bool = True,
4694 ) -> None:
4695 """Creates a new FrameClause specifying the bounds of a window frame.
4697 :param start: The start value.
4698 :param end: The end value.
4699 :param start_frame_type: The :class:`FrameClauseType` for the
4700 start value.
4701 :param end_frame_type: The :class:`FrameClauseType` for the end value.
4702 """
4703 self.lower_bind = self._as_literal(start)
4704 self.upper_bind = self._as_literal(end)
4705 self.lower_type = FrameClauseType(start_frame_type)
4706 self.upper_type = FrameClauseType(end_frame_type)
4707 if _validate:
4708 if (
4709 self.lower_type in _require_none
4710 and self.lower_bind is not None
4711 ):
4712 raise exc.ArgumentError(
4713 "Cannot specify a value for start with frame type "
4714 f"{self.lower_type.name}"
4715 )
4716 if (
4717 self.upper_type in _require_none
4718 and self.upper_bind is not None
4719 ):
4720 raise exc.ArgumentError(
4721 "Cannot specify a value for end with frame type "
4722 f"{self.upper_type.name}"
4723 )
4725 @classmethod
4726 def _as_literal(cls, value: Any) -> BindParameter[Any] | None:
4727 if value is None:
4728 return None
4729 elif isinstance(value, int):
4730 return literal(value, type_api.INTEGERTYPE)
4731 elif isinstance(value, BindParameter):
4732 return value
4733 else:
4734 return literal(value) # let the default type resolution occur
4736 @classmethod
4737 def _handle_int(
4738 cls, value: Any | None, coerce_int: bool
4739 ) -> tuple[int | None, FrameClauseType]:
4740 if value is None:
4741 return None, FrameClauseType.UNBOUNDED
4743 if coerce_int:
4744 try:
4745 integer = int(value)
4746 except ValueError as err:
4747 raise exc.ArgumentError(
4748 "Integer or None expected for values in rows/groups frame"
4749 ) from err
4750 elif not isinstance(value, int):
4751 raise exc.ArgumentError(
4752 "When using a tuple to specify a range only integer or none "
4753 "values are allowed in the range frame. To specify a "
4754 "different type use the FrameClause directly."
4755 )
4756 else:
4757 integer = value
4758 if integer == 0:
4759 return None, FrameClauseType.CURRENT
4760 elif integer < 0:
4761 return abs(integer), FrameClauseType.PRECEDING
4762 else:
4763 return integer, FrameClauseType.FOLLOWING
4765 @classmethod
4766 def _parse(
4767 cls,
4768 range_: _FrameIntTuple | FrameClause | None,
4769 coerce_int: bool,
4770 ) -> FrameClause | None:
4771 if range_ is None or isinstance(range_, FrameClause):
4772 return range_
4774 try:
4775 r0, r1 = range_
4776 except (ValueError, TypeError) as ve:
4777 raise exc.ArgumentError(
4778 "2-tuple expected for range/rows/groups"
4779 ) from ve
4781 l_b, l_t = cls._handle_int(r0, coerce_int)
4782 u_b, u_t = cls._handle_int(r1, coerce_int)
4784 return FrameClause(
4785 start=l_b,
4786 end=u_b,
4787 start_frame_type=l_t,
4788 end_frame_type=u_t,
4789 _validate=False,
4790 )
4793class AggregateOrderBy(WrapsColumnExpression[_T]):
4794 """Represent an aggregate ORDER BY expression.
4796 This is a special operator against aggregate functions such as
4797 ``array_agg()``, ``json_arrayagg()`` ``string_agg()``, etc. that provides
4798 for an ORDER BY expression, using a syntax that's compatible with
4799 the backend.
4801 :class:`.AggregateOrderBy` is a generalized version of the
4802 :class:`.WithinGroup` construct, the latter of which always provides a
4803 "WITHIN GROUP (ORDER BY ...)" expression. :class:`.AggregateOrderBy` will
4804 also compile to "WITHIN GROUP (ORDER BY ...)" on backends such as Oracle
4805 and SQL Server that don't have another style of aggregate function
4806 ordering.
4808 .. versionadded:: 2.1
4811 """
4813 __visit_name__ = "aggregateorderby"
4815 _traverse_internals: _TraverseInternalsType = [
4816 ("element", InternalTraversal.dp_clauseelement),
4817 ("order_by", InternalTraversal.dp_clauseelement),
4818 ]
4820 order_by: ClauseList
4822 def __init__(
4823 self,
4824 element: Union[FunctionElement[_T], FunctionFilter[_T]],
4825 *order_by: _ColumnExpressionArgument[Any],
4826 ):
4827 self.element = element
4828 if not order_by:
4829 raise TypeError("at least one ORDER BY element is required")
4830 self.order_by = ClauseList(
4831 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4832 )
4834 if not TYPE_CHECKING:
4836 @util.memoized_property
4837 def type(self) -> TypeEngine[_T]: # noqa: A001
4838 return self.element.type
4840 @property
4841 def wrapped_column_expression(self) -> ColumnElement[_T]:
4842 return self.element
4844 def __reduce__(self):
4845 return self.__class__, (self.element,) + (
4846 tuple(self.order_by) if self.order_by is not None else ()
4847 )
4849 def over(
4850 self,
4851 *,
4852 partition_by: _ByArgument | None = None,
4853 order_by: _ByArgument | None = None,
4854 rows: _FrameIntTuple | FrameClause | None = None,
4855 range_: _FrameIntTuple | FrameClause | None = None,
4856 groups: _FrameIntTuple | FrameClause | None = None,
4857 exclude: str | None = None,
4858 ) -> Over[_T]:
4859 """Produce an OVER clause against this :class:`.WithinGroup`
4860 construct.
4862 This function has the same signature as that of
4863 :meth:`.FunctionElement.over`.
4865 """
4866 return Over(
4867 self,
4868 partition_by=partition_by,
4869 order_by=order_by,
4870 range_=range_,
4871 rows=rows,
4872 groups=groups,
4873 exclude=exclude,
4874 )
4876 @overload
4877 def filter(self) -> Self: ...
4879 @overload
4880 def filter(
4881 self,
4882 __criterion0: _ColumnExpressionArgument[bool],
4883 *criterion: _ColumnExpressionArgument[bool],
4884 ) -> FunctionFilter[_T]: ...
4886 def filter(
4887 self, *criterion: _ColumnExpressionArgument[bool]
4888 ) -> Union[Self, FunctionFilter[_T]]:
4889 """Produce a FILTER clause against this function."""
4890 if not criterion:
4891 return self
4892 return FunctionFilter(self, *criterion)
4894 @util.ro_non_memoized_property
4895 def _from_objects(self) -> List[FromClause]:
4896 return list(
4897 itertools.chain(
4898 *[
4899 c._from_objects
4900 for c in (self.element, self.order_by)
4901 if c is not None
4902 ]
4903 )
4904 )
4907class WithinGroup(AggregateOrderBy[_T]):
4908 """Represent a WITHIN GROUP (ORDER BY) clause.
4910 This is a special operator against so-called
4911 "ordered set aggregate" and "hypothetical
4912 set aggregate" functions, including ``percentile_cont()``,
4913 ``rank()``, ``dense_rank()``, etc.
4915 It's supported only by certain database backends, such as PostgreSQL,
4916 Oracle Database and MS SQL Server.
4918 The :class:`.WithinGroup` construct extracts its type from the
4919 method :meth:`.FunctionElement.within_group_type`. If this returns
4920 ``None``, the function's ``.type`` is used.
4922 """
4924 __visit_name__ = "withingroup"
4925 inherit_cache = True
4927 if not TYPE_CHECKING:
4929 @util.memoized_property
4930 def type(self) -> TypeEngine[_T]: # noqa: A001
4931 wgt = self.element.within_group_type(self)
4932 if wgt is not None:
4933 return wgt
4934 else:
4935 return self.element.type
4938class FunctionFilter(Generative, ColumnElement[_T]):
4939 """Represent a function FILTER clause.
4941 This is a special operator against aggregate and window functions,
4942 which controls which rows are passed to it.
4943 It's supported only by certain database backends.
4945 Invocation of :class:`.FunctionFilter` is via
4946 :meth:`.FunctionElement.filter`::
4948 func.count(1).filter(True)
4950 .. seealso::
4952 :meth:`.FunctionElement.filter`
4954 """
4956 __visit_name__ = "funcfilter"
4958 _traverse_internals: _TraverseInternalsType = [
4959 ("func", InternalTraversal.dp_clauseelement),
4960 ("criterion", InternalTraversal.dp_clauseelement),
4961 ]
4963 criterion: Optional[ColumnElement[bool]] = None
4965 def __init__(
4966 self,
4967 func: Union[FunctionElement[_T], AggregateOrderBy[_T]],
4968 *criterion: _ColumnExpressionArgument[bool],
4969 ):
4970 self.func = func
4971 self.filter.non_generative(self, *criterion) # type: ignore[attr-defined] # noqa: E501
4973 @_generative
4974 def filter(self, *criterion: _ColumnExpressionArgument[bool]) -> Self:
4975 """Produce an additional FILTER against the function.
4977 This method adds additional criteria to the initial criteria
4978 set up by :meth:`.FunctionElement.filter`.
4980 Multiple criteria are joined together at SQL render time
4981 via ``AND``.
4984 """
4986 for crit in list(criterion):
4987 crit = coercions.expect(roles.WhereHavingRole, crit)
4989 if self.criterion is not None:
4990 self.criterion = self.criterion & crit
4991 else:
4992 self.criterion = crit
4994 return self
4996 def over(
4997 self,
4998 partition_by: _ByArgument | None = None,
4999 order_by: _ByArgument | None = None,
5000 range_: _FrameIntTuple | FrameClause | None = None,
5001 rows: _FrameIntTuple | FrameClause | None = None,
5002 groups: _FrameIntTuple | FrameClause | None = None,
5003 exclude: str | None = None,
5004 ) -> Over[_T]:
5005 """Produce an OVER clause against this filtered function.
5007 Used against aggregate or so-called "window" functions,
5008 for database backends that support window functions.
5010 The expression::
5012 func.rank().filter(MyClass.y > 5).over(order_by="x")
5014 is shorthand for::
5016 from sqlalchemy import over, funcfilter
5018 over(funcfilter(func.rank(), MyClass.y > 5), order_by="x")
5020 See :func:`_expression.over` for a full description.
5022 """
5023 return Over(
5024 self,
5025 partition_by=partition_by,
5026 order_by=order_by,
5027 range_=range_,
5028 rows=rows,
5029 groups=groups,
5030 exclude=exclude,
5031 )
5033 def within_group(
5034 self, *order_by: _ColumnExpressionArgument[Any]
5035 ) -> WithinGroup[_T]:
5036 """Produce a WITHIN GROUP (ORDER BY expr) clause against
5037 this function.
5038 """
5039 return WithinGroup(self, *order_by)
5041 def within_group_type(
5042 self, within_group: WithinGroup[_T]
5043 ) -> Optional[TypeEngine[_T]]:
5044 return None
5046 def self_group(
5047 self, against: Optional[OperatorType] = None
5048 ) -> Union[Self, Grouping[_T]]:
5049 if operators.is_precedent(operators.filter_op, against):
5050 return Grouping(self)
5051 else:
5052 return self
5054 if not TYPE_CHECKING:
5056 @util.memoized_property
5057 def type(self) -> TypeEngine[_T]: # noqa: A001
5058 return self.func.type
5060 @util.ro_non_memoized_property
5061 def _from_objects(self) -> List[FromClause]:
5062 return list(
5063 itertools.chain(
5064 *[
5065 c._from_objects
5066 for c in (self.func, self.criterion)
5067 if c is not None
5068 ]
5069 )
5070 )
5073class NamedColumn(KeyedColumnElement[_T]):
5074 is_literal = False
5075 table: Optional[FromClause] = None
5076 name: str
5077 key: str
5079 def _compare_name_for_result(self, other):
5080 return (hasattr(other, "name") and self.name == other.name) or (
5081 hasattr(other, "_label") and self._label == other._label
5082 )
5084 @util.ro_memoized_property
5085 def description(self) -> str:
5086 return self.name
5088 @HasMemoized.memoized_attribute
5089 def _tq_key_label(self) -> Optional[str]:
5090 """table qualified label based on column key.
5092 for table-bound columns this is <tablename>_<column key/proxy key>;
5094 all other expressions it resolves to key/proxy key.
5096 """
5097 proxy_key = self._proxy_key
5098 if proxy_key and proxy_key != self.name:
5099 return self._gen_tq_label(proxy_key)
5100 else:
5101 return self._tq_label
5103 @HasMemoized.memoized_attribute
5104 def _tq_label(self) -> Optional[str]:
5105 """table qualified label based on column name.
5107 for table-bound columns this is <tablename>_<columnname>; all other
5108 expressions it resolves to .name.
5110 """
5111 return self._gen_tq_label(self.name)
5113 @HasMemoized.memoized_attribute
5114 def _render_label_in_columns_clause(self):
5115 return True
5117 @HasMemoized.memoized_attribute
5118 def _non_anon_label(self):
5119 return self.name
5121 def _gen_tq_label(
5122 self, name: str, dedupe_on_key: bool = True
5123 ) -> Optional[str]:
5124 return name
5126 def _bind_param(
5127 self,
5128 operator: OperatorType,
5129 obj: Any,
5130 type_: Optional[TypeEngine[_T]] = None,
5131 expanding: bool = False,
5132 ) -> BindParameter[_T]:
5133 return BindParameter(
5134 self.key,
5135 obj,
5136 _compared_to_operator=operator,
5137 _compared_to_type=self.type,
5138 type_=type_,
5139 unique=True,
5140 expanding=expanding,
5141 )
5143 def _make_proxy(
5144 self,
5145 selectable: FromClause,
5146 *,
5147 primary_key: ColumnSet,
5148 foreign_keys: Set[KeyedColumnElement[Any]],
5149 name: Optional[str] = None,
5150 key: Optional[str] = None,
5151 name_is_truncatable: bool = False,
5152 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5153 disallow_is_literal: bool = False,
5154 **kw: Any,
5155 ) -> typing_Tuple[str, ColumnClause[_T]]:
5156 c = ColumnClause(
5157 (
5158 coercions.expect(roles.TruncatedLabelRole, name or self.name)
5159 if name_is_truncatable
5160 else (name or self.name)
5161 ),
5162 type_=self.type,
5163 _selectable=selectable,
5164 is_literal=False,
5165 )
5167 c._propagate_attrs = selectable._propagate_attrs
5168 if name is None:
5169 c.key = self.key
5170 if compound_select_cols:
5171 c._proxies = list(compound_select_cols)
5172 else:
5173 c._proxies = [self]
5175 if selectable._is_clone_of is not None:
5176 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
5177 return c.key, c
5180_PS = ParamSpec("_PS")
5183class Label(roles.LabeledColumnExprRole[_T], NamedColumn[_T]):
5184 """Represents a column label (AS).
5186 Represent a label, as typically applied to any column-level
5187 element using the ``AS`` sql keyword.
5189 """
5191 __visit_name__ = "label"
5193 _traverse_internals: _TraverseInternalsType = [
5194 ("name", InternalTraversal.dp_anon_name),
5195 ("type", InternalTraversal.dp_type),
5196 ("_element", InternalTraversal.dp_clauseelement),
5197 ]
5199 _cache_key_traversal = [
5200 ("name", InternalTraversal.dp_anon_name),
5201 ("_element", InternalTraversal.dp_clauseelement),
5202 ]
5204 _element: ColumnElement[_T]
5205 name: str
5207 def __init__(
5208 self,
5209 name: Optional[str],
5210 element: _ColumnExpressionArgument[_T],
5211 type_: Optional[_TypeEngineArgument[_T]] = None,
5212 ):
5213 orig_element = element
5214 element = coercions.expect(
5215 roles.ExpressionElementRole,
5216 element,
5217 apply_propagate_attrs=self,
5218 )
5219 while isinstance(element, Label):
5220 # TODO: this is only covered in test_text.py, but nothing
5221 # fails if it's removed. determine rationale
5222 element = element.element
5224 if name:
5225 self.name = name
5226 else:
5227 self.name = _anonymous_label.safe_construct(
5228 id(self), getattr(element, "name", "anon")
5229 )
5230 if isinstance(orig_element, Label):
5231 # TODO: no coverage for this block, again would be in
5232 # test_text.py where the resolve_label concept is important
5233 self._resolve_label = orig_element._label
5235 self.key = self._tq_label = self._tq_key_label = self.name
5236 self._element = element
5238 self.type = (
5239 type_api.to_instance(type_)
5240 if type_ is not None
5241 else self._element.type
5242 )
5244 self._proxies = [element]
5246 def __reduce__(self):
5247 return self.__class__, (self.name, self._element, self.type)
5249 @HasMemoized.memoized_attribute
5250 def _render_label_in_columns_clause(self):
5251 return True
5253 def _bind_param(self, operator, obj, type_=None, expanding=False):
5254 return BindParameter(
5255 None,
5256 obj,
5257 _compared_to_operator=operator,
5258 type_=type_,
5259 _compared_to_type=self.type,
5260 unique=True,
5261 expanding=expanding,
5262 )
5264 @util.memoized_property
5265 def _is_implicitly_boolean(self):
5266 return self.element._is_implicitly_boolean
5268 @HasMemoized.memoized_attribute
5269 def _allow_label_resolve(self):
5270 return self.element._allow_label_resolve
5272 @property
5273 def _order_by_label_element(self):
5274 return self
5276 def as_reference(self) -> _label_reference[_T]:
5277 """refer to this labeled expression in a clause such as GROUP BY,
5278 ORDER BY etc. as the label name itself, without expanding
5279 into the full expression.
5281 .. versionadded:: 2.1
5283 """
5284 return _label_reference(self)
5286 @HasMemoized.memoized_attribute
5287 def element(self) -> ColumnElement[_T]:
5288 return self._element.self_group(against=operators.as_)
5290 def self_group(self, against: Optional[OperatorType] = None) -> Label[_T]:
5291 return self._apply_to_inner(self._element.self_group, against=against)
5293 def _negate(self):
5294 return self._apply_to_inner(self._element._negate)
5296 def _apply_to_inner(
5297 self,
5298 fn: Callable[_PS, ColumnElement[_T]],
5299 *arg: _PS.args,
5300 **kw: _PS.kwargs,
5301 ) -> Label[_T]:
5302 sub_element = fn(*arg, **kw)
5303 if sub_element is not self._element:
5304 return Label(self.name, sub_element, type_=self.type)
5305 else:
5306 return self
5308 @property
5309 def primary_key(self): # type: ignore[override]
5310 return self.element.primary_key
5312 @property
5313 def foreign_keys(self): # type: ignore[override]
5314 return self.element.foreign_keys
5316 def _copy_internals(
5317 self,
5318 *,
5319 clone: _CloneCallableType = _clone,
5320 anonymize_labels: bool = False,
5321 **kw: Any,
5322 ) -> None:
5323 self._reset_memoizations()
5324 self._element = clone(self._element, **kw)
5325 if anonymize_labels:
5326 self.name = _anonymous_label.safe_construct(
5327 id(self), getattr(self.element, "name", "anon")
5328 )
5329 self.key = self._tq_label = self._tq_key_label = self.name
5331 @util.ro_non_memoized_property
5332 def _from_objects(self) -> List[FromClause]:
5333 return self.element._from_objects
5335 def _make_proxy(
5336 self,
5337 selectable: FromClause,
5338 *,
5339 primary_key: ColumnSet,
5340 foreign_keys: Set[KeyedColumnElement[Any]],
5341 name: Optional[str] = None,
5342 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5343 **kw: Any,
5344 ) -> typing_Tuple[str, ColumnClause[_T]]:
5345 name = self.name if not name else name
5347 key, e = self.element._make_proxy(
5348 selectable,
5349 name=name,
5350 disallow_is_literal=True,
5351 name_is_truncatable=isinstance(name, _truncated_label),
5352 compound_select_cols=compound_select_cols,
5353 primary_key=primary_key,
5354 foreign_keys=foreign_keys,
5355 )
5357 # there was a note here to remove this assertion, which was here
5358 # to determine if we later could support a use case where
5359 # the key and name of a label are separate. But I don't know what
5360 # that case was. For now, this is an unexpected case that occurs
5361 # when a label name conflicts with other columns and select()
5362 # is attempting to disambiguate an explicit label, which is not what
5363 # the user would want. See issue #6090.
5364 if key != self.name and not isinstance(self.name, _anonymous_label):
5365 raise exc.InvalidRequestError(
5366 "Label name %s is being renamed to an anonymous label due "
5367 "to disambiguation "
5368 "which is not supported right now. Please use unique names "
5369 "for explicit labels." % (self.name)
5370 )
5372 e._propagate_attrs = selectable._propagate_attrs
5373 e._proxies.append(self)
5374 if self.type is not None:
5375 e.type = self.type
5377 return self.key, e
5380class ColumnClause(
5381 roles.DDLReferredColumnRole,
5382 roles.LabeledColumnExprRole[_T],
5383 roles.StrAsPlainColumnRole,
5384 Immutable,
5385 NamedColumn[_T],
5386):
5387 """Represents a column expression from any textual string.
5389 The :class:`.ColumnClause`, a lightweight analogue to the
5390 :class:`_schema.Column` class, is typically invoked using the
5391 :func:`_expression.column` function, as in::
5393 from sqlalchemy import column
5395 id, name = column("id"), column("name")
5396 stmt = select(id, name).select_from("user")
5398 The above statement would produce SQL like:
5400 .. sourcecode:: sql
5402 SELECT id, name FROM user
5404 :class:`.ColumnClause` is the immediate superclass of the schema-specific
5405 :class:`_schema.Column` object. While the :class:`_schema.Column`
5406 class has all the
5407 same capabilities as :class:`.ColumnClause`, the :class:`.ColumnClause`
5408 class is usable by itself in those cases where behavioral requirements
5409 are limited to simple SQL expression generation. The object has none of
5410 the associations with schema-level metadata or with execution-time
5411 behavior that :class:`_schema.Column` does,
5412 so in that sense is a "lightweight"
5413 version of :class:`_schema.Column`.
5415 Full details on :class:`.ColumnClause` usage is at
5416 :func:`_expression.column`.
5418 .. seealso::
5420 :func:`_expression.column`
5422 :class:`_schema.Column`
5424 """
5426 table: Optional[FromClause]
5427 is_literal: bool
5429 __visit_name__ = "column"
5431 _traverse_internals: _TraverseInternalsType = [
5432 ("name", InternalTraversal.dp_anon_name),
5433 ("type", InternalTraversal.dp_type),
5434 ("table", InternalTraversal.dp_clauseelement),
5435 ("is_literal", InternalTraversal.dp_boolean),
5436 ]
5438 onupdate: Optional[DefaultGenerator] = None
5439 default: Optional[DefaultGenerator] = None
5440 server_default: Optional[FetchedValue] = None
5441 server_onupdate: Optional[FetchedValue] = None
5443 _is_multiparam_column = False
5445 @property
5446 def _is_star(self): # type: ignore[override]
5447 return self.is_literal and self.name == "*"
5449 def __init__(
5450 self,
5451 text: str,
5452 type_: Optional[_TypeEngineArgument[_T]] = None,
5453 is_literal: bool = False,
5454 _selectable: Optional[FromClause] = None,
5455 ):
5456 self.key = self.name = text
5457 self.table = _selectable
5459 # if type is None, we get NULLTYPE, which is our _T. But I don't
5460 # know how to get the overloads to express that correctly
5461 self.type = type_api.to_instance(type_) # type: ignore[assignment]
5463 self.is_literal = is_literal
5465 def get_children(self, *, column_tables=False, **kw):
5466 # override base get_children() to not return the Table
5467 # or selectable that is parent to this column. Traversals
5468 # expect the columns of tables and subqueries to be leaf nodes.
5469 return []
5471 @property
5472 def entity_namespace(self):
5473 if self.table is not None:
5474 return self.table.entity_namespace
5475 else:
5476 return super().entity_namespace
5478 def _clone(self, detect_subquery_cols=False, **kw):
5479 if (
5480 detect_subquery_cols
5481 and self.table is not None
5482 and self.table._is_subquery
5483 ):
5484 clone = kw.pop("clone")
5485 table = clone(self.table, **kw)
5486 new = table.c.corresponding_column(self)
5487 return new
5489 return super()._clone(**kw)
5491 @HasMemoized_ro_memoized_attribute
5492 def _from_objects(self) -> List[FromClause]:
5493 t = self.table
5494 if t is not None:
5495 return [t]
5496 else:
5497 return []
5499 @HasMemoized.memoized_attribute
5500 def _render_label_in_columns_clause(self):
5501 return self.table is not None
5503 @property
5504 def _ddl_label(self):
5505 return self._gen_tq_label(self.name, dedupe_on_key=False)
5507 def _compare_name_for_result(self, other):
5508 if (
5509 self.is_literal
5510 or self.table is None
5511 or self.table._is_textual
5512 or not hasattr(other, "proxy_set")
5513 or (
5514 isinstance(other, ColumnClause)
5515 and (
5516 other.is_literal
5517 or other.table is None
5518 or other.table._is_textual
5519 )
5520 )
5521 ):
5522 return (hasattr(other, "name") and self.name == other.name) or (
5523 hasattr(other, "_tq_label")
5524 and self._tq_label == other._tq_label
5525 )
5526 else:
5527 return other.proxy_set.intersection(self.proxy_set)
5529 def _gen_tq_label(
5530 self, name: str, dedupe_on_key: bool = True
5531 ) -> Optional[str]:
5532 """generate table-qualified label
5534 for a table-bound column this is <tablename>_<columnname>.
5536 used primarily for LABEL_STYLE_TABLENAME_PLUS_COL
5537 as well as the .columns collection on a Join object.
5539 """
5540 label: str
5541 t = self.table
5542 if self.is_literal:
5543 return None
5544 elif t is not None and is_named_from_clause(t):
5545 if has_schema_attr(t) and t.schema:
5546 label = (
5547 t.schema.replace(".", "_") + "_" + t.name + ("_" + name)
5548 )
5549 else:
5550 assert not TYPE_CHECKING or isinstance(t, NamedFromClause)
5551 label = t.name + ("_" + name)
5553 # propagate name quoting rules for labels.
5554 if is_quoted_name(name) and name.quote is not None:
5555 if is_quoted_name(label):
5556 label.quote = name.quote
5557 else:
5558 label = quoted_name(label, name.quote)
5559 elif is_quoted_name(t.name) and t.name.quote is not None:
5560 # can't get this situation to occur, so let's
5561 # assert false on it for now
5562 assert not isinstance(label, quoted_name)
5563 label = quoted_name(label, t.name.quote)
5565 if dedupe_on_key:
5566 # ensure the label name doesn't conflict with that of an
5567 # existing column. note that this implies that any Column
5568 # must **not** set up its _label before its parent table has
5569 # all of its other Column objects set up. There are several
5570 # tables in the test suite which will fail otherwise; example:
5571 # table "owner" has columns "name" and "owner_name". Therefore
5572 # column owner.name cannot use the label "owner_name", it has
5573 # to be "owner_name_1".
5574 if label in t.c:
5575 _label = label
5576 counter = 1
5577 while _label in t.c:
5578 _label = label + f"_{counter}"
5579 counter += 1
5580 label = _label
5582 return coercions.expect(roles.TruncatedLabelRole, label)
5584 else:
5585 return name
5587 def _make_proxy(
5588 self,
5589 selectable: FromClause,
5590 *,
5591 primary_key: ColumnSet,
5592 foreign_keys: Set[KeyedColumnElement[Any]],
5593 name: Optional[str] = None,
5594 key: Optional[str] = None,
5595 name_is_truncatable: bool = False,
5596 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5597 disallow_is_literal: bool = False,
5598 **kw: Any,
5599 ) -> typing_Tuple[str, ColumnClause[_T]]:
5600 # the "is_literal" flag normally should never be propagated; a proxied
5601 # column is always a SQL identifier and never the actual expression
5602 # being evaluated. however, there is a case where the "is_literal" flag
5603 # might be used to allow the given identifier to have a fixed quoting
5604 # pattern already, so maintain the flag for the proxy unless a
5605 # :class:`.Label` object is creating the proxy. See [ticket:4730].
5606 is_literal = (
5607 not disallow_is_literal
5608 and self.is_literal
5609 and (
5610 # note this does not accommodate for quoted_name differences
5611 # right now
5612 name is None
5613 or name == self.name
5614 )
5615 )
5616 c = self._constructor(
5617 (
5618 coercions.expect(roles.TruncatedLabelRole, name or self.name)
5619 if name_is_truncatable
5620 else (name or self.name)
5621 ),
5622 type_=self.type,
5623 _selectable=selectable,
5624 is_literal=is_literal,
5625 )
5626 c._propagate_attrs = selectable._propagate_attrs
5627 if name is None:
5628 c.key = self.key
5629 if compound_select_cols:
5630 c._proxies = list(compound_select_cols)
5631 else:
5632 c._proxies = [self]
5634 if selectable._is_clone_of is not None:
5635 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
5636 return c.key, c
5639class TableValuedColumn(NamedColumn[_T]):
5640 __visit_name__ = "table_valued_column"
5642 _traverse_internals: _TraverseInternalsType = [
5643 ("name", InternalTraversal.dp_anon_name),
5644 ("type", InternalTraversal.dp_type),
5645 ("scalar_alias", InternalTraversal.dp_clauseelement),
5646 ]
5648 def __init__(self, scalar_alias: NamedFromClause, type_: TypeEngine[_T]):
5649 self.scalar_alias = scalar_alias
5650 self.key = self.name = scalar_alias.name
5651 self.type = type_
5653 def _copy_internals(
5654 self, clone: _CloneCallableType = _clone, **kw: Any
5655 ) -> None:
5656 self.scalar_alias = clone(self.scalar_alias, **kw)
5657 self.key = self.name = self.scalar_alias.name
5659 @util.ro_non_memoized_property
5660 def _from_objects(self) -> List[FromClause]:
5661 return [self.scalar_alias]
5664class CollationClause(ColumnElement[str]):
5665 __visit_name__ = "collation"
5667 _traverse_internals: _TraverseInternalsType = [
5668 ("collation", InternalTraversal.dp_string),
5669 ("collation_schema", InternalTraversal.dp_string),
5670 ]
5672 @classmethod
5673 @util.preload_module("sqlalchemy.sql.sqltypes")
5674 def _create_collation_expression(
5675 cls,
5676 expression: _ColumnExpressionArgument[str],
5677 collation: str,
5678 collation_schema: Optional[str] = None,
5679 ) -> BinaryExpression[str]:
5681 sqltypes = util.preloaded.sql_sqltypes
5683 expr = coercions.expect(roles.ExpressionElementRole[str], expression)
5685 if expr.type._type_affinity is sqltypes.String:
5686 collate_type = expr.type._with_collation(
5687 collation, collation_schema
5688 )
5689 else:
5690 collate_type = expr.type
5692 return BinaryExpression(
5693 expr,
5694 CollationClause(collation, collation_schema),
5695 operators.collate,
5696 type_=collate_type,
5697 )
5699 def __init__(self, collation, collation_schema=None):
5700 self.collation = collation
5701 self.collation_schema = collation_schema
5704class _IdentifiedClause(Executable, ClauseElement):
5705 __visit_name__ = "identified"
5707 def __init__(self, ident):
5708 self.ident = ident
5711class SavepointClause(_IdentifiedClause):
5712 __visit_name__ = "savepoint"
5713 inherit_cache = False
5716class RollbackToSavepointClause(_IdentifiedClause):
5717 __visit_name__ = "rollback_to_savepoint"
5718 inherit_cache = False
5721class ReleaseSavepointClause(_IdentifiedClause):
5722 __visit_name__ = "release_savepoint"
5723 inherit_cache = False
5726class quoted_name(util.MemoizedSlots, str):
5727 """Represent a SQL identifier combined with quoting preferences.
5729 :class:`.quoted_name` is a Python unicode/str subclass which
5730 represents a particular identifier name along with a
5731 ``quote`` flag. This ``quote`` flag, when set to
5732 ``True`` or ``False``, overrides automatic quoting behavior
5733 for this identifier in order to either unconditionally quote
5734 or to not quote the name. If left at its default of ``None``,
5735 quoting behavior is applied to the identifier on a per-backend basis
5736 based on an examination of the token itself.
5738 A :class:`.quoted_name` object with ``quote=True`` is also
5739 prevented from being modified in the case of a so-called
5740 "name normalize" option. Certain database backends, such as
5741 Oracle Database, Firebird, and DB2 "normalize" case-insensitive names
5742 as uppercase. The SQLAlchemy dialects for these backends
5743 convert from SQLAlchemy's lower-case-means-insensitive convention
5744 to the upper-case-means-insensitive conventions of those backends.
5745 The ``quote=True`` flag here will prevent this conversion from occurring
5746 to support an identifier that's quoted as all lower case against
5747 such a backend.
5749 The :class:`.quoted_name` object is normally created automatically
5750 when specifying the name for key schema constructs such as
5751 :class:`_schema.Table`, :class:`_schema.Column`, and others.
5752 The class can also be
5753 passed explicitly as the name to any function that receives a name which
5754 can be quoted, such as :meth:`.Inspector.has_table` with an
5755 unconditionally quoted name::
5757 from sqlalchemy import create_engine
5758 from sqlalchemy import inspect
5759 from sqlalchemy.sql import quoted_name
5761 engine = create_engine("oracle+oracledb://some_dsn")
5762 print(inspect(engine).has_table(quoted_name("some_table", True)))
5764 The above logic will run the "has table" logic against the Oracle Database
5765 backend, passing the name exactly as ``"some_table"`` without converting to
5766 upper case.
5768 A :class:`.quoted_name` object with ``quote=False`` may be passed to APIs
5769 that apply automatic quoting in order to keep the given name unquoted,
5770 such as when a PostgreSQL ``INHERITS`` option refers to a schema-qualified
5771 table name like ``my_schema.some_table``.
5773 """
5775 __slots__ = "quote", "lower", "upper"
5777 quote: Optional[bool]
5779 @overload
5780 @classmethod
5781 def construct(cls, value: str, quote: Optional[bool]) -> quoted_name: ...
5783 @overload
5784 @classmethod
5785 def construct(cls, value: None, quote: Optional[bool]) -> None: ...
5787 @classmethod
5788 def construct(
5789 cls, value: Optional[str], quote: Optional[bool]
5790 ) -> Optional[quoted_name]:
5791 if value is None:
5792 return None
5793 else:
5794 return quoted_name(value, quote)
5796 def __new__(cls, value: str, quote: Optional[bool]) -> Self:
5797 assert (
5798 value is not None
5799 ), "use quoted_name.construct() for None passthrough"
5800 if isinstance(value, cls) and (quote is None or value.quote == quote):
5801 return value
5802 self = super().__new__(cls, value)
5804 self.quote = quote
5805 return self
5807 def __reduce__(self):
5808 return quoted_name, (str(self), self.quote)
5810 def _memoized_method_lower(self):
5811 if self.quote:
5812 return self
5813 else:
5814 return str(self).lower()
5816 def _memoized_method_upper(self):
5817 if self.quote:
5818 return self
5819 else:
5820 return str(self).upper()
5823def _find_columns(clause: ClauseElement) -> Set[ColumnClause[Any]]:
5824 """locate Column objects within the given expression."""
5826 cols: Set[ColumnClause[Any]] = set()
5827 traverse(clause, {}, {"column": cols.add})
5828 return cols
5831def _type_from_args(args: Sequence[ColumnElement[_T]]) -> TypeEngine[_T]:
5832 for a in args:
5833 if not a.type._isnull:
5834 return a.type
5835 else:
5836 return type_api.NULLTYPE # type: ignore[return-value]
5839def _corresponding_column_or_error(fromclause, column, require_embedded=False):
5840 c = fromclause.corresponding_column(
5841 column, require_embedded=require_embedded
5842 )
5843 if c is None:
5844 raise exc.InvalidRequestError(
5845 "Given column '%s', attached to table '%s', "
5846 "failed to locate a corresponding column from table '%s'"
5847 % (column, getattr(column, "table", None), fromclause.description)
5848 )
5849 return c
5852class _memoized_property_but_not_nulltype(
5853 util.memoized_property["TypeEngine[_T]"]
5854):
5855 """memoized property, but dont memoize NullType"""
5857 def __get__(self, obj, cls):
5858 if obj is None:
5859 return self
5860 result = self.fget(obj)
5861 if not result._isnull:
5862 obj.__dict__[self.__name__] = result
5863 return result
5866class AnnotatedColumnElement(Annotated):
5867 _Annotated__element: ColumnElement[Any]
5869 def __init__(self, element, values):
5870 Annotated.__init__(self, element, values)
5871 for attr in (
5872 "comparator",
5873 "_proxy_key",
5874 "_tq_key_label",
5875 "_tq_label",
5876 "_non_anon_label",
5877 "type",
5878 ):
5879 self.__dict__.pop(attr, None)
5880 for attr in ("name", "key", "table"):
5881 if self.__dict__.get(attr, False) is None:
5882 self.__dict__.pop(attr)
5884 def _with_annotations(self, values):
5885 clone = super()._with_annotations(values)
5886 for attr in (
5887 "comparator",
5888 "_proxy_key",
5889 "_tq_key_label",
5890 "_tq_label",
5891 "_non_anon_label",
5892 ):
5893 clone.__dict__.pop(attr, None)
5894 return clone
5896 @util.memoized_property
5897 def name(self):
5898 """pull 'name' from parent, if not present"""
5899 return self._Annotated__element.name
5901 @_memoized_property_but_not_nulltype
5902 def type(self):
5903 """pull 'type' from parent and don't cache if null.
5905 type is routinely changed on existing columns within the
5906 mapped_column() initialization process, and "type" is also consulted
5907 during the creation of SQL expressions. Therefore it can change after
5908 it was already retrieved. At the same time we don't want annotated
5909 objects having overhead when expressions are produced, so continue
5910 to memoize, but only when we have a non-null type.
5912 """
5913 return self._Annotated__element.type
5915 @util.memoized_property
5916 def table(self):
5917 """pull 'table' from parent, if not present"""
5918 return self._Annotated__element.table
5920 @util.memoized_property
5921 def key(self):
5922 """pull 'key' from parent, if not present"""
5923 return self._Annotated__element.key
5925 @util.memoized_property
5926 def info(self) -> _InfoType:
5927 if TYPE_CHECKING:
5928 assert isinstance(self._Annotated__element, Column)
5929 return self._Annotated__element.info
5931 @util.memoized_property
5932 def _anon_name_label(self) -> str:
5933 return self._Annotated__element._anon_name_label
5936class _truncated_label(quoted_name):
5937 """A unicode subclass used to identify symbolic "
5938 "names that may require truncation."""
5940 __slots__ = ()
5942 def __new__(cls, value: str, quote: Optional[bool] = None) -> Self:
5943 quote = getattr(value, "quote", quote)
5944 # return super(_truncated_label, cls).__new__(cls, value, quote, True)
5945 return super().__new__(cls, value, quote)
5947 def __reduce__(self) -> Any:
5948 return self.__class__, (str(self), self.quote)
5950 def apply_map(self, map_: Mapping[str, Any]) -> str:
5951 return self
5954class conv(_truncated_label):
5955 """Mark a string indicating that a name has already been converted
5956 by a naming convention.
5958 This is a string subclass that indicates a name that should not be
5959 subject to any further naming conventions.
5961 E.g. when we create a :class:`.Constraint` using a naming convention
5962 as follows::
5964 m = MetaData(
5965 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5966 )
5967 t = Table(
5968 "t", m, Column("x", Integer), CheckConstraint("x > 5", name="x5")
5969 )
5971 The name of the above constraint will be rendered as ``"ck_t_x5"``.
5972 That is, the existing name ``x5`` is used in the naming convention as the
5973 ``constraint_name`` token.
5975 In some situations, such as in migration scripts, we may be rendering
5976 the above :class:`.CheckConstraint` with a name that's already been
5977 converted. In order to make sure the name isn't double-modified, the
5978 new name is applied using the :func:`_schema.conv` marker. We can
5979 use this explicitly as follows::
5982 m = MetaData(
5983 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5984 )
5985 t = Table(
5986 "t",
5987 m,
5988 Column("x", Integer),
5989 CheckConstraint("x > 5", name=conv("ck_t_x5")),
5990 )
5992 Where above, the :func:`_schema.conv` marker indicates that the constraint
5993 name here is final, and the name will render as ``"ck_t_x5"`` and not
5994 ``"ck_t_ck_t_x5"``
5996 .. seealso::
5998 :ref:`constraint_naming_conventions`
6000 """
6002 __slots__ = ()
6005# for backwards compatibility in case
6006# someone is re-implementing the
6007# _truncated_identifier() sequence in a custom
6008# compiler
6009_generated_label = _truncated_label
6010_anonymous_label_escape = re.compile(r"[%\(\) \$]+")
6011# for bind parameter keys, additionally escape the characters that
6012# SQLCompiler.bindname_escape_characters would otherwise escape only at
6013# compile time, after the uniquifying counter has already been applied.
6014# escaping them up front is what allows names like "a.b" and "a_b" to be
6015# disambiguated as "a_b_1" / "a_b_2" rather than colliding. see #13534
6016_bind_key_escape = re.compile(r"[%\(\) \$\.\[\]:]+")
6019class _anonymous_label(_truncated_label):
6020 """A unicode subclass used to identify anonymously
6021 generated names."""
6023 __slots__ = ()
6025 @classmethod
6026 def safe_construct_with_key(
6027 cls, seed: int | str, body: str, sanitize_key: bool = False
6028 ) -> typing_Tuple[_anonymous_label, str]:
6029 # need to escape chars that interfere with format
6030 # strings in any case, issue #8724
6031 if sanitize_key:
6032 # sanitize_key is an extra step used by BindParameter, which
6033 # also escapes the characters that would otherwise be escaped
6034 # only at compile time; issue #13534
6035 body = _bind_key_escape.sub("_", body).strip("_")
6036 else:
6037 body = _anonymous_label_escape.sub("_", body)
6039 key = f"{seed} {body.replace('%', '%%')}"
6040 label = _anonymous_label(f"%({key})s")
6041 return label, key
6043 @classmethod
6044 def safe_construct(
6045 cls, seed: int | str, body: str, sanitize_key: bool = False
6046 ) -> _anonymous_label:
6047 # need to escape chars that interfere with format
6048 # strings in any case, issue #8724
6049 if sanitize_key:
6050 # sanitize_key is an extra step used by BindParameter, which
6051 # also escapes the characters that would otherwise be escaped
6052 # only at compile time; issue #13534
6053 body = _bind_key_escape.sub("_", body).strip("_")
6054 else:
6055 body = _anonymous_label_escape.sub("_", body)
6057 return _anonymous_label(f"%({seed} {body.replace('%', '%%')})s")
6059 def __add__(self, other: str) -> _anonymous_label:
6060 if "%" in other and not isinstance(other, _anonymous_label):
6061 other = str(other).replace("%", "%%")
6062 else:
6063 other = str(other)
6065 return _anonymous_label(
6066 quoted_name(
6067 str.__add__(self, other),
6068 self.quote,
6069 )
6070 )
6072 def __radd__(self, other: str) -> _anonymous_label:
6073 if "%" in other and not isinstance(other, _anonymous_label):
6074 other = str(other).replace("%", "%%")
6075 else:
6076 other = str(other)
6078 return _anonymous_label(
6079 quoted_name(
6080 str.__add__(other, self),
6081 self.quote,
6082 )
6083 )
6085 def apply_map(self, map_: Mapping[str, Any]) -> str:
6086 if self.quote is not None:
6087 # preserve quoting only if necessary
6088 return quoted_name(self % map_, self.quote)
6089 else:
6090 # else skip the constructor call
6091 return self % map_