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